diff --git a/.env.example b/.env.example index 4c2e5557d..76a94ddcc 100644 --- a/.env.example +++ b/.env.example @@ -687,9 +687,11 @@ PLUGINS_DIR=./data/plugins # Plugin directory (default: ./data/plugins) # PLUGINS_DIR, which is where plugin packages are installed. It moves plugin state only: the # databases, sessions, media and auth dirs each carry their own path knob and none of them follow # this one. -# Moving it does NOT carry the existing registry across: point it at an empty directory and the +# Moving it does NOT carry the existing state across: point it at an empty directory and the # gateway starts with no record of installed plugins, even though their packages are still under -# PLUGINS_DIR. Move data/plugins/registry.json yourself when you change this. +# PLUGINS_DIR. Move the whole `/plugins` tree yourself when you change this, not just +# registry.json: each plugin's persisted ctx.storage lives beside it as `/key-*.json`, +# so carrying the registry alone leaves every plugin enabled but with its state gone. # PLUGIN_STATE_DIR=./data # Cap on a plugin .zip downloaded by install-from-URL (matches the 5 MB upload limit). A non-positive # or non-numeric value falls back to the default. diff --git a/CHANGELOG.md b/CHANGELOG.md index db42eddd2..1045acaf7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,14 +14,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed - `POST /sessions/{sessionId}/chats/unread` publishes its own `MarkChatUnreadDto` rather than sharing `MarkChatReadDto`. The body is unchanged (`chatId` alone), but a generated client sees the schema under a new name. -- ⚠️ **Breaking (Go and Java clients).** `subscribePresence` takes its own `SubscribePresenceRequest` rather than the shared `MarkChatRequest`, which now serves `markUnread` alone. Swap the type at the call site; the wire body is unchanged. `SubscribePresenceDto` had no contract-gate coverage while one type stood for two routes. +- ⚠️ **Breaking (Go, Java and typed Python callers).** `markRead` and `subscribePresence` each take their own request type rather than the shared `MarkChatRequest`, which now serves `markUnread` alone. Swap the type at both call sites; the wire body is unchanged and the JavaScript and PHP clients are unaffected. `SubscribePresenceDto` had no contract-gate coverage while one type stood for two routes. ### Fixed +- `POST /chats/read` answers 400 for `"messageIds": null` instead of 500. `@IsOptional` skips every validator for null as well as undefined, so the value reached the Baileys adapter and was dereferenced there. The published schema now carries `minItems` too, so it no longer advertises an empty array the server refuses. +- A read receipt goes only to the chat the caller named. A message id belonging to another chat in the same session carried that chat's address out of the message store, so the receipt landed there while the route reported success for the chat in the path. +- The Go client can express an empty `messageIds` again. `omitempty` on a plain slice dropped it, so a caller asking for nothing to be acknowledged silently acknowledged the newest message; the field is a pointer, so absent and empty are distinct on the wire. - The dashboard CSP nonce is substituted at every occurrence in the served document, not only the first. One placeholder exists today, so a second would have been left reading the literal text and its script refused by the browser. - Outbound webhook deliveries survive a hard crash. Fan-out was fire-and-forget, so a crash between persisting a message and completing its POST lost the delivery, against a documented at-least-once contract. Deliveries are now recorded before they are attempted, and a bounded sweep replays whatever is stranded under its stored idempotency key. +- A stranded webhook delivery now gets the replay budget it was promised. The reconciler read success from a call that cannot fail, so a replay that never delivered was retired as dispatched on the first sweep and its payload dropped. Delivery reports an outcome instead, and a failed replay stays pending. +- Restoring a backup no longer aborts when the target already holds the outbound delivery records. The table has no session foreign key, so the replace never cleared it and every overlapping row collided, rolling the whole import back. - Settled outbound delivery records are pruned after `WEBHOOK_OUTBOX_RETENTION_DAYS` (default 7). A record that can still be replayed is never pruned on age, and a non-positive window falls back to the default rather than letting the table grow without bound. - `PLUGIN_STATE_DIR` moves the plugin registry and per-plugin storage off the default `./data`. It was the one piece of state with no path knob, so a test run rewrote the developer's own registry. +- `backup.sh` and `restore.sh` follow `PLUGIN_STATE_DIR`. Both hardcoded the plugin state under the data dir, so with the knob set the archive carried neither the registry nor any plugin's persisted storage, and a restore put nothing back. The knob's own note now says to move the whole tree, not just `registry.json`. - The e2e lane sweeps the throwaway state roots it creates. Each suite gets its own, nothing removed them, and the temp directory accumulated hundreds of entries over a few days of runs. - e2e assertions are no longer answered by unrelated processes on the host. supertest binds its per-request listener to the wildcard address and then dials 127.0.0.1, which on macOS lets a process holding that port on 127.0.0.1 answer instead. Each suite's server now listens on loopback during init, which supertest reuses. - A stalled `apt-get` can no longer hold a CI run open. The scripts-smoke job installed sqlite3 and shellcheck unbounded, so a slow mirror held two main runs past an hour with every other job already green. Both steps now time out and skip the install when the runner already ships the tool. diff --git a/openapi.json b/openapi.json index fcd120859..ba8ff23e8 100644 --- a/openapi.json +++ b/openapi.json @@ -10265,6 +10265,7 @@ "messageIds": { "description": "Specific message IDs to mark read. Baileys acknowledges individual messages, so without this only the newest message the engine still holds in memory gets a receipt — a burst leaves its earlier messages unread forever, and a restarted session has no message to acknowledge at all. Callers that persist inbound message IDs should send them here. Ignored by whatsapp-web.js, whose own sendSeen is chat-level.", "maxItems": 100, + "minItems": 1, "example": [ "3EB0C767D26B8A3F1A2B", "3EB0C767D26B8A3F1A2C" diff --git a/scripts/backup.sh b/scripts/backup.sh index cf3b023c8..b3e834743 100755 --- a/scripts/backup.sh +++ b/scripts/backup.sh @@ -64,7 +64,14 @@ MEDIA_DIR="$(openwa_resolve STORAGE_LOCAL_PATH "$DATA_DIR/media")" # registry and each plugin's ctx.storage below — so an unset PLUGINS_DIR must resolve there # too, or the archive silently omits the plugin packages. PLUGIN_PACKAGES_DIR="$(openwa_resolve PLUGINS_DIR "$DATA_DIR/plugins")" -PLUGIN_STATE_DIR="$DATA_DIR/plugins" +# Plugin registry + every plugin's persisted ctx.storage. The app puts them at /plugins, +# where dataDir is PLUGIN_STATE_DIR when that is set and ./data otherwise, so the knob has to be +# resolved here exactly like PLUGINS_DIR above. Hardcoding $DATA_DIR/plugins meant an operator who +# moved plugin state got an archive with neither the registry nor any plugin's storage in it, and +# a restore that put nothing back. Resolved under its own name because the knob names the ROOT, +# not the plugins directory inside it. +PLUGIN_STATE_ROOT="$(openwa_resolve PLUGIN_STATE_DIR "$DATA_DIR")" +PLUGIN_STATE_DIR="$PLUGIN_STATE_ROOT/plugins" GENERATED_ENV="$DATA_DIR/.env.generated" ADMIN_KEY_FILE="$DATA_DIR/.api-key" diff --git a/scripts/restore.sh b/scripts/restore.sh index 25dd4e873..1b2173f9d 100755 --- a/scripts/restore.sh +++ b/scripts/restore.sh @@ -79,7 +79,14 @@ MEDIA_DIR="$(openwa_resolve STORAGE_LOCAL_PATH "$DATA_DIR/media")" # registry and each plugin's ctx.storage below — so an unset PLUGINS_DIR must resolve there # too, or the archive silently omits the plugin packages. PLUGIN_PACKAGES_DIR="$(openwa_resolve PLUGINS_DIR "$DATA_DIR/plugins")" -PLUGIN_STATE_DIR="$DATA_DIR/plugins" +# Plugin registry + every plugin's persisted ctx.storage. The app puts them at /plugins, +# where dataDir is PLUGIN_STATE_DIR when that is set and ./data otherwise, so the knob has to be +# resolved here exactly like PLUGINS_DIR above. Hardcoding $DATA_DIR/plugins meant an operator who +# moved plugin state got an archive with neither the registry nor any plugin's storage in it, and +# a restore that put nothing back. Resolved under its own name because the knob names the ROOT, +# not the plugins directory inside it. +PLUGIN_STATE_ROOT="$(openwa_resolve PLUGIN_STATE_DIR "$DATA_DIR")" +PLUGIN_STATE_DIR="$PLUGIN_STATE_ROOT/plugins" RESTORE_TIMESTAMP="$(date +%Y%m%d-%H%M%S)" RESOLVED_CWD="$(pwd -P)" diff --git a/scripts/smoke-test-backup-restore.sh b/scripts/smoke-test-backup-restore.sh index 91a08da3a..3a226428a 100755 --- a/scripts/smoke-test-backup-restore.sh +++ b/scripts/smoke-test-backup-restore.sh @@ -293,5 +293,42 @@ if [ "$(db_fingerprint "$F/extract2/main.sqlite")" != "STALE-main" ]; then fi pass "(f) data/.env.generated resolves paths for both scripts, and the environment still wins" +echo "" +echo "==> (g) PLUGIN_STATE_DIR moves the registry and ctx.storage, and both scripts follow it" +# The knob names the ROOT; the app keeps plugin state at /plugins. Both scripts hardcoded +# $OPENWA_DATA_DIR/plugins, so with the knob set the archive carried neither the registry nor any +# plugin's persisted storage, and the restore put nothing back. Silent both ways: an empty source +# directory simply produces no plugin-state entry. +G="$WORK/g" +mkdir -p "$G/state" "$G/elsewhere/plugins/chatwoot" "$G/extract" "$G/restore/state" +make_fixture "$G/state/main.sqlite" "golf-main" +make_fixture "$G/state/openwa.sqlite" "golf-data" +printf '{"plugins":[{"id":"chatwoot"}]}' >"$G/elsewhere/plugins/registry.json" +printf 'mapped-conversation' >"$G/elsewhere/plugins/chatwoot/key-Zm9v.json" +( + cd "$G" + OPENWA_DATA_DIR="$G/state" PLUGIN_STATE_DIR="$G/elsewhere" BACKUP_DIR="$G/out" \ + MAIN_DATABASE_NAME="$G/state/main.sqlite" DATABASE_NAME="$G/state/openwa.sqlite" "$BACKUP" >/dev/null +) +ARCHIVE_G="$(ls "$G"/out/openwa-backup-*.tar.gz)" +tar -xzf "$ARCHIVE_G" -C "$G/extract" +if [ ! -f "$G/extract/plugin-state/registry.json" ]; then + fail "(g) backup ignored PLUGIN_STATE_DIR: the plugin registry is missing from the archive" +fi +if [ ! -f "$G/extract/plugin-state/chatwoot/key-Zm9v.json" ]; then + fail "(g) backup ignored PLUGIN_STATE_DIR: a plugin's persisted ctx.storage is missing" +fi +# And the restore has to put them back where the knob points, not under the default data dir. +( + cd "$G" + OPENWA_DATA_DIR="$G/restore/state" PLUGIN_STATE_DIR="$G/restored-elsewhere" \ + MAIN_DATABASE_NAME="$G/restore/state/main.sqlite" DATABASE_NAME="$G/restore/state/openwa.sqlite" \ + "$RESTORE" "$ARCHIVE_G" --force >/dev/null +) +if [ ! -f "$G/restored-elsewhere/plugins/registry.json" ]; then + fail "(g) restore ignored PLUGIN_STATE_DIR: the registry did not land under the configured root" +fi +pass "(g) PLUGIN_STATE_DIR is honoured by backup and by restore" + echo "" echo "All smoke tests passed!" diff --git a/sdk/go/mark_read_body_test.go b/sdk/go/mark_read_body_test.go new file mode 100644 index 000000000..b73560b36 --- /dev/null +++ b/sdk/go/mark_read_body_test.go @@ -0,0 +1,50 @@ +package openwa + +import ( + "context" + "strings" + "testing" +) + +// The three states of messageIds differ on the wire, and a plain []string with omitempty could not +// tell two of them apart: an empty slice was dropped, so a caller asking for nothing to be +// acknowledged silently acknowledged the newest message instead. The pointer makes "absent" and +// "empty" distinct, and this pins all three so the tag cannot quietly go back. +func TestMarkReadBodyDistinguishesAbsentFromEmpty(t *testing.T) { + empty := []string{} + named := []string{"3EB0C767D26B8A3F1A2B"} + + cases := []struct { + name string + body MarkChatReadRequest + want string + }{ + {"absent omits the key", MarkChatReadRequest{ChatID: "628123@c.us"}, `{"chatId":"628123@c.us"}`}, + {"empty sends []", MarkChatReadRequest{ChatID: "628123@c.us", MessageIDs: &empty}, `"messageIds":[]`}, + {"named sends the ids", MarkChatReadRequest{ChatID: "628123@c.us", MessageIDs: &named}, `"messageIds":["3EB0C767D26B8A3F1A2B"]`}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rt := &recordTransport{status: 200, body: `{"success":true}`} + c := newTestClient(t, rt) + + if _, err := c.Chats.MarkRead(context.Background(), "s1", tc.body); err != nil { + t.Fatalf("MarkRead: %v", err) + } + if got := string(rt.lastRaw); !strings.Contains(got, tc.want) { + t.Fatalf("body = %s, want it to contain %s", got, tc.want) + } + }) + } + + // The absent case must ALSO not carry the key at all, which "contains" cannot express. + rt := &recordTransport{status: 200, body: `{"success":true}`} + c := newTestClient(t, rt) + if _, err := c.Chats.MarkRead(context.Background(), "s1", MarkChatReadRequest{ChatID: "628123@c.us"}); err != nil { + t.Fatalf("MarkRead: %v", err) + } + if strings.Contains(string(rt.lastRaw), "messageIds") { + t.Fatalf("absent body must omit messageIds entirely, got %s", rt.lastRaw) + } +} diff --git a/sdk/go/types_chat.go b/sdk/go/types_chat.go index 09a46c65c..9f5e7d3a1 100644 --- a/sdk/go/types_chat.go +++ b/sdk/go/types_chat.go @@ -35,7 +35,13 @@ type MarkChatReadRequest struct { // MessageIDs are the messages to acknowledge (at most 100; an empty list is refused). Baileys // acknowledges individual messages, so without this only the newest message the engine still // holds in memory gets a receipt. Ignored by whatsapp-web.js, whose own sendSeen is chat-level. - MessageIDs []string `json:"messageIds,omitempty"` + // + // A POINTER because the three states differ on the wire and a plain slice cannot tell two of + // them apart: nil omits the key (acknowledge the newest), &[]string{} sends [] (the server + // refuses it with a 400), and a populated slice names the messages. With `[]string` plus + // omitempty an empty slice was dropped, so a caller asking for nothing to be acknowledged + // silently acknowledged the newest message instead. + MessageIDs *[]string `json:"messageIds,omitempty"` } // ChatState is the typing indicator a chat shows. diff --git a/src/engine/adapters/baileys-contacts.ts b/src/engine/adapters/baileys-contacts.ts index c176bef37..a68bf41cb 100644 --- a/src/engine/adapters/baileys-contacts.ts +++ b/src/engine/adapters/baileys-contacts.ts @@ -355,7 +355,9 @@ export class BaileysContacts { * synthesised key, which is what the 1:1 case ran on before. */ private async receiptKeys(chatId: string, messageIds?: string[]): Promise { - if (messageIds === undefined) { + // null as well as undefined: the REST body rejects an explicit null, but this is the engine + // boundary and an internal caller reaching it with one used to dereference it below as a 500. + if (messageIds === undefined || messageIds === null) { const last = this.host.lastMessage(chatId); return last ? [last.key] : []; } @@ -364,7 +366,17 @@ export class BaileysContacts { } const remoteJid = this.host.toEngineJid(chatId); const stored = (await this.host.getStoredMessages(messageIds)) ?? []; - const keyById = new Map(stored.filter(msg => msg.key?.id).map(msg => [msg.key.id as string, msg.key])); + // A stored key is only usable when it belongs to THIS chat. Without the check, an id from + // another chat in the same session carried that chat's remoteJid into readMessages, so the + // receipt landed there while the route answered success for the chat the caller named. Both + // sides fold through toEngineJid so the @c.us and @s.whatsapp.net spellings of one chat still + // match; anything that still differs falls back to the synthesised key for the ADDRESSED chat, + // which is exactly what every id ran on before stored keys existed. + const keyById = new Map( + stored + .filter(msg => msg.key?.id && msg.key.remoteJid && this.host.toEngineJid(msg.key.remoteJid) === remoteJid) + .map(msg => [msg.key.id as string, msg.key]), + ); return messageIds.map(id => keyById.get(id) ?? { remoteJid, id, fromMe: false }); } diff --git a/src/engine/adapters/baileys-send-seen.spec.ts b/src/engine/adapters/baileys-send-seen.spec.ts index b9e735399..ed7693951 100644 --- a/src/engine/adapters/baileys-send-seen.spec.ts +++ b/src/engine/adapters/baileys-send-seen.spec.ts @@ -69,6 +69,48 @@ describe('sendSeen', () => { ]); }); + it('never sends the receipt to another chat, even when the store resolves the id', async () => { + // The stored key carries its own remoteJid. Used unchecked, an id belonging to a different chat + // in the same session acknowledged THAT chat while the route answered success for the one the + // caller named, and the caller's own chat stayed unread. + const readMessages = jest.fn().mockResolvedValue(undefined); + const host = makeHost({ + getSocket: () => ({ readMessages }) as unknown as WASocket, + getStoredMessages: () => + Promise.resolve([stored({ id: 'X1', remoteJid: '628999@s.whatsapp.net', fromMe: false })]), + }); + + await expect(new BaileysContacts(host, 500).sendSeen('628123@c.us', ['X1'])).resolves.toBe(true); + + // Falls back to the synthesised key for the ADDRESSED chat rather than the stored one. + expect(readMessages).toHaveBeenCalledWith([{ remoteJid: '628123@s.whatsapp.net', id: 'X1', fromMe: false }]); + }); + + it('still uses a stored key whose chat is spelled in the other dialect', async () => { + // Control for the check above: the same chat written @c.us must still resolve, or the fix would + // have cost every receipt its participant and fromMe. + const readMessages = jest.fn().mockResolvedValue(undefined); + const host = makeHost({ + getSocket: () => ({ readMessages }) as unknown as WASocket, + getStoredMessages: () => Promise.resolve([stored({ id: 'D1', remoteJid: '628123@c.us', fromMe: true })]), + }); + + await expect(new BaileysContacts(host, 500).sendSeen('628123@s.whatsapp.net', ['D1'])).resolves.toBe(true); + expect(readMessages).toHaveBeenCalledWith([{ id: 'D1', remoteJid: '628123@c.us', fromMe: true }]); + }); + + it('treats a null id list as absent rather than dereferencing it', async () => { + // The REST body rejects an explicit null, but this is the engine boundary; it used to throw a + // TypeError here and surface as a 500. + const readMessages = jest.fn().mockResolvedValue(undefined); + const host = makeHost({ getSocket: () => ({ readMessages }) as unknown as WASocket }); + + await expect(new BaileysContacts(host, 500).sendSeen('628123@c.us', null as unknown as string[])).resolves.toBe( + true, + ); + expect(readMessages).toHaveBeenCalledWith([{ id: 'M1', remoteJid: '628123@s.whatsapp.net' }]); + }); + it('carries the participant on a group receipt, which a synthesised key cannot', async () => { // A group receipt with no `participant` names no sender, so WhatsApp cannot attribute it and // the message stays unread on the sender's side while the API answers success: true. diff --git a/src/modules/infra/infra-data.controller.spec.ts b/src/modules/infra/infra-data.controller.spec.ts index ef7cf2697..67642fb8a 100644 --- a/src/modules/infra/infra-data.controller.spec.ts +++ b/src/modules/infra/infra-data.controller.spec.ts @@ -1268,6 +1268,38 @@ describe('InfraDataController.import/export preserves every data-DB table', () = expect((await lidRepo.findOneByOrFail({ lid: '222' })).phone).toBeNull(); }); + // Restoring ONTO the instance that produced the archive is the rollback flow, and it is the one the + // outbox broke. The table carries no FK to sessions, so the sessions DELETE never reached it, and + // UNIQUE(webhookId, idempotencyKey) then collided on every row until the all-or-nothing gate rolled + // the entire import back. Every other table's test clears first, which is why nothing caught it; + // this one deliberately does not. + it('restores webhook_outbox_events onto an instance that already holds them', async () => { + await seedSession('s1'); + const outboxRepo = ds.getRepository(WebhookOutboxEvent); + await outboxRepo.save( + outboxRepo.create({ + webhookId: 'wh-1', + sessionId: 's1', + event: 'message.received', + idempotencyKey: 'key-1', + deliveryId: 'del-1', + payload: { from: '628111@c.us' }, + state: 'pending', + attempts: 0, + }), + ); + + const dump = await controller.exportData(); + expect((dump.tables as unknown as { webhookOutboxEvents?: unknown[] }).webhookOutboxEvents).toHaveLength(1); + + const res = await controller.importData({ tables: dump.tables }); + + expect(res.warnings).toEqual([]); + expect(res.imported).toBe(true); + expect(await outboxRepo.count()).toBe(1); + expect((await outboxRepo.findOneByOrFail({ idempotencyKey: 'key-1' })).state).toBe('pending'); + }); + // The messages import column list must carry every later-added column; `author` (the group // sender identity) was the one that drifted — a restored backup silently lost all attribution. it('restores the group-sender author column on a backup→restore', async () => { diff --git a/src/modules/infra/infra-data.service.ts b/src/modules/infra/infra-data.service.ts index 3e1c5f2f0..8d19c8d62 100644 --- a/src/modules/infra/infra-data.service.ts +++ b/src/modules/infra/infra-data.service.ts @@ -602,6 +602,12 @@ export class InfraDataService { await clearTable('conversation_mappings'); await clearTable('ingress_events'); await clearTable('webhook_delivery_failures'); + // Same rule, and it bites harder here: webhook_outbox_events carries UNIQUE(webhookId, + // idempotencyKey), so without this clear a restore onto an instance that already holds the + // archive's rows collides on every one of them, and the all-or-nothing gate below rolls the + // whole import back. Restoring a backup onto the instance that produced it is exactly the + // rollback flow, so leaving it out broke the recovery path rather than a corner of it. + await clearTable('webhook_outbox_events'); await clearTable('integration_delivery_failures'); // status_updates has no FK to sessions; clear it explicitly so the replace is complete. await clearTable('status_updates'); diff --git a/src/modules/session/dto/mark-chat-read.dto.spec.ts b/src/modules/session/dto/mark-chat-read.dto.spec.ts index 24a88ff3f..7d65e11e9 100644 --- a/src/modules/session/dto/mark-chat-read.dto.spec.ts +++ b/src/modules/session/dto/mark-chat-read.dto.spec.ts @@ -34,6 +34,12 @@ describe('MarkChatReadDto messageIds validation', () => { expect(messageIdErrors(['3EB0C767D26B8A3F1A2B', '3EB0C767D26B8A3F1A2C'])).toBe(0); }); + it('rejects an explicit null instead of letting it reach the engine', () => { + // @IsOptional() skips every validator for null as well as undefined, so `"messageIds": null` + // used to pass validation and then dereference in the adapter as a 500. + expect(messageIdErrors(null)).toBeGreaterThan(0); + }); + it('rejects an empty list rather than reading it as "the newest message"', () => { // The engine treats a missing list as "acknowledge the newest message". A caller that computed // its unread set and got none back must not land in that branch. diff --git a/src/modules/session/dto/mark-chat-read.dto.ts b/src/modules/session/dto/mark-chat-read.dto.ts index e993de6fa..662b43e56 100644 --- a/src/modules/session/dto/mark-chat-read.dto.ts +++ b/src/modules/session/dto/mark-chat-read.dto.ts @@ -1,5 +1,5 @@ import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; -import { ArrayMaxSize, ArrayNotEmpty, IsArray, IsNotEmpty, IsOptional, IsString, Matches } from 'class-validator'; +import { ArrayMaxSize, ArrayNotEmpty, IsArray, IsNotEmpty, IsString, Matches, ValidateIf } from 'class-validator'; /** * Ceiling on one request's receipt batch, so a single call cannot hand the engine an unbounded key @@ -45,9 +45,15 @@ export class MarkChatReadDto { // this the published schema advertises an unbounded array and a caller batching more than the // cap discovers the limit as a 400. maxItems: MARK_READ_MESSAGE_IDS_MAX, + // @ArrayNotEmpty rejects [], so the published schema has to say so too; without minItems the + // contract advertised an empty array as valid against a server that answers 400. + minItems: 1, example: ['3EB0C767D26B8A3F1A2B', '3EB0C767D26B8A3F1A2C'], }) - @IsOptional() + // Not @IsOptional: that skips every validator for null as well as undefined, so an explicit + // `"messageIds": null` reached the engine unchecked and dereferenced there as a 500. Absent stays + // absent; present-but-null falls through to @IsArray and answers 400. + @ValidateIf((_object, value) => value !== undefined) @IsArray() // An empty array asks for nothing to be acknowledged. Rejected rather than accepted, because the // engine reads a missing list as "the newest message" and the two must not collapse: a caller that diff --git a/src/modules/webhook/webhook-delivery.service.spec.ts b/src/modules/webhook/webhook-delivery.service.spec.ts index 9a3143be1..60299310c 100644 --- a/src/modules/webhook/webhook-delivery.service.spec.ts +++ b/src/modules/webhook/webhook-delivery.service.spec.ts @@ -608,6 +608,29 @@ describe('WebhookDeliveryService', () => { expect(hookManager.execute).not.toHaveBeenCalledWith('webhook:error', expect.anything(), expect.anything()); }); + // The reconciler branches on this VALUE, never on a throw: every failing path inside redeliver + // dead-letters in place, so nothing reaches the caller as an exception. When this reported + // nothing, a replay that never delivered was retired 'dispatched' on the first sweep and the + // documented WEBHOOK_RECONCILE_MAX_ATTEMPTS budget was unreachable. + it('redeliver reports failed when the receiver never accepts, and delivered when it does', async () => { + const webhook = createMockWebhook({ events: ['message.received'], retryCount: 1 }); + (hookManager.execute as jest.Mock).mockResolvedValue({ continue: true, data: {} }); + + mockFetch.mockReset(); + mockFetch.mockRejectedValue(new Error('receiver down')); + await expect( + service.redeliver(webhook, 'sess-1', 'message.received', 'stored-key-1', { from: 'x@c.us' }), + ).resolves.toBe('failed'); + + // Control: the same call on a receiver that answers 2xx must NOT report 'failed', otherwise + // the assertion above is satisfied by a method that reports failure unconditionally. + mockFetch.mockReset(); + mockFetch.mockResolvedValue({ ok: true, status: 200 }); + await expect( + service.redeliver(webhook, 'sess-1', 'message.received', 'stored-key-2', { from: 'x@c.us' }), + ).resolves.toBe('delivered'); + }); + it("isolates each webhook's data so an in-place before-hook mutation cannot bleed across webhooks", async () => { const a = createMockWebhook({ id: 'wh-a', events: ['message.received'] }); const b = createMockWebhook({ id: 'wh-b', events: ['message.received'] }); diff --git a/src/modules/webhook/webhook-delivery.service.ts b/src/modules/webhook/webhook-delivery.service.ts index af35ac00e..6ba2fd7b7 100644 --- a/src/modules/webhook/webhook-delivery.service.ts +++ b/src/modules/webhook/webhook-delivery.service.ts @@ -60,6 +60,13 @@ const DEFAULT_WEBHOOK_MAX_PAYLOAD_BYTES = 1024 * 1024; const DEFAULT_WEBHOOK_SHUTDOWN_DRAIN_MS = 5000; /** Per-event-occurrence context threaded through the dispatch pipeline stages (was closure state). */ +/** + * The result of one delivery attempt. Reported, not thrown: every failure below is already handled + * in place, so nothing reaches a caller as an exception and a try/catch cannot tell a delivered + * event from a dead-lettered one. + */ +export type WebhookDeliveryOutcome = 'delivered' | 'enqueued' | 'failed'; + interface DispatchEventContext { sessionId: string; event: string; @@ -392,23 +399,32 @@ export class WebhookDeliveryService implements OnModuleInit, OnModuleDestroy { } } + /** + * What became of one delivery attempt, reported rather than thrown. + * + * Every failure path here is already handled in place (a dead-letter row, a hook, a log), so none + * of them reach the caller as an exception. The reconciler has to tell a delivered event from a + * dead-lettered one to know whether the outbox row may be retired, and a caught throw cannot tell + * it: there is none. This mirrors the inbound twin, where `ingressEnqueue.enqueue` returns an + * outcome and the caller retires the payload only when it is not 'failed'. + */ private async deliverOne( webhook: Webhook, deliveryId: string, idempotencyKey: string, ctx: DispatchEventContext, - ): Promise { + ): Promise { const preflight = await this.preflightDelivery(webhook, deliveryId, idempotencyKey, ctx); if (!preflight) { - return; + // Preflight records its own undelivered row and returns null; nothing left the process. + return 'failed'; } const { finalPayload, body, headers } = preflight; // Use queue if available, otherwise fallback to direct delivery if (this.queueEnabled && this.webhookQueue) { - await this.enqueueWithFallback(webhook, finalPayload, body, headers, deliveryId, idempotencyKey, ctx); - } else { - await this.deliverDirect(webhook, finalPayload, body, headers, deliveryId, ctx); + return this.enqueueWithFallback(webhook, finalPayload, body, headers, deliveryId, idempotencyKey, ctx); } + return this.deliverDirect(webhook, finalPayload, body, headers, deliveryId, ctx); } private async enqueueWithFallback( @@ -419,7 +435,7 @@ export class WebhookDeliveryService implements OnModuleInit, OnModuleDestroy { deliveryId: string, idempotencyKey: string, ctx: DispatchEventContext, - ): Promise { + ): Promise { const { sessionId, event } = ctx; try { // Sign the exact pre-serialized body from preflight. The processor re-serializes the same @@ -515,8 +531,13 @@ export class WebhookDeliveryService implements OnModuleInit, OnModuleDestroy { webhookId: webhook.id, action: 'webhook_queue_fallback_failed', }); + return 'failed'; } + // The queue never took it, but the fallback POST did. + return 'delivered'; } + // Handed to BullMQ, which owns the retries and the dead-letter row from here. + return 'enqueued'; } /** Direct delivery when the queue is disabled. */ @@ -527,7 +548,7 @@ export class WebhookDeliveryService implements OnModuleInit, OnModuleDestroy { headers: Record, deliveryId: string, ctx: DispatchEventContext, - ): Promise { + ): Promise { const { sessionId, event } = ctx; try { await this.deliverWebhook(webhook, finalPayload, headers, body); @@ -557,7 +578,9 @@ export class WebhookDeliveryService implements OnModuleInit, OnModuleDestroy { webhookId: webhook.id, action: 'webhook_delivery_failed', }); + return 'failed'; } + return 'delivered'; } /** @@ -651,9 +674,9 @@ export class WebhookDeliveryService implements OnModuleInit, OnModuleDestroy { event: string, idempotencyKey: string, data: Record, - ): Promise { + ): Promise { const deliveryId = generateDeliveryId(); - await this.deliverOne(webhook, deliveryId, idempotencyKey, { sessionId, event, baseData: data }); + return this.deliverOne(webhook, deliveryId, idempotencyKey, { sessionId, event, baseData: data }); } /** diff --git a/src/modules/webhook/webhook-reconciler.service.spec.ts b/src/modules/webhook/webhook-reconciler.service.spec.ts index 083db916c..6d9188ada 100644 --- a/src/modules/webhook/webhook-reconciler.service.spec.ts +++ b/src/modules/webhook/webhook-reconciler.service.spec.ts @@ -43,7 +43,7 @@ describe('WebhookReconcilerService', () => { beforeEach(() => { outbox = { findStale: jest.fn().mockResolvedValue([]), close: jest.fn(), countAttempt: jest.fn() }; - delivery = { redeliver: jest.fn().mockResolvedValue(undefined) }; + delivery = { redeliver: jest.fn().mockResolvedValue('delivered') }; webhooks = { findOne: jest.fn().mockResolvedValue({ id: 'wh-1', active: true }) }; service = new WebhookReconcilerService(webhooks as never, outbox as never, delivery as never); }); @@ -66,9 +66,13 @@ describe('WebhookReconcilerService', () => { expect(stats).toMatchObject({ scanned: 1, replayed: 1 }); }); - it('counts the attempt BEFORE replaying, so a delivery that always throws still exhausts its budget', async () => { + // The REAL shape of a delivery failure. redeliver resolves 'failed' rather than rejecting, because + // every failing path inside it already dead-letters and logs; an earlier version of this test + // mocked a rejection instead, which the collaborator cannot produce, so the budget it claimed to + // guard was never exercised and a dead-lettered event was retired as 'dispatched' on sweep one. + it('leaves a row pending when the replay did not deliver, so the budget is actually spent', async () => { outbox.findStale.mockResolvedValue([row({ attempts: 1 })]); - delivery.redeliver.mockRejectedValue(new Error('receiver down')); + delivery.redeliver.mockResolvedValue('failed'); const stats = await service.sweep(OPTS); @@ -81,6 +85,26 @@ describe('WebhookReconcilerService', () => { expect(stats).toMatchObject({ replayed: 0, failed: 1 }); }); + it('retires the row when the replay was handed to the queue rather than delivered inline', async () => { + outbox.findStale.mockResolvedValue([row({ attempts: 1 })]); + delivery.redeliver.mockResolvedValue('enqueued'); + + const stats = await service.sweep(OPTS); + + expect(outbox.close).toHaveBeenCalledWith('wh-1', 'stored-key_wh-1', 'dispatched'); + expect(stats).toMatchObject({ replayed: 1, failed: 0 }); + }); + + it('keeps a row pending when the replay throws an unexpected fault', async () => { + outbox.findStale.mockResolvedValue([row({ attempts: 1 })]); + delivery.redeliver.mockRejectedValue(new Error('boom')); + + const stats = await service.sweep(OPTS); + + expect(outbox.close).not.toHaveBeenCalled(); + expect(stats).toMatchObject({ replayed: 0, failed: 1 }); + }); + it('stops replaying once the budget is spent instead of looping forever', async () => { outbox.findStale.mockResolvedValue([row({ attempts: 3 })]); diff --git a/src/modules/webhook/webhook-reconciler.service.ts b/src/modules/webhook/webhook-reconciler.service.ts index 354fcec8e..834540ded 100644 --- a/src/modules/webhook/webhook-reconciler.service.ts +++ b/src/modules/webhook/webhook-reconciler.service.ts @@ -110,11 +110,28 @@ export class WebhookReconcilerService implements OnModuleInit, OnModuleDestroy { } await this.outbox.countAttempt(row.id, row.attempts); try { - await this.delivery.redeliver(webhook, row.sessionId, row.event, row.idempotencyKey, row.payload); + // The outcome is a RETURN VALUE, not an exception. Every delivery failure is handled in + // place (dead-letter row, hook, log), so redeliver resolves either way and a catch here + // would see nothing: retiring on resolve alone marked dead-lettered events 'dispatched' + // and nulled their payload, spending the whole budget on one sweep. + const outcome = await this.delivery.redeliver( + webhook, + row.sessionId, + row.event, + row.idempotencyKey, + row.payload, + ); + if (outcome === 'failed') { + // Left 'pending' on purpose: the next sweep retries it until the budget is spent. + this.logger.warn(`Replay of ${row.event} to webhook ${row.webhookId} did not deliver`); + stats.failed++; + continue; + } await this.outbox.close(row.webhookId, row.idempotencyKey, 'dispatched'); stats.replayed++; } catch (error) { - // Left 'pending' on purpose: the next sweep retries it until the budget is spent. + // An exception is an unexpected fault rather than a delivery failure; the row stays + // pending either way. this.logger.warn(`Replay of ${row.event} to webhook ${row.webhookId} failed: ${String(error)}`); stats.failed++; }