Configurable idempotency TTL, presence debounce, file-message delivery pipeline, VAPID key endpoint

.env.example

@@ -28,10 +28,16 @@ OPENAI_API_KEY=
 
 # Messaging
 XMTP_ENV=dev
+# Seconds an eventId is remembered for replay/duplicate protection. Defaults to 86400 (24h).
+IDEMPOTENCY_TTL_SECONDS=86400
+# Milliseconds an offline transition is held before broadcasting user_offline,
+# so a brief disconnect+reconnect blip produces no offline/online pair. Defaults to 5000 (5s).
+PRESENCE_OFFLINE_GRACE_MS=5000
 
 # Push Notifications (VAPID)
+# The frontend fetches VAPID_PUBLIC_KEY from GET /push/vapid-public-key at
+# runtime (#349) rather than a separate NEXT_PUBLIC_* build-time var, so the
+# two can never drift out of sync.
 VAPID_PUBLIC_KEY=
 VAPID_PRIVATE_KEY=
-VAPID_SUBJECT=mailto:admin@example.com
-# Exposed to the browser via Next.js
-NEXT_PUBLIC_VAPID_PUBLIC_KEY=
+VAPID_SUBJECT=mailto:admin@example.com

apps/backend/src/__tests__/dispatcher.test.ts

@@ -1,4 +1,4 @@
-import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
 import { EventEmitter } from 'events';
 import { EventDispatcher } from '../socket/dispatcher.js';
 import type { AuthSocket } from '../middleware/socketAuth.js';
@@ -187,3 +187,115 @@ describe('EventDispatcher.listen — envelope routing', () => {
     expect(errors.length).toBeGreaterThan(0);
   });
 });
+
+describe('EventDispatcher — configurable idempotency TTL (#344)', () => {
+  const ORIGINAL_TTL = process.env.IDEMPOTENCY_TTL_SECONDS;
+
+  afterEach(() => {
+    if (ORIGINAL_TTL === undefined) delete process.env.IDEMPOTENCY_TTL_SECONDS;
+    else process.env.IDEMPOTENCY_TTL_SECONDS = ORIGINAL_TTL;
+  });
+
+  it('defaults the Redis SET EX TTL to 86400 seconds when unset', async () => {
+    delete process.env.IDEMPOTENCY_TTL_SECONDS;
+    const { socket, trigger } = makeSocket();
+    const redis = makeRedis('OK');
+    const dispatcher = new EventDispatcher(makeIo(), socket, redis as never);
+    dispatcher.register('join_room', vi.fn());
+    dispatcher.listen();
+
+    trigger('dispatch', {
+      eventId: 'evt-default-ttl',
+      type: 'join_room',
+      timestamp: Date.now(),
+      payload: {},
+    });
+
+    await new Promise((r) => setTimeout(r, 10));
+    expect(redis.set).toHaveBeenCalledWith(
+      'event:idempotency:evt-default-ttl',
+      '1',
+      'EX',
+      86_400,
+      'NX',
+    );
+  });
+
+  it('reads the Redis SET EX TTL from IDEMPOTENCY_TTL_SECONDS when configured', async () => {
+    process.env.IDEMPOTENCY_TTL_SECONDS = '120';
+    const { socket, trigger } = makeSocket();
+    const redis = makeRedis('OK');
+    const dispatcher = new EventDispatcher(makeIo(), socket, redis as never);
+    dispatcher.register('join_room', vi.fn());
+    dispatcher.listen();
+
+    trigger('dispatch', {
+      eventId: 'evt-custom-ttl',
+      type: 'join_room',
+      timestamp: Date.now(),
+      payload: {},
+    });
+
+    await new Promise((r) => setTimeout(r, 10));
+    expect(redis.set).toHaveBeenCalledWith(
+      'event:idempotency:evt-custom-ttl',
+      '1',
+      'EX',
+      120,
+      'NX',
+    );
+  });
+
+  it('falls back to the default TTL for an invalid (non-numeric) override', async () => {
+    process.env.IDEMPOTENCY_TTL_SECONDS = 'not-a-number';
+    const { socket, trigger } = makeSocket();
+    const redis = makeRedis('OK');
+    const dispatcher = new EventDispatcher(makeIo(), socket, redis as never);
+    dispatcher.register('join_room', vi.fn());
+    dispatcher.listen();
+
+    trigger('dispatch', {
+      eventId: 'evt-invalid-ttl',
+      type: 'join_room',
+      timestamp: Date.now(),
+      payload: {},
+    });
+
+    await new Promise((r) => setTimeout(r, 10));
+    expect(redis.set).toHaveBeenCalledWith(
+      'event:idempotency:evt-invalid-ttl',
+      '1',
+      'EX',
+      86_400,
+      'NX',
+    );
+  });
+
+  it('rejects a duplicate eventId within the TTL window regardless of configured TTL', async () => {
+    process.env.IDEMPOTENCY_TTL_SECONDS = '300';
+    const { socket, trigger } = makeSocket();
+    // null == Redis SET NX found the key already present (still within TTL)
+    const redis = makeRedis(null);
+    const dispatcher = new EventDispatcher(makeIo(), socket, redis as never);
+    const handler = vi.fn();
+    dispatcher.register('join_room', handler);
+    dispatcher.listen();
+
+    trigger('dispatch', {
+      eventId: 'evt-duplicate-within-ttl',
+      type: 'join_room',
+      timestamp: Date.now(),
+      payload: {},
+    });
+
+    await new Promise((r) => setTimeout(r, 10));
+    expect(handler).not.toHaveBeenCalled();
+    expect(redis.set).toHaveBeenCalledWith(
+      'event:idempotency:evt-duplicate-within-ttl',
+      '1',
+      'EX',
+      300,
+      'NX',
+    );
+  });
+});

