feat: add localId

prisma/migrations/20250718031550_add_local_id_to_session_message/migration.sql

@@ -0,0 +1,11 @@
+/*
+  Warnings:
+
+  - A unique constraint covering the columns `[sessionId,localId]` on the table `SessionMessage` will be added. If there are existing duplicate values, this will fail.
+
+*/
+-- AlterTable
+ALTER TABLE "SessionMessage" ADD COLUMN     "localId" TEXT;
+
+-- CreateIndex
+CREATE UNIQUE INDEX "SessionMessage_sessionId_localId_key" ON "SessionMessage"("sessionId", "localId");

prisma/schema.prisma

@@ -55,11 +55,13 @@ model SessionMessage {
     id        String   @id @default(cuid())
     sessionId String
     session   Session  @relation(fields: [sessionId], references: [id])
+    localId   String?
     seq       Int
     /// [SessionMessageContent]
     content   Json
     createdAt DateTime @default(now())
     updatedAt DateTime @updatedAt
+    @@unique([sessionId, localId])
 }
 
 model Update {

sources/app/api.ts

@@ -94,7 +94,7 @@ export async function startApi() {
     });
 
     // Send session update to all relevant connections
-    let emitUpdateToInterestedClients = ({event, userId, sessionId, payload, skipSenderConnection}: {
+    let emitUpdateToInterestedClients = ({ event, userId, sessionId, payload, skipSenderConnection }: {
         event: string,
         userId: string,
         sessionId: string,
@@ -595,7 +595,7 @@ export async function startApi() {
         });
 
         socket.on('message', async (data: any) => {
-            const { sid, message } = data;
+            const { sid, message, localId } = data;
 
             log({ module: 'websocket' }, `Received message from socket ${socket.id}: ${sid} ${message.length} bytes`);
 
@@ -606,6 +606,7 @@ export async function startApi() {
             if (!session) {
                 return;
             }
+            let useLocalId = typeof localId === 'string' ? localId : null;
 
             // Create encrypted message
             const msgContent: PrismaJson.SessionMessageContent = {
@@ -641,12 +642,20 @@ export async function startApi() {
                 const msgSeq = session.seq + 1;
                 const updSeq = user.seq + 1;
 
+                if (useLocalId) {
+                    const existing = await tx.sessionMessage.findFirst({
+                        where: { sessionId: sid, localId: useLocalId }
+                    });
+                    return { msg: existing, update: null };
+                }
+
                 // Create message
                 const msg = await tx.sessionMessage.create({
                     data: {
                         sessionId: sid,
                         seq: msgSeq,
-                        content: msgContent
+                        content: msgContent,
+                        localId: useLocalId
                     }
                 });
 
@@ -690,9 +699,13 @@ export async function startApi() {
                 throw error;
             });
 
-            if (!result) return;
+            // If no update, we're done
+            if (!result) {
+                return;
+            }
             
             // Emit update to relevant clients
+            if (result.update) {
                 emitUpdateToInterestedClients({
                     event: 'update',
                     userId,
@@ -705,6 +718,7 @@ export async function startApi() {
                     },
                     skipSenderConnection: connection
                 });
+            }
         });
 
         socket.on('update-metadata', async (data: any, callback: (response: any) => void) => {