diff --git a/.compliance/ropa.yaml b/.compliance/ropa.yaml index 664dfa62..af0ee016 100644 --- a/.compliance/ropa.yaml +++ b/.compliance/ropa.yaml @@ -695,7 +695,7 @@ processing_activities: - representation_attendee # tredjepart: namn + företag i representationssvar data_categories: - user.contact.phone_number # HMAC-hash med peppar + AES-256-GCM + mask; aldrig i klartext i DB - - user.communication_content # chattext och rå webhook-payload, 90 dagars retention + - user.communication_content # chattext och webhook-payload utan avsändarnummer, 90 dagars retention - user.document # kvittofiler, 7 år WORM enligt BFL 7 kap - user.name # deltagarnamn i representationssvar (tredjepart) - user.activity_timestamp @@ -729,6 +729,12 @@ processing_activities: # phone_masked bevaras för unikhetshistorik respektive visning) # kvittofilen 7 år, WORM (BFL 7 kap), # hanteras av dokumentarkivet, ej av denna cron + # kontoradering (art 17) omedelbart, inte via cron: + # anonymize_user_account revokerar och kryptoshreddar kopplingen, + # nollställer body_text/raw_payload på kopplingens meddelanden och + # raderar engångskoder (migration 20260803090000). auth.users + # gravsätts i stället för att raderas, så tabellens ON DELETE + # CASCADE utlöses aldrig av sig själv duration: differentiated_per_data_category basis: gdpr_storage_limitation_and_bfl_7_kap stored_in: @@ -752,3 +758,5 @@ processing_activities: - llm_reads_answers_as_untrusted_data_no_tools - receipt_files_worm_protected_bfl - opt_out_stop_keyword_honored + - muted_senders_no_content_persistence + - account_deletion_revokes_and_shreds_whatsapp_link diff --git a/DECISIONS.md b/DECISIONS.md index 135a7028..8e271bd4 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -785,3 +785,11 @@ One line per decision: `[YYYY-MM-DD] : `. Appended by agents and [2026-08-02] whatsapp-inbox PR4 schema: ONE new migration (20260802210000) adds whatsapp_messages.acked_at. Burst-ack membership (which ingested rows the single combined M4/M5 covered) must be derivable relationally across webhook invocations; deriving it from "last outbound ack timestamp" breaks the moment one outbound insert fails, and conversation-context accumulation races. [2026-08-02] whatsapp-inbox M6 body omits merchant/amount (spec showed them): for multi-company senders the item is created only after the company answer, so nothing is extracted when the question is asked. Body degrades to "Vilket företag gäller kvittot/kvittona?". [2026-08-02] whatsapp-inbox PR5b retention: revoked links crypto-shred by setting phone_enc='' (column is NOT NULL; empty string is the cleared marker, guarded by neq for one-shot idempotency) while phone_hash and phone_masked survive on purpose (re-link uniqueness history + audit display). Scoped to whatsapp_* tables only: inbox_rate_counters (the per-company sibling) has NO cleanup anywhere and stays that way here; adding one is a separate change, noted in the PR body. +[2026-08-03] whatsapp-inbox GDPR erasure: anonymize_user_account REVOKES and crypto-shreds the phone link instead of DELETEing it. The declared auth.users ON DELETE CASCADE never fires (Accounted tombstones the auth row for ~100 years), and revocation-not-deletion is the channel's existing discipline: lookupActiveLink filters revoked_at, so a revoked link makes every further message take the content-free unknown-sender path, while phone_hash survives as non-reversible (HMAC + pepper) re-link history, exactly like the deliberately retained auth.users.email. +[2026-08-03] whatsapp-inbox company question: kept the guarded state commit BEFORE the M6 send (a fast tap must find the state) and added a rollback on send failure, rather than sending first. Send-first would let two concurrent burst workers both send M6, losing the single-ask guarantee the .neq('state','awaiting_company') guard exists for. +[2026-08-03] whatsapp-inbox company_choice_expired: at the 48h TTL the parked receipts are NO LONGER discarded. The options stay in the conversation context and the rows stay staged, so a late digit/tap still files them (Meta serves the media ~30 days); only rows older than that get the terminal marker. Chosen over "notify at expiry" because the 24h service window is closed by then and v1 sends no templates, so there is no way to tell the user in chat; and over "auto-file into the default company" because there is no default (that is why the question was asked). +[2026-08-03] whatsapp-inbox concurrency: conversation writes go through updateConversation(), an optimistic compare-and-set on the existing updated_at column (the update trigger makes it a revision counter), instead of adding a version column or moving pending_question/question_queue/budget into real columns. No migration, works for every writer including the sweep, and the mutation re-runs against fresh state instead of failing. +[2026-08-03] whatsapp-inbox inline dispositions (stop/start/byt/company choice): the durable side effect now runs BEFORE the terminal wamid row is written, with a SELECT pre-check for dedupe. Insert-first-then-effect was at-most-once (a crash in between lost the action forever: the redelivery 23505s and the sweep never claims 'done' rows). Accepting a possibly repeated idempotent effect plus a duplicate confirmation is the right trade against silently dropping an opt-out. Not routed through the received/claim lifecycle instead, because the sweep re-dispatches text rows to the answer worker, which would turn a crashed 'stopp' into an M16 fallback. +[2026-08-03] whatsapp-inbox M11 ('stopp'): copy changed to say PAUSED rather than disconnected; the keyword only sets muted_at. Muting also stops persisting content (body_text/raw_payload null on muted senders, matching the unknown-sender discipline) so the promise is true in the data too. Actually revoking on 'stopp' was rejected: 'start' must be able to re-open the same binding, and the settings panel already owns real disconnection ("Koppla från"). +[2026-08-03] whatsapp-inbox raw_payload: the sender's plaintext E.164 number is stripped before persisting (redactRawPayload) and replies resolve the recipient by decrypting the link's phone_enc. Storing the number verbatim on every message row defeated the point of the AES-256-GCM column and contradicted the RoPA claim that it is never in the clear. Legacy rows still holding `from` keep working via a fallback read. +[2026-08-05] Representation clarifying question has NO amount floor (removed the 150 kr gate after the Swedish compliance review on PR #1340): documenting deltagare + syfte is what makes a representation expense deductible at all (BFL 5 kap 6-7 §) and that duty is not conditioned on any sum; the 300 kr/person figure is the VAT-deduction base cap, an unrelated rule. Noise is bounded by the triggers instead (receipt-shaped + restaurant/cafe/hotel merchant, <=1 question per receipt, <=2 per burst, <=6 per sender per day, one "nej" dismisses). diff --git a/extensions/general/whatsapp-inbox/__tests__/answer-flow.test.ts b/extensions/general/whatsapp-inbox/__tests__/answer-flow.test.ts index 0497c1b1..f73686b3 100644 --- a/extensions/general/whatsapp-inbox/__tests__/answer-flow.test.ts +++ b/extensions/general/whatsapp-inbox/__tests__/answer-flow.test.ts @@ -272,7 +272,7 @@ describe('answer flow (text rows through processInboundMessage)', () => { expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m10ContextConfirm) // No conversation update ever wrote a company id. for (const args of findCalls('whatsapp_conversations', 'update')) { - expect(Object.keys(args[0] as Record)).not.toContain('company_id') + expect((args[0] as Record).company_id).toBeUndefined() } }) diff --git a/extensions/general/whatsapp-inbox/__tests__/company-question.test.ts b/extensions/general/whatsapp-inbox/__tests__/company-question.test.ts index 7a612d0f..e1fd77d6 100644 --- a/extensions/general/whatsapp-inbox/__tests__/company-question.test.ts +++ b/extensions/general/whatsapp-inbox/__tests__/company-question.test.ts @@ -217,6 +217,7 @@ describe('applyCompanyChoice', () => { }) expect(applied).toEqual({ + ok: true, companyId: 'company-2', companyName: 'Bolag B AB', stagedMessageIds: ['stg-1', 'stg-2'], @@ -266,8 +267,10 @@ describe('applyCompanyChoice', () => { replyBase, }) - expect(applied?.companyId).toBe('company-1') - expect(applied?.stagedMessageIds).toEqual([]) + expect(applied.ok).toBe(true) + if (!applied.ok) throw new Error('expected the choice to apply') + expect(applied.companyId).toBe('company-1') + expect(applied.stagedMessageIds).toEqual([]) }) it('rejects a company the sender is not a member of (forged/stale payload)', async () => { @@ -283,7 +286,7 @@ describe('applyCompanyChoice', () => { replyBase, }) - expect(applied).toBeNull() + expect(applied).toEqual({ ok: false, reason: 'not_member' }) expect(sendTextMock).not.toHaveBeenCalled() expect(findCalls('whatsapp_conversations', 'update')).toHaveLength(0) }) @@ -298,9 +301,45 @@ describe('applyCompanyChoice', () => { to: '46701234567', replyBase, }) - expect(applied).toBeNull() + // Ordinary user input (a typo), so the CALLER re-prompts the options + // rather than the silence reserved for forged payloads. + expect(applied).toEqual({ ok: false, reason: 'invalid_option' }) expect(sendTextMock).not.toHaveBeenCalled() }) + + it('treats a transient membership-query error as unresolved, not as "not a member"', async () => { + const { supabase, enqueue, findCalls } = createQueuedMockSupabase() + enqueue({ error: { message: 'connection reset' } }) + + const applied = await applyCompanyChoice(supabase as unknown as SupabaseClient, { + conversation: awaitingConversation(), + link: makeLink(), + choice: { companyId: 'company-1' }, + via: 'button', + to: '46701234567', + replyBase, + }) + + expect(applied).toEqual({ ok: false, reason: 'lookup_failed' }) + expect(findCalls('whatsapp_conversations', 'update')).toHaveLength(0) + }) + + it('a second tap after the choice was applied confirms nothing (options are gone)', async () => { + const { supabase, findCalls } = createQueuedMockSupabase() + + const applied = await applyCompanyChoice(supabase as unknown as SupabaseClient, { + conversation: makeConversation({ state: 'idle', context: {} }), + link: makeLink(), + choice: { companyId: 'company-1' }, + via: 'button', + to: '46701234567', + replyBase, + }) + + expect(applied).toEqual({ ok: false, reason: 'already_applied' }) + expect(sendTextMock).not.toHaveBeenCalled() + expect(findCalls('whatsapp_conversations', 'update')).toHaveLength(0) + }) }) describe('truncateTitle', () => { diff --git a/extensions/general/whatsapp-inbox/__tests__/hardening.test.ts b/extensions/general/whatsapp-inbox/__tests__/hardening.test.ts new file mode 100644 index 00000000..32eea118 --- /dev/null +++ b/extensions/general/whatsapp-inbox/__tests__/hardening.test.ts @@ -0,0 +1,552 @@ +/** + * Regression tests for the adversarial-review hardening pass. + * + * Every case here fails against the pre-fix code: a failed send stranding the + * company question, an ignored ack send result, SEK-labelled foreign amounts, + * unguarded terminal status writes, quoted corrections landing on the wrong + * receipt, blind whole-context conversation writes, the plaintext phone in + * raw_payload, text answers to the re-send question, and the 48h expiry + * discarding parked receipts. + */ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { createQueuedMockSupabase } from '@/tests/helpers' +import type { SupabaseClient } from '@supabase/supabase-js' + +vi.mock('@/extensions/general/whatsapp-inbox/lib/graph-api', async () => { + const actual = await vi.importActual< + typeof import('@/extensions/general/whatsapp-inbox/lib/graph-api') + >('@/extensions/general/whatsapp-inbox/lib/graph-api') + return { + ...actual, + sendText: vi.fn().mockResolvedValue({ ok: true, wamid: 'wamid.OUT' }), + sendReplyButtons: vi.fn().mockResolvedValue({ ok: true, wamid: 'wamid.OUT' }), + sendList: vi.fn().mockResolvedValue({ ok: true, wamid: 'wamid.OUT' }), + markReadWithTyping: vi.fn().mockResolvedValue(undefined), + downloadMedia: vi.fn(), + } +}) + +vi.mock('@/lib/processing-history/append', () => ({ + appendProcessingHistory: vi.fn().mockResolvedValue('event-1'), +})) + +vi.mock('@/lib/rate-limits/agent', () => ({ + checkAgentRateLimit: vi.fn().mockResolvedValue({ ok: true }), +})) + +vi.mock('@/extensions/general/whatsapp-inbox/lib/interpret-answer', () => ({ + interpretChatAnswer: vi.fn().mockResolvedValue({ ok: false }), +})) + +import { sendText, sendReplyButtons } from '@/extensions/general/whatsapp-inbox/lib/graph-api' +import { askCompanyQuestion } from '@/extensions/general/whatsapp-inbox/lib/company-question' +import { + finalizeBurst, + processInboundMessage, +} from '@/extensions/general/whatsapp-inbox/lib/process-inbound' +import { + redactRawPayload, + resolveAnswerTarget, + resolveRecipient, + updateConversation, +} from '@/extensions/general/whatsapp-inbox/lib/conversation' +import { encryptPhone } from '@/extensions/general/whatsapp-inbox/lib/phone-crypto' +import { runSweep } from '@/extensions/general/whatsapp-inbox/lib/sweep' +import { botCopy, TEMPLATE } from '@/extensions/general/whatsapp-inbox/lib/messages' + +const sendTextMock = vi.mocked(sendText) +const sendButtonsMock = vi.mocked(sendReplyButtons) + +function makeConversation(overrides: Record = {}) { + return { + id: 'conv-1', + phone_link_id: 'link-1', + state: 'idle', + context: {}, + company_id: 'company-1', + service_window_expires_at: new Date(Date.now() + 60 * 60 * 1000).toISOString(), + debounce_until: new Date(Date.now() - 1000).toISOString(), + pending_ack: true, + last_inbound_at: null, + last_outbound_at: null, + created_at: '2026-08-01T09:00:00Z', + updated_at: '2026-08-01T09:00:00Z', + ...overrides, + } +} + +function makeLink(overrides: Record = {}) { + return { + id: 'link-1', + user_id: 'user-1', + phone_hash: 'hash-1', + phone_enc: 'enc', + phone_masked: '+46 70 *** ** 67', + default_company_id: null, + last_company_id: null, + verified_at: '2026-08-01T09:00:00Z', + revoked_at: null, + muted_at: null, + ...overrides, + } +} + +function makeRow(overrides: Record = {}) { + return { + id: 'msg-1', + direction: 'inbound', + wamid: 'wamid.IN1', + sender_phone_hash: 'hash-1', + phone_link_id: 'link-1', + conversation_id: 'conv-1', + message_type: 'image', + body_text: null, + media_id: 'media-1', + media_mime: 'image/jpeg', + media_sha256: 'abc', + media_filename: 'kvitto.jpg', + raw_payload: { from: '46701234567' }, + processing_status: 'received', + attempts: 0, + error_message: null, + inbox_item_id: null, + delivery_status: null, + correlation_id: 'corr-1', + acked_at: null, + created_at: '2026-08-01T10:00:00Z', + updated_at: '2026-08-01T10:00:00Z', + ...overrides, + } +} + +beforeEach(() => { + vi.clearAllMocks() + process.env.WHATSAPP_PHONE_HASH_KEY = 'test-pepper' + process.env.WHATSAPP_PHONE_ENCRYPTION_KEY = 'a'.repeat(64) + sendTextMock.mockResolvedValue({ ok: true, wamid: 'wamid.OUT' }) + sendButtonsMock.mockResolvedValue({ ok: true, wamid: 'wamid.OUT' }) +}) + +describe('company question is not one-shot when the send fails', () => { + function enqueueAsk(mock: ReturnType) { + mock.enqueue({ data: [{ company_id: 'company-1' }, { company_id: 'company-2' }] }) + mock.enqueue({ + data: [ + { id: 'company-1', name: 'Bolag A AB' }, + { id: 'company-2', name: 'Bolag B AB' }, + ], + }) + mock.enqueue({ data: [{ ...makeConversation(), state: 'awaiting_company' }] }) // guarded commit + } + + it('rolls the question back so the next receipt re-asks', async () => { + sendButtonsMock.mockResolvedValue({ ok: false, wamid: null }) + const mock = createQueuedMockSupabase() + enqueueAsk(mock) + mock.enqueue({ + data: [ + { + ...makeConversation(), + state: 'awaiting_company', + context: { company_options: [{ id: 'company-1', name: 'Bolag A AB' }] }, + }, + ], + }) + + const asked = await askCompanyQuestion(mock.supabase as unknown as SupabaseClient, { + conversation: makeConversation() as never, + link: makeLink() as never, + to: '46701234567', + replyBase: {}, + stagedCount: 3, + }) + + expect(asked).toBe(false) + const updates = mock + .findCalls('whatsapp_conversations', 'update') + .map((args) => args[0] as { state?: string; context?: Record }) + // First write arms the question, the second undoes it. + expect(updates[0].state).toBe('awaiting_company') + expect(updates[1].state).toBe('idle') + expect(updates[1].context?.company_options).toBeUndefined() + expect(updates[1].context?.pending_question).toBeUndefined() + }) + + it('keeps the question when the send succeeded', async () => { + const mock = createQueuedMockSupabase() + enqueueAsk(mock) + + const asked = await askCompanyQuestion(mock.supabase as unknown as SupabaseClient, { + conversation: makeConversation() as never, + link: makeLink() as never, + to: '46701234567', + replyBase: {}, + stagedCount: 1, + }) + + expect(asked).toBe(true) + expect(mock.findCalls('whatsapp_conversations', 'update')).toHaveLength(1) + }) +}) + +describe('combined ack', () => { + function enqueueBurst( + mock: ReturnType, + extracted: Record, + ) { + mock.enqueue({ data: [{ id: 'conv-1' }] }) // claimAck + mock.enqueue({ data: makeConversation() }) + mock.enqueue({ + data: [ + { + ...makeRow(), + processing_status: 'done', + inbox_item_id: 'item-1', + }, + ], + }) + mock.enqueue({ + data: [ + { + id: 'item-1', + company_id: 'company-1', + extracted_data: extracted, + channel_context: { channel: 'whatsapp' }, + document_id: null, + }, + ], + }) + mock.enqueue({ data: [makeConversation()] }) // guarded state write + } + + it('states a foreign total in its own currency, never as kronor', async () => { + const mock = createQueuedMockSupabase() + enqueueBurst(mock, { + documentKind: 'receipt', + legibility: 'good', + supplier: { name: 'Hotel Adlon' }, + invoice: { invoiceDate: '2026-07-30', currency: 'EUR' }, + totals: { total: 250 }, + lineItems: [], + }) + + await finalizeBurst(mock.supabase as unknown as SupabaseClient, 'conv-1') + + expect(sendTextMock).toHaveBeenCalledTimes(1) + const body = sendTextMock.mock.calls[0][1].body + expect(body).toContain('250 EUR') + expect(body).not.toContain('250 kr') + }) + + it('still says kr for SEK and for an unstated currency', async () => { + const mock = createQueuedMockSupabase() + enqueueBurst(mock, { + documentKind: 'receipt', + legibility: 'good', + supplier: { name: 'ICA Maxi' }, + invoice: { invoiceDate: '2026-07-30' }, + totals: { total: 234 }, + lineItems: [], + }) + + await finalizeBurst(mock.supabase as unknown as SupabaseClient, 'conv-1') + expect(sendTextMock.mock.calls[0][1].body).toContain('234 kr') + }) + + it('leaves the rows unacked when the send failed so the sweep retries', async () => { + sendTextMock.mockResolvedValue({ ok: false, wamid: null }) + const mock = createQueuedMockSupabase() + enqueueBurst(mock, { + documentKind: 'receipt', + legibility: 'good', + supplier: { name: 'ICA Maxi' }, + invoice: { invoiceDate: '2026-07-30' }, + totals: { total: 234 }, + lineItems: [], + }) + + await finalizeBurst(mock.supabase as unknown as SupabaseClient, 'conv-1') + + const messageUpdates = mock + .findCalls('whatsapp_messages', 'update') + .map((args) => args[0] as Record) + expect(messageUpdates.some((patch) => 'acked_at' in patch)).toBe(false) + }) +}) + +describe('terminal status writes hold the claim', () => { + it('guards markStatus on processing_status=processing', async () => { + const mock = createQueuedMockSupabase() + mock.enqueue({ data: makeRow() }) + mock.enqueue({ data: { id: 'msg-1' } }) // claim + mock.enqueue({ data: makeLink({ revoked_at: '2026-08-01T10:00:00Z' }) }) + mock.enqueue({ data: null }) // markStatus skipped + + await processInboundMessage(mock.supabase as unknown as SupabaseClient, 'msg-1') + + const guards = mock + .findCalls('whatsapp_messages', 'eq') + .filter((args) => args[0] === 'processing_status') + // The claim guards on 'received'; the terminal write must guard on + // 'processing' so a losing worker cannot clobber the winner's row. + expect(guards.map((args) => args[1])).toContain('processing') + }) +}) + +describe('answer targeting', () => { + const askedAt = new Date(Date.now() - 60 * 60 * 1000).toISOString() + + function conversationWith(recent: Record[], overrides = {}) { + return makeConversation({ + state: 'awaiting_context', + context: { + pending_question: { type: 'context', inbox_item_id: 'item-B', asked_at: askedAt }, + recent_questions: recent, + }, + ...overrides, + }) + } + + it('binds a quoted follow-up to the quoted receipt, not to another open question', async () => { + const mock = createQueuedMockSupabase() + mock.enqueue({ data: { inbox_item_id: 'item-A' } }) // quoted wamid resolves + + const target = await resolveAnswerTarget( + mock.supabase as unknown as SupabaseClient, + conversationWith([ + { type: 'representation', inbox_item_id: 'item-A', asked_at: askedAt, status: 'answered' }, + { type: 'context', inbox_item_id: 'item-B', asked_at: askedAt, status: 'open' }, + ]) as never, + 'wamid.QUOTED', + ) + + expect(target?.inboxItemId).toBe('item-A') + expect(target?.followUp).toBe(true) + }) + + it('quoting the open question is still an ordinary answer, not a late one', async () => { + const mock = createQueuedMockSupabase() + mock.enqueue({ data: { inbox_item_id: 'item-B' } }) + + const target = await resolveAnswerTarget( + mock.supabase as unknown as SupabaseClient, + conversationWith([ + { type: 'context', inbox_item_id: 'item-B', asked_at: askedAt, status: 'open' }, + ]) as never, + 'wamid.QUOTED', + ) + + expect(target).toEqual({ type: 'context', inboxItemId: 'item-B', late: false }) + }) +}) + +describe('conversation writes are compare-and-set', () => { + it('re-applies the mutation against fresh state after losing a race', async () => { + const mock = createQueuedMockSupabase() + mock.enqueue({ data: [] }) // guard did not match: someone else wrote first + mock.enqueue({ data: makeConversation({ context: { budget: { day_key: 'x', count: 2 } } }) }) + mock.enqueue({ data: [makeConversation()] }) + + const seen: number[] = [] + const result = await updateConversation( + mock.supabase as unknown as SupabaseClient, + makeConversation() as never, + (_current, context) => { + seen.push(context.budget?.count ?? 0) + return { context: { ...context, question_queue: [] } } + }, + ) + + expect(result).not.toBeNull() + // Second pass saw the concurrent writer's state, not the stale snapshot. + expect(seen).toEqual([0, 2]) + }) + + it('gives up rather than looping forever', async () => { + const mock = createQueuedMockSupabase() + for (let i = 0; i < 12; i++) { + mock.enqueue({ data: [] }) + mock.enqueue({ data: makeConversation() }) + } + + const result = await updateConversation( + mock.supabase as unknown as SupabaseClient, + makeConversation() as never, + (_current, context) => ({ context }), + ) + expect(result).toBeNull() + }) +}) + +describe('phone PII', () => { + it('strips the sender number from the persisted payload but keeps the quote id', () => { + const redacted = redactRawPayload({ + from: '46701234567', + id: 'wamid.IN1', + type: 'text', + context: { id: 'wamid.QUOTED' }, + }) + expect(redacted).not.toHaveProperty('from') + expect(redacted).toMatchObject({ id: 'wamid.IN1', context: { id: 'wamid.QUOTED' } }) + }) + + it('resolves the reply address from the encrypted link when the payload is redacted', () => { + const row = makeRow({ raw_payload: { id: 'wamid.IN1', type: 'text' } }) + const link = { phone_enc: encryptPhone('46701234567') } + expect(resolveRecipient(row as never, link)).toBe('46701234567') + }) + + it('returns null instead of throwing on a shredded link', () => { + const row = makeRow({ raw_payload: {} }) + expect(resolveRecipient(row as never, { phone_enc: '' })).toBeNull() + }) +}) + +describe('text reply while a re-send question is open', () => { + it('keeps it as a note on that receipt and leaves the question open', async () => { + const mock = createQueuedMockSupabase() + mock.enqueue({ + data: makeRow({ + id: 'msg-t1', + message_type: 'text', + body_text: 'det var lunch på Ica, 240 kr', + media_id: null, + }), + }) + mock.enqueue({ data: { id: 'msg-t1' } }) // claim + mock.enqueue({ data: makeLink() }) + mock.enqueue({ + data: makeConversation({ + state: 'awaiting_resend', + context: { + pending_question: { + type: 'resend', + inbox_item_id: 'item-9', + asked_at: new Date().toISOString(), + }, + }, + }), + }) + mock.enqueue({ data: { channel_context: { channel: 'whatsapp' } } }) // load item context + mock.enqueue({ data: null }) // item update + mock.enqueue({ data: null }) // markStatus done + + await processInboundMessage(mock.supabase as unknown as SupabaseClient, 'msg-t1') + + expect(sendTextMock).toHaveBeenCalledTimes(1) + expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m9NoteSaved) + const itemPatch = mock.findCall('invoice_inbox_items', 'update') as [ + { channel_context: Record }, + ] + expect(itemPatch[0].channel_context.user_note).toContain('lunch på Ica') + // The question is untouched: only a sharper file can answer it. + expect(mock.findCalls('whatsapp_conversations', 'update')).toHaveLength(0) + }) +}) + +describe('answer worker re-claims', () => { + it('does not follow a delivered confirm with the M16 fallback', async () => { + // A worker that died after applying the answer leaves the row + // 'processing'; the sweep re-runs it and the question now resolves to + // nothing, which used to send "I did not understand" right after the + // confirmation the user already received. + const mock = createQueuedMockSupabase() + mock.enqueue({ + data: makeRow({ + id: 'msg-t1', + message_type: 'text', + body_text: 'Lunch med Anna Berg', + media_id: null, + attempts: 1, + }), + }) + mock.enqueue({ data: { id: 'msg-t1' } }) // claim + mock.enqueue({ data: makeLink() }) + mock.enqueue({ data: makeConversation({ state: 'idle', context: {} }) }) + mock.enqueue({ data: null }) // markStatus done + + await processInboundMessage(mock.supabase as unknown as SupabaseClient, 'msg-t1') + + expect(sendTextMock).not.toHaveBeenCalled() + }) + + it('still explains itself on a first attempt with nothing to answer', async () => { + const mock = createQueuedMockSupabase() + mock.enqueue({ + data: makeRow({ + id: 'msg-t1', + message_type: 'text', + body_text: 'kan du bokföra allt åt mig?', + media_id: null, + attempts: 0, + }), + }) + mock.enqueue({ data: { id: 'msg-t1' } }) + mock.enqueue({ data: makeLink() }) + mock.enqueue({ data: makeConversation({ state: 'idle', context: {} }) }) + mock.enqueue({ data: null }) + + await processInboundMessage(mock.supabase as unknown as SupabaseClient, 'msg-t1') + + expect(sendTextMock).toHaveBeenCalledTimes(1) + expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m16Fallback) + }) +}) + +describe('48h company-question expiry', () => { + it('keeps the parked receipts recoverable by a late answer', async () => { + const mock = createQueuedMockSupabase() + mock.enqueue({ data: [] }) // 1a stuck received + mock.enqueue({ data: [] }) // 1b stuck processing + mock.enqueue({ data: [] }) // 2a stale pending_ack + mock.enqueue({ data: [] }) // 2b unacked rows + mock.enqueue({ + data: [ + makeConversation({ + state: 'awaiting_company', + context: { + company_options: [{ id: 'company-1', name: 'Bolag A AB' }], + pending_question: { + type: 'company', + inbox_item_id: null, + asked_at: new Date(Date.now() - 72 * 60 * 60 * 1000).toISOString(), + }, + }, + }), + ], + }) // 3 question TTL + mock.enqueue({ data: null }) // stamp only rows past the media window + mock.enqueue({ count: 2 }) // rows still staged + mock.enqueue({ data: null }) // conversation back to idle + mock.enqueue({ data: [] }) // 4 pins + + await runSweep(mock.supabase as unknown as SupabaseClient) + + const expiryStamp = mock + .findCalls('whatsapp_messages', 'update') + .map((args) => args[0] as Record) + .find((patch) => patch.error_message === 'company_choice_expired') + // The stamp is now bounded by how long Meta still serves the media. + expect(expiryStamp).toBeTruthy() + const stampFilters = mock.findCalls('whatsapp_messages', 'lt') + expect(stampFilters.some((args) => args[0] === 'created_at')).toBe(true) + + const idleUpdate = mock + .findCalls('whatsapp_conversations', 'update') + .map((args) => args[0] as { state?: string; context?: Record }) + .find((patch) => patch.state === 'idle') + // company_options survive, so a late digit still maps to a company. + expect(idleUpdate?.context?.company_options).toBeTruthy() + expect(idleUpdate?.context?.pending_question).toBeUndefined() + }) +}) + +describe('stop copy', () => { + it('does not claim the number is disconnected: stopp only pauses', () => { + for (const locale of ['sv', 'en'] as const) { + const body = botCopy(locale).m11Stop() + expect(body).not.toMatch(/kopplas från|is disconnected/) + expect(body).toMatch(/pausar|pausing/) + } + }) +}) diff --git a/extensions/general/whatsapp-inbox/__tests__/linking.test.ts b/extensions/general/whatsapp-inbox/__tests__/linking.test.ts index 267776ba..1d537fa7 100644 --- a/extensions/general/whatsapp-inbox/__tests__/linking.test.ts +++ b/extensions/general/whatsapp-inbox/__tests__/linking.test.ts @@ -3,6 +3,8 @@ import crypto from 'crypto' import { createQueuedMockSupabase } from '@/tests/helpers' import { CODE_ALPHABET, + LinkCodeRateLimitError, + MAX_CODES_PER_TTL_WINDOW, normalizeLinkCode, looksLikeLinkCode, hashLinkCode, @@ -64,15 +66,31 @@ describe('linking', () => { }) describe('hashLinkCode', () => { - it('is sha256 hex of the code', () => { - const expected = crypto.createHash('sha256').update('AC-7KP4QF').digest('hex') - expect(hashLinkCode('AC-7KP4QF')).toBe(expected) + it('is an HMAC under the server pepper, never a bare sha256', () => { + // 30^6 codes behind a fixed 'AC-' prefix is enumerable offline in about + // a second, so an unpeppered digest would protect nothing at rest. + const bare = crypto.createHash('sha256').update('AC-7KP4QF').digest('hex') + const peppered = crypto + .createHmac('sha256', Buffer.from('test-pepper', 'utf8')) + .update('AC-7KP4QF') + .digest('hex') + expect(hashLinkCode('AC-7KP4QF')).toBe(peppered) + expect(hashLinkCode('AC-7KP4QF')).not.toBe(bare) + }) + + it('changes with the pepper', () => { + const withA = hashLinkCode('AC-7KP4QF') + process.env.WHATSAPP_PHONE_HASH_KEY = 'another-pepper' + expect(hashLinkCode('AC-7KP4QF')).not.toBe(withA) + process.env.WHATSAPP_PHONE_HASH_KEY = 'test-pepper' }) }) describe('mintLinkCode', () => { it('mints an AC- prefixed code from the ambiguity-free alphabet with a 10 min TTL', async () => { const { supabase, enqueue, findCall } = createQueuedMockSupabase() + enqueue({ count: 0 }) // mint-rate check + enqueue({ data: null, error: null }) // burn earlier unused codes enqueue({ data: null, error: null }) const before = Date.now() @@ -93,11 +111,34 @@ describe('linking', () => { it('throws when the insert fails', async () => { const { supabase, enqueue } = createQueuedMockSupabase() + enqueue({ count: 0 }) + enqueue({ data: null, error: null }) enqueue({ data: null, error: { message: 'boom' } }) await expect( mintLinkCode(supabase as unknown as SupabaseClient, 'user-1'), ).rejects.toThrow(/boom/) }) + + it('burns the caller earlier unused codes so only the newest one works', async () => { + const { supabase, enqueue, findCall } = createQueuedMockSupabase() + enqueue({ count: 1 }) + enqueue({ data: null, error: null }) + enqueue({ data: null, error: null }) + + await mintLinkCode(supabase as unknown as SupabaseClient, 'user-1') + + const burn = findCall('whatsapp_link_codes', 'update') as [Record] + expect(burn[0].used_at).toBeTruthy() + }) + + it('refuses to mint once the caller has burned through the window quota', async () => { + const { supabase, enqueue } = createQueuedMockSupabase() + enqueue({ count: MAX_CODES_PER_TTL_WINDOW }) + + await expect( + mintLinkCode(supabase as unknown as SupabaseClient, 'user-1'), + ).rejects.toBeInstanceOf(LinkCodeRateLimitError) + }) }) describe('consumeLinkCode', () => { diff --git a/extensions/general/whatsapp-inbox/__tests__/process-inbound.test.ts b/extensions/general/whatsapp-inbox/__tests__/process-inbound.test.ts index 0e1ef785..c2b109e6 100644 --- a/extensions/general/whatsapp-inbox/__tests__/process-inbound.test.ts +++ b/extensions/general/whatsapp-inbox/__tests__/process-inbound.test.ts @@ -411,6 +411,7 @@ describe('processInboundMessage (media intake)', () => { enqueue({ data: makeLink() }) enqueue({ data: makeConversation() }) enqueue({ data: [{ company_id: 'company-1' }] }) + enqueue({ data: null }) // no inbox item for this message yet enqueue({ data: { id: 'existing-doc' } }) // dup found enqueue({ data: null }) // markStatus skipped @@ -492,7 +493,9 @@ describe('processInboundMessage (media intake)', () => { enqueue({ data: makeLink() }) enqueue({ data: makeConversation() }) enqueue({ data: [{ company_id: 'company-1' }] }) + enqueue({ data: null }) // no inbox item yet enqueue({ data: null }) // markStatus error + enqueue({ data: null }) // no M18 sent yet await processInboundMessage(supabase as unknown as SupabaseClient, 'msg-1') @@ -503,7 +506,10 @@ describe('processInboundMessage (media intake)', () => { expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m18Error) }) - it('suppresses M18 on re-claims (attempts > 1)', async () => { + it('still sends M18 on a re-claim when the first attempt died before the catch', async () => { + // Gating M18 on `attempt <= 1` made it unreachable for exactly the crash + // the sweep exists for: the dead first attempt already burned its + // attempt, so the sender heard nothing about that receipt at all. downloadMediaMock.mockRejectedValue(new GraphApiError('still failing')) const { supabase, enqueue, findCalls } = createQueuedMockSupabase() enqueue({ data: makeRow({ attempts: 1 }) }) @@ -511,11 +517,51 @@ describe('processInboundMessage (media intake)', () => { enqueue({ data: makeLink() }) enqueue({ data: makeConversation() }) enqueue({ data: [{ company_id: 'company-1' }] }) + enqueue({ data: null }) // no inbox item yet enqueue({ data: null }) // markStatus error + enqueue({ data: null }) // no M18 sent for this message yet + + await processInboundMessage(supabase as unknown as SupabaseClient, 'msg-1') + + expect(lastUpdate(findCalls).processing_status).toBe('error') + expect(sendTextMock).toHaveBeenCalledTimes(1) + expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m18Error) + }) + + it('never repeats M18 when one already went out for the same message', async () => { + downloadMediaMock.mockRejectedValue(new GraphApiError('still failing')) + const { supabase, enqueue, findCalls } = createQueuedMockSupabase() + enqueue({ data: makeRow({ attempts: 2 }) }) + enqueue({ data: { id: 'msg-1' } }) + enqueue({ data: makeLink() }) + enqueue({ data: makeConversation() }) + enqueue({ data: [{ company_id: 'company-1' }] }) + enqueue({ data: null }) // no inbox item yet + enqueue({ data: null }) // markStatus error + enqueue({ data: { id: 'out-1' } }) // an M18 for this correlation exists await processInboundMessage(supabase as unknown as SupabaseClient, 'msg-1') expect(lastUpdate(findCalls).processing_status).toBe('error') expect(sendTextMock).not.toHaveBeenCalled() }) + + it('adopts an item a concurrent worker already created instead of ingesting twice', async () => { + const { supabase, enqueue, findCalls } = createQueuedMockSupabase() + enqueue({ data: makeRow() }) + enqueue({ data: { id: 'msg-1' } }) + enqueue({ data: makeLink() }) + enqueue({ data: makeConversation() }) + enqueue({ data: [{ company_id: 'company-1' }] }) + enqueue({ data: { id: 'item-winner' } }) // the winner's item + enqueue({ data: null }) // markStatus done + + await processInboundMessage(supabase as unknown as SupabaseClient, 'msg-1') + + expect(downloadMediaMock).not.toHaveBeenCalled() + expect(uploadAndExtractMock).not.toHaveBeenCalled() + const finalUpdate = lastUpdate(findCalls) + expect(finalUpdate.processing_status).toBe('done') + expect(finalUpdate.inbox_item_id).toBe('item-winner') + }) }) diff --git a/extensions/general/whatsapp-inbox/__tests__/questions.test.ts b/extensions/general/whatsapp-inbox/__tests__/questions.test.ts index b6955a87..d427c592 100644 --- a/extensions/general/whatsapp-inbox/__tests__/questions.test.ts +++ b/extensions/general/whatsapp-inbox/__tests__/questions.test.ts @@ -2,7 +2,6 @@ import { describe, it, expect } from 'vitest' import { evaluateQuestion, QUESTION_PRIORITY, - REPRESENTATION_MIN_TOTAL, type QuestionInput, } from '@/extensions/general/whatsapp-inbox/lib/questions' import type { InvoiceExtractionResult } from '@/types' @@ -77,7 +76,7 @@ describe('evaluateQuestion', () => { ).toBeNull() }) - it('restaurant/cafe/hotel receipts >= 150 kr trigger representation', () => { + it('restaurant/cafe/hotel receipts trigger representation', () => { for (const category of ['restaurant', 'cafe', 'hotel'] as const) { expect( evaluateQuestion( @@ -87,17 +86,34 @@ describe('evaluateQuestion', () => { } }) - it('representation needs the minimum total: a coffee below the floor is left alone', () => { + it('asks about a small business meal too: the documentation duty has no amount floor', () => { + // BFL 5 kap 6-7 §: deltagare + syfte is what makes representation + // deductible, regardless of the sum. The 300 kr/person figure is the + // VAT-base cap, a different rule. Flagged by the Swedish compliance + // review on PR #1340; the old 150 kr floor silently skipped it. expect( evaluateQuestion( input({ extracted: extraction({ merchantCategory: 'cafe', - totals: { subtotal: null, vatAmount: null, total: REPRESENTATION_MIN_TOTAL - 1 }, + totals: { subtotal: null, vatAmount: null, total: 120 }, }), }), ), - ).toBeNull() + ).toEqual({ type: 'representation' }) + }) + + it('still asks when the total is unreadable, rather than skipping the trail', () => { + expect( + evaluateQuestion( + input({ + extracted: extraction({ + merchantCategory: 'restaurant', + totals: { subtotal: null, vatAmount: null, total: null }, + }), + }), + ), + ).toEqual({ type: 'representation' }) }) it('a supplier_invoice never triggers representation, whatever the merchant', () => { diff --git a/extensions/general/whatsapp-inbox/__tests__/webhook-post.test.ts b/extensions/general/whatsapp-inbox/__tests__/webhook-post.test.ts index b8409e5e..92a49316 100644 --- a/extensions/general/whatsapp-inbox/__tests__/webhook-post.test.ts +++ b/extensions/general/whatsapp-inbox/__tests__/webhook-post.test.ts @@ -363,10 +363,26 @@ describe('POST /webhook', () => { mock.enqueue({ data: null }) // conversation window } + /** Dispositions with a durable side effect run it BEFORE the terminal row + * is written, so the wamid dedupe becomes a pre-check and the insert + * lands last (see handleLinkedSender). */ + function enqueueDurablePreamble( + mock: ReturnType, + link: Record, + conversation: Record = { id: 'conv-1', state: 'idle', context: {} }, + ) { + mock.enqueue({ data: link }) // link lookup + mock.enqueue({ data: conversation }) // conversation + mock.enqueue({ data: null }) // wamid dedupe pre-check: not seen yet + mock.enqueue({ data: null }) // link last_message_at + mock.enqueue({ data: null }) // conversation window + } + it('stopp mutes the link and confirms with M11', async () => { const mock = mockSupabase() - enqueueLinkedTextPreamble(mock, makeLink()) + enqueueDurablePreamble(mock, makeLink()) mock.enqueue({ data: null }) // muted_at update + mock.enqueue({ data: null }) // terminal row insert await route.handler(signedRequest(envelope({ messages: [textMessage('Stopp')] }))) @@ -406,8 +422,9 @@ describe('POST /webhook', () => { it('start unmutes and welcomes back with M12', async () => { const mock = mockSupabase() - enqueueLinkedTextPreamble(mock, makeLink({ muted_at: '2026-08-01T10:00:00Z' })) + enqueueDurablePreamble(mock, makeLink({ muted_at: '2026-08-01T10:00:00Z' })) mock.enqueue({ data: null }) // muted_at cleared + mock.enqueue({ data: null }) // terminal row insert await route.handler(signedRequest(envelope({ messages: [textMessage('start')] }))) @@ -513,13 +530,14 @@ describe('POST /webhook', () => { const mock = mockSupabase() mock.enqueue({ data: makeLink() }) mock.enqueue({ data: awaitingCompany() }) - mock.enqueue({ data: { id: 'msg-row-1' } }) // insert + mock.enqueue({ data: null }) // wamid dedupe pre-check mock.enqueue({ data: null }) // link last_message_at mock.enqueue({ data: null }) // conversation window update mock.enqueue({ data: { company_id: 'company-2' } }) // membership check - mock.enqueue({ data: null }) // conversation pin update + mock.enqueue({ data: null }) // guarded conversation pin update mock.enqueue({ data: null }) // link last_company_id mock.enqueue({ data: [{ id: 'stg-1' }, { id: 'stg-2' }] }) // staged reopen + mock.enqueue({ data: null }) // terminal row insert await route.handler(signedRequest(envelope({ messages: [textMessage('2')] }))) @@ -534,13 +552,14 @@ describe('POST /webhook', () => { const mock = mockSupabase() mock.enqueue({ data: makeLink() }) mock.enqueue({ data: awaitingCompany() }) - mock.enqueue({ data: { id: 'msg-row-1' } }) + mock.enqueue({ data: null }) // wamid dedupe pre-check mock.enqueue({ data: null }) mock.enqueue({ data: null }) mock.enqueue({ data: { company_id: 'company-1' } }) // membership check - mock.enqueue({ data: null }) // conversation pin update + mock.enqueue({ data: null }) // guarded conversation pin update mock.enqueue({ data: null }) // link last_company_id mock.enqueue({ data: [{ id: 'stg-1' }] }) // staged reopen + mock.enqueue({ data: null }) // terminal row insert await route.handler( signedRequest( @@ -609,10 +628,11 @@ describe('POST /webhook', () => { company_id: 'company-2', }, }) - mock.enqueue({ data: { id: 'msg-row-1' } }) + mock.enqueue({ data: null }) // wamid dedupe pre-check mock.enqueue({ data: null }) // link update mock.enqueue({ data: null }) // conversation window update - mock.enqueue({ data: null }) // pin clear update + mock.enqueue({ data: null }) // guarded pin clear update + mock.enqueue({ data: null }) // terminal row insert await route.handler(signedRequest(envelope({ messages: [textMessage('byt')] }))) @@ -695,6 +715,159 @@ describe('POST /webhook', () => { }) }) + describe('hardening: dispositions, late company answers, payload redaction', () => { + const withOptions = (state: string) => ({ + id: 'conv-1', + state, + context: { + company_options: [ + { id: 'company-1', name: 'Bolag A AB' }, + { id: 'company-2', name: 'Bolag B AB' }, + ], + }, + company_id: null, + }) + + it('an out-of-range digit gets the options repeated instead of silence', async () => { + const mock = mockSupabase() + mock.enqueue({ data: makeLink() }) + mock.enqueue({ data: withOptions('awaiting_company') }) + mock.enqueue({ data: null }) // wamid dedupe pre-check + mock.enqueue({ data: null }) // link last_message_at + mock.enqueue({ data: null }) // conversation window + mock.enqueue({ data: null }) // terminal row insert + + await route.handler(signedRequest(envelope({ messages: [textMessage('9')] }))) + + expect(sendTextMock).toHaveBeenCalledTimes(1) + expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m6CompanyRetry) + expect(sendTextMock.mock.calls[0][1].body).toContain('Bolag B AB') + }) + + it('a typed company name gets the options repeated, not the M16 lecture', async () => { + const mock = mockSupabase() + mock.enqueue({ data: makeLink() }) + mock.enqueue({ data: withOptions('awaiting_company') }) + mock.enqueue({ data: { id: 'msg-row-1' } }) // insert (reply-only disposition) + mock.enqueue({ data: null }) + mock.enqueue({ data: null }) + + await route.handler(signedRequest(envelope({ messages: [textMessage('Bolag B AB')] }))) + + expect(sendTextMock).toHaveBeenCalledTimes(1) + expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m6CompanyRetry) + }) + + it('accepts a LATE company answer after the 48h reset (options still present)', async () => { + const mock = mockSupabase() + mock.enqueue({ data: makeLink() }) + mock.enqueue({ data: withOptions('idle') }) + mock.enqueue({ data: null }) // wamid dedupe pre-check + mock.enqueue({ data: null }) // link last_message_at + mock.enqueue({ data: null }) // conversation window + mock.enqueue({ data: { company_id: 'company-2' } }) // membership check + mock.enqueue({ data: null }) // guarded pin write + mock.enqueue({ data: null }) // link last_company_id + mock.enqueue({ data: [{ id: 'stg-1' }] }) // staged reopen + mock.enqueue({ data: null }) // terminal row insert + + await route.handler(signedRequest(envelope({ messages: [textMessage('2')] }))) + + expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m6CompanyConfirm) + expect(kickMock).toHaveBeenCalledWith(['stg-1'], { companySelectedVia: 'numbered' }) + }) + + it("recognises 'byt' inside an awaiting state, the word m6-confirm teaches", async () => { + const mock = mockSupabase() + mock.enqueue({ data: makeLink() }) + mock.enqueue({ + data: { + id: 'conv-1', + state: 'awaiting_context', + context: { + pin_expires_at: new Date(Date.now() + 60 * 60 * 1000).toISOString(), + pending_question: { + type: 'context', + inbox_item_id: 'item-1', + asked_at: new Date().toISOString(), + }, + }, + company_id: 'company-2', + }, + }) + mock.enqueue({ data: null }) // wamid dedupe pre-check + mock.enqueue({ data: null }) // link last_message_at + mock.enqueue({ data: null }) // conversation window + mock.enqueue({ data: null }) // guarded pin clear + mock.enqueue({ data: null }) // terminal row insert + + await route.handler(signedRequest(envelope({ messages: [textMessage('byt')] }))) + + expect(sendTextMock).toHaveBeenCalledTimes(1) + expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m6BytPin) + expect(kickMock).toHaveBeenCalledWith([]) + }) + + it('applies STOP before writing its terminal row, so a crash cannot lose the opt-out', async () => { + const mock = mockSupabase() + mock.enqueue({ data: makeLink() }) + mock.enqueue({ data: { id: 'conv-1', state: 'idle', context: {} } }) + mock.enqueue({ data: null }) // wamid dedupe pre-check + mock.enqueue({ data: null }) // link last_message_at + mock.enqueue({ data: null }) // conversation window + mock.enqueue({ data: null }) // muted_at + mock.enqueue({ data: null }) // terminal row insert + + await route.handler(signedRequest(envelope({ messages: [textMessage('stopp')] }))) + + const order = mock.calls + .filter( + (c) => + (c.table === 'whatsapp_phone_links' && + c.method === 'update' && + (c.args[0] as Record).muted_at != null) || + (c.table === 'whatsapp_messages' && c.method === 'insert'), + ) + .map((c) => `${c.table}.${c.method}`) + expect(order).toEqual(['whatsapp_phone_links.update', 'whatsapp_messages.insert']) + const [row] = mock.findCall('whatsapp_messages', 'insert') as [Record] + expect(row.processing_status).toBe('done') + }) + + it('never persists the sender plaintext number in raw_payload', async () => { + const mock = mockSupabase() + mock.enqueue({ data: makeLink() }) + mock.enqueue({ data: { id: 'conv-1', state: 'idle', context: {} } }) + mock.enqueue({ data: { id: 'msg-row-1' } }) + mock.enqueue({ data: null }) + mock.enqueue({ data: null }) + + await route.handler(signedRequest(envelope({ messages: [imageMessage()] }))) + + const [row] = mock.findCall('whatsapp_messages', 'insert') as [Record] + expect(JSON.stringify(row.raw_payload)).not.toContain('46701234567') + expect(row.raw_payload).toMatchObject({ id: 'wamid.IN1', type: 'image' }) + }) + + it('a muted sender contributes no chat content at all', async () => { + const mock = mockSupabase() + mock.enqueue({ data: makeLink({ muted_at: '2026-08-01T10:00:00Z' }) }) + mock.enqueue({ data: { id: 'conv-1', state: 'idle', context: {} } }) + mock.enqueue({ data: { id: 'msg-row-1' } }) + mock.enqueue({ data: null }) + mock.enqueue({ data: null }) + + await route.handler( + signedRequest(envelope({ messages: [textMessage('känslig text till en död linje')] })), + ) + + const [row] = mock.findCall('whatsapp_messages', 'insert') as [Record] + expect(row.processing_status).toBe('skipped') + expect(row.body_text).toBeNull() + expect(row.raw_payload).toBeNull() + }) + }) + it('acks signed-but-unparseable bodies without redelivery bait', async () => { mockSupabase() const raw = 'not json' diff --git a/extensions/general/whatsapp-inbox/index.ts b/extensions/general/whatsapp-inbox/index.ts index 652e166d..57eb260c 100644 --- a/extensions/general/whatsapp-inbox/index.ts +++ b/extensions/general/whatsapp-inbox/index.ts @@ -40,6 +40,7 @@ import { hashPhone } from './lib/phone-crypto' import { consumeLinkCode, createPhoneLink, + LinkCodeRateLimitError, looksLikeLinkCode, lookupActiveLink, mintLinkCode, @@ -51,7 +52,9 @@ import { DEBOUNCE_WINDOW_MS, getContext, getOrCreateConversation, + redactRawPayload, resolveAnswerTarget, + updateConversation, type ConversationContext, } from './lib/conversation' import { applyCompanyChoice, type CompanyChoiceVia } from './lib/company-question' @@ -75,8 +78,10 @@ const SERVICE_WINDOW_MS = 24 * 60 * 60 * 1000 const STOP_KEYWORDS = new Set(['stopp', 'stop', 'avsluta']) const HELP_KEYWORDS = new Set(['hjälp', 'hjalp', 'help', 'support', 'människa', 'manniska']) const START_KEYWORD = 'start' -// Clears the 8h company pin. Recognized in idle only: inside an awaiting -// state the word could plausibly be answer text. +// Clears the 8h company pin. m6CompanyConfirm literally teaches this word, so +// it is recognized in every state EXCEPT awaiting_company (where the expected +// reply is a company, and 'byt' gets the re-prompt instead of being swallowed +// as answer text on someone's receipt). const BYT_KEYWORD = 'byt' const DefaultCompanySchema = z.object({ @@ -217,6 +222,7 @@ type Disposition = | { kind: 'byt' } | { kind: 'company_digit'; digit: number } | { kind: 'company_interactive'; companyId: string } + | { kind: 'company_retry' } | { kind: 'answer' } | { kind: 'text_open' } | { kind: 'voice' } @@ -224,12 +230,28 @@ type Disposition = | { kind: 'fallback' } | { kind: 'silence' } +/** Dispositions with a durable side effect (mute, pin, company routing). + * Their effect runs BEFORE the terminal row is written: see + * handleLinkedSender. Reply-only dispositions keep the persist-first shape. */ +const DURABLE_DISPOSITIONS: ReadonlySet = new Set([ + 'stop', + 'start', + 'byt', + 'company_digit', + 'company_interactive', +]) + function classify( msg: ParsedInboundMessage, muted: boolean, conversation: WhatsAppConversation | null, ): Disposition { const state = conversation?.state ?? 'idle' + // company_options outlive the awaiting_company state: the 48h TTL resets the + // state but keeps the options so a LATE company answer still lands (media + // stays fetchable from Meta for ~30 days, well past the question TTL). + const companyChoiceOpen = + (conversation ? (getContext(conversation).company_options?.length ?? 0) : 0) > 0 if (msg.type === 'text') { const normalized = (msg.text ?? '').trim().toLowerCase() if (muted) { @@ -239,22 +261,32 @@ function classify( if (STOP_KEYWORDS.has(normalized)) return { kind: 'stop' } if (HELP_KEYWORDS.has(normalized)) return { kind: 'help' } if (normalized === START_KEYWORD) return { kind: 'start' } - if (normalized === BYT_KEYWORD && state === 'idle') return { kind: 'byt' } - if (state === 'awaiting_company') { + if (normalized === BYT_KEYWORD && state !== 'awaiting_company') return { kind: 'byt' } + if (state === 'awaiting_company' || companyChoiceOpen) { // The answer is a tapped button/row (interactive) or a typed digit. if (/^\d{1,2}$/.test(normalized)) return { kind: 'company_digit', digit: Number(normalized) } - return { kind: 'fallback' } + // Anything else while the question is on screen: repeat the options. + // The M16 fallback ("I cannot answer questions") right after the bot + // asked one is the wrong answer to a typed company name. + if (state === 'awaiting_company') return { kind: 'company_retry' } } - if (state === 'awaiting_representation' || state === 'awaiting_context') { + if ( + state === 'awaiting_representation' || + state === 'awaiting_context' || + // Words cannot answer the re-send question, but they are still about + // that receipt: the worker keeps them as a note (M9 note) instead of + // binding them to some other receipt's open question. + state === 'awaiting_resend' + ) { return { kind: 'answer' } } - // Idle / awaiting_resend free text: maybe a late answer to an earlier - // question (quoted reply or the most recent open one); resolved below. + // Idle free text: maybe a late answer to an earlier question (quoted + // reply or the most recent open one); resolved below. return { kind: 'text_open' } } if (muted) return { kind: 'silence' } if (msg.type === 'interactive') { - if (conversation?.state === 'awaiting_company' && msg.interactiveReplyId) { + if ((conversation?.state === 'awaiting_company' || companyChoiceOpen) && msg.interactiveReplyId) { return { kind: 'company_interactive', companyId: msg.interactiveReplyId } } // Stale button tap after the question closed: silence beats lecturing. @@ -272,6 +304,60 @@ function classify( return { kind: 'silence' } } +/** True when this inbound wamid was already persisted (Meta redelivery). */ +async function inboundAlreadyPersisted( + supabase: SupabaseClient, + wamid: string | null, +): Promise { + if (!wamid) return false + const { data } = await supabase + .from('whatsapp_messages') + .select('id') + .eq('wamid', wamid) + .eq('direction', 'inbound') + .limit(1) + .maybeSingle() + return data != null +} + +/** last_message_at + the 24h service window, touched for every inbound. */ +async function touchInbound( + supabase: SupabaseClient, + link: WhatsAppPhoneLink, + conversationId: string | null, + now: Date, + armBurst: boolean, +): Promise { + await supabase + .from('whatsapp_phone_links') + .update({ last_message_at: now.toISOString() }) + .eq('id', link.id) + if (!conversationId) return + // Media arms the burst debounce: each file pushes the deadline forward, + // and the deferred worker whose deadline survives claims the ONE ack. + // Two literal payloads (not one built at runtime) so the phantom-column + // scanner can verify both shapes. + if (armBurst) { + await supabase + .from('whatsapp_conversations') + .update({ + last_inbound_at: now.toISOString(), + service_window_expires_at: new Date(now.getTime() + SERVICE_WINDOW_MS).toISOString(), + debounce_until: new Date(now.getTime() + DEBOUNCE_WINDOW_MS).toISOString(), + pending_ack: true, + }) + .eq('id', conversationId) + } else { + await supabase + .from('whatsapp_conversations') + .update({ + last_inbound_at: now.toISOString(), + service_window_expires_at: new Date(now.getTime() + SERVICE_WINDOW_MS).toISOString(), + }) + .eq('id', conversationId) + } +} + async function handleLinkedSender( supabase: SupabaseClient, msg: ParsedInboundMessage, @@ -281,7 +367,8 @@ async function handleLinkedSender( ): Promise { const conversation = await getOrCreateConversation(supabase, link.id) const conversationId = conversation?.id ?? null - let disposition = classify(msg, link.muted_at != null, conversation) + const muted = link.muted_at != null + let disposition = classify(msg, muted, conversation) // Late-answer probe: free text with no open awaiting state still answers a // recent question when it quotes one of its messages or one is open <=7d. @@ -294,15 +381,126 @@ async function handleLinkedSender( const correlationId = crypto.randomUUID() const copy = botCopy('sv') + const now = new Date() + const replyBase = { + senderPhoneHash: phoneHash, + phoneLinkId: link.id, + conversationId, + correlationId, + } - // Persist-first, dedupe on the inbound-wamid partial unique index. A - // redelivered wamid violates it (23505): already handled, stop entirely. + // ── Durable dispositions: effect first, terminal row after ── + // Writing the row 'done' up front made these at-most-once: an instance that + // died between the insert and the effect lost the action forever, because + // Meta's redelivery hits the wamid dedupe and the sweep never claims 'done' + // rows. A dropped STOP is a dropped opt-out, so the order is inverted here + // and dedupe becomes a pre-check: the worst case is now a repeated (and + // idempotent) effect plus a duplicate confirmation, never a lost one. + if (DURABLE_DISPOSITIONS.has(disposition.kind)) { + if (await inboundAlreadyPersisted(supabase, msg.wamid)) return + await touchInbound(supabase, link, conversationId, now, false) + + switch (disposition.kind) { + case 'company_digit': + case 'company_interactive': { + if (!conversation) return + const options = getContext(conversation).company_options ?? [] + const via: CompanyChoiceVia = + disposition.kind === 'company_digit' + ? 'numbered' + : options.length <= MAX_REPLY_BUTTONS + ? 'button' + : 'list' + const applied = await applyCompanyChoice(supabase, { + conversation, + link, + choice: + disposition.kind === 'company_digit' + ? { digit: disposition.digit } + : { companyId: disposition.companyId }, + via, + to: msg.from, + replyBase, + }) + if (applied.ok) { + if (applied.stagedMessageIds.length > 0) { + kickInboundProcessing(applied.stagedMessageIds, { companySelectedVia: via }) + } + } else if (applied.reason === 'invalid_option' && options.length > 0) { + // A typed digit outside the range is a typo, not a forged payload: + // silence just leaves the receipts parked until they expire. + await sendText(supabase, { + to: msg.from, + body: copy.m6CompanyRetry({ options: options.map((o) => o.name) }), + template: TEMPLATE.m6CompanyRetry, + ...replyBase, + }) + } + // Other rejections stay silent (stale or forged payloads). + break + } + case 'byt': { + if (conversation) { + await updateConversation(supabase, conversation, (_current, context) => { + const next: ConversationContext = { ...context } + delete next.pin_expires_at + delete next.pin_source + return { company_id: null, context: next } + }) + } + await sendText(supabase, { to: msg.from, body: copy.m6BytPin(), template: TEMPLATE.m6BytPin, ...replyBase }) + break + } + case 'stop': + await supabase + .from('whatsapp_phone_links') + .update({ muted_at: now.toISOString() }) + .eq('id', link.id) + .is('muted_at', null) + await sendText(supabase, { to: msg.from, body: copy.m11Stop(), template: TEMPLATE.m11Stop, ...replyBase }) + break + case 'start': + if (muted) { + await supabase + .from('whatsapp_phone_links') + .update({ muted_at: null }) + .eq('id', link.id) + } + await sendText(supabase, { to: msg.from, body: copy.m12Start(), template: TEMPLATE.m12Start, ...replyBase }) + break + } + + // Terminal row last. A concurrent redelivery that won the race 23505s. + const { error: durableInsertError } = await supabase.from('whatsapp_messages').insert({ + direction: 'inbound', + wamid: msg.wamid, + sender_phone_hash: phoneHash, + phone_link_id: link.id, + conversation_id: conversationId, + message_type: msg.type, + body_text: msg.type === 'text' ? msg.text : (msg.caption ?? null), + raw_payload: redactRawPayload(msg.raw), + processing_status: 'done', + correlation_id: correlationId, + }) + if (durableInsertError && durableInsertError.code !== '23505') { + log.error('Failed to persist handled WhatsApp message', durableInsertError) + } + return + } + + // ── Everything else: persist-first ────────────────────────── + // Dedupe on the inbound-wamid partial unique index. A redelivered wamid + // violates it (23505): already handled, stop entirely. const initialStatus = disposition.kind === 'media' || disposition.kind === 'answer' ? 'received' : disposition.kind === 'silence' ? 'skipped' : 'done' + // A muted sender is told the channel is paused, so nothing they send is + // read: match the unknown-sender discipline and keep no content at all. + const keepContent = !muted const { data: inserted, error: insertError } = await supabase .from('whatsapp_messages') .insert({ @@ -312,12 +510,15 @@ async function handleLinkedSender( phone_link_id: link.id, conversation_id: conversationId, message_type: msg.type, - body_text: msg.type === 'text' ? msg.text : (msg.caption ?? null), - media_id: msg.media?.id ?? null, - media_mime: msg.media?.mime ?? null, - media_sha256: msg.media?.sha256 ?? null, - media_filename: msg.media?.filename ?? null, - raw_payload: msg.raw as Record, + body_text: keepContent ? (msg.type === 'text' ? msg.text : (msg.caption ?? null)) : null, + media_id: keepContent ? (msg.media?.id ?? null) : null, + media_mime: keepContent ? (msg.media?.mime ?? null) : null, + media_sha256: keepContent ? (msg.media?.sha256 ?? null) : null, + media_filename: keepContent ? (msg.media?.filename ?? null) : null, + // The sender's plaintext E.164 number lived in here on every single + // row, which defeated the AES-256-GCM phone_enc on the link. Replies + // decrypt the link instead (resolveRecipient). + raw_payload: keepContent ? redactRawPayload(msg.raw) : null, processing_status: initialStatus, correlation_id: correlationId, }) @@ -331,43 +532,7 @@ async function handleLinkedSender( } const messageId = (inserted as { id: string } | null)?.id ?? null - const now = new Date() - await supabase - .from('whatsapp_phone_links') - .update({ last_message_at: now.toISOString() }) - .eq('id', link.id) - if (conversationId) { - // Media arms the burst debounce: each file pushes the deadline forward, - // and the deferred worker whose deadline survives claims the ONE ack. - // Two literal payloads (not one built at runtime) so the phantom-column - // scanner can verify both shapes. - if (disposition.kind === 'media') { - await supabase - .from('whatsapp_conversations') - .update({ - last_inbound_at: now.toISOString(), - service_window_expires_at: new Date(now.getTime() + SERVICE_WINDOW_MS).toISOString(), - debounce_until: new Date(now.getTime() + DEBOUNCE_WINDOW_MS).toISOString(), - pending_ack: true, - }) - .eq('id', conversationId) - } else { - await supabase - .from('whatsapp_conversations') - .update({ - last_inbound_at: now.toISOString(), - service_window_expires_at: new Date(now.getTime() + SERVICE_WINDOW_MS).toISOString(), - }) - .eq('id', conversationId) - } - } - - const replyBase = { - senderPhoneHash: phoneHash, - phoneLinkId: link.id, - conversationId, - correlationId, - } + await touchInbound(supabase, link, conversationId, now, disposition.kind === 'media') switch (disposition.kind) { case 'media': @@ -376,61 +541,16 @@ async function handleLinkedSender( // path may call the LLM; the webhook must 200 in seconds). if (messageId) deferredMessageIds.push(messageId) return - case 'company_digit': - case 'company_interactive': { - if (!conversation) return - const via: CompanyChoiceVia = - disposition.kind === 'company_digit' - ? 'numbered' - : (getContext(conversation).company_options?.length ?? 0) <= MAX_REPLY_BUTTONS - ? 'button' - : 'list' - const applied = await applyCompanyChoice(supabase, { - conversation, - link, - choice: - disposition.kind === 'company_digit' - ? { digit: disposition.digit } - : { companyId: disposition.companyId }, - via, + case 'company_retry': { + const options = conversation ? (getContext(conversation).company_options ?? []) : [] + await sendText(supabase, { to: msg.from, - replyBase, + body: options.length > 0 ? copy.m6CompanyRetry({ options: options.map((o) => o.name) }) : copy.m16Fallback(), + template: options.length > 0 ? TEMPLATE.m6CompanyRetry : TEMPLATE.m16Fallback, + ...replyBase, }) - // Silent on invalid/unauthorized choices (stale or forged payloads). - if (applied && applied.stagedMessageIds.length > 0) { - kickInboundProcessing(applied.stagedMessageIds, { companySelectedVia: via }) - } return } - case 'byt': { - if (conversation) { - const context: ConversationContext = { ...getContext(conversation) } - delete context.pin_expires_at - delete context.pin_source - await supabase - .from('whatsapp_conversations') - .update({ company_id: null, context: context as Record }) - .eq('id', conversation.id) - } - await sendText(supabase, { to: msg.from, body: copy.m6BytPin(), template: TEMPLATE.m6BytPin, ...replyBase }) - return - } - case 'stop': - await supabase - .from('whatsapp_phone_links') - .update({ muted_at: now.toISOString() }) - .eq('id', link.id) - await sendText(supabase, { to: msg.from, body: copy.m11Stop(), template: TEMPLATE.m11Stop, ...replyBase }) - return - case 'start': - if (link.muted_at != null) { - await supabase - .from('whatsapp_phone_links') - .update({ muted_at: null }) - .eq('id', link.id) - } - await sendText(supabase, { to: msg.from, body: copy.m12Start(), template: TEMPLATE.m12Start, ...replyBase }) - return case 'help': await sendText(supabase, { to: msg.from, body: copy.m13Help(), template: TEMPLATE.m13Help, ...replyBase }) return @@ -572,7 +692,18 @@ export const whatsappInboxExtension: Extension = { // whatsapp_link_codes is service-role only (RLS with no policies). const serviceClient = createServiceClient() - const minted = await mintLinkCode(serviceClient, ctx.userId) + let minted: Awaited> + try { + minted = await mintLinkCode(serviceClient, ctx.userId) + } catch (err) { + if (err instanceof LinkCodeRateLimitError) { + return NextResponse.json( + { error: 'För många koder begärda. Vänta en stund och försök igen.' }, + { status: 429 }, + ) + } + throw err + } // The wa.me link needs the real number, not the Graph object id. // WHATSAPP_PUBLIC_NUMBER (E.164 digits) is authoritative when set; diff --git a/extensions/general/whatsapp-inbox/lib/company-question.ts b/extensions/general/whatsapp-inbox/lib/company-question.ts index 947b3e5c..62b65c1b 100644 --- a/extensions/general/whatsapp-inbox/lib/company-question.ts +++ b/extensions/general/whatsapp-inbox/lib/company-question.ts @@ -27,6 +27,7 @@ import { COMPANY_PIN_TTL_MS, STAGED_AWAITING_COMPANY, getContext, + updateConversation, type ConversationContext, } from './conversation' @@ -73,6 +74,12 @@ export async function loadCompanyOptions( * transition (WHERE state <> 'awaiting_company') makes concurrent workers of * one burst produce exactly one M6; losers stage silently. * + * The state is committed BEFORE the send (a fast tap must find it) but rolled + * back when the send fails: sends are best-effort and never throw, so without + * the rollback an expired token or a Graph 5xx left the conversation parked on + * a question the user never received, with the guard suppressing every later + * ask and the 48h TTL eventually discarding the staged receipts. + * * Returns true when this caller sent the question. */ export async function askCompanyQuestion( @@ -97,11 +104,13 @@ export async function askCompanyQuestion( } const now = new Date() + const askedAt = now.toISOString() + const previousState = args.conversation.state const context = getContext(args.conversation) const nextContext: ConversationContext = { ...context, company_options: options, - pending_question: { type: 'company', inbox_item_id: null, asked_at: now.toISOString() }, + pending_question: { type: 'company', inbox_item_id: null, asked_at: askedAt }, } const { data: won } = await supabase @@ -109,28 +118,37 @@ export async function askCompanyQuestion( .update({ state: 'awaiting_company', context: nextContext as Record }) .eq('id', args.conversation.id) .neq('state', 'awaiting_company') - .select('id') + .select('*') if (!Array.isArray(won) || won.length === 0) return false + // Take updated_at (the revision guard) from the echoed row, but state and + // context from what we just wrote: that is what a rollback must match. + const committed: WhatsAppConversation = { + ...args.conversation, + ...((won[0] as WhatsAppConversation | undefined) ?? {}), + state: 'awaiting_company', + context: nextContext as Record, + } const copy = botCopy('sv') const body = copy.m6CompanyQuestion({ count: args.stagedCount }) const base = { to: args.to, template: TEMPLATE.m6CompanyQuestion, ...args.replyBase } + let sent: { ok: boolean } if (options.length <= MAX_REPLY_BUTTONS) { - await sendReplyButtons(supabase, { + sent = await sendReplyButtons(supabase, { ...base, body, buttons: options.map((o) => ({ id: o.id, title: o.name })), }) } else if (options.length <= MAX_LIST_ROWS) { - await sendList(supabase, { + sent = await sendList(supabase, { ...base, body, buttonLabel: 'Välj företag', rows: options.map((o) => ({ id: o.id, title: o.name })), }) } else { - await sendText(supabase, { + sent = await sendText(supabase, { ...base, body: copy.m6CompanyQuestionNumbered({ count: args.stagedCount, @@ -138,22 +156,60 @@ export async function askCompanyQuestion( }), }) } + + if (!sent.ok) { + // Undo the ask so the next receipt re-triggers it. Only our own question + // is rolled back: a newer one (different asked_at) is left alone. + await updateConversation(supabase, committed, (current, currentContext) => { + if ( + current.state !== 'awaiting_company' || + currentContext.pending_question?.asked_at !== askedAt + ) { + return null + } + const rolledBack: ConversationContext = { ...currentContext } + delete rolledBack.company_options + delete rolledBack.pending_question + return { state: previousState === 'awaiting_company' ? 'idle' : previousState, context: rolledBack } + }) + log.warn('company question send failed; question rolled back', { + conversationId: args.conversation.id, + }) + return false + } return true } export interface AppliedCompanyChoice { + ok: true companyId: string companyName: string /** Parked message rows re-opened for processing (run through the kick). */ stagedMessageIds: string[] } +export interface RejectedCompanyChoice { + ok: false + /** + * - invalid_option: a typed digit outside the numbered range. Ordinary + * user input (a typo), so the caller re-prompts instead of going silent. + * - not_member / lookup_failed / already_applied: stale or forged payloads + * and races. Silence. + */ + reason: 'invalid_option' | 'not_member' | 'lookup_failed' | 'already_applied' +} + +export type CompanyChoiceResult = AppliedCompanyChoice | RejectedCompanyChoice + /** * Apply a company answer (button/list id or typed digit). Validates * membership, pins the company for 8h, confirms with M6-confirm, re-opens - * the parked rows and arms the combined ack. Returns null on an invalid or - * unauthorized choice (callers stay silent: only stale or forged payloads - * get here). + * the parked rows and arms the combined ack. + * + * The open question, not the conversation state, is the thing being claimed: + * company_options are deleted by the winning write, so a double tap delivered + * into parallel invocations confirms exactly once, and a LATE answer (the 48h + * TTL reset the state to idle but kept the options) still lands. */ export async function applyCompanyChoice( supabase: SupabaseClient, @@ -165,9 +221,10 @@ export async function applyCompanyChoice( to: string replyBase: ReplyBase }, -): Promise { +): Promise { const context = getContext(args.conversation) const options = context.company_options ?? [] + if (options.length === 0) return { ok: false, reason: 'already_applied' } let companyId: string | null = null if ('companyId' in args.choice) { @@ -175,23 +232,31 @@ export async function applyCompanyChoice( } else { const option = options[args.choice.digit - 1] companyId = option?.id ?? null + if (!companyId) return { ok: false, reason: 'invalid_option' } } - if (!companyId) return null + if (!companyId) return { ok: false, reason: 'invalid_option' } // Defense in depth: the chosen company must be one the sender belongs to, - // whatever the payload claimed. - const { data: membership } = await supabase + // whatever the payload claimed. A query ERROR is not a missing membership: + // treating a transient failure as "not a member" silently drops the answer. + const { data: membership, error: membershipError } = await supabase .from('company_members') .select('company_id') .eq('user_id', args.link.user_id) .eq('company_id', companyId) .limit(1) .maybeSingle() + if (membershipError) { + log.error('company choice membership lookup failed', membershipError, { + conversationId: args.conversation.id, + }) + return { ok: false, reason: 'lookup_failed' } + } if (!membership) { log.warn('company choice rejected: sender is not a member', { conversationId: args.conversation.id, }) - return null + return { ok: false, reason: 'not_member' } } const known = options.find((o) => o.id === companyId) @@ -206,24 +271,25 @@ export async function applyCompanyChoice( } const now = new Date() - const nextContext: ConversationContext = { ...context } - delete nextContext.company_options - delete nextContext.pending_question - nextContext.pin_expires_at = new Date(now.getTime() + COMPANY_PIN_TTL_MS).toISOString() - nextContext.pin_source = args.via - // Pin + arm the combined ack: the staged receipts process right after, // and the finalize step claims `pending_ack AND debounce_until <= now()`. - await supabase - .from('whatsapp_conversations') - .update({ + // Guarded: whoever deletes company_options first owns the answer. + const applied = await updateConversation(supabase, args.conversation, (_current, currentContext) => { + if ((currentContext.company_options?.length ?? 0) === 0) return null + const nextContext: ConversationContext = { ...currentContext } + delete nextContext.company_options + delete nextContext.pending_question + nextContext.pin_expires_at = new Date(now.getTime() + COMPANY_PIN_TTL_MS).toISOString() + nextContext.pin_source = args.via + return { state: 'idle', company_id: companyId, - context: nextContext as Record, + context: nextContext, pending_ack: true, debounce_until: now.toISOString(), - }) - .eq('id', args.conversation.id) + } + }) + if (!applied) return { ok: false, reason: 'already_applied' } await supabase .from('whatsapp_phone_links') @@ -246,6 +312,7 @@ export async function applyCompanyChoice( .select('id') return { + ok: true, companyId, companyName, stagedMessageIds: ((reopened ?? []) as { id: string }[]).map((r) => r.id), diff --git a/extensions/general/whatsapp-inbox/lib/conversation.ts b/extensions/general/whatsapp-inbox/lib/conversation.ts index a11cc03c..c7709c5a 100644 --- a/extensions/general/whatsapp-inbox/lib/conversation.ts +++ b/extensions/general/whatsapp-inbox/lib/conversation.ts @@ -12,16 +12,22 @@ * - recent_questions asked-question log for late answers (<= 7 days) * - budget daily content-question counter (Europe/Stockholm) * - * Concurrency note: context is read-modify-write via PostgREST. Single-writer - * discipline keeps that safe: only the burst-ack winner, the answer worker, - * the company-choice handler and the sweep write it, and the pending_ack / - * processing_status claims serialize them. Media staging deliberately does - * NOT go through context (staged refs live as whatsapp_messages rows) so - * parallel webhook invocations never race on this column. + * Concurrency note: context is read-modify-write via PostgREST, and the + * writers (burst-ack winner, answer worker, company-choice handler, sweep) + * hold DIFFERENT claims (pending_ack vs processing_status), so they are not + * serialized against each other. Every context write therefore goes through + * updateConversation(), which guards on the updated_at it read and re-applies + * the mutation against fresh state when it loses. Media staging deliberately + * does NOT go through context (staged refs live as whatsapp_messages rows) so + * parallel webhook invocations never race on this column at all. */ import type { SupabaseClient } from '@supabase/supabase-js' +import { createLogger } from '@/lib/logger' import type { WhatsAppConversation, WhatsAppMessage } from '@/types' +import { decryptPhone } from './phone-crypto' + +const log = createLogger('whatsapp-inbox/conversation') export const COMPANY_PIN_TTL_MS = 8 * 60 * 60 * 1000 export const QUESTION_TTL_MS = 48 * 60 * 60 * 1000 @@ -103,6 +109,73 @@ export async function loadConversation( return (data as WhatsAppConversation | null) ?? null } +/** Fields a conversation write may touch. Anything omitted stays as it is. */ +export interface ConversationPatch { + state?: WhatsAppConversation['state'] + context?: ConversationContext + company_id?: string | null + last_outbound_at?: string + pending_ack?: boolean + debounce_until?: string +} + +/** + * Optimistic-concurrency write of a conversation row. + * + * `context` is a whole-jsonb column written read-modify-write, and its writers + * hold different claims (the ack winner holds pending_ack, an answer worker + * holds its own row's processing_status, the sweep holds nothing), so a blind + * `.eq('id')` update silently clobbers whatever landed in between: resurrected + * questions, wiped pending_question, dropped queue entries. + * + * The updated_at trigger makes that column a revision counter, so guarding on + * the value the mutation was derived from turns every write into a + * compare-and-set. On a lost race the row is reloaded and `mutate` runs again + * against fresh state; returning null from it aborts (the work is already + * done, or no longer applies). + */ +export async function updateConversation( + supabase: SupabaseClient, + conversation: WhatsAppConversation, + mutate: ( + current: WhatsAppConversation, + context: ConversationContext, + ) => ConversationPatch | null, + maxAttempts = 4, +): Promise { + let current: WhatsAppConversation | null = conversation + for (let attempt = 0; attempt < maxAttempts; attempt++) { + if (!current) return null + const patch = mutate(current, getContext(current)) + if (!patch) return null + // Literal payload so the phantom-column scanner can verify the column + // names; undefined values drop out of the JSON body PostgREST receives. + const { data } = await supabase + .from('whatsapp_conversations') + .update({ + state: patch.state, + context: patch.context as Record | undefined, + company_id: patch.company_id, + last_outbound_at: patch.last_outbound_at, + pending_ack: patch.pending_ack, + debounce_until: patch.debounce_until, + }) + .eq('id', current.id) + .eq('updated_at', current.updated_at) + .select('*') + // PostgREST returns [] only when the revision guard did not match. + if (!Array.isArray(data) || data.length > 0) { + const written = Array.isArray(data) ? (data[0] as WhatsAppConversation) : null + return written ?? ({ ...current, ...patch } as WhatsAppConversation) + } + current = await loadConversation(supabase, current.id) + } + log.warn('conversation write gave up after concurrent modifications', { + conversationId: conversation.id, + }) + return null +} + /** * Atomic burst-ack claim. Exactly one caller per debounce window gets a row * back; everyone else stays silent. `debounce_until <= now` uses the caller's @@ -147,7 +220,8 @@ export function bumpBudget( ): ConversationContext['budget'] { const dayKey = stockholmDayKey(now) const current = context.budget?.day_key === dayKey ? context.budget.count : 0 - return { day_key: dayKey, count: current + asked } + // Negative `asked` refunds a question whose send failed; never below zero. + return { day_key: dayKey, count: Math.max(0, current + asked) } } /** True when the company pin on the conversation is present and unexpired. */ @@ -178,17 +252,30 @@ export interface AnswerTarget { inboxItemId: string /** True when matched outside an awaiting_* state (quoted or recent). */ late: boolean + /** The quoted receipt's question was already answered: this is a follow-up + * correction, appended to the note instead of overwriting the answer. */ + followUp?: boolean } const TEXT_ANSWERABLE: ReadonlySet = new Set(['representation', 'context']) +function isTextAnswerable( + question: RecentQuestion, +): question is RecentQuestion & { type: TextAnswerableQuestionType } { + return TEXT_ANSWERABLE.has(question.type) +} + /** * Resolve which question a free-text message answers. * - * 1. An awaiting_representation/awaiting_context state answers its own - * pending question. - * 2. Otherwise a quoted reply (context.message_id -> whatsapp_messages.wamid, - * either direction) resolves through that row's inbox_item_id. + * 1. A quoted reply (context.message_id -> whatsapp_messages.wamid, either + * direction) names its receipt, so it wins over everything else, including + * an unrelated pending question and including a question that receipt has + * already answered (the reply is then a follow-up correction). Binding a + * quoted correction to some OTHER receipt because the quoted one was + * settled is how the wrong item gets the note. + * 2. Otherwise an awaiting_representation/awaiting_context state answers its + * own pending question. * 3. Otherwise the single most recent question asked within 7 days that has * not been answered yet (moved_to_app still accepts a late answer). */ @@ -200,26 +287,20 @@ export async function resolveAnswerTarget( ): Promise { const context = getContext(conversation) - if ( + const pending = (conversation.state === 'awaiting_representation' || conversation.state === 'awaiting_context') && context.pending_question && TEXT_ANSWERABLE.has(context.pending_question.type) && context.pending_question.inbox_item_id - ) { - return { - type: context.pending_question.type as TextAnswerableQuestionType, - inboxItemId: context.pending_question.inbox_item_id, - late: false, - } - } + ? { + type: context.pending_question.type as TextAnswerableQuestionType, + inboxItemId: context.pending_question.inbox_item_id, + } + : null - const recent = (context.recent_questions ?? []).filter( - (q): q is RecentQuestion & { type: TextAnswerableQuestionType } => - TEXT_ANSWERABLE.has(q.type) && - q.status !== 'answered' && - now.getTime() - new Date(q.asked_at).getTime() <= LATE_ANSWER_MAX_AGE_MS, - ) + const withinWindow = (question: RecentQuestion): boolean => + now.getTime() - new Date(question.asked_at).getTime() <= LATE_ANSWER_MAX_AGE_MS if (quotedWamid) { const { data: quotedRow } = await supabase @@ -231,11 +312,31 @@ export async function resolveAnswerTarget( .maybeSingle() const quotedItemId = (quotedRow as { inbox_item_id: string | null } | null)?.inbox_item_id if (quotedItemId) { - const match = recent.find((q) => q.inbox_item_id === quotedItemId) - if (match) return { type: match.type, inboxItemId: match.inbox_item_id, late: true } + // Quoting the very question that is open: an ordinary answer, not late. + if (pending && pending.inboxItemId === quotedItemId) { + return { ...pending, late: false } + } + const quoted = [...(context.recent_questions ?? [])] + .filter((q) => isTextAnswerable(q) && withinWindow(q) && q.inbox_item_id === quotedItemId) + .sort((a, b) => new Date(b.asked_at).getTime() - new Date(a.asked_at).getTime())[0] + if (quoted && isTextAnswerable(quoted)) { + return { + type: quoted.type, + inboxItemId: quoted.inbox_item_id, + late: true, + followUp: quoted.status === 'answered', + } + } } } + if (pending) return { ...pending, late: false } + + const recent = (context.recent_questions ?? []).filter( + (q): q is RecentQuestion & { type: TextAnswerableQuestionType } => + isTextAnswerable(q) && q.status !== 'answered' && withinWindow(q), + ) + if (recent.length > 0) { const latest = [...recent].sort( (a, b) => new Date(b.asked_at).getTime() - new Date(a.asked_at).getTime(), @@ -270,9 +371,48 @@ export function appendRecentQuestion( return [...kept, entry].slice(-10) } -/** The recipient phone (E.164 digits) for replies, read back from the - * persisted raw payload (the row never stores the raw number elsewhere). */ +/** + * Legacy recipient read: rows persisted before the raw payload was redacted + * still carry the sender's plaintext number under `from`. New rows do not, + * so this returns null for them and the caller decrypts the phone link. + */ export function extractRecipient(row: WhatsAppMessage): string | null { const raw = row.raw_payload as { from?: unknown } | null return raw && typeof raw.from === 'string' && raw.from.length > 0 ? raw.from : null } + +/** + * The recipient phone (E.164 digits) for replies. + * + * The authoritative copy is whatsapp_phone_links.phone_enc (AES-256-GCM): + * persisting the plaintext number in every message's raw_payload defeated the + * whole point of encrypting it once on the link. Falls back to the legacy + * raw_payload copy for rows written before the redaction. + */ +export function resolveRecipient( + row: WhatsAppMessage, + link: { phone_enc?: string | null } | null, +): string | null { + const legacy = extractRecipient(row) + if (legacy) return legacy + if (!link?.phone_enc) return null + try { + const phone = decryptPhone(link.phone_enc) + return phone.length > 0 ? phone : null + } catch (err) { + // Shredded (retention/erasure sets phone_enc = '') or a key rotation: + // there is no one left to reply to, and that must not throw here. + log.warn('could not decrypt phone link for reply', { + error: err instanceof Error ? err.message : String(err), + }) + return null + } +} + +/** Strip the sender's plaintext phone number out of the payload we persist. + * Everything else (type, timestamp, quoted context.id) is kept verbatim. */ +export function redactRawPayload(raw: unknown): Record | null { + if (!raw || typeof raw !== 'object') return null + const { from: _from, ...rest } = raw as Record + return rest +} diff --git a/extensions/general/whatsapp-inbox/lib/linking.ts b/extensions/general/whatsapp-inbox/lib/linking.ts index ceb9d070..60aafb48 100644 --- a/extensions/general/whatsapp-inbox/lib/linking.ts +++ b/extensions/general/whatsapp-inbox/lib/linking.ts @@ -4,7 +4,11 @@ * The code proves control of an Accounted account (minted in an authenticated * settings panel) and sending it proves possession of the phone; the webhook * binds the two. Invite-token pattern (lib/auth/invite-tokens.ts): the raw - * code exists only in the user's chat, the DB stores its sha256. + * code exists only in the user's chat, the DB stores a hash. That hash is + * HMAC-peppered, NOT a plain sha256: invite tokens are 256-bit random, but a + * link code is one of 30^6 values behind a fixed 'AC-' prefix, and enumerating + * that space offline takes about a second (the exact reason phone-crypto.ts + * peppers the phone hash). * * All functions here take a SERVICE-ROLE client: whatsapp_link_codes has RLS * enabled with no policies, and link INSERTs are service-role only by design @@ -14,7 +18,7 @@ import crypto from 'crypto' import type { SupabaseClient } from '@supabase/supabase-js' import type { WhatsAppPhoneLink } from '@/types' -import { encryptPhone, hashPhone, maskPhone } from './phone-crypto' +import { encryptPhone, hashPhone, hashSecret, maskPhone } from './phone-crypto' /** Uppercased twin of generate_inbox_local_part's ambiguity-free alphabet * (no I/L/O/U, no 0/1): codes survive being read aloud or retyped. */ @@ -22,9 +26,20 @@ export const CODE_ALPHABET = 'ABCDEFGHJKMNPQRSTVWXYZ23456789' export const CODE_PREFIX = 'AC-' export const CODE_LENGTH = 6 export const CODE_TTL_MS = 10 * 60 * 1000 +/** Codes a single account may mint inside the TTL window before it must wait. + * The panel mints one per visit; anything above this is a script. */ +export const MAX_CODES_PER_TTL_WINDOW = 5 export function hashLinkCode(code: string): string { - return crypto.createHash('sha256').update(code).digest('hex') + return hashSecret(code) +} + +/** Thrown by mintLinkCode when the caller is minting far too fast. */ +export class LinkCodeRateLimitError extends Error { + readonly name = 'LinkCodeRateLimitError' + constructor() { + super('Too many link codes requested') + } } /** @@ -59,12 +74,32 @@ export interface MintedCode { expiresAt: string } -/** Mint a fresh code for the settings panel. Earlier unused codes stay valid - * until their own 10-minute expiry; the webhook consumes whichever arrives. */ +/** + * Mint a fresh code for the settings panel. + * + * Exactly one code per account is live at a time: minting burns the caller's + * earlier unused codes, so the panel showing a code is the only code that + * works. Minting is also capped per TTL window, because the route is an + * authenticated unbounded INSERT otherwise. + */ export async function mintLinkCode( serviceClient: SupabaseClient, userId: string, ): Promise { + const windowStart = new Date(Date.now() - CODE_TTL_MS).toISOString() + const { count } = await serviceClient + .from('whatsapp_link_codes') + .select('id', { count: 'exact', head: true }) + .eq('user_id', userId) + .gte('created_at', windowStart) + if ((count ?? 0) >= MAX_CODES_PER_TTL_WINDOW) throw new LinkCodeRateLimitError() + + await serviceClient + .from('whatsapp_link_codes') + .update({ used_at: new Date().toISOString() }) + .eq('user_id', userId) + .is('used_at', null) + let body = '' for (let i = 0; i < CODE_LENGTH; i++) { body += CODE_ALPHABET[crypto.randomInt(CODE_ALPHABET.length)] diff --git a/extensions/general/whatsapp-inbox/lib/messages.ts b/extensions/general/whatsapp-inbox/lib/messages.ts index fc5ac32b..da28e59e 100644 --- a/extensions/general/whatsapp-inbox/lib/messages.ts +++ b/extensions/general/whatsapp-inbox/lib/messages.ts @@ -36,12 +36,14 @@ export const TEMPLATE = { m5BurstAck: 'm5_burst_ack', m6CompanyQuestion: 'm6_company_question', m6CompanyConfirm: 'm6_company_confirm', + m6CompanyRetry: 'm6_company_retry', m6BytPin: 'm6_byt_pin', m7Representation: 'm7_representation', m8RepConfirmed: 'm8_rep_confirmed', m8RepPartial: 'm8_rep_partial', m8RepDenied: 'm8_rep_denied', m9Resend: 'm9_resend', + m9NoteSaved: 'm9_note_saved', m10Context: 'm10_context', m10ContextConfirm: 'm10_context_confirm', m11Stop: 'm11_stop', @@ -51,6 +53,7 @@ export const TEMPLATE = { m15Unsupported: 'm15_unsupported', m16Fallback: 'm16_fallback', m17RateLimited: 'm17_rate_limited', + m17RateLimitedDay: 'm17_rate_limited_day', m18Error: 'm18_error', } as const @@ -111,6 +114,12 @@ const SV = { m6CompanyConfirm: ({ companyName }: { companyName: string }) => `*${companyName}*, noterat. Jag minns valet i 8 timmar (skriv *byt* för att ändra).`, + // Typo or a typed company name while the question is open: repeat the list + // instead of the M16 lecture or silence. + m6CompanyRetry: ({ options }: { options: string[] }) => + `Jag känner inte igen svaret. Svara med en siffra 1-${options.length}:\n\n` + + options.map((name, i) => `${i + 1}. ${name}`).join('\n'), + m6BytPin: () => 'Okej, jag frågar vilket företag nästa gång du skickar ett kvitto.', m7Representation: ({ ref }: { ref?: string | null } = {}) => @@ -131,6 +140,11 @@ const SV = { : 'Bilden blev för lågupplöst för att läsas av, WhatsApp komprimerar foton hårt.\n\n' + 'Skicka gärna samma kvitto igen som *dokument*: gem-ikonen -> _Dokument_ -> välj bilden. Då behålls full skärpa. Kvittot ligger kvar i Underlag så länge.', + // Text arrived while a re-send question is open: the words cannot replace + // the file, so they are kept as a note and the question stays open. + m9NoteSaved: () => + 'Tack! Sparat som anteckning på kvittot. Skicka gärna samma kvitto igen som *dokument* (gem-ikonen -> _Dokument_) så kan jag läsa av beloppen.', + m10Context: ({ ordinal }: { ordinal?: number | null } = {}) => ordinal != null ? `Kvitto ${ordinal}: jag kunde inte läsa av allt. Vad gäller köpet? Skriv en kort rad (t.ex. _taxi till kundmöte, 240 kr_) så sparar jag det på kvittot.` @@ -138,8 +152,12 @@ const SV = { m10ContextConfirm: () => 'Tack! Sparat som anteckning på kvittot.', + // Honest wording: `stopp` PAUSES the channel (muted_at), it does not revoke + // the binding. Actually disconnecting is the settings panel's "Koppla från", + // and the panel shows the paused state as "Pausad": the copy must match. m11Stop: () => - 'Okej, jag slutar svara här och ditt nummer kopplas från. Dina underlag i Accounted påverkas inte. Skicka *start* om du vill aktivera igen.', + 'Okej, jag pausar och slutar svara här. Numret är kvar kopplat men jag tar inte emot något mer. Dina underlag i Accounted påverkas inte.\n\n' + + 'Skicka *start* när du vill aktivera igen, eller koppla från numret helt under *Inställningar -> WhatsApp* i Accounted.', m12Start: () => 'Välkommen tillbaka! Ditt nummer är aktivt igen, skicka kvitton när du vill.', @@ -159,6 +177,11 @@ const SV = { m17RateLimited: ({ minutes = 10 }: LinkedTemplateArgs) => `Det blev många filer på kort tid, jag pausar mottagningen en stund. Försök igen om cirka ${minutes} minuter.`, + // The 500/day company quota resets at midnight, so "cirka 10 minuter" is + // simply wrong there: never promise a minute count we cannot keep. + m17RateLimitedDay: () => + 'Dagens gräns för mottagna filer är nådd för det här företaget. Jag tar emot fler i morgon, eller ladda upp direkt i appen under *Underlag* så kommer de in nu.', + m18Error: () => 'Något gick fel när jag tog emot filen. Försök igen om en stund, eller ladda upp kvittot direkt i appen under *Underlag*. Skriv *hjälp* om det fortsätter.', } @@ -220,6 +243,10 @@ const EN: typeof SV = { m6CompanyConfirm: ({ companyName }: { companyName: string }) => `*${companyName}*, noted. I will remember the choice for 8 hours (type *byt* to change).`, + m6CompanyRetry: ({ options }: { options: string[] }) => + `I did not recognise that answer. Reply with a number 1-${options.length}:\n\n` + + options.map((name, i) => `${i + 1}. ${name}`).join('\n'), + m6BytPin: () => 'Okay, I will ask which company the next time you send a receipt.', m7Representation: ({ ref }: { ref?: string | null } = {}) => @@ -240,6 +267,9 @@ const EN: typeof SV = { : 'The image came through at too low a resolution to read, WhatsApp compresses photos hard.\n\n' + 'Please send the same receipt again as a *document*: paperclip icon -> _Document_ -> pick the image. That keeps full sharpness. The receipt stays in Underlag meanwhile.', + m9NoteSaved: () => + 'Thanks! Saved as a note on the receipt. Please send the same receipt again as a *document* (paperclip icon -> _Document_) so I can read the amounts.', + m10Context: ({ ordinal }: { ordinal?: number | null } = {}) => ordinal != null ? `Receipt ${ordinal}: I could not read everything. What was the purchase? Write a short line (e.g. _taxi to client meeting, 240 kr_) and I will save it on the receipt.` @@ -248,7 +278,8 @@ const EN: typeof SV = { m10ContextConfirm: () => 'Thanks! Saved as a note on the receipt.', m11Stop: () => - 'Okay, I will stop replying here and your number is disconnected. Your documents in Accounted are not affected. Send *start* to activate again.', + 'Okay, I am pausing and will stop replying here. Your number stays linked but I will not receive anything more. Your documents in Accounted are not affected.\n\n' + + 'Send *start* whenever you want to activate again, or disconnect the number entirely under *Settings -> WhatsApp* in Accounted.', m12Start: () => 'Welcome back! Your number is active again, send receipts whenever you like.', @@ -268,6 +299,9 @@ const EN: typeof SV = { m17RateLimited: ({ minutes = 10 }: LinkedTemplateArgs) => `That was a lot of files in a short time, I am pausing intake for a bit. Try again in about ${minutes} minutes.`, + m17RateLimitedDay: () => + 'This company has reached its daily limit for received files. I will accept more tomorrow, or upload directly in the app under *Underlag* to get them in now.', + m18Error: () => 'Something went wrong receiving the file. Try again in a moment, or upload the receipt directly in the app under *Underlag*. Type *hjälp* if it keeps happening.', } diff --git a/extensions/general/whatsapp-inbox/lib/phone-crypto.ts b/extensions/general/whatsapp-inbox/lib/phone-crypto.ts index fdf90195..b41c3698 100644 --- a/extensions/general/whatsapp-inbox/lib/phone-crypto.ts +++ b/extensions/general/whatsapp-inbox/lib/phone-crypto.ts @@ -40,6 +40,17 @@ export function hashPhone(rawPhone: string): string { return crypto.createHmac('sha256', getHashKey()).update(normalizePhone(rawPhone)).digest('hex') } +/** + * Peppered hash for any other low-entropy secret stored in this channel + * (link codes: 30^6 values, a smaller space than the phone numbers the + * pepper exists for). Same key, same reason: a plain sha256 over a space + * that small is enumerable offline in seconds, so hashing at rest would + * protect nothing. + */ +export function hashSecret(value: string): string { + return crypto.createHmac('sha256', getHashKey()).update(value).digest('hex') +} + /** AES-256-GCM encrypt a phone number. Returns hex(iv | tag | ciphertext) for a text column. */ export function encryptPhone(rawPhone: string): string { const key = getEncryptionKey() diff --git a/extensions/general/whatsapp-inbox/lib/process-inbound.ts b/extensions/general/whatsapp-inbox/lib/process-inbound.ts index f314828b..b7959879 100644 --- a/extensions/general/whatsapp-inbox/lib/process-inbound.ts +++ b/extensions/general/whatsapp-inbox/lib/process-inbound.ts @@ -54,7 +54,9 @@ import { markRecentQuestion, questionsAskedToday, resolveAnswerTarget, + resolveRecipient, serviceWindowOpen, + updateConversation, type ConversationContext, type QueuedQuestion, type QuestionType, @@ -112,11 +114,19 @@ function fallbackFilename(mime: string): string { return `whatsapp-${new Date().toISOString().slice(0, 10)}.${ext}` } -function formatSek(amount: number): string { - return `${new Intl.NumberFormat('sv-SE', { +/** + * Amount as the ack states it. The extraction pipeline preserves the document + * currency ("Do NOT default to SEK"), so printing every total with 'kr' turned + * a EUR 250 hotel receipt into "250 kr" on the one surface that tells the user + * their receipt was filed correctly. + */ +function formatAmount(amount: number, currency: string | null | undefined): string { + const formatted = new Intl.NumberFormat('sv-SE', { minimumFractionDigits: 0, maximumFractionDigits: 2, - }).format(amount)} kr` + }).format(amount) + const code = (currency ?? '').trim().toUpperCase() + return code === '' || code === 'SEK' ? `${formatted} kr` : `${formatted} ${code}` } async function loadRow( @@ -169,13 +179,21 @@ async function rateLimitNoticeAlreadySent( .select('id') .eq('direction', 'outbound') .eq('sender_phone_hash', senderPhoneHash) - .eq('raw_payload->>template', TEMPLATE.m17RateLimited) + // Both m17 variants (minute + day scope) share the window. + .like('raw_payload->>template', `${TEMPLATE.m17RateLimited}%`) .gte('created_at', since) .limit(1) .maybeSingle() return data != null } +/** + * Terminal status write. Guarded on processing_status='processing': the row is + * only ours while we hold the claim, and an unguarded write let a late loser + * (a sweep re-claim that raced a live worker) overwrite the winner's 'done' + * and null its inbox_item_id, which silently loses the burst ack and breaks + * quoted-reply resolution. + */ async function markStatus( supabase: SupabaseClient, messageId: string, @@ -191,6 +209,69 @@ async function markStatus( inbox_item_id: extra.inboxItemId ?? null, }) .eq('id', messageId) + .eq('processing_status', 'processing') +} + +/** The Underlag item this chat message already produced, if any. The partial + * unique index invoice_inbox_items_whatsapp_msg makes this the item-level + * idempotency key for a re-processed row. */ +async function existingInboxItemId( + supabase: SupabaseClient, + whatsappMessageId: string, +): Promise { + const { data } = await supabase + .from('invoice_inbox_items') + .select('id') + .eq('whatsapp_message_id', whatsappMessageId) + .limit(1) + .maybeSingle() + return (data as { id: string } | null)?.id ?? null +} + +/** True when an M18 failure notice already went out for this message. */ +async function errorNoticeAlreadySent( + supabase: SupabaseClient, + correlationId: string | null, +): Promise { + if (!correlationId) return false + const { data } = await supabase + .from('whatsapp_messages') + .select('id') + .eq('direction', 'outbound') + .eq('correlation_id', correlationId) + .eq('raw_payload->>template', TEMPLATE.m18Error) + .limit(1) + .maybeSingle() + return data != null +} + +/** + * Tell the sender their file failed, at most once per message. Gating this on + * `attempt <= 1` made it unreachable for exactly the failure the sweep exists + * for: a first attempt that dies with the instance already counted its + * attempt, so every re-claim suppressed the notice and the row ended in + * 'error' with the sender never hearing about that receipt at all. + */ +export async function sendErrorNoticeOnce( + supabase: SupabaseClient, + args: { + to: string + senderPhoneHash: string | null + phoneLinkId: string | null + conversationId: string | null + correlationId: string | null + }, +): Promise { + if (await errorNoticeAlreadySent(supabase, args.correlationId)) return + await sendText(supabase, { + to: args.to, + body: botCopy('sv').m18Error(), + template: TEMPLATE.m18Error, + senderPhoneHash: args.senderPhoneHash, + phoneLinkId: args.phoneLinkId, + conversationId: args.conversationId, + correlationId: args.correlationId, + }) } // ── Company resolution (PR4 ladder) ────────────────────────── @@ -214,21 +295,19 @@ async function resolveCompanyTarget( if (conversation && hasLivePin(conversation, now) && conversation.company_id) { if (await isMember(supabase, link.user_id, conversation.company_id)) { - // Sliding TTL. Refresh at most once a minute: the pin write is a plain - // context read-modify-write, and throttling it shrinks the window in - // which it could race a question-state write from the ack winner. + // Sliding TTL, refreshed at most once a minute. The write is guarded + // (updateConversation): a blind whole-context replacement here would + // wipe a pending_question the ack winner wrote in between, and the + // sweep would then reset the state a minute later. const context = getContext(conversation) const expiresAt = context.pin_expires_at ? new Date(context.pin_expires_at).getTime() : 0 if (expiresAt - now.getTime() < COMPANY_PIN_TTL_MS - 60 * 1000) { - await supabase - .from('whatsapp_conversations') - .update({ - context: { - ...context, - pin_expires_at: new Date(now.getTime() + COMPANY_PIN_TTL_MS).toISOString(), - } as Record, - }) - .eq('id', conversation.id) + await updateConversation(supabase, conversation, (_current, currentContext) => ({ + context: { + ...currentContext, + pin_expires_at: new Date(now.getTime() + COMPANY_PIN_TTL_MS).toISOString(), + }, + })) } return { companyId: conversation.company_id, via: selectedViaOverride ?? 'pin' } } @@ -300,7 +379,6 @@ async function processMediaMessage( opts: KickOptions, ): Promise { const copy = botCopy('sv') - const to = extractRecipient(row) const replyBase = { senderPhoneHash: row.sender_phone_hash, @@ -308,9 +386,10 @@ async function processMediaMessage( conversationId: row.conversation_id, correlationId: row.correlation_id, } + let to: string | null = null try { - if (!row.phone_link_id || !row.media_id || !to) { + if (!row.phone_link_id || !row.media_id) { await markStatus(supabase, row.id, 'error', { errorMessage: 'Message row is missing link or media reference', }) @@ -326,6 +405,13 @@ async function processMediaMessage( }) return { kind: 'none' } } + to = resolveRecipient(row, link) + if (!to) { + await markStatus(supabase, row.id, 'error', { + errorMessage: 'No reply address for message row', + }) + return { kind: 'none' } + } const conversation = row.conversation_id ? await loadConversation(supabase, row.conversation_id) @@ -394,12 +480,33 @@ async function processMediaMessage( log.error('RateLimitedDropped append failed', err) } if (row.sender_phone_hash && !(await rateLimitNoticeAlreadySent(supabase, row.sender_phone_hash))) { - await sendText(supabase, { to, body: copy.m17RateLimited({}), template: TEMPLATE.m17RateLimited, ...replyBase }) + // The daily company quota resets at Stockholm midnight, so the + // template's 10-minute default is simply false there. + const dayScope = limit.scope === 'day' + await sendText(supabase, { + to, + body: dayScope + ? copy.m17RateLimitedDay() + : copy.m17RateLimited({ minutes: Math.max(1, Math.ceil((limit.retryAfterSec ?? 600) / 60)) }), + template: dayScope ? TEMPLATE.m17RateLimitedDay : TEMPLATE.m17RateLimited, + ...replyBase, + }) } await markStatus(supabase, row.id, 'skipped', { errorMessage: 'Rate limited' }) return { kind: 'media_processed', conversationId: conversation?.id ?? null } } + // ── Already ingested? (idempotency at the item level) ── + // A sweep re-claim that raced a live worker must adopt the item the + // winner created, not create a second one: the WORM document is written + // before the item insert and can never be deleted (7-year retention), + // and the unique index would make the second insert throw. + const alreadyIngested = await existingInboxItemId(supabase, row.id) + if (alreadyIngested) { + await markStatus(supabase, row.id, 'done', { inboxItemId: alreadyIngested }) + return { kind: 'media_processed', conversationId: conversation?.id ?? null } + } + // ── Download (fresh URL per media id; 10 MB stream-checked cap) ── const media = await downloadMedia(row.media_id) @@ -419,22 +526,33 @@ async function processMediaMessage( } // ── Upload + extract (the shared invoice-inbox funnel) ── - const result = await uploadAndExtract( - supabase, - link.user_id, - companyId, - { name: row.media_filename || fallbackFilename(mime), buffer: media.buffer, type: mime }, - 'whatsapp', - undefined, - undefined, - { - channelMeta: { whatsappMessageId: row.id, caption: row.body_text ?? null }, - actorId: 'whatsapp-inbound', - }, - ) + let inboxItemId: string + try { + const result = await uploadAndExtract( + supabase, + link.user_id, + companyId, + { name: row.media_filename || fallbackFilename(mime), buffer: media.buffer, type: mime }, + 'whatsapp', + undefined, + undefined, + { + channelMeta: { whatsappMessageId: row.id, caption: row.body_text ?? null }, + actorId: 'whatsapp-inbound', + }, + ) + inboxItemId = result.inbox_item_id + } catch (uploadErr) { + // invoice_inbox_items_whatsapp_msg is a unique index: a concurrent + // worker that got there first is a success, not a failure. + const adopted = await existingInboxItemId(supabase, row.id) + if (!adopted) throw uploadErr + await markStatus(supabase, row.id, 'done', { inboxItemId: adopted }) + return { kind: 'media_processed', conversationId: conversation?.id ?? null } + } // Record how the company was chosen (audit + FieldsRail in PR5). - await updateItemContext(supabase, result.inbox_item_id, (context) => ({ + await updateItemContext(supabase, inboxItemId, (context) => ({ ...context, company_selected_via: resolved.via, })) @@ -449,19 +567,19 @@ async function processMediaMessage( await resolveResendQuestion(supabase, conversation) } - await markStatus(supabase, row.id, 'done', { inboxItemId: result.inbox_item_id }) + await markStatus(supabase, row.id, 'done', { inboxItemId }) // No ack here: the burst winner sends ONE combined M4/M5 (+ question). return { kind: 'media_processed', conversationId: conversation?.id ?? null } } catch (err) { const message = err instanceof GraphApiError || err instanceof Error ? err.message : String(err) - log.error('WhatsApp intake processing failed', err, { messageId: row.id }) + log.error('WhatsApp intake processing failed', err, { messageId: row.id, attempt }) try { await markStatus(supabase, row.id, 'error', { errorMessage: message.slice(0, 500) }) - // M18 once per message: only on the first attempt, so a sweep re-claim - // of the same row never spams the sender. - if (to && attempt <= 1) { - await sendText(supabase, { to, body: botCopy('sv').m18Error(), template: TEMPLATE.m18Error, ...replyBase }) + // M18 at most once per message, tracked by the outbound row rather than + // the attempt counter (a first attempt that died still burned its). + if (to) { + await sendErrorNoticeOnce(supabase, { to, ...replyBase }) } } catch (innerErr) { log.error('Failed to record WhatsApp processing error', innerErr, { messageId: row.id }) @@ -492,15 +610,14 @@ async function resolveResendQuestion( : { type: 'resend', asked_at: pending.asked_at, status: 'answered' }, })) - const nextContext: ConversationContext = { - ...context, - recent_questions: markRecentQuestion(context, oldItemId, 'answered'), - } - delete nextContext.pending_question - await supabase - .from('whatsapp_conversations') - .update({ state: 'idle', context: nextContext as Record }) - .eq('id', conversation.id) + await updateConversation(supabase, conversation, (_current, currentContext) => { + const nextContext: ConversationContext = { + ...currentContext, + recent_questions: markRecentQuestion(currentContext, oldItemId, 'answered'), + } + delete nextContext.pending_question + return { state: 'idle', context: nextContext } + }) await appendQuestionHistory(supabase, { inboxItemId: oldItemId, eventType: 'ChannelQuestionAnswered', @@ -524,7 +641,8 @@ function ackLine(entry: BurstEntry): string { const merchant = entry.extracted?.supplier?.name ?? entry.row.media_filename ?? 'Kvitto' const total = entry.extracted?.totals?.total ?? null - return `${merchant}, ${total != null ? formatSek(total) : 'belopp saknas'}` + const currency = entry.extracted?.invoice?.currency ?? null + return `${merchant}, ${total != null ? formatAmount(total, currency) : 'belopp saknas'}` } /** @@ -558,7 +676,13 @@ export async function finalizeBurst( const nowIso = now.toISOString() const rowIds = rows.map((r) => r.id) - const to = rows.map(extractRecipient).find((value) => value != null) ?? null + // Legacy rows still carry the sender number; redacted ones do not, and + // then the reply address comes from the link's encrypted copy. + let to = rows.map(extractRecipient).find((value) => value != null) ?? null + if (!to) { + const link = await loadLink(supabase, conversation.phone_link_id) + to = rows.map((r) => resolveRecipient(r, link)).find((value) => value != null) ?? null + } // Never initiate outside the 24h service window (no templates in v1): // if the window closed, hand off silently and mark the rows covered. if (!to || !serviceWindowOpen(conversation, now)) { @@ -677,7 +801,7 @@ export async function finalizeBurst( total != null ? copy.m4Ack({ merchant: entry.extracted?.supplier?.name ?? null, - amount: formatSek(total), + amount: formatAmount(total, entry.extracted?.invoice?.currency ?? null), date: entry.extracted?.invoice?.invoiceDate ?? null, }) : copy.m4AckEmpty() @@ -707,30 +831,35 @@ export async function finalizeBurst( } // ── Persist state BEFORE sending (a fast answer must find it) ── - const nextContext: ConversationContext = { ...context, question_queue: queue } - let nextState = conversation.state - if (toAsk) { - nextContext.pending_question = { - type: toAsk.type, - inbox_item_id: toAsk.entry.itemId, - asked_at: nowIso, - } - nextContext.recent_questions = appendRecentQuestion( - context, - { type: toAsk.type, inbox_item_id: toAsk.entry.itemId, asked_at: nowIso, status: 'open' }, - now, - ) - nextContext.budget = bumpBudget(context, 1, now) - nextState = STATE_FOR_QUESTION[toAsk.type] - } - await supabase - .from('whatsapp_conversations') - .update({ - state: nextState, - context: nextContext as Record, - last_outbound_at: nowIso, - }) - .eq('id', conversationId) + // Guarded write: this runs seconds after the reads above, and the answer + // worker (a different claim entirely) may have settled a question in + // between. Replacing the whole jsonb blindly resurrected answered + // questions and dropped queue entries. + const previousState = conversation.state + const askedItemId = toAsk?.entry.itemId ?? null + const committed = await updateConversation( + supabase, + conversation, + (current, currentContext) => { + const nextContext: ConversationContext = { ...currentContext, question_queue: queue } + let nextState = current.state + if (toAsk) { + nextContext.pending_question = { + type: toAsk.type, + inbox_item_id: toAsk.entry.itemId, + asked_at: nowIso, + } + nextContext.recent_questions = appendRecentQuestion( + currentContext, + { type: toAsk.type, inbox_item_id: toAsk.entry.itemId, asked_at: nowIso, status: 'open' }, + now, + ) + nextContext.budget = bumpBudget(currentContext, 1, now) + nextState = STATE_FOR_QUESTION[toAsk.type] + } + return { state: nextState, context: nextContext, last_outbound_at: nowIso } + }, + ) if (toAsk) { await updateItemContext(supabase, toAsk.entry.itemId, (itemContext) => ({ @@ -754,7 +883,7 @@ export async function finalizeBurst( } const last = rows[rows.length - 1] - await sendText(supabase, { + const sent = await sendText(supabase, { to, body, template, @@ -765,6 +894,34 @@ export async function finalizeBurst( inboxItemId: ackItemId, }) + if (!sent.ok) { + // Sends never throw, so an ignored result meant a Graph failure left the + // user with no ack ever (acked_at blocks the sweep's re-arm) AND the + // conversation parked on a question they never received, which the next + // unrelated text would then be interpreted as answering. Undo the + // question and leave the rows unacked so sweep pass 2b retries. + if (toAsk && committed && askedItemId) { + await updateConversation(supabase, committed, (current, currentContext) => { + if (currentContext.pending_question?.asked_at !== nowIso) return null + const rolledBack: ConversationContext = { ...currentContext } + delete rolledBack.pending_question + rolledBack.recent_questions = (currentContext.recent_questions ?? []).filter( + (q) => !(q.inbox_item_id === askedItemId && q.asked_at === nowIso), + ) + rolledBack.budget = bumpBudget(currentContext, -1, now) + return { state: current.state === STATE_FOR_QUESTION[toAsk.type] ? previousState : current.state, context: rolledBack } + }) + await updateItemContext(supabase, askedItemId, (itemContext) => { + if (itemContext.pending_question?.asked_at !== nowIso) return itemContext + const reverted = { ...itemContext } + delete reverted.pending_question + return reverted + }) + } + log.warn('combined ack send failed; rows left unacked for the sweep', { conversationId }) + return + } + await supabase.from('whatsapp_messages').update({ acked_at: nowIso }).in('id', rowIds) } catch (err) { log.error('burst finalize failed', err, { conversationId }) @@ -781,12 +938,23 @@ function renderParticipants( .join(', ') } +/** Append a chat line to the item's note, keeping whatever is already there. */ +async function appendUserNote( + supabase: SupabaseClient, + inboxItemId: string, + text: string, +): Promise { + await updateItemContext(supabase, inboxItemId, (itemContext) => ({ + ...itemContext, + user_note: itemContext.user_note ? `${itemContext.user_note}\n${text}` : text, + })) +} + async function processAnswerMessage( supabase: SupabaseClient, row: WhatsAppMessage, ): Promise { const copy = botCopy('sv') - const to = extractRecipient(row) const replyBase = { senderPhoneHash: row.sender_phone_hash, phoneLinkId: row.phone_link_id, @@ -795,7 +963,7 @@ async function processAnswerMessage( } try { - if (!row.phone_link_id || !to) { + if (!row.phone_link_id) { await markStatus(supabase, row.id, 'error', { errorMessage: 'Answer row is missing link' }) return { kind: 'none' } } @@ -804,6 +972,11 @@ async function processAnswerMessage( await markStatus(supabase, row.id, 'skipped', { errorMessage: 'Phone link revoked' }) return { kind: 'none' } } + const to = resolveRecipient(row, link) + if (!to) { + await markStatus(supabase, row.id, 'error', { errorMessage: 'No reply address for answer row' }) + return { kind: 'none' } + } const conversation = row.conversation_id ? await loadConversation(supabase, row.conversation_id) : null @@ -815,15 +988,59 @@ async function processAnswerMessage( const raw = row.raw_payload as { context?: { id?: unknown } } | null const quotedWamid = typeof raw?.context?.id === 'string' ? raw.context.id : null + + // Words cannot answer the re-send question (it needs a sharper file), but + // they ARE about that receipt. Keep them there and leave the question + // open, instead of letting the late-answer probe bind them to some other + // receipt's question and confirm it as that receipt's note. + const resendPending = getContext(conversation).pending_question + if ( + conversation.state === 'awaiting_resend' && + resendPending?.type === 'resend' && + resendPending.inbox_item_id + ) { + await appendUserNote(supabase, resendPending.inbox_item_id, text) + await sendText(supabase, { + to, + body: copy.m9NoteSaved(), + template: TEMPLATE.m9NoteSaved, + ...replyBase, + inboxItemId: resendPending.inbox_item_id, + }) + await markStatus(supabase, row.id, 'done', { inboxItemId: resendPending.inbox_item_id }) + return { kind: 'answer', conversationId: conversation.id } + } + const target = await resolveAnswerTarget(supabase, conversation, quotedWamid) if (!target) { // The question disappeared between webhook routing and processing - // (TTL sweep or a concurrent answer): plain fallback. - await sendText(supabase, { to, body: copy.m16Fallback(), template: TEMPLATE.m16Fallback, ...replyBase }) + // (TTL sweep or a concurrent answer): plain fallback. NOT on a + // re-claim though: a worker that died after applying the answer and + // sending its confirm leaves the row 'processing', and the retry would + // then follow that confirm with "I did not understand". + if (row.attempts === 0) { + await sendText(supabase, { to, body: copy.m16Fallback(), template: TEMPLATE.m16Fallback, ...replyBase }) + } await markStatus(supabase, row.id, 'done') return { kind: 'answer', conversationId: conversation.id } } + if (target.followUp) { + // The user quoted a receipt whose question is already settled: this is + // an addition ('glömde: Bo Ek var också med'), so it is appended to + // that receipt instead of overwriting its confirmed answer. + await appendUserNote(supabase, target.inboxItemId, text) + await sendText(supabase, { + to, + body: copy.m10ContextConfirm(), + template: TEMPLATE.m10ContextConfirm, + ...replyBase, + inboxItemId: target.inboxItemId, + }) + await markStatus(supabase, row.id, 'done', { inboxItemId: target.inboxItemId }) + return { kind: 'answer', conversationId: conversation.id } + } + const nowIso = new Date().toISOString() let confirmBody: string let confirmTemplate: TemplateId @@ -922,6 +1139,11 @@ async function processAnswerMessage( await updateItemContext(supabase, target.inboxItemId, (itemContext) => ({ ...itemContext, user_note: note, + // The note is an LLM paraphrase; keep what the human actually + // wrote (and when) next to it, the way the representation branch + // does. Without this the only copy of the raw wording is + // whatsapp_messages.body_text, which the retention cron purges. + context_answer: { raw_answer: text, answered_at: nowIso }, pending_question: itemContext.pending_question ? { ...itemContext.pending_question, status: 'answered' } : undefined, @@ -932,21 +1154,23 @@ async function processAnswerMessage( } // Conversation bookkeeping: settle the question, then surface the next - // queued one (if any and the budget allows). - const context = getContext(conversation) - const nextContext: ConversationContext = { - ...context, - recent_questions: markRecentQuestion(context, target.inboxItemId, 'answered'), - } + // queued one (if any and the budget allows). Guarded: the LLM call above + // takes seconds, and a burst finalize or a second answer fragment may + // have written the row in between (a blind write resurrected the + // question or dropped the queue). let nextState = conversation.state - if (!target.late && context.pending_question?.inbox_item_id === target.inboxItemId) { - delete nextContext.pending_question - nextState = 'idle' - } - await supabase - .from('whatsapp_conversations') - .update({ state: nextState, context: nextContext as Record }) - .eq('id', conversation.id) + await updateConversation(supabase, conversation, (current, currentContext) => { + const patchContext: ConversationContext = { + ...currentContext, + recent_questions: markRecentQuestion(currentContext, target.inboxItemId, 'answered'), + } + nextState = current.state + if (!target.late && currentContext.pending_question?.inbox_item_id === target.inboxItemId) { + delete patchContext.pending_question + nextState = 'idle' + } + return { state: nextState, context: patchContext } + }) await appendQuestionHistory(supabase, { inboxItemId: target.inboxItemId, @@ -970,8 +1194,11 @@ async function processAnswerMessage( await markStatus(supabase, row.id, 'done', { inboxItemId: target.inboxItemId }) return { kind: 'answer', conversationId: conversation.id } } catch (err) { - // No M18 here: that copy is about files. An answer that failed to - // process stays silent; the sweep may retry (attempts < 3). + // No M18 here: that copy is about files. This catch is a programming + // -error net, not a retry hook: every dependency on the answer path + // degrades instead of throwing (interpretChatAnswer returns {ok:false}, + // sends never throw, supabase-js returns errors), and 'error' is + // terminal for the sweep, which claims only 'received'/'processing'. log.error('WhatsApp answer processing failed', err, { messageId: row.id }) try { await markStatus(supabase, row.id, 'error', { @@ -999,7 +1226,8 @@ async function askNextQueuedQuestion( const conversation = await loadConversation(supabase, conversationId) if (!conversation || conversation.state !== 'idle') return const context = getContext(conversation) - const queue: QueuedQuestion[] = [...(context.question_queue ?? [])] + const originalQueue: QueuedQuestion[] = context.question_queue ?? [] + const queue: QueuedQuestion[] = [...originalQueue] if (queue.length === 0) return const now = new Date() @@ -1025,13 +1253,45 @@ async function askNextQueuedQuestion( })) } if (!next) { - await supabase - .from('whatsapp_conversations') - .update({ context: { ...context, question_queue: [] } as Record }) - .eq('id', conversationId) + await updateConversation(supabase, conversation, (_current, currentContext) => ({ + context: { ...currentContext, question_queue: [] }, + })) return } + // Claim the pop BEFORE composing and sending. The idle check plus shift + // above is a plain read-modify-write, so two answer workers landing + // together both popped the same question and asked it twice. This guarded + // write elects one of them and settles the whole question state in one go. + const claimed = await updateConversation(supabase, conversation, (current, currentContext) => { + if (current.state !== 'idle') return null + if ( + JSON.stringify(currentContext.question_queue ?? []) !== JSON.stringify(originalQueue) + ) { + return null + } + return { + state: STATE_FOR_QUESTION[next!.type], + context: { + ...currentContext, + question_queue: queue, + pending_question: { + type: next!.type, + inbox_item_id: next!.inbox_item_id, + asked_at: nowIso, + }, + recent_questions: appendRecentQuestion( + currentContext, + { type: next!.type, inbox_item_id: next!.inbox_item_id, asked_at: nowIso, status: 'open' }, + now, + ), + budget: bumpBudget(currentContext, 1, now), + }, + last_outbound_at: nowIso, + } + }) + if (!claimed) return + const { data: itemData } = await supabase .from('invoice_inbox_items') .select('id, extracted_data') @@ -1060,26 +1320,6 @@ async function askNextQueuedQuestion( template = TEMPLATE.m10Context } - const nextContext: ConversationContext = { - ...context, - question_queue: queue, - pending_question: { type: next.type, inbox_item_id: next.inbox_item_id, asked_at: nowIso }, - recent_questions: appendRecentQuestion( - context, - { type: next.type, inbox_item_id: next.inbox_item_id, asked_at: nowIso, status: 'open' }, - now, - ), - budget: bumpBudget(context, 1, now), - } - await supabase - .from('whatsapp_conversations') - .update({ - state: STATE_FOR_QUESTION[next.type], - context: nextContext as Record, - last_outbound_at: nowIso, - }) - .eq('id', conversationId) - await updateItemContext(supabase, next.inbox_item_id, (itemContext) => ({ ...itemContext, pending_question: { type: next!.type, asked_at: nowIso, status: 'open' }, diff --git a/extensions/general/whatsapp-inbox/lib/questions.ts b/extensions/general/whatsapp-inbox/lib/questions.ts index 24f7d78c..cee28c24 100644 --- a/extensions/general/whatsapp-inbox/lib/questions.ts +++ b/extensions/general/whatsapp-inbox/lib/questions.ts @@ -13,9 +13,20 @@ import { MEAL_PATTERN } from '@/lib/tax/expense-warnings' import type { InvoiceExtractionResult } from '@/types' import type { QuestionType } from './conversation' -/** Representation question only above this total (SEK): Skatteverket-grade - * documentation for a 45 kr coffee is noise, not compliance. */ -export const REPRESENTATION_MIN_TOTAL = 150 +/** No amount floor on the representation question, deliberately. + * + * An earlier version gated it at 150 kr to avoid asking about a 45 kr + * coffee. That was wrong: what makes a representation expense deductible + * at all is documenting deltagare and syfte (BFL 5 kap 6-7 §, verifikation + * content requirements), and that duty is not conditioned on any amount. + * The 300 kr/person figure people remember is the VAT-deduction BASE cap, + * a different rule entirely. A 120 kr business lunch booked with no + * participant trail is exactly the deduction Skatteverket denies later. + * + * Noise is controlled where it belongs instead: the question fires only for + * receipt-shaped documents from restaurant/cafe/hotel merchants, at most + * once per receipt, at most twice per burst and six times per sender per + * day, and a single "nej" dismisses it for good. */ /** Compressed-chat-photo signal: WhatsApp chat photos are JPEG, nameless and * small. Below this size with an empty extraction, assume the compression @@ -82,9 +93,6 @@ function isUnreadable(input: QuestionInput): boolean { function isRepresentation(input: QuestionInput): boolean { const extracted = input.extracted - const total = extracted?.totals?.total ?? null - if (total == null || total < REPRESENTATION_MIN_TOTAL) return false - const kind = extracted?.documentKind ?? null const category = extracted?.merchantCategory ?? null diff --git a/extensions/general/whatsapp-inbox/lib/sweep.ts b/extensions/general/whatsapp-inbox/lib/sweep.ts index 887b8f02..b033f01b 100644 --- a/extensions/general/whatsapp-inbox/lib/sweep.ts +++ b/extensions/general/whatsapp-inbox/lib/sweep.ts @@ -7,37 +7,55 @@ * lost message: * * 1. Re-claim whatsapp_messages stuck in 'received' (>60s) or 'processing' - * (>90s); after MAX_ATTEMPTS they land in 'error'. + * (>5 min, safely above the worst-case live worker); after MAX_ATTEMPTS + * they land in 'error' and the sender gets one M18. * 2. Claim stale pending_ack conversations (debounce crash) and send the * combined ack; re-arm conversations whose winner died after claiming * but before sending (done rows left unacked). * 3. Expire questions past the 48h TTL: conversation back to idle, the * item's pending_question -> moved_to_app. NEVER sends anything: the 24h - * service window is long gone, and v1 sends no templates. + * service window is long gone, and v1 sends no templates. Company + * questions keep their options and their parked receipts, so a late + * answer still files them (see the pass itself). * 4. Clear expired 8h company pins. */ import type { SupabaseClient } from '@supabase/supabase-js' import { createLogger } from '@/lib/logger' -import type { WhatsAppConversation } from '@/types' +import type { WhatsAppConversation, WhatsAppMessage } from '@/types' import { COMPANY_CHOICE_EXPIRED, QUESTION_TTL_MS, STAGED_AWAITING_COMPANY, getContext, + resolveRecipient, + updateConversation, type ConversationContext, } from './conversation' -import { finalizeBurst, processInboundMessage } from './process-inbound' +import { finalizeBurst, processInboundMessage, sendErrorNoticeOnce } from './process-inbound' import { appendQuestionHistory, updateItemContext } from './item-context' const log = createLogger('whatsapp-inbox/sweep') const RECEIVED_STUCK_MS = 60 * 1000 -const PROCESSING_STUCK_MS = 90 * 1000 +/** + * A 'processing' row is only stuck if no live worker can still be on it. + * The enforced step budget of one media row is markRead (10s) + media lookup + * (10s) + download (30s) + Bedrock extraction (the cron route budgets 10-60s, + * with no short SDK timeout), under a maxDuration of 300s, and there is no + * heartbeat between the claim and the terminal write. 90s therefore re-claimed + * live workers on ordinary large PDFs and ran two of them on the same message. + * A crashed row waiting five minutes is a latency regression; two concurrent + * workers are a correctness problem. + */ +const PROCESSING_STUCK_MS = 5 * 60 * 1000 const ACK_STALE_MS = 60 * 1000 const UNACKED_REARM_MS = 120 * 1000 const MAX_ATTEMPTS = 3 const BATCH = 25 +/** Staged receipts stay answerable while Meta still serves their media + * (~30 days). Past that the marker is honest: nothing can recover them. */ +const STAGED_MEDIA_MAX_AGE_MS = 30 * 24 * 60 * 60 * 1000 export interface SweepSummary { reclaimedReceived: number @@ -52,18 +70,58 @@ interface StuckRow { id: string attempts: number conversation_id: string | null + direction: string + message_type: string + sender_phone_hash: string | null + phone_link_id: string | null + correlation_id: string | null + raw_payload: Record | null } +/** + * Park a row that ran out of attempts, and tell the sender once. Without the + * notice a file whose FIRST attempt died with the instance ends terminally + * with no ack and no error: the burst ack only lists ingested rows, so that + * receipt simply vanishes from the conversation. + */ async function markMaxAttempts( supabase: SupabaseClient, row: StuckRow, fromStatus: 'received' | 'processing', ): Promise { - await supabase + const { data: parked } = await supabase .from('whatsapp_messages') .update({ processing_status: 'error', error_message: 'Max attempts exceeded' }) .eq('id', row.id) .eq('processing_status', fromStatus) + .select('id') + if (Array.isArray(parked) && parked.length === 0) return + if (row.message_type === 'text') return // M18 is about files + + const link = row.phone_link_id + ? await loadPhoneLink(supabase, row.phone_link_id) + : null + const to = resolveRecipient(row as unknown as WhatsAppMessage, link) + if (!to) return + await sendErrorNoticeOnce(supabase, { + to, + senderPhoneHash: row.sender_phone_hash, + phoneLinkId: row.phone_link_id, + conversationId: row.conversation_id, + correlationId: row.correlation_id, + }) +} + +async function loadPhoneLink( + supabase: SupabaseClient, + phoneLinkId: string, +): Promise<{ phone_enc: string | null } | null> { + const { data } = await supabase + .from('whatsapp_phone_links') + .select('phone_enc') + .eq('id', phoneLinkId) + .maybeSingle() + return (data as { phone_enc: string | null } | null) ?? null } /** Run one sweep pass. Never throws. */ @@ -84,7 +142,9 @@ export async function runSweep(supabase: SupabaseClient): Promise const cutoff = new Date(now - RECEIVED_STUCK_MS).toISOString() const { data } = await supabase .from('whatsapp_messages') - .select('id, attempts, conversation_id') + .select( + 'id, attempts, conversation_id, direction, message_type, sender_phone_hash, phone_link_id, correlation_id, raw_payload', + ) .eq('processing_status', 'received') .lt('created_at', cutoff) .order('created_at', { ascending: true }) @@ -110,7 +170,9 @@ export async function runSweep(supabase: SupabaseClient): Promise const cutoff = new Date(now - PROCESSING_STUCK_MS).toISOString() const { data } = await supabase .from('whatsapp_messages') - .select('id, attempts, conversation_id') + .select( + 'id, attempts, conversation_id, direction, message_type, sender_phone_hash, phone_link_id, correlation_id, raw_payload', + ) .eq('processing_status', 'processing') .lt('updated_at', cutoff) .order('updated_at', { ascending: true }) @@ -173,13 +235,18 @@ export async function runSweep(supabase: SupabaseClient): Promise ...new Set(((data ?? []) as { conversation_id: string }[]).map((r) => r.conversation_id)), ] for (const conversationId of conversationIds) { - // Re-arm only when no claim is pending (pending_ack=false): a pending - // one is already covered by 2a or a live worker. + // pending_ack=false plus unacked rows is ALSO the state of a live + // claimant between claimAck and its acked_at stamp, and the 120s cutoff + // above measures the ROWS' done-stamp, not when the ack was claimed. So + // the conversation's own updated_at (which claimAck bumps) is the second + // condition: without it the sweep re-armed under a working finalize and + // a second combined ack went out. await supabase .from('whatsapp_conversations') .update({ pending_ack: true, debounce_until: new Date().toISOString() }) .eq('id', conversationId) .eq('pending_ack', false) + .lt('updated_at', cutoff) finalizeConversations.add(conversationId) } } catch (err) { @@ -222,13 +289,31 @@ export async function runSweep(supabase: SupabaseClient): Promise questionType: pending.type, }) } + // Company questions are the one kind whose expiry used to DESTROY work: + // the parked receipts were stamped company_choice_expired, a marker no + // code reads, so they never became Underlag rows and nothing ever told + // the user. The 24h service window is long gone at 48h and v1 sends no + // templates, so the honest recovery is to keep accepting a LATE answer: + // the rows stay staged and company_options stay in the context, which + // classify() treats as an open choice even in idle. Only when Meta has + // stopped serving the media (~30 days) does the marker become true. + let keepCompanyOptions = false if (conversation.state === 'awaiting_company') { + const staleCutoff = new Date(now - STAGED_MEDIA_MAX_AGE_MS).toISOString() await supabase .from('whatsapp_messages') .update({ error_message: COMPANY_CHOICE_EXPIRED }) .eq('conversation_id', conversation.id) .eq('processing_status', 'skipped') .eq('error_message', STAGED_AWAITING_COMPANY) + .lt('created_at', staleCutoff) + const { count: stillStaged } = await supabase + .from('whatsapp_messages') + .select('id', { count: 'exact', head: true }) + .eq('conversation_id', conversation.id) + .eq('processing_status', 'skipped') + .eq('error_message', STAGED_AWAITING_COMPANY) + keepCompanyOptions = (stillStaged ?? 0) > 0 } // Queued questions expire with the episode. for (const queued of context.question_queue ?? []) { @@ -251,7 +336,7 @@ export async function runSweep(supabase: SupabaseClient): Promise ), } delete nextContext.pending_question - delete nextContext.company_options + if (!keepCompanyOptions) delete nextContext.company_options delete nextContext.question_queue await supabase .from('whatsapp_conversations') @@ -275,14 +360,21 @@ export async function runSweep(supabase: SupabaseClient): Promise const context = getContext(conversation) const expiresAt = context.pin_expires_at if (expiresAt != null && new Date(expiresAt).getTime() > now) continue - const nextContext: ConversationContext = { ...context } - delete nextContext.pin_expires_at - delete nextContext.pin_source - await supabase - .from('whatsapp_conversations') - .update({ company_id: null, context: nextContext as Record }) - .eq('id', conversation.id) - summary.clearedPins++ + // Guarded, and re-checked against fresh state: this loop awaits a + // network round trip per row, so a company choice applied in between + // used to be reverted (company_id nulled, the pre-choice context + // restored) and the question re-asked seconds after the user answered. + const cleared = await updateConversation(supabase, conversation, (_current, currentContext) => { + const stillExpired = + currentContext.pin_expires_at == null || + new Date(currentContext.pin_expires_at).getTime() <= Date.now() + if (!stillExpired) return null + const nextContext: ConversationContext = { ...currentContext } + delete nextContext.pin_expires_at + delete nextContext.pin_source + return { company_id: null, context: nextContext } + }) + if (cleared) summary.clearedPins++ } } catch (err) { log.error('sweep: pin expiry pass failed', err) diff --git a/supabase/migrations/20260803090000_anonymize_whatsapp_channel.sql b/supabase/migrations/20260803090000_anonymize_whatsapp_channel.sql new file mode 100644 index 00000000..6582c29a --- /dev/null +++ b/supabase/migrations/20260803090000_anonymize_whatsapp_channel.sql @@ -0,0 +1,157 @@ +-- Extend anonymize_user_account with the WhatsApp channel. +-- +-- WHY +-- --- +-- whatsapp_phone_links.user_id is declared REFERENCES auth.users(id) ON DELETE +-- CASCADE (migration 20260802090000), and the channel relied on that cascade +-- for erasure. Accounted never deletes auth.users: account deletion is +-- app/api/account/delete/route.ts -> anonymize_user_account plus a ~100-year +-- ban, which deliberately KEEPS the auth row as a tombstone. The cascade +-- therefore never fires, and nothing else revokes the link: no extension +-- subscribes to account.deleted either. +-- +-- Consequence before this migration: after erasure the phone link stayed +-- ACTIVE (revoked_at NULL) with a decryptable phone_enc and the WhatsApp +-- profile name, lookupActiveLink kept resolving the number, and every further +-- inbound message from the erased data subject was persisted with body_text +-- and the verbatim raw_payload while the bot kept replying. That is continued +-- collection with no lawful basis and a GDPR Art 17 gap. +-- +-- WHAT +-- ---- +-- The RPC is re-created verbatim from 20260724150000 with one added block. +-- The link is REVOKED and crypto-shredded rather than deleted, matching the +-- channel's own revocation-not-deletion discipline and the retention cron's +-- shredding shape (phone_enc = '' is the cleared marker on a NOT NULL column): +-- * revoked_at set -> lookupActiveLink stops resolving the number, so any +-- further message falls to the unknown-sender path, +-- which persists no content at all. +-- * phone_enc = '' -> the number is no longer recoverable from the row. +-- * wa_profile_name, phone_masked, default/last company -> cleared. +-- * phone_hash STAYS: it is an HMAC under a server-side pepper (not +-- reversible), and it is what keeps "this phone was once linked" honest +-- for the uniqueness history the same way auth.users.email is kept. +-- * conversation state/context reset (pinned company, pending questions). +-- * body_text + raw_payload nulled on every message of that link: same +-- shape as the 90-day retention purge, only immediate. +-- * unused link codes deleted (they mint a NEW link for the erased user). +-- +-- Idempotent: every predicate only matches rows still carrying the data, and +-- the RPC already refuses a second run against an anonymized profile. + +CREATE OR REPLACE FUNCTION public.anonymize_user_account(target_user_id uuid) + RETURNS void + LANGUAGE plpgsql + SECURITY DEFINER + SET search_path TO 'public' +AS $function$ +DECLARE + blocker_count int; +BEGIN + IF auth.uid() IS DISTINCT FROM target_user_id THEN + RAISE EXCEPTION 'Can only delete your own account'; + END IF; + + -- Reject repeat invocations against an already-anonymized tombstone: the + -- account is gone, re-running would only churn the scrubbed row. + IF EXISTS ( + SELECT 1 FROM public.profiles + WHERE id = target_user_id AND anonymized_at IS NOT NULL + ) THEN + RAISE EXCEPTION 'Account is already deleted' USING ERRCODE = 'P0002'; + END IF; + + SELECT count(*) INTO blocker_count + FROM public.company_members cm + JOIN public.companies c ON c.id = cm.company_id + WHERE cm.user_id = target_user_id + AND cm.role = 'owner' + AND c.archived_at IS NULL; + + IF blocker_count > 0 THEN + RAISE EXCEPTION 'Cannot delete account: user still owns % active compan(y/ies)', blocker_count + USING ERRCODE = 'P0001'; + END IF; + + DELETE FROM public.company_members WHERE user_id = target_user_id; + DELETE FROM public.team_members WHERE user_id = target_user_id; + DELETE FROM public.bankid_identities WHERE user_id = target_user_id; + + DELETE FROM public.user_preferences WHERE user_id = target_user_id; + DELETE FROM public.api_keys WHERE user_id = target_user_id; + + -- WhatsApp channel (see header): the auth.users cascade never fires here. + DELETE FROM public.whatsapp_link_codes WHERE user_id = target_user_id; + + UPDATE public.whatsapp_messages m + SET body_text = NULL, + raw_payload = NULL + FROM public.whatsapp_phone_links l + WHERE l.user_id = target_user_id + AND m.phone_link_id = l.id + AND (m.body_text IS NOT NULL OR m.raw_payload IS NOT NULL); + + UPDATE public.whatsapp_conversations c + SET state = 'idle', + context = '{}'::jsonb, + company_id = NULL + FROM public.whatsapp_phone_links l + WHERE l.user_id = target_user_id + AND c.phone_link_id = l.id; + + UPDATE public.whatsapp_phone_links + SET revoked_at = coalesce(revoked_at, now()), + phone_enc = '', + phone_masked = '+** *** ** **', + wa_profile_name = NULL, + default_company_id = NULL, + last_company_id = NULL + WHERE user_id = target_user_id; + + UPDATE public.profiles + SET email = NULL, + full_name = NULL, + avatar_url = NULL, + deleted_at = now(), + anonymized_at = now(), + updated_at = now() + WHERE id = target_user_id; + + -- Scrub PII from the auth tombstone. auth.users.email is intentionally + -- kept (blocks re-signup + lets support verify identity for BFL-retained + -- data recovery; documented legitimate interest, see + -- app/api/account/delete/route.ts). + UPDATE auth.users + SET raw_user_meta_data = '{}'::jsonb, + raw_app_meta_data = coalesce(raw_app_meta_data, '{}'::jsonb) - 'bankid_linked' - 'has_password' + WHERE id = target_user_id; +END; +$function$; + +REVOKE ALL ON FUNCTION public.anonymize_user_account(uuid) FROM PUBLIC, anon; +GRANT EXECUTE ON FUNCTION public.anonymize_user_account(uuid) TO authenticated; + +-- Repair pass: tombstones anonymized before this migration whose WhatsApp +-- link is still live. Guarded by anonymized_at, so live users are untouched. +UPDATE public.whatsapp_messages m + SET body_text = NULL, + raw_payload = NULL + FROM public.whatsapp_phone_links l + JOIN public.profiles p ON p.id = l.user_id + WHERE p.anonymized_at IS NOT NULL + AND m.phone_link_id = l.id + AND (m.body_text IS NOT NULL OR m.raw_payload IS NOT NULL); + +UPDATE public.whatsapp_phone_links l + SET revoked_at = coalesce(l.revoked_at, now()), + phone_enc = '', + phone_masked = '+** *** ** **', + wa_profile_name = NULL, + default_company_id = NULL, + last_company_id = NULL + FROM public.profiles p + WHERE p.id = l.user_id + AND p.anonymized_at IS NOT NULL + AND (l.revoked_at IS NULL OR l.phone_enc <> '' OR l.wa_profile_name IS NOT NULL); + +NOTIFY pgrst, 'reload schema'; diff --git a/tests/pg/whatsapp-account-deletion.pg.test.ts b/tests/pg/whatsapp-account-deletion.pg.test.ts new file mode 100644 index 00000000..d8ae5a95 --- /dev/null +++ b/tests/pg/whatsapp-account-deletion.pg.test.ts @@ -0,0 +1,183 @@ +import { randomUUID, createHash } from 'node:crypto' +import { describe, expect, it } from 'vitest' +import { getPool, withUserContext } from './setup' +import { insertAuthUser } from './fixtures' + +/** + * Migration 20260803090000: anonymize_user_account must erase the WhatsApp + * channel too. + * + * whatsapp_phone_links.user_id declares ON DELETE CASCADE, but Accounted + * never deletes auth.users (the row is tombstoned for ~100 years), so the + * cascade never fires. Before the migration an erased data subject kept an + * ACTIVE phone link with a decryptable phone_enc, and every further inbound + * message was persisted with its content: GDPR Art 17 plus continued + * collection with no lawful basis. These tests fail against the pre-migration + * definition of the RPC. + */ + +function hash(value: string): string { + return createHash('sha256').update(value).digest('hex') +} + +interface SeededChannel { + userId: string + linkId: string + conversationId: string + messageId: string + codeId: string +} + +async function seedLinkedUser(): Promise { + const userId = await insertAuthUser() + await getPool().query( + `INSERT INTO public.profiles (id, email, full_name) + VALUES ($1, $2, 'PG Real') + ON CONFLICT (id) DO UPDATE SET email = EXCLUDED.email, full_name = EXCLUDED.full_name`, + [userId, `pg-real-${userId}@test.invalid`], + ) + + const linkId = randomUUID() + await getPool().query( + `INSERT INTO public.whatsapp_phone_links + (id, user_id, phone_hash, phone_enc, phone_masked, wa_profile_name) + VALUES ($1, $2, $3, 'deadbeefcafe', '+46 70 *** ** 67', 'Erased Person')`, + [linkId, userId, hash(randomUUID())], + ) + + const conversationId = randomUUID() + await getPool().query( + `INSERT INTO public.whatsapp_conversations (id, phone_link_id, state, context) + VALUES ($1, $2, 'awaiting_company', '{"company_options": [{"id": "x", "name": "Bolag AB"}]}'::jsonb)`, + [conversationId, linkId], + ) + + const messageId = randomUUID() + await getPool().query( + `INSERT INTO public.whatsapp_messages + (id, direction, wamid, sender_phone_hash, phone_link_id, conversation_id, + message_type, body_text, raw_payload, processing_status) + VALUES ($1, 'inbound', $2, $3, $4, $5, 'text', 'lunch med Anna', + '{"from": "46701234567", "type": "text"}'::jsonb, 'done')`, + [messageId, `wamid.${randomUUID()}`, hash('sender'), linkId, conversationId], + ) + + const codeId = randomUUID() + await getPool().query( + `INSERT INTO public.whatsapp_link_codes (id, user_id, code_hash, expires_at) + VALUES ($1, $2, $3, now() + interval '10 minutes')`, + [codeId, userId, hash(randomUUID())], + ) + + return { userId, linkId, conversationId, messageId, codeId } +} + +describe('anonymize_user_account: WhatsApp channel erasure (pg)', () => { + it('revokes and shreds the phone link so the number stops resolving', async () => { + const seeded = await seedLinkedUser() + + await withUserContext(seeded.userId, async (client) => { + await client.query('SELECT public.anonymize_user_account($1)', [seeded.userId]) + await client.query('RESET ROLE') + + const { rows } = await client.query<{ + revoked_at: string | null + phone_enc: string + phone_masked: string + wa_profile_name: string | null + default_company_id: string | null + last_company_id: string | null + }>( + `SELECT revoked_at, phone_enc, phone_masked, wa_profile_name, + default_company_id, last_company_id + FROM public.whatsapp_phone_links WHERE id = $1`, + [seeded.linkId], + ) + expect(rows).toHaveLength(1) + // revoked_at is what lookupActiveLink filters on: a revoked link makes + // every further inbound message take the unknown-sender path, which + // persists no content at all. + expect(rows[0]!.revoked_at).not.toBeNull() + expect(rows[0]!.phone_enc).toBe('') + expect(rows[0]!.phone_masked).toBe('+** *** ** **') + expect(rows[0]!.wa_profile_name).toBeNull() + expect(rows[0]!.default_company_id).toBeNull() + expect(rows[0]!.last_company_id).toBeNull() + }) + }) + + it('nulls body_text and raw_payload on every message of that link', async () => { + const seeded = await seedLinkedUser() + + await withUserContext(seeded.userId, async (client) => { + await client.query('SELECT public.anonymize_user_account($1)', [seeded.userId]) + await client.query('RESET ROLE') + + const { rows } = await client.query<{ + body_text: string | null + raw_payload: unknown + wamid: string | null + }>( + `SELECT body_text, raw_payload, wamid + FROM public.whatsapp_messages WHERE id = $1`, + [seeded.messageId], + ) + expect(rows).toHaveLength(1) + expect(rows[0]!.body_text).toBeNull() + expect(rows[0]!.raw_payload).toBeNull() + // The skeleton survives: "a message existed" stays auditable. + expect(rows[0]!.wamid).not.toBeNull() + }) + }) + + it('resets the conversation and deletes outstanding link codes', async () => { + const seeded = await seedLinkedUser() + + await withUserContext(seeded.userId, async (client) => { + await client.query('SELECT public.anonymize_user_account($1)', [seeded.userId]) + await client.query('RESET ROLE') + + const conversation = await client.query<{ + state: string + context: Record + company_id: string | null + }>( + `SELECT state, context, company_id + FROM public.whatsapp_conversations WHERE id = $1`, + [seeded.conversationId], + ) + expect(conversation.rows[0]!.state).toBe('idle') + expect(conversation.rows[0]!.context).toEqual({}) + expect(conversation.rows[0]!.company_id).toBeNull() + + const codes = await client.query<{ n: number }>( + `SELECT count(*)::int AS n FROM public.whatsapp_link_codes WHERE user_id = $1`, + [seeded.userId], + ) + expect(codes.rows[0]!.n).toBe(0) + }) + }) + + it('leaves another user WhatsApp data untouched', async () => { + const erased = await seedLinkedUser() + const bystander = await seedLinkedUser() + + await withUserContext(erased.userId, async (client) => { + await client.query('SELECT public.anonymize_user_account($1)', [erased.userId]) + await client.query('RESET ROLE') + + const { rows } = await client.query<{ revoked_at: string | null; phone_enc: string }>( + `SELECT revoked_at, phone_enc FROM public.whatsapp_phone_links WHERE id = $1`, + [bystander.linkId], + ) + expect(rows[0]!.revoked_at).toBeNull() + expect(rows[0]!.phone_enc).toBe('deadbeefcafe') + + const message = await client.query<{ body_text: string | null }>( + `SELECT body_text FROM public.whatsapp_messages WHERE id = $1`, + [bystander.messageId], + ) + expect(message.rows[0]!.body_text).toBe('lunch med Anna') + }) + }) +}) diff --git a/types/index.ts b/types/index.ts index 18a46fb9..62c38a11 100644 --- a/types/index.ts +++ b/types/index.ts @@ -2751,6 +2751,14 @@ export interface InboxChannelContext { denied?: boolean } user_note?: string | null + /** What the user actually typed when answering a context question, kept + * next to the LLM paraphrase in user_note. The paraphrase is what renders; + * this is the durable human answer, mirroring the representation branch + * (whatsapp_messages.body_text is purged at 90 days, so it is no trail). */ + context_answer?: { + raw_answer: string + answered_at: string + } quality?: { resend_requested_at: string resent?: boolean