#303 WebSocket client: envelope protocol, resume + sync FIXED

apps/backend/src/lib/eventEnvelope.ts

@@ -1,55 +1,12 @@
-import { randomUUID } from 'node:crypto';
 import { z } from 'zod';
-
-// Central registry of all valid socket event types.
-export const KNOWN_EVENT_TYPES = new Set([
-  // Inbound (client → server)
-  'join_room',
-  'send_message',
-  'message_history',
-  'delete_message',
-  'message_read',
-  'create_conversation',
-  'typing_start',
-  'typing_stop',
-  'ask_assistant',
-  'resume',
-  'join_device_channel',
-  // Outbound (server → client) — registered so the registry is the single source of truth
-  'room_joined',
-  'new_message',
-  'message_ack',
-  'message_deleted',
-  'read_receipt',
-  'conversation_created',
-  'ephemeral_replay',
-  'resume_complete',
-  'device_envelope',
-  'error',
-]);
-
-export const EventEnvelopeSchema = z.object({
-  eventId: z.string().min(1, 'eventId is required'),
-  type: z.string().min(1, 'type is required'),
-  timestamp: z.number().int().positive('timestamp must be a positive integer'),
-  payload: z.record(z.string(), z.unknown()).optional().default({}),
-});
-
+export declare const KNOWN_EVENT_TYPES: Set<string>;
+export declare const EventEnvelopeSchema: z.ZodObject<{
+    eventId: z.ZodString;
+    type: z.ZodString;
+    timestamp: z.ZodNumber;
+    payload: z.ZodDefault<z.ZodOptional<z.ZodRecord<z.ZodString, z.ZodUnknown>>>;
+}, z.core.$strip>;
 export type EventEnvelope = z.infer<typeof EventEnvelopeSchema>;
-
-export function isKnownEventType(type: string): boolean {
-  return KNOWN_EVENT_TYPES.has(type);
-}
-
-export function createEnvelope(
-  type: string,
-  payload: Record<string, unknown>,
-  eventId?: string,
-): EventEnvelope {
-  return {
-    eventId: eventId ?? randomUUID(),
-    type,
-    timestamp: Date.now(),
-    payload,
-  };
-}
+export declare function isKnownEventType(type: string): boolean;
+export declare function createEnvelope(type: string, payload: Record<string, unknown>, eventId?: string): EventEnvelope;
+//# sourceMappingURL=eventEnvelope.d.ts.map

apps/backend/src/middleware/auth.ts

@@ -45,6 +45,7 @@ export async function requireAuth(
     return;
   }
 
+
   req.auth = payload;
   next();
 }

apps/backend/src/middleware/socketAuth.ts

@@ -39,6 +39,7 @@ export async function socketAuthMiddleware(
     return;
   }
 
+
   socket.auth = payload;
   next();
 }

apps/backend/src/routes/auth.ts

@@ -4,7 +4,7 @@ import type { Request, Response, IRouter } from 'express';
 import rateLimit, { type RateLimitRequestHandler } from 'express-rate-limit';
 import { Keypair } from '@stellar/stellar-sdk';
 import { db } from '../db/index.js';
-import { users, wallets, devices } from '../db/schema.js';
+import { users, wallets, devices, userDevices } from '../db/schema.js';
 import { eq, and } from 'drizzle-orm';
 import { createNonce, consumeNonce } from '../lib/nonce.js';
 import { signToken } from '../lib/jwt.js';
@@ -57,7 +57,7 @@ authRouter.post(
   verifyLimiter,
   validate(VerifySchema),
   async (req: Request, res: Response) => {
-    const { walletAddress, signature, nonce, identityPublicKey } = req.body as VerifyBody;
+    const { walletAddress, signature, nonce, identityPublicKey, deviceId: clientDeviceId, deviceName, platform, registrationId } = req.body as VerifyBody;
 
     // Validate and consume nonce
     const valid = consumeNonce(walletAddress, nonce);
@@ -135,7 +135,42 @@ authRouter.post(
       deviceId = newDevice.id;
     }
 
-    const token = signToken({ userId, walletAddress, deviceId });
-    res.json({ token });
+    let tokenDeviceId = deviceId;
+
+    if (clientDeviceId) {
+      const [userDevice] = await db
+        .insert(userDevices)
+        .values({
+          userId,
+          deviceId: clientDeviceId,
+          deviceName: deviceName ?? 'Web browser',
+          platform: platform ?? 'web',
+          identityPublicKey,
+          registrationId: registrationId ? Number(registrationId) : null,
+          lastSeenAt: new Date(),
+        })
+        .onConflictDoUpdate({
+          target: [userDevices.userId, userDevices.deviceId],
+          set: {
+            deviceName: deviceName ?? 'Web browser',
+            platform: platform ?? 'web',
+            identityPublicKey,
+            registrationId: registrationId ? Number(registrationId) : null,
+            lastSeenAt: new Date(),
+            revokedAt: null,
+          },
+        })
+        .returning({ id: userDevices.id });
+
+      if (!userDevice) {
+        res.status(500).json({ error: 'Failed to register messaging device' });
+        return;
+      }
+
+      tokenDeviceId = userDevice.id;
+    }
+
+    const token = signToken({ userId, walletAddress, deviceId: tokenDeviceId });
+    res.json({ token, deviceId: tokenDeviceId });
   },
 );