apps/backend/src/__tests__/file.messages.test.ts

@@ -35,7 +35,10 @@
 const mockMemberFindMany = vi.fn();
 const mockMessageFindFirst = vi.fn();
 const mockFileFindFirst = vi.fn();
+const mockMessageFindFirst = vi.fn();
 const mockDevicesFindMany = vi.fn();
+const mockInsert = vi.fn();
+const mockFindMany = vi.fn();
 const mockUpdate = vi.fn();
 
 /** Every insert(table).values(rows) call, so envelope rows can be asserted. */
@@ -185,10 +188,13 @@
 const FILE_ID = 'file-abc';
 const MESSAGE_ID = 'msg-client-supplied';
 
-// The content is an E2EE envelope ciphertext. The server treats it as an
-// opaque string — it must NOT parse or store the embedded fileKey.
-const ENVELOPE_CIPHERTEXT =
-  'encrypted:{"fileId":"file-abc","fileName":"photo.jpg","mimeType":"image/jpeg","size":204800,"fileKey":"SUPER_SECRET_KEY_NEVER_STORED"}';
+// The content is an E2EE envelope ciphertext for the message body. The server
+// treats it as an opaque string. The file's symmetric encryption key must
+// NEVER appear here — it only ever lives inside `envelopes[].ciphertext`.
+const ENVELOPE_CIPHERTEXT = 'encrypted:{"fileId":"file-abc","fileName":"photo.jpg"}';
+const FILE_KEY_ENVELOPES = [
+  { recipientDeviceId: RECIPIENT_DEVICE_ID, ciphertext: 'sealed:SUPER_SECRET_KEY_NEVER_STORED' },
+];
 
 function readyFile(
   overrides: Partial<{
@@ -553,16 +559,60 @@
   it('rejects when sender is not a member of the conversation', async () => {
     mockMemberFindFirst.mockResolvedValue(undefined);
 
-    const socket = makeSocket('non-member');
+    await handler(basePayload({ messageId: '' }));
+
+    expect(socket.emit).toHaveBeenCalledWith(
+      'error',
+      expect.objectContaining({
+        event: 'send_file_message',
+        message: expect.stringContaining('messageId'),
+      }),
+    );
+    expect(mockInsert).not.toHaveBeenCalled();
+  });
+
+  it('rejects when envelopes are missing (the file key has nowhere safe to travel)', async () => {
+    mockMemberFindFirst.mockResolvedValueOnce({
+      id: 'm1',
+      userId: SENDER_ID,
+      conversationId: CONVERSATION_ID,
+    });
+
+    const socket = makeSocket(SENDER_ID);
     const io = makeIo();
     const handler = await getHandler(socket, io);
 
     await handler({
       conversationId: CONVERSATION_ID,
-      fileId: FILE_ID,
-      content: ENVELOPE_CIPHERTEXT,
-      contentType: 'file',
     });
+    mockFileFindFirst.mockResolvedValueOnce(readyFile());
+    // fetchSiblingDeviceIds finds one sibling device the payload didn't cover.
+    mockDevicesFindMany.mockResolvedValueOnce([{ id: 'device-sibling', userId: SENDER_ID }]);
+
+    const socket = makeSocket(SENDER_ID);
+    const io = makeIo();
+    const handler = await getHandler(socket, io);
+
+    await handler(basePayload());
+
+    expect(socket.emit).toHaveBeenCalledWith(
+      'error',
+      expect.objectContaining({
+        event: 'device_set_mismatch',
+        missingDeviceIds: ['device-sibling'],
+      }),
+    );
+    expect(mockInsert).not.toHaveBeenCalled();
+  });
+
+  it('rejects when sender is not a member of the conversation', async () => {
+    mockMemberFindFirst.mockResolvedValueOnce(undefined); // no membership
+
+    const socket = makeSocket('non-member');
+    const io = makeIo();
+    const handler = await getHandler(socket, io);
+
+    await handler(basePayload());
 
     expect(socket.emit).toHaveBeenCalledWith(
       'error',

apps/backend/src/__tests__/presence.test.ts

@@ -1,10 +1,13 @@
-import { describe, it, expect, beforeEach } from 'vitest';
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
 import {
   setOnline,
   refreshPresence,
   setOffline,
   markDeviceOffline,
   isOnline,
+  scheduleOfflineBroadcast,
+  cancelPendingOfflineBroadcast,
+  __resetOfflineBroadcastsForTesting,
 } from '../services/presence.js';
 
 class FakeRedis {
@@ -64,22 +67,22 @@
   });
 
   it('tracks a device entry in the per-user hash and refreshes its ttl', async () => {
     await setOnline(redis as any, 'user-1', 'device-1', '1710000000000');
 
     expect(redis.hashes.get('presence:user:user-1')).toEqual({ 'device-1': '1710000000000' });
     expect(redis.expires.get('presence:user:user-1:device:device-1')).toBe(90);
 
     await refreshPresence(redis as any, 'user-1', 'device-1', '1710000001000');
 
     expect(redis.hashes.get('presence:user:user-1')).toEqual({ 'device-1': '1710000001000' });
     expect(redis.expires.get('presence:user:user-1:device:device-1')).toBe(90);
   });
 
   it('removes a device entry when it disconnects and marks the user offline once all devices are gone', async () => {
     await setOnline(redis as any, 'user-1', 'device-1', '1710000000000');
     await setOnline(redis as any, 'user-1', 'device-2', '1710000000100');
 
     const firstRemoved = await setOffline(redis as any, 'user-1', 'device-1');
     expect(firstRemoved).toBe(false);
     expect(redis.hashes.get('presence:user:user-1')).toEqual({ 'device-2': '1710000000100' });
 
@@ -96,3 +99,74 @@
     expect(redis.deleted.has('presence:user:user-1:device:device-1')).toBe(true);
   });
 });
+
+describe('presence offline-broadcast debounce (#345)', () => {
+  const ORIGINAL_GRACE = process.env.PRESENCE_OFFLINE_GRACE_MS;
+
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(() => {
+    __resetOfflineBroadcastsForTesting();
+    vi.useRealTimers();
+    if (ORIGINAL_GRACE === undefined) delete process.env.PRESENCE_OFFLINE_GRACE_MS;
+    else process.env.PRESENCE_OFFLINE_GRACE_MS = ORIGINAL_GRACE;
+  });
+
+  it('does not broadcast if cancelled (reconnect) before the grace window elapses', async () => {
+    process.env.PRESENCE_OFFLINE_GRACE_MS = '5000';
+    const broadcast = vi.fn();
+
+    scheduleOfflineBroadcast('user-1', broadcast);
+    vi.advanceTimersByTime(4999);
+    const cancelled = cancelPendingOfflineBroadcast('user-1');
+
+    await vi.runAllTimersAsync();
+
+    expect(cancelled).toBe(true);
+    expect(broadcast).not.toHaveBeenCalled();
+  });
+
+  it('broadcasts once the grace window fully elapses without a reconnect', async () => {
+    process.env.PRESENCE_OFFLINE_GRACE_MS = '5000';
+    const broadcast = vi.fn();
+
+    scheduleOfflineBroadcast('user-1', broadcast);
+    await vi.advanceTimersByTimeAsync(5000);
+
+    expect(broadcast).toHaveBeenCalledTimes(1);
+    expect(cancelPendingOfflineBroadcast('user-1')).toBe(false);
+  });
+
+  it('cancelling with nothing pending is a no-op that reports false', () => {
+    expect(cancelPendingOfflineBroadcast('user-with-no-pending-broadcast')).toBe(false);
+  });
+
+  it('re-scheduling for the same user replaces (does not stack) the pending broadcast', async () => {
+    process.env.PRESENCE_OFFLINE_GRACE_MS = '5000';
+    const first = vi.fn();
+    const second = vi.fn();
+
+    scheduleOfflineBroadcast('user-1', first);
+    vi.advanceTimersByTime(2000);
+    scheduleOfflineBroadcast('user-1', second);
+
+    await vi.advanceTimersByTimeAsync(5000);
+
+    expect(first).not.toHaveBeenCalled();
+    expect(second).toHaveBeenCalledTimes(1);
+  });
+
+  it('defaults the grace window to 5000ms when PRESENCE_OFFLINE_GRACE_MS is unset', async () => {
+    delete process.env.PRESENCE_OFFLINE_GRACE_MS;
+    const broadcast = vi.fn();
+
+    scheduleOfflineBroadcast('user-1', broadcast);
+    vi.advanceTimersByTime(4999);
+    expect(broadcast).not.toHaveBeenCalled();
+
+    await vi.advanceTimersByTimeAsync(1);
+    expect(broadcast).toHaveBeenCalledTimes(1);
+  });
+});