From f60dbf794a8e578bcc54e2841a00bb9b666afd2f Mon Sep 17 00:00:00 2001 From: ayo-mide55 Date: Thu, 1 Oct 2026 19:58:09 +0100 Subject: [PATCH] feat(sockets): ACK-gated offline message queue with flush-on-reconnect (#147) --- .env.example | 13 +- src/config/env.ts | 6 + src/sockets/messageQueue.ts | 354 ++++++++++++++++++++++++++++++++-- src/sockets/socket.service.ts | 22 ++- tests/messageQueue.test.ts | 241 +++++++++++++++++++++++ 5 files changed, 610 insertions(+), 26 deletions(-) diff --git a/.env.example b/.env.example index e207230..16949b8 100644 --- a/.env.example +++ b/.env.example @@ -308,10 +308,19 @@ SOCKET_PING_INTERVAL_MS=25000 # [default: 2] SOCKET_MAX_MISSED_PONGS=2 -# How long, in ms, a client has to ACK a queued socket message before it is -# dropped. [default: 15000] +# How long, in ms, a client has to ACK a queued socket message before the +# server re-delivers it (retry). [default: 15000] SOCKET_MESSAGE_ACK_TIMEOUT_MS=15000 +# Maximum number of unacknowledged messages retained per driver. When the +# bound is exceeded the oldest queued message is evicted. [default: 100] +SOCKET_MESSAGE_QUEUE_MAX=100 + +# Maximum number of automatic re-delivery attempts per queued message after +# ACK timeouts. Once exhausted the message stays queued for the next reconnect +# flush but is no longer retried automatically. [default: 5] +SOCKET_MESSAGE_MAX_RETRIES=5 + # ─── Shutdown ──────────────────────────────────────────────────────────────── # How long, in ms, to wait for in-flight HTTP/Socket/DB work during SIGTERM / diff --git a/src/config/env.ts b/src/config/env.ts index 3ecf080..9aa5237 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -85,6 +85,10 @@ interface EnvConfig { SOCKET_MAX_MISSED_PONGS: number; /** Time (ms) a queued socket message waits for an ack before retry. Default: 15000 */ SOCKET_MESSAGE_ACK_TIMEOUT_MS: number; + /** Max unacknowledged messages retained per driver before oldest are evicted. Default: 100 */ + SOCKET_MESSAGE_QUEUE_MAX: number; + /** Max automatic re-delivery attempts per queued message after ACK timeouts. Default: 5 */ + SOCKET_MESSAGE_MAX_RETRIES: number; /** Interval (ms) between periodic socket token expiry checks. Default: 60000 */ SOCKET_TOKEN_CHECK_INTERVAL_MS: number; /** Grace period (ms) granted after a socket token expires. Default: 30000 */ @@ -298,6 +302,8 @@ const envSchema = z.object({ SOCKET_PING_TIMEOUT_MS: z.coerce.number().int().min(1000).default(20000), SOCKET_MAX_MISSED_PONGS: z.coerce.number().int().min(1).max(10).default(2), SOCKET_MESSAGE_ACK_TIMEOUT_MS: z.coerce.number().int().min(1000).default(15000), + SOCKET_MESSAGE_QUEUE_MAX: z.coerce.number().int().min(1).default(100), + SOCKET_MESSAGE_MAX_RETRIES: z.coerce.number().int().min(0).max(20).default(5), SOCKET_TOKEN_CHECK_INTERVAL_MS: z.coerce.number().int().min(1000).default(60000), SOCKET_TOKEN_GRACE_PERIOD_MS: z.coerce.number().int().min(0).default(30000), SYNC_BATCH_SIZE_LIMIT: z.coerce.number().int().min(1).max(10000).default(500), diff --git a/src/sockets/messageQueue.ts b/src/sockets/messageQueue.ts index b3ccb81..d33c8a1 100644 --- a/src/sockets/messageQueue.ts +++ b/src/sockets/messageQueue.ts @@ -1,6 +1,27 @@ import { randomUUID } from 'crypto'; import env from '../config/env'; +import logger from '../config/logger'; +/** + * Callback handed to an emitter so the client can acknowledge receipt. + * A message is only removed from the queue when this is invoked with a + * non-null/non-undefined value (an empty object counts as an ACK). + */ +export type MessageAckCallback = (ack?: unknown) => void; + +/** + * Emitter signature used to (re)deliver a queued message to a client. + * Matches the subset of `Socket.emit` the queue relies on. + */ +export type MessageEmitter = ( + event: string, + payload: T, + ackCallback?: MessageAckCallback, +) => void; + +/** + * A single message tracked by the delivery-guarantee queue. + */ export interface QueuedSocketMessage { id: string; userId: string; @@ -8,21 +29,100 @@ export interface QueuedSocketMessage { payload: T; queuedAt: number; updatedAt: number; + /** When true the message is retained until the client ACKs it. */ ackRequired: boolean; + /** Number of automatic re-deliveries performed after ACK timeouts. */ retries: number; ackTimeoutMs: number; + /** Upper bound on automatic re-deliveries before retrying is paused. */ + maxRetries: number; + /** + * Connection generation this message was last delivered on, or `null` when + * it has not yet been delivered on the current connection. Used to make + * reconnects idempotent (see {@link MessageQueueService.flush}). + */ + deliveredConnectionId: number | null; + /** + * Delivery lifecycle: + * - `pending` – awaiting its first/next delivery attempt. + * - `in-flight` – delivered, awaiting an ACK (a timer is armed). + * - `exhausted` – retry budget spent; retained for the next reconnect. + */ + status: 'pending' | 'in-flight' | 'exhausted'; + /** Timestamp (ms epoch) of the most recent delivery attempt, if any. */ + lastAttemptAt: number | null; } export interface EnqueueSocketMessageOptions { ackRequired?: boolean; ackTimeoutMs?: number; + /** Overrides the configured automatic retry budget for this message. */ + maxRetries?: number; payload?: T; } +export interface MessageQueueOptions { + /** Default ACK timeout applied when an enqueue call omits one. */ + defaultAckTimeoutMs?: number; + /** Per-driver queue bound; oldest messages are evicted past this size. */ + maxQueueSize?: number; + /** Default automatic retry budget for queued messages. */ + maxRetries?: number; +} + +/** + * In-memory offline message queue providing at-least-once delivery for + * driver-bound socket messages. + * + * Delivery-guarantee invariants: + * 1. **ACK gating** – an ACK-required message is never removed from its + * driver's queue until the client ACKs it or a fixed bound evicts it. + * 2. **Timeout/retry** – each delivery arms a timer; if no ACK arrives + * within `ackTimeoutMs` the message is re-delivered, up to `maxRetries`. + * Once the budget is spent the message is *retained* (`exhausted`), + * never silently dropped, so a later reconnect can still flush it. + * 3. **Bounded persistence** – at most `maxQueueSize` unacknowledged + * messages are retained per driver; the oldest is evicted, so an + * abusive or permanently-offline client cannot grow memory unboundedly. + * 4. **Idempotent reconnect flush** – {@link flush} only emits messages not + * already in-flight on the *same* connection generation. Re-delivery + * across a genuine reconnect happens exactly once, and an ACKed message + * is never re-sent. + * 5. **Per-driver isolation** – queues, timers, emitters and ACKs are keyed + * by `userId`; acknowledging one driver never touches another. + * 6. **Disconnect safety** – {@link handleDisconnect} drops the dead + * emitter, stops its timers and rewinds in-flight messages to `pending` + * so they are flushed when the driver reconnects. + */ export class MessageQueueService { private readonly queues = new Map(); - private readonly defaultAckTimeoutMs = env.SOCKET_MESSAGE_ACK_TIMEOUT_MS; + /** Armed ACK-timeout timers, keyed by `${userId}\u0000${messageId}`. */ + private readonly timers = new Map>(); + /** Most recent delivery target per driver, used to perform retries. */ + private readonly emitters = new Map(); + /** Active connection generation per driver. */ + private readonly activeConnectionId = new Map(); + /** Stable connection generations for emitter/socket owners. */ + private readonly connectionIds = new WeakMap(); + private connectionSeq = 0; + + private readonly defaultAckTimeoutMs: number; + private readonly maxQueueSize: number; + private readonly maxRetries: number; + constructor(options: MessageQueueOptions = {}) { + this.defaultAckTimeoutMs = options.defaultAckTimeoutMs ?? env.SOCKET_MESSAGE_ACK_TIMEOUT_MS; + this.maxQueueSize = options.maxQueueSize ?? env.SOCKET_MESSAGE_QUEUE_MAX; + this.maxRetries = options.maxRetries ?? env.SOCKET_MESSAGE_MAX_RETRIES; + } + + /** + * Queue a message for a driver. + * + * The message is retained until it is ACKed (when `ackRequired`), retried + * past its ACK timeout, or evicted by the per-driver bound. Enqueuing never + * emits; delivery happens through {@link flush} or {@link sendWithAck}. + */ public enqueue( userId: string, event: string, @@ -48,20 +148,44 @@ export class MessageQueueService { ackRequired: options.ackRequired ?? true, retries: 0, ackTimeoutMs: options.ackTimeoutMs ?? this.defaultAckTimeoutMs, + maxRetries: options.maxRetries ?? this.maxRetries, + deliveredConnectionId: null, + status: 'pending', + lastAttemptAt: null, }; const existing = this.queues.get(normalizedUserId) ?? []; + + // Bound the queue: evict the oldest messages so a permanently-offline or + // abusive client cannot grow memory without limit. + while (existing.length >= this.maxQueueSize) { + const evicted = existing.shift(); + if (evicted) { + this.clearTimer(normalizedUserId, evicted.id); + logger.warn( + `[MessageQueue] Evicted oldest message for userId=${normalizedUserId} ` + + `messageId=${evicted.id} event=${evicted.event} (bound=${this.maxQueueSize})`, + ); + } + } + existing.push(queued); this.queues.set(normalizedUserId, existing); return queued; } + /** Return a copy of a driver's pending/unacknowledged messages. */ public getPending(userId: string): QueuedSocketMessage[] { const queue = this.queues.get(userId?.trim() ?? ''); return queue ? [...queue] : []; } + /** + * Remove a single message after the client acknowledged it. + * + * @returns `true` when a matching message was found and removed. + */ public acknowledge(userId: string, messageId: string): boolean { const normalizedUserId = userId?.trim(); const queue = normalizedUserId ? this.queues.get(normalizedUserId) : undefined; @@ -76,42 +200,72 @@ export class MessageQueueService { } queue.splice(index, 1); + this.clearTimer(normalizedUserId as string, messageId); if (queue.length === 0) { - this.queues.delete(normalizedUserId); + this.queues.delete(normalizedUserId as string); } return true; } + /** + * Deliver every unacknowledged message for a driver to the given emitter. + * + * Idempotent within a connection: messages already delivered (and awaiting + * an ACK) on the same connection generation are skipped, so calling this + * twice for one socket cannot double-deliver. A genuine reconnect resolves + * to a new connection generation, so previously in-flight messages are + * re-delivered exactly once. + * + * @param userId - Driver whose queue should be flushed. + * @param emit - Delivery function (typically `socket.emit`). + * @param owner - Stable identity for the connection (e.g. the socket). + * Defaults to `emit`; pass the socket so {@link sendWithAck} + * and reconnect flushes share one generation. + * @returns Number of messages emitted during this call. + */ public flush( userId: string, - emit: (event: string, payload: T, ackCallback?: (ack?: unknown) => void) => void, + emit: MessageEmitter, + owner?: object, ): number { const normalizedUserId = userId?.trim(); const queue = normalizedUserId ? this.queues.get(normalizedUserId) : undefined; - if (!queue || queue.length === 0) { + if (!normalizedUserId || !queue || queue.length === 0) { return 0; } + const connectionId = this.resolveConnectionId(owner ?? emit); + this.emitters.set(normalizedUserId, emit as MessageEmitter); + this.activeConnectionId.set(normalizedUserId, connectionId); + const pending = [...queue]; + let delivered = 0; for (const message of pending) { - emit(message.event, message.payload as T, (ack?: unknown) => { - if (ack !== undefined && ack !== null) { - this.acknowledge(normalizedUserId, message.id); - } - }); + // Skip anything already in-flight (or exhausted) on this connection. + if (message.deliveredConnectionId === connectionId) { + continue; + } + + this.deliver(normalizedUserId, message, emit as MessageEmitter, connectionId); + delivered += 1; } - return pending.length; + return delivered; } + /** + * Queue and immediately deliver a message to a live socket, arming the + * ACK-timeout retry loop. Returns the queued record, or `null` when the + * socket is not associated with an authenticated driver. + */ public sendWithAck( socket: { data?: { userId?: string }; - emit: (event: string, payload: T, ack?: (ack?: unknown) => void) => void; + emit: (event: string, payload: T, ack?: MessageAckCallback) => void; }, event: string, payload: T, @@ -123,19 +277,69 @@ export class MessageQueueService { const queued = this.enqueue(userId, event, payload, { ackRequired: true }); - socket.emit(event, payload, (ack?: unknown) => { - if (ack !== undefined && ack !== null) { - this.acknowledge(userId, queued.id); - } - }); + const emit = socket.emit as MessageEmitter; + const connectionId = this.resolveConnectionId(socket); + + this.emitters.set(userId, emit as MessageEmitter); + this.activeConnectionId.set(userId, connectionId); + this.deliver(userId, queued, emit as MessageEmitter, connectionId); return queued; } + /** + * Handle a driver disconnect: drop the dead emitter, cancel pending retry + * timers and rewind in-flight messages to `pending` so they are flushed on + * reconnect. Safe to call for unknown/anonymous users. + * + * @returns Number of messages rewound to `pending`. + */ + public handleDisconnect(userId?: string): number { + const normalizedUserId = userId?.trim(); + if (!normalizedUserId) { + return 0; + } + + this.emitters.delete(normalizedUserId); + this.activeConnectionId.delete(normalizedUserId); + + const queue = this.queues.get(normalizedUserId); + if (!queue || queue.length === 0) { + return 0; + } + + let rewound = 0; + for (const message of queue) { + this.clearTimer(normalizedUserId, message.id); + if (message.status === 'in-flight') { + message.status = 'pending'; + message.deliveredConnectionId = null; + message.updatedAt = Date.now(); + rewound += 1; + } + } + + return rewound; + } + + /** + * Drop queued messages (and their timers) for one driver, or for every + * driver when `userId` is omitted. + * + * @returns Number of messages removed. + */ public clear(userId?: string): number { if (!userId) { - const total = Array.from(this.queues.values()).reduce((sum, queue) => sum + queue.length, 0); + let total = 0; + for (const queue of this.queues.values()) { + total += queue.length; + for (const message of queue) { + this.clearTimer(message.userId, message.id); + } + } this.queues.clear(); + this.emitters.clear(); + this.activeConnectionId.clear(); return total; } @@ -144,11 +348,127 @@ export class MessageQueueService { const size = queue?.length ?? 0; if (queue) { + for (const message of queue) { + this.clearTimer(normalizedUserId, message.id); + } this.queues.delete(normalizedUserId); } + this.emitters.delete(normalizedUserId); + this.activeConnectionId.delete(normalizedUserId); + return size; } + + // ─── Private helpers ──────────────────────────────────────────────────────── + + /** Emit one message and arm its ACK timeout (when an ACK is required). */ + private deliver( + userId: string, + message: QueuedSocketMessage, + emit: MessageEmitter, + connectionId: number | null, + ): void { + const now = Date.now(); + message.updatedAt = now; + message.lastAttemptAt = now; + message.deliveredConnectionId = connectionId; + + // Fire-and-forget messages carry no ACK contract, so retain nothing. + if (!message.ackRequired) { + emit(message.event, message.payload); + this.acknowledge(userId, message.id); + return; + } + + message.status = 'in-flight'; + emit(message.event, message.payload, (ack?: unknown) => { + if (ack !== undefined && ack !== null) { + this.acknowledge(userId, message.id); + } + }); + + this.armAckTimer(userId, message); + } + + /** (Re)arm the ACK-timeout timer for a message. */ + private armAckTimer(userId: string, message: QueuedSocketMessage): void { + if (!message.ackRequired || message.ackTimeoutMs <= 0) { + return; + } + + this.clearTimer(userId, message.id); + + const timer = setTimeout(() => { + this.handleAckTimeout(userId, message.id); + }, message.ackTimeoutMs); + + // Never let a retry timer keep the process alive on its own. + (timer as { unref?: () => void }).unref?.(); + + this.timers.set(this.timerKey(userId, message.id), timer); + } + + /** Called when a message's ACK window elapses without acknowledgement. */ + private handleAckTimeout(userId: string, messageId: string): void { + this.timers.delete(this.timerKey(userId, messageId)); + + const message = this.queues.get(userId)?.find((m) => m.id === messageId); + if (!message || !message.ackRequired || message.status === 'exhausted') { + return; + } + + if (message.retries >= message.maxRetries) { + // Budget spent: keep the message for the next reconnect flush instead + // of dropping it, but stop the automatic retry loop. + message.status = 'exhausted'; + message.updatedAt = Date.now(); + logger.warn( + `[MessageQueue] Retry budget exhausted for userId=${userId} ` + + `messageId=${messageId} retries=${message.retries}`, + ); + return; + } + + const emit = this.emitters.get(userId); + if (!emit) { + // Driver is offline: park the message until the next reconnect flush. + message.status = 'pending'; + message.deliveredConnectionId = null; + return; + } + + message.retries += 1; + logger.debug( + `[MessageQueue] ACK timeout — retrying userId=${userId} ` + + `messageId=${messageId} attempt=${message.retries}/${message.maxRetries}`, + ); + + this.deliver(userId, message, emit, this.activeConnectionId.get(userId) ?? null); + } + + private timerKey(userId: string, messageId: string): string { + return `${userId}\u0000${messageId}`; + } + + private clearTimer(userId: string, messageId: string): void { + const key = this.timerKey(userId, messageId); + const timer = this.timers.get(key); + if (timer) { + clearTimeout(timer); + this.timers.delete(key); + } + } + + private resolveConnectionId(owner: object): number { + let id = this.connectionIds.get(owner); + if (id === undefined) { + id = ++this.connectionSeq; + this.connectionIds.set(owner, id); + } + return id; + } } +/** Singleton instance shared across the sockets layer. */ export const messageQueueService = new MessageQueueService(); diff --git a/src/sockets/socket.service.ts b/src/sockets/socket.service.ts index cf0afdb..e5df1ad 100644 --- a/src/sockets/socket.service.ts +++ b/src/sockets/socket.service.ts @@ -60,13 +60,17 @@ export class SocketService { this.connections.set(socket.id, meta); if (userId) { - const flushed = messageQueueService.flush(userId, (event, payload, ackCallback) => { - (socket.emit as (event: string, payload: unknown, ack?: unknown) => void)( - event, - payload, - ackCallback, - ); - }); + const flushed = messageQueueService.flush( + userId, + (event, payload, ackCallback) => { + (socket.emit as (event: string, payload: unknown, ack?: unknown) => void)( + event, + payload, + ackCallback, + ); + }, + socket, + ); if (flushed > 0) { logger.info( @@ -134,6 +138,10 @@ export class SocketService { `remaining=${this.connections.size - 1}`, ); + // Rewind any in-flight queued messages so they flush on reconnect, and + // stop retry timers so nothing is delivered into the dead socket. + messageQueueService.handleDisconnect(meta.userId); + // Remove the connection record this.connections.delete(socket.id); } diff --git a/tests/messageQueue.test.ts b/tests/messageQueue.test.ts index a84dd03..e24ceb9 100644 --- a/tests/messageQueue.test.ts +++ b/tests/messageQueue.test.ts @@ -1,12 +1,29 @@ import { MessageQueueService } from '../src/sockets/messageQueue'; +// Silence the queue's own logging during tests. +jest.mock('../src/config/logger', () => ({ + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), +})); + describe('MessageQueueService', () => { + afterEach(() => { + jest.useRealTimers(); + jest.clearAllMocks(); + }); + + // ── Core ACK gating ──────────────────────────────────────────────────────── + it('queues a message for a driver and keeps it pending until ack', () => { const service = new MessageQueueService(); const queued = service.enqueue('driver-123', 'delivery:status', { deliveryId: 'abc' }); expect(queued.userId).toBe('driver-123'); + expect(queued.ackRequired).toBe(true); + expect(queued.status).toBe('pending'); expect(service.getPending('driver-123')).toHaveLength(1); expect(service.acknowledge('driver-123', queued.id)).toBe(true); expect(service.getPending('driver-123')).toHaveLength(0); @@ -34,4 +51,228 @@ describe('MessageQueueService', () => { expect(service.getPending('driver-789')).toHaveLength(0); }); + + it('removes a message when the emit ack callback is invoked', () => { + const service = new MessageQueueService(); + const emit = jest.fn(); + + service.enqueue('driver-ack-cb', 'delivery:status', { deliveryId: 'ready' }); + service.flush('driver-ack-cb', emit); + + const ackCallback = emit.mock.calls[0][2] as (ack?: unknown) => void; + ackCallback({ ok: true }); + + expect(service.getPending('driver-ack-cb')).toHaveLength(0); + }); + + // ── Idempotent flush / reconnect ─────────────────────────────────────────── + + it('does not re-deliver an in-flight message when flush is repeated on the same connection', () => { + const service = new MessageQueueService(); + const emit = jest.fn(); + + service.enqueue('driver-idem', 'delivery:status', { deliveryId: 'a' }); + + expect(service.flush('driver-idem', emit)).toBe(1); + // Same emit function => same connection generation => no duplicate delivery. + expect(service.flush('driver-idem', emit)).toBe(0); + expect(emit).toHaveBeenCalledTimes(1); + }); + + it('flushes unacked messages exactly once on reconnect (no duplicates, acked skipped)', () => { + const service = new MessageQueueService(); + const firstConnection = jest.fn(); + const reconnected = jest.fn(); + + const first = service.enqueue('driver-reconnect', 'delivery:status', { n: 1 }); + service.enqueue('driver-reconnect', 'delivery:status', { n: 2 }); + + expect(service.flush('driver-reconnect', firstConnection)).toBe(2); + + // Driver acks only the first message while still connected. + expect(service.acknowledge('driver-reconnect', first.id)).toBe(true); + + // Reconnect: only the unacked message is re-delivered. + expect(service.flush('driver-reconnect', reconnected)).toBe(1); + expect(reconnected).toHaveBeenCalledTimes(1); + expect(reconnected.mock.calls[0][1]).toEqual({ n: 2 }); + + // A repeated flush on the same reconnect must not duplicate. + expect(service.flush('driver-reconnect', reconnected)).toBe(0); + expect(reconnected).toHaveBeenCalledTimes(1); + }); + + // ── ACK timeout / retry ──────────────────────────────────────────────────── + + it('retries a message when its ACK timeout elapses', () => { + jest.useFakeTimers(); + const service = new MessageQueueService({ defaultAckTimeoutMs: 1000, maxRetries: 3 }); + const emit = jest.fn(); + + service.enqueue('driver-retry', 'delivery:status', { deliveryId: 'r' }); + service.flush('driver-retry', emit); + expect(emit).toHaveBeenCalledTimes(1); + + jest.advanceTimersByTime(1000); + expect(emit).toHaveBeenCalledTimes(2); + expect(service.getPending('driver-retry')[0].retries).toBe(1); + }); + + it('stops retrying once the budget is exhausted but keeps the message for reconnect', () => { + jest.useFakeTimers(); + const service = new MessageQueueService({ defaultAckTimeoutMs: 500, maxRetries: 1 }); + const emit = jest.fn(); + + service.enqueue('driver-exhaust', 'delivery:status', { deliveryId: 'x' }); + service.flush('driver-exhaust', emit); + + jest.advanceTimersByTime(500); // one retry + expect(emit).toHaveBeenCalledTimes(2); + + jest.advanceTimersByTime(500 * 10); // budget spent, no further retries + expect(emit).toHaveBeenCalledTimes(2); + + const [pending] = service.getPending('driver-exhaust'); + expect(pending.status).toBe('exhausted'); + + // A reconnect still flushes the retained message exactly once. + const reconnected = jest.fn(); + expect(service.flush('driver-exhaust', reconnected)).toBe(1); + expect(reconnected).toHaveBeenCalledTimes(1); + }); + + it('cancels the retry timer when a message is acknowledged', () => { + jest.useFakeTimers(); + const service = new MessageQueueService({ defaultAckTimeoutMs: 1000 }); + const emit = jest.fn((_event: string, _payload: unknown, ack?: (ack?: unknown) => void) => { + ack?.({ ok: true }); + }); + + service.enqueue('driver-ack-timer', 'delivery:status', { deliveryId: 't' }); + service.flush('driver-ack-timer', emit); + + expect(service.getPending('driver-ack-timer')).toHaveLength(0); + + jest.advanceTimersByTime(10_000); + expect(emit).toHaveBeenCalledTimes(1); + }); + + // ── Bounded queue / eviction ─────────────────────────────────────────────── + + it('evicts the oldest message when the per-driver bound is exceeded', () => { + const service = new MessageQueueService({ maxQueueSize: 3 }); + + const oldest = service.enqueue('driver-bounded', 'delivery:status', { n: 1 }); + service.enqueue('driver-bounded', 'delivery:status', { n: 2 }); + service.enqueue('driver-bounded', 'delivery:status', { n: 3 }); + service.enqueue('driver-bounded', 'delivery:status', { n: 4 }); + + const pending = service.getPending('driver-bounded'); + expect(pending).toHaveLength(3); + expect(pending.map((m) => m.payload)).toEqual([{ n: 2 }, { n: 3 }, { n: 4 }]); + expect(pending.some((m) => m.id === oldest.id)).toBe(false); + }); + + // ── Per-driver isolation ─────────────────────────────────────────────────── + + it('isolates queues, acks and flushes per driver', () => { + const service = new MessageQueueService(); + + const driverA = service.enqueue('driver-A', 'delivery:status', { n: 'a' }); + service.enqueue('driver-B', 'delivery:status', { n: 'b' }); + + // A cross-driver ack must not remove another driver's message. + expect(service.acknowledge('driver-B', driverA.id)).toBe(false); + expect(service.getPending('driver-B')).toHaveLength(1); + + const emitA = jest.fn(); + expect(service.flush('driver-A', emitA)).toBe(1); + expect(service.flush('driver-B', jest.fn())).toBe(1); + + expect(service.acknowledge('driver-A', driverA.id)).toBe(true); + expect(service.getPending('driver-A')).toHaveLength(0); + expect(service.getPending('driver-B')).toHaveLength(1); + }); + + // ── Disconnect safety ────────────────────────────────────────────────────── + + it('rewinds in-flight messages on disconnect and flushes them on reconnect', () => { + jest.useFakeTimers(); + const service = new MessageQueueService({ defaultAckTimeoutMs: 1000 }); + const deadSocket = jest.fn(); + + service.enqueue('driver-offline', 'delivery:status', { n: 1 }); + service.flush('driver-offline', deadSocket); + + expect(service.handleDisconnect('driver-offline')).toBe(1); + expect(service.getPending('driver-offline')[0].status).toBe('pending'); + + // A pending timer must not deliver into the disconnected socket. + jest.advanceTimersByTime(1000); + expect(deadSocket).toHaveBeenCalledTimes(1); + + const reconnected = jest.fn(); + expect(service.flush('driver-offline', reconnected)).toBe(1); + expect(reconnected).toHaveBeenCalledTimes(1); + }); + + it('is a no-op when disconnecting an unknown or anonymous user', () => { + const service = new MessageQueueService(); + expect(service.handleDisconnect('nobody')).toBe(0); + expect(service.handleDisconnect(undefined)).toBe(0); + expect(() => service.handleDisconnect()).not.toThrow(); + }); + + // ── sendWithAck helper ───────────────────────────────────────────────────── + + it('queues, delivers and retries a sendWithAck message to the same socket', () => { + jest.useFakeTimers(); + const service = new MessageQueueService({ defaultAckTimeoutMs: 1000 }); + const socket = { data: { userId: 'driver-send' }, emit: jest.fn() }; + + const queued = service.sendWithAck(socket, 'delivery:status', { n: 1 }); + + expect(queued).not.toBeNull(); + expect(socket.emit).toHaveBeenCalledTimes(1); + expect(service.getPending('driver-send')).toHaveLength(1); + + jest.advanceTimersByTime(1000); // retry to the same socket + expect(socket.emit).toHaveBeenCalledTimes(2); + + const ackCallback = socket.emit.mock.calls[1][2] as (ack?: unknown) => void; + ackCallback({ ok: true }); + expect(service.getPending('driver-send')).toHaveLength(0); + + jest.advanceTimersByTime(5000); + expect(socket.emit).toHaveBeenCalledTimes(2); + }); + + it('returns null from sendWithAck for an unauthenticated socket', () => { + const service = new MessageQueueService(); + const socket = { data: {}, emit: jest.fn() }; + + expect(service.sendWithAck(socket, 'delivery:status', { n: 1 })).toBeNull(); + }); + + // ── Validation & clear ───────────────────────────────────────────────────── + + it('rejects enqueue calls without a userId or event', () => { + const service = new MessageQueueService(); + + expect(() => service.enqueue('', 'delivery:status', {})).toThrow(/userId is required/); + expect(() => service.enqueue('driver-x', ' ', {})).toThrow(/event name is required/); + }); + + it('clears a single driver queue or every queue', () => { + const service = new MessageQueueService(); + service.enqueue('driver-1', 'e', { n: 1 }); + service.enqueue('driver-2', 'e', { n: 2 }); + + expect(service.clear('driver-1')).toBe(1); + expect(service.getPending('driver-1')).toHaveLength(0); + expect(service.getPending('driver-2')).toHaveLength(1); + + expect(service.clear()).toBe(1); + expect(service.getPending('driver-2')).toHaveLength(0); + }); });