apps/backend/src/routes/messages.ts

@@ -31,7 +31,12 @@ messagesRouter.post('/', validate(SendMessageSchema), async (req: AuthRequest, r
   };
 
   // ── content-type-specific validation ──────────────────────────────────────
-  const validation = validateMessagePayload({ contentType, ciphertext, envelopes, fileId });
+  const validation = validateMessagePayload({
+    ...(contentType !== undefined ? { contentType } : {}),
+    ...(ciphertext !== undefined ? { ciphertext } : {}),
+    ...(envelopes !== undefined ? { envelopes } : {}),
+    ...(fileId !== undefined ? { fileId } : {}),
+  });
   if (!validation.ok) {
     res.status(validation.code).json({ error: validation.message });
     return;

apps/backend/src/routes/sync.ts

@@ -80,6 +80,9 @@ syncRouter.get('/', async (req: AuthRequest, res) => {
       createdAt: messageEnvelopes.createdAt,
       sequenceNumber: messages.sequenceNumber,
       conversationId: messages.conversationId,
+      senderId: messages.senderId,
+      senderDeviceId: messages.senderDeviceId,
+      contentType: messages.contentType,
     })
     .from(messageEnvelopes)
     .innerJoin(messages, eq(messageEnvelopes.messageId, messages.id))
@@ -116,6 +119,9 @@ syncRouter.get('/', async (req: AuthRequest, res) => {
       id: r.id,
       messageId: r.messageId,
       conversationId: r.conversationId,
+      senderId: r.senderId,
+      senderDeviceId: r.senderDeviceId,
+      contentType: r.contentType,
       ciphertext: r.ciphertext,
       sequenceNumber: r.sequenceNumber,
       deliveredAt: r.deliveredAt,

apps/backend/src/schemas/auth.schemas.ts

@@ -5,6 +5,13 @@ export const ChallengeSchema = z.object({
   walletAddress: z.string().min(1, 'walletAddress is required'),
 });
 
+const DeviceRegistrationSchema = z.object({
+  deviceId: z.string().min(1, 'deviceId is required').optional(),
+  deviceName: z.string().min(1, 'deviceName is required').optional(),
+  platform: z.enum(['web', 'ios', 'android']).optional(),
+  registrationId: z.string().optional(),
+});
+
 export const DeviceSchema = z.object({
   deviceId: z.string().min(1, 'deviceId is required'),
   deviceName: z.string().min(1, 'deviceName is required'),
@@ -13,16 +20,18 @@ export const DeviceSchema = z.object({
   registrationId: z.string().optional(),
 });
 
-export const VerifySchema = z.object({
-  walletAddress: z.string().min(1, 'walletAddress is required'),
-  signature: z.string().min(1, 'signature is required'),
-  nonce: z.string().min(1, 'nonce is required'),
-  /**
-   * Base64-encoded Ed25519 SPKI DER identity public key (44 bytes).
-   * Validated for correct base64 and exact byte length before any crypto operation.
-   */
-  identityPublicKey: IdentityPublicKeySchema,
-});
+export const VerifySchema = z
+  .object({
+    walletAddress: z.string().min(1, 'walletAddress is required'),
+    signature: z.string().min(1, 'signature is required'),
+    nonce: z.string().min(1, 'nonce is required'),
+    /**
+     * Base64-encoded Ed25519 SPKI DER identity public key (44 bytes).
+     * Validated for correct base64 and exact byte length before any crypto operation.
+     */
+    identityPublicKey: IdentityPublicKeySchema,
+  })
+  .merge(DeviceRegistrationSchema);
 
 export type ChallengeBody = z.infer<typeof ChallengeSchema>;
 export type DeviceBody = z.infer<typeof DeviceSchema>;

apps/backend/src/socket/messaging.ts

@@ -66,6 +66,9 @@ export function registerMessagingHandlers(io: Server, socket: AuthSocket): void
 
   // ── send_message ───────────────────────────────────────────────────────────
   dispatcher.register('send_message', async (payload) => {
+
+    const { conversationId, messageId, content, contentType, ciphertext, envelopes, fileId } = payload as {
+
     const {
       conversationId,
       messageId,
@@ -75,6 +78,7 @@ export function registerMessagingHandlers(io: Server, socket: AuthSocket): void
       envelopes,
       fileId: payloadFileId,
     } = payload as {
+
       conversationId: string;
       messageId?: string;
       content?: string;
@@ -109,10 +113,17 @@ export function registerMessagingHandlers(io: Server, socket: AuthSocket): void
     const effectiveCiphertext = ciphertext ?? content ?? undefined;
 
     const validation = validateMessagePayload({
+
+      ...(contentType !== undefined ? { contentType } : {}),
+      ...(effectiveCiphertext !== undefined ? { ciphertext: effectiveCiphertext } : {}),
+      ...(envelopes !== undefined ? { envelopes } : {}),
+      ...(fileId !== undefined ? { fileId } : {}),
+
       contentType,
       ciphertext: effectiveCiphertext,
       envelopes,
       fileId: payloadFileId,
+
     });
     if (!validation.ok) {
       socket.emit('error', {
@@ -145,15 +156,25 @@ export function registerMessagingHandlers(io: Server, socket: AuthSocket): void
       return;
     }
 
+
+    const resolvedContentType = contentType?.trim().toLowerCase() || 'text';
+    let persistedFileId = fileId;
+    if (FILE_CONTENT_TYPES.has(resolvedContentType) && !persistedFileId) {
+
     let fileId: string | undefined = payloadFileId;
     const resolvedContentType = contentType || 'text/plain';
     if (FILE_CONTENT_TYPES.has(resolvedContentType)) {
+
       const [fileRow] = await db
         .insert(files)
         .values({ storageKey: messageId })
         .onConflictDoUpdate({ target: files.storageKey, set: { storageKey: messageId } })
         .returning({ id: files.id });
+
+      persistedFileId = fileRow?.id;
+
       fileId = fileRow?.id ?? payloadFileId;
+
     }
 
     const [message] = await db
@@ -165,7 +186,7 @@ export function registerMessagingHandlers(io: Server, socket: AuthSocket): void
         senderDeviceId: deviceId,
         contentType: resolvedContentType,
         ciphertext: effectiveCiphertext,
-        fileId: fileId ?? null,
+        fileId: persistedFileId ?? null,
       })
       .returning();
 
@@ -211,10 +232,15 @@ export function registerMessagingHandlers(io: Server, socket: AuthSocket): void
       return;
     }
 
+    if (message) {
+      await deliverMessage(io, message, conversationId);
+    }
+
     socket.emit('message_ack', { messageId, sequenceNumber: message.sequenceNumber });
 
     await deliverMessage(io, message, conversationId);
 
+
     const members = await db.query.conversationMembers.findMany({
       where: eq(conversationMembers.conversationId, conversationId),
       columns: { userId: true },
@@ -455,7 +481,7 @@ export function registerMessagingHandlers(io: Server, socket: AuthSocket): void
         ),
       );
 
-    io.to(conversationId).volatile.emit('read_receipt', { userId, lastReadMessageId });
+    io.to(conversationId).volatile.emit('read_receipt', { conversationId, userId, lastReadMessageId });
 
     if (redis) {
       const members = await db.query.conversationMembers.findMany({
@@ -470,6 +496,71 @@ export function registerMessagingHandlers(io: Server, socket: AuthSocket): void
     }
   });
 
+  // ── message_delivered ──────────────────────────────────────────────────────
+  dispatcher.register('message_delivered', async (payload) => {
+    const { conversationId, messageId, envelopeId, sequenceNumber } = payload as {
+      conversationId?: string;
+      messageId?: string;
+      envelopeId?: string;
+      sequenceNumber?: number;
+    };
+
+    if (!conversationId || !messageId) {
+      socket.emit('error', {
+        event: 'message_delivered',
+        message: 'conversationId and messageId are required',
+      });
+      return;
+    }
+
+    const membership = await db.query.conversationMembers.findFirst({
+      where: and(
+        eq(conversationMembers.conversationId, conversationId),
+        eq(conversationMembers.userId, userId),
+      ),
+    });
+
+    if (!membership) {
+      socket.emit('error', { event: 'message_delivered', message: 'Not a member of this conversation' });
+      return;
+    }
+
+    await db
+      .update(messageEnvelopes)
+      .set({ deliveredAt: new Date() })
+      .where(
+        and(
+          eq(messageEnvelopes.messageId, messageId),
+          eq(messageEnvelopes.recipientDeviceId, socket.auth!.deviceId),
+        ),
+      );
+
+    io.to(conversationId).volatile.emit('delivery_receipt', {
+      conversationId,
+      messageId,
+      envelopeId,
+      userId,
+      deviceId: socket.auth!.deviceId,
+      sequenceNumber,
+      deliveredAt: new Date().toISOString(),
+    });
+
+    if (redis) {
+      const members = await db.query.conversationMembers.findMany({
+        where: eq(conversationMembers.conversationId, conversationId),
+        columns: { userId: true },
+      });
+      await publishEphemeral(
+        redis,
+        members.map((member) => member.userId),
+        {
+          type: 'delivery_receipt',
+          data: { conversationId, messageId, envelopeId, userId, deviceId: socket.auth!.deviceId, sequenceNumber },
+        },
+      );
+    }
+  });
+
   // ── resume ─────────────────────────────────────────────────────────────────
   dispatcher.register('resume', async (payload) => {
     if (!redis) {