feat(inbox): staged upload ack, HEIC/HEIF validation, WhatsApp silence fixes (#1605)
* fix(whatsapp): app-side unmute, close silent intake paths, health visibility - add POST /link/unmute and a Reactivate control on the Pausad state - company resolution: transient query errors release the row for sweep retry; genuine zero-options sends M19 instead of parking silently - media from unlinked senders bypasses the hourly greeting throttle (10 min burst window, daily cap kept) - GET /link returns 7-day failed-delivery and parked-inbound counts; sweep summary logs outboundFailed24h Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(documents): real HEIC/HEIF magic-byte validation, bilingual upload errors - detect ISO-BMFF ftyp brands (heic/heix/heim/heis/hevc/hevx/hevm/hevs, mif1/msf1) instead of exempting image/heic from validation; declared heic/heif accepts either family member (iOS labels vary) - new INBOX_UPLOAD_* structured error codes replace raw English strings on the inbox upload and attach-document routes - registry doc corrected to the real 10 MB cap Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(inbox): staged upload with instant ack and deferred AI extraction - web uploads insert the inbox item as status processing and respond immediately; Bedrock extraction and supplier match run via after() with a CAS flip to received (email and WhatsApp channels keep the synchronous path) - widen invoice_inbox_items.status CHECK to include processing (migration 20260813180000, pg-real test included) - crash-recovery sweep cron (*/2) flips stale processing rows; bulk-book skips extraction_in_progress items - workspace: processing chip, in-flight rows disable actions, realtime flip, retry-extraction button for empty extractions - picker accept list drops HEIC/HEIF so iOS transcodes library photos to JPEG; server allowlists unchanged (supersedes 2026-08-01 HEIC decision, see DECISIONS.md) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(migrations): bump inbox processing-status migration past main's latest Main merged 20260813210000 while this PR was in flight; an inserted version older than the latest applied aborts the prod db push at merge. Renamed 20260813180000 to 20260813213000 and updated references. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs(decisions): log preview-tracker orphan repair after migration rename Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Mattsson
co-authored by
Claude Fable 5
Claude Fable 5
parent
456366fa70
commit
4a9fa5e6c5
@@ -41,8 +41,17 @@ vi.mock('@/lib/entitlements/has-capability', async (importOriginal) => {
|
||||
return { ...actual, hasCapability: vi.fn().mockResolvedValue(true) }
|
||||
})
|
||||
|
||||
// The deferred extraction worker (staged upload) builds its own cookieless
|
||||
// service client; route it to the same mock supabase so the flow is
|
||||
// observable in the one test that reaches Bedrock.
|
||||
vi.mock('@/lib/auth/api-keys', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import('@/lib/auth/api-keys')>()
|
||||
return { ...actual, createServiceClientNoCookies: vi.fn() }
|
||||
})
|
||||
|
||||
import { extractInvoiceFields } from '@/extensions/general/invoice-inbox/lib/extract-invoice-fields'
|
||||
import { hasCapability } from '@/lib/entitlements/has-capability'
|
||||
import { createServiceClientNoCookies } from '@/lib/auth/api-keys'
|
||||
|
||||
function findRoute(method: string, path: string) {
|
||||
return invoiceInboxExtension.apiRoutes!.find(
|
||||
@@ -86,7 +95,7 @@ function makeMultipartRequest(form: FormData, path: string): Request {
|
||||
// suppliers (match) in that order via separate .from() chains.
|
||||
function makeUploadSupabase(opts: {
|
||||
isSandbox: boolean
|
||||
captured: { row?: Record<string, unknown> }
|
||||
captured: { row?: Record<string, unknown>; flip?: Record<string, unknown> }
|
||||
}) {
|
||||
const settingsChain = {
|
||||
select: vi.fn().mockReturnThis(),
|
||||
@@ -100,6 +109,11 @@ function makeUploadSupabase(opts: {
|
||||
limit: vi.fn().mockReturnThis(),
|
||||
maybeSingle: vi.fn().mockResolvedValue({ data: null }),
|
||||
}
|
||||
// The deferred worker's CAS flip: .update(p).eq().eq().select('id').
|
||||
const updateChain = {
|
||||
eq: vi.fn(() => updateChain),
|
||||
select: vi.fn().mockResolvedValue({ data: [{ id: 'inbox-1' }], error: null }),
|
||||
}
|
||||
const inboxChain = {
|
||||
insert: vi.fn((row: Record<string, unknown>) => {
|
||||
opts.captured.row = row
|
||||
@@ -112,8 +126,9 @@ function makeUploadSupabase(opts: {
|
||||
}),
|
||||
}
|
||||
}),
|
||||
update: vi.fn().mockReturnValue({
|
||||
eq: vi.fn().mockReturnThis(),
|
||||
update: vi.fn((payload: Record<string, unknown>) => {
|
||||
opts.captured.flip = payload
|
||||
return updateChain
|
||||
}),
|
||||
}
|
||||
return {
|
||||
@@ -151,9 +166,10 @@ describe('Sandbox companies skip Bedrock extraction', () => {
|
||||
expect(captured.row?.extraction_skipped).toBe(true)
|
||||
})
|
||||
|
||||
it('runs extraction normally for non-sandbox companies', async () => {
|
||||
const captured: { row?: Record<string, unknown> } = {}
|
||||
it('defers extraction for non-sandbox companies: responds processing, then flips', async () => {
|
||||
const captured: { row?: Record<string, unknown>; flip?: Record<string, unknown> } = {}
|
||||
const supabase = makeUploadSupabase({ isSandbox: false, captured })
|
||||
vi.mocked(createServiceClientNoCookies).mockReturnValue(supabase as never)
|
||||
vi.mocked(extractInvoiceFields).mockResolvedValueOnce({
|
||||
data: {
|
||||
supplier: { name: null, orgNumber: null, vatNumber: null, address: null, bankgiro: null, plusgiro: null },
|
||||
@@ -174,10 +190,19 @@ describe('Sandbox companies skip Bedrock extraction', () => {
|
||||
const res = await uploadRoute.handler(makeMultipartRequest(form, '/upload'), buildCtx(supabase))
|
||||
const { status, body } = await parseJsonResponse<{ data: Record<string, unknown> }>(res)
|
||||
|
||||
// Staged upload: the response is the receipt ack, extraction runs after.
|
||||
expect(status).toBe(200)
|
||||
expect(extractInvoiceFields).toHaveBeenCalledOnce()
|
||||
expect(body.data.status).toBe('processing')
|
||||
expect(body.data.extracted_data).toBeNull()
|
||||
expect(body.data.extraction_skipped).toBe(false)
|
||||
expect(body.data.skip_reason).toBeNull()
|
||||
expect(captured.row?.status).toBe('processing')
|
||||
expect(captured.row?.extracted_data).toBeNull()
|
||||
|
||||
await vi.waitFor(() => expect(extractInvoiceFields).toHaveBeenCalledOnce())
|
||||
await vi.waitFor(() => expect(captured.flip).toBeDefined())
|
||||
expect(captured.flip?.status).toBe('received')
|
||||
expect(captured.flip?.extraction_skipped).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
import { describe, it, expect, vi, beforeEach } from 'vitest'
|
||||
import { runInboxSweep, PROCESSING_STUCK_MS } from '../lib/sweep'
|
||||
import { emptyResult } from '../lib/extract-invoice-fields'
|
||||
|
||||
/**
|
||||
* The staged-upload crash recovery: only rows that are (a) still
|
||||
* 'processing', (b) older than the stuck threshold, and (c) still carrying a
|
||||
* NULL extracted_data are flipped, and the flip writes the same empty
|
||||
* skeleton the synchronous swallow-failure path persists. No re-extraction.
|
||||
*/
|
||||
|
||||
interface Captured {
|
||||
selectFilters: Array<{ method: string; args: unknown[] }>
|
||||
updatePayload?: Record<string, unknown>
|
||||
updateFilters: Array<{ method: string; args: unknown[] }>
|
||||
}
|
||||
|
||||
function makeSupabase(opts: {
|
||||
staleIds: string[]
|
||||
claimedIds?: string[]
|
||||
selectError?: { message: string }
|
||||
updateError?: { message: string }
|
||||
}) {
|
||||
const captured: Captured = { selectFilters: [], updateFilters: [] }
|
||||
|
||||
const selectChain = {
|
||||
eq: vi.fn((...args: unknown[]) => {
|
||||
captured.selectFilters.push({ method: 'eq', args })
|
||||
return selectChain
|
||||
}),
|
||||
lt: vi.fn((...args: unknown[]) => {
|
||||
captured.selectFilters.push({ method: 'lt', args })
|
||||
return selectChain
|
||||
}),
|
||||
limit: vi.fn().mockResolvedValue(
|
||||
opts.selectError
|
||||
? { data: null, error: opts.selectError }
|
||||
: { data: opts.staleIds.map((id) => ({ id })), error: null },
|
||||
),
|
||||
}
|
||||
|
||||
const updateChain = {
|
||||
in: vi.fn((...args: unknown[]) => {
|
||||
captured.updateFilters.push({ method: 'in', args })
|
||||
return updateChain
|
||||
}),
|
||||
eq: vi.fn((...args: unknown[]) => {
|
||||
captured.updateFilters.push({ method: 'eq', args })
|
||||
return updateChain
|
||||
}),
|
||||
is: vi.fn((...args: unknown[]) => {
|
||||
captured.updateFilters.push({ method: 'is', args })
|
||||
return updateChain
|
||||
}),
|
||||
select: vi.fn().mockResolvedValue(
|
||||
opts.updateError
|
||||
? { data: null, error: opts.updateError }
|
||||
: { data: (opts.claimedIds ?? opts.staleIds).map((id) => ({ id })), error: null },
|
||||
),
|
||||
}
|
||||
|
||||
const supabase = {
|
||||
from: vi.fn(() => ({
|
||||
select: vi.fn(() => selectChain),
|
||||
update: vi.fn((payload: Record<string, unknown>) => {
|
||||
captured.updatePayload = payload
|
||||
return updateChain
|
||||
}),
|
||||
})),
|
||||
}
|
||||
|
||||
return { supabase: supabase as never, captured }
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
})
|
||||
|
||||
describe('runInboxSweep', () => {
|
||||
it('flips stale processing rows to received with the empty skeleton', async () => {
|
||||
const { supabase, captured } = makeSupabase({ staleIds: ['i1', 'i2'] })
|
||||
|
||||
const summary = await runInboxSweep(supabase)
|
||||
|
||||
expect(summary).toEqual({ flipped: 2 })
|
||||
// The stale scan targets processing rows older than the threshold.
|
||||
expect(captured.selectFilters).toEqual([
|
||||
{ method: 'eq', args: ['status', 'processing'] },
|
||||
{ method: 'lt', args: ['created_at', expect.any(String)] },
|
||||
])
|
||||
const cutoff = new Date(captured.selectFilters[1].args[1] as string).getTime()
|
||||
expect(Date.now() - cutoff).toBeGreaterThanOrEqual(PROCESSING_STUCK_MS - 1000)
|
||||
// The flip is a guarded claim: status still processing, data still NULL.
|
||||
expect(captured.updateFilters).toEqual([
|
||||
{ method: 'in', args: ['id', ['i1', 'i2']] },
|
||||
{ method: 'eq', args: ['status', 'processing'] },
|
||||
{ method: 'is', args: ['extracted_data', null] },
|
||||
])
|
||||
expect(captured.updatePayload).toEqual({
|
||||
status: 'received',
|
||||
extracted_data: emptyResult(),
|
||||
extraction_skipped: false,
|
||||
})
|
||||
})
|
||||
|
||||
it('counts only the rows the guarded update actually claimed', async () => {
|
||||
// A worker finished i2 between the scan and the flip: its result wins.
|
||||
const { supabase } = makeSupabase({ staleIds: ['i1', 'i2'], claimedIds: ['i1'] })
|
||||
|
||||
const summary = await runInboxSweep(supabase)
|
||||
|
||||
expect(summary).toEqual({ flipped: 1 })
|
||||
})
|
||||
|
||||
it('does nothing when no processing row is stale', async () => {
|
||||
const { supabase, captured } = makeSupabase({ staleIds: [] })
|
||||
|
||||
const summary = await runInboxSweep(supabase)
|
||||
|
||||
expect(summary).toEqual({ flipped: 0 })
|
||||
expect(captured.updatePayload).toBeUndefined()
|
||||
})
|
||||
|
||||
it('never throws: a failed select reports zero flips', async () => {
|
||||
const { supabase } = makeSupabase({ staleIds: [], selectError: { message: 'boom' } })
|
||||
|
||||
await expect(runInboxSweep(supabase)).resolves.toEqual({ flipped: 0 })
|
||||
})
|
||||
|
||||
it('never throws: a failed update reports zero flips', async () => {
|
||||
const { supabase } = makeSupabase({ staleIds: ['i1'], updateError: { message: 'boom' } })
|
||||
|
||||
await expect(runInboxSweep(supabase)).resolves.toEqual({ flipped: 0 })
|
||||
})
|
||||
})
|
||||
@@ -37,7 +37,16 @@ vi.mock('@/lib/entitlements/has-capability', async (importOriginal) => {
|
||||
return { ...actual, hasCapability: vi.fn().mockResolvedValue(true) }
|
||||
})
|
||||
|
||||
// The deferred extraction worker builds its own cookieless service client
|
||||
// (the request-scoped one may be gone once the response is flushed). Route
|
||||
// it to the same mock supabase so the CAS flip is observable.
|
||||
vi.mock('@/lib/auth/api-keys', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import('@/lib/auth/api-keys')>()
|
||||
return { ...actual, createServiceClientNoCookies: vi.fn() }
|
||||
})
|
||||
|
||||
import { extractInvoiceFields, emptyResult } from '@/extensions/general/invoice-inbox/lib/extract-invoice-fields'
|
||||
import { createServiceClientNoCookies } from '@/lib/auth/api-keys'
|
||||
|
||||
function findRoute(method: string, path: string) {
|
||||
return invoiceInboxExtension.apiRoutes!.find(
|
||||
@@ -47,10 +56,19 @@ function findRoute(method: string, path: string) {
|
||||
|
||||
const uploadRoute = findRoute('POST', '/upload')
|
||||
|
||||
// Build the supabase mock the upload handler needs:
|
||||
interface FlipCapture {
|
||||
payload?: Record<string, unknown>
|
||||
filters: Array<[string, unknown]>
|
||||
}
|
||||
|
||||
// Build the supabase mock the upload handler and the deferred worker need:
|
||||
// .from('invoice_inbox_items').insert(row).select('*').single() → { data: row, error: null }
|
||||
// .from('invoice_inbox_items').update(p).eq().eq().select('id') → CAS flip (captured)
|
||||
// .from('suppliers').select().eq().eq()... .maybeSingle() → { data: null }
|
||||
function makeSupabase(captured: { row?: Record<string, unknown> }) {
|
||||
function makeSupabase(
|
||||
captured: { row?: Record<string, unknown> },
|
||||
flip: FlipCapture = { filters: [] },
|
||||
) {
|
||||
const supplierChain = {
|
||||
select: vi.fn().mockReturnThis(),
|
||||
eq: vi.fn().mockReturnThis(),
|
||||
@@ -58,6 +76,13 @@ function makeSupabase(captured: { row?: Record<string, unknown> }) {
|
||||
limit: vi.fn().mockReturnThis(),
|
||||
maybeSingle: vi.fn().mockResolvedValue({ data: null }),
|
||||
}
|
||||
const updateChain = {
|
||||
eq: vi.fn((...args: [string, unknown]) => {
|
||||
flip.filters.push(args)
|
||||
return updateChain
|
||||
}),
|
||||
select: vi.fn().mockResolvedValue({ data: [{ id: 'inbox-1' }], error: null }),
|
||||
}
|
||||
const inboxChain = {
|
||||
insert: vi.fn((row: Record<string, unknown>) => {
|
||||
captured.row = row
|
||||
@@ -70,6 +95,10 @@ function makeSupabase(captured: { row?: Record<string, unknown> }) {
|
||||
}),
|
||||
}
|
||||
}),
|
||||
update: vi.fn((payload: Record<string, unknown>) => {
|
||||
flip.payload = payload
|
||||
return updateChain
|
||||
}),
|
||||
}
|
||||
return {
|
||||
from: vi.fn((table: string) => {
|
||||
@@ -120,10 +149,12 @@ beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
})
|
||||
|
||||
describe('POST /upload: page-count gate (issue #553)', () => {
|
||||
it('slices long PDFs to the first 3 pages and extracts, instead of skipping', async () => {
|
||||
describe('POST /upload: staged extraction + page-count gate (issue #553)', () => {
|
||||
it('defers long PDFs: responds processing, then slices to 3 pages and flips to received', async () => {
|
||||
const captured: { row?: Record<string, unknown> } = {}
|
||||
const supabase = makeSupabase(captured)
|
||||
const flip: FlipCapture = { filters: [] }
|
||||
const supabase = makeSupabase(captured, flip)
|
||||
vi.mocked(createServiceClientNoCookies).mockReturnValue(supabase as never)
|
||||
vi.mocked(extractInvoiceFields).mockResolvedValueOnce({
|
||||
data: emptyResult(),
|
||||
rawText: 'ok',
|
||||
@@ -131,28 +162,46 @@ describe('POST /upload: page-count gate (issue #553)', () => {
|
||||
|
||||
const req = await makeUploadRequest(6)
|
||||
const res = await uploadRoute.handler(req, buildCtx(supabase))
|
||||
const { status, body } = await parseJsonResponse<{
|
||||
data: { extracted_data?: { pages?: { total: number; analyzed: number } } } & Record<string, unknown>
|
||||
}>(res)
|
||||
const { status, body } = await parseJsonResponse<{ data: Record<string, unknown> }>(res)
|
||||
|
||||
// The response is the receipt ack: the row exists, extraction has not
|
||||
// landed, nothing is reported skipped.
|
||||
expect(status).toBe(200)
|
||||
expect(extractInvoiceFields).toHaveBeenCalledOnce()
|
||||
// The buffer handed to Bedrock is the sliced copy, not the original.
|
||||
const sentBuffer = vi.mocked(extractInvoiceFields).mock.calls[0][0].buffer
|
||||
const sentPdf = await PDFDocument.load(sentBuffer)
|
||||
expect(sentPdf.getPageCount()).toBe(3)
|
||||
// The row is a normal extracted row: the truncation is recorded in
|
||||
// extracted_data.pages rather than as a skip.
|
||||
expect(captured.row?.extraction_skipped).toBe(false)
|
||||
expect(body.data.status).toBe('processing')
|
||||
expect(body.data.extracted_data).toBeNull()
|
||||
expect(body.data.extraction_skipped).toBe(false)
|
||||
expect(body.data.skip_reason).toBeNull()
|
||||
expect(body.data.page_count).toBe(6)
|
||||
expect(body.data.extracted_data?.pages).toEqual({ total: 6, analyzed: 3 })
|
||||
expect(captured.row?.status).toBe('processing')
|
||||
expect(captured.row?.extracted_data).toBeNull()
|
||||
expect(captured.row?.extraction_skipped).toBe(false)
|
||||
|
||||
// The deferred worker extracts from the sliced copy, not the original.
|
||||
await vi.waitFor(() => expect(extractInvoiceFields).toHaveBeenCalledOnce())
|
||||
const sentBuffer = vi.mocked(extractInvoiceFields).mock.calls[0][0].buffer
|
||||
const sentPdf = await PDFDocument.load(sentBuffer)
|
||||
expect(sentPdf.getPageCount()).toBe(3)
|
||||
|
||||
// ...and CAS-flips the processing row to received, with the truncation
|
||||
// recorded in extracted_data.pages rather than as a skip.
|
||||
await vi.waitFor(() => expect(flip.payload).toBeDefined())
|
||||
expect(flip.payload?.status).toBe('received')
|
||||
expect(flip.payload?.extraction_skipped).toBe(false)
|
||||
expect((flip.payload?.extracted_data as { pages?: unknown })?.pages).toEqual({
|
||||
total: 6,
|
||||
analyzed: 3,
|
||||
})
|
||||
expect(flip.filters).toEqual([
|
||||
['id', 'inbox-1'],
|
||||
['status', 'processing'],
|
||||
])
|
||||
})
|
||||
|
||||
it('runs extraction normally for PDFs at or below the page-count limit', async () => {
|
||||
it('defers PDFs at or below the page-count limit and extracts the full buffer', async () => {
|
||||
const captured: { row?: Record<string, unknown> } = {}
|
||||
const supabase = makeSupabase(captured)
|
||||
const flip: FlipCapture = { filters: [] }
|
||||
const supabase = makeSupabase(captured, flip)
|
||||
vi.mocked(createServiceClientNoCookies).mockReturnValue(supabase as never)
|
||||
vi.mocked(extractInvoiceFields).mockResolvedValueOnce({
|
||||
data: emptyResult(),
|
||||
rawText: 'ok',
|
||||
@@ -163,14 +212,26 @@ describe('POST /upload: page-count gate (issue #553)', () => {
|
||||
const { status, body } = await parseJsonResponse<{ data: Record<string, unknown> }>(res)
|
||||
|
||||
expect(status).toBe(200)
|
||||
expect(extractInvoiceFields).toHaveBeenCalledOnce()
|
||||
expect(captured.row?.extraction_skipped).toBe(false)
|
||||
expect(body.data.status).toBe('processing')
|
||||
expect(body.data.extraction_skipped).toBe(false)
|
||||
expect(body.data.skip_reason).toBeNull()
|
||||
expect(body.data.page_count).toBe(2)
|
||||
|
||||
await vi.waitFor(() => expect(extractInvoiceFields).toHaveBeenCalledOnce())
|
||||
const sentBuffer = vi.mocked(extractInvoiceFields).mock.calls[0][0].buffer
|
||||
const sentPdf = await PDFDocument.load(sentBuffer)
|
||||
expect(sentPdf.getPageCount()).toBe(2)
|
||||
|
||||
await vi.waitFor(() => expect(flip.payload).toBeDefined())
|
||||
expect(flip.payload?.status).toBe('received')
|
||||
// No slice happened, so no pages truncation marker.
|
||||
expect((flip.payload?.extracted_data as { pages?: unknown })?.pages).toBeUndefined()
|
||||
})
|
||||
|
||||
it('honors client-side skip_extraction=true with skip_reason=client_opt_out', async () => {
|
||||
it('honors client-side skip_extraction=true synchronously with skip_reason=client_opt_out', async () => {
|
||||
// The BYO-extraction opt-out must stay on the synchronous path: the
|
||||
// caller PUTs its parsed fields right after upload, and a deferred flip
|
||||
// would overwrite them.
|
||||
const captured: { row?: Record<string, unknown> } = {}
|
||||
const supabase = makeSupabase(captured)
|
||||
|
||||
@@ -185,7 +246,9 @@ describe('POST /upload: page-count gate (issue #553)', () => {
|
||||
const { body } = await parseJsonResponse<{ data: Record<string, unknown> }>(res)
|
||||
|
||||
expect(extractInvoiceFields).not.toHaveBeenCalled()
|
||||
expect(body.data.status).toBe('received')
|
||||
expect(body.data.extraction_skipped).toBe(true)
|
||||
expect(body.data.skip_reason).toBe('client_opt_out')
|
||||
expect(captured.row?.status).toBe('received')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -189,10 +189,17 @@ export const invoiceInboxExtension: Extension = {
|
||||
if (!limit.ok) {
|
||||
return NextResponse.json(
|
||||
{
|
||||
error:
|
||||
limit.scope === 'minute'
|
||||
? 'För många uppladdningar på kort tid. Försök igen om en stund.'
|
||||
: 'Dagsgränsen för uppladdningar är nådd. Försök igen imorgon.',
|
||||
error: {
|
||||
code: 'RATE_LIMITED',
|
||||
message:
|
||||
limit.scope === 'minute'
|
||||
? 'För många uppladdningar på kort tid. Försök igen om en stund.'
|
||||
: 'Dagsgränsen för uppladdningar är nådd. Försök igen imorgon.',
|
||||
message_en:
|
||||
limit.scope === 'minute'
|
||||
? 'Too many uploads in a short time. Try again in a moment.'
|
||||
: 'The daily upload limit has been reached. Try again tomorrow.',
|
||||
},
|
||||
retry_after: limit.retryAfterSec,
|
||||
},
|
||||
{ status: 429, headers: { 'Retry-After': String(limit.retryAfterSec ?? 60) } },
|
||||
@@ -213,15 +220,18 @@ export const invoiceInboxExtension: Extension = {
|
||||
formData.get('skip_extraction') === 'true' ||
|
||||
formData.get('skip_extraction') === '1'
|
||||
|
||||
if (!file) return NextResponse.json({ error: 'No file provided' }, { status: 400 })
|
||||
if (!file) return errorResponseFromCode('INBOX_UPLOAD_NO_FILE', ctx.log)
|
||||
if (file.size > MAX_FILE_SIZE) {
|
||||
return NextResponse.json({ error: `File too large (max ${MAX_FILE_SIZE / 1024 / 1024} MB)` }, { status: 400 })
|
||||
return errorResponseFromCode('INBOX_UPLOAD_TOO_LARGE', ctx.log, {
|
||||
messageSv: `Filen är för stor. Maxstorlek är ${MAX_FILE_SIZE / 1024 / 1024} MB.`,
|
||||
messageEn: `File exceeds the ${MAX_FILE_SIZE / 1024 / 1024} MB size limit.`,
|
||||
})
|
||||
}
|
||||
if (!UPLOAD_ALLOWED_MIME_TYPES.has(file.type)) {
|
||||
return NextResponse.json(
|
||||
{ error: `Unsupported file type: ${file.type}. Allowed: PDF, JPEG, PNG, HEIC, WebP` },
|
||||
{ status: 400 }
|
||||
)
|
||||
return errorResponseFromCode('INBOX_UPLOAD_UNSUPPORTED_TYPE', ctx.log, {
|
||||
messageSv: `Filtypen stöds inte: ${file.type || 'okänd'}. Tillåtna format: PDF, JPEG, PNG, HEIC och WebP.`,
|
||||
messageEn: `Unsupported file type: ${file.type || 'unknown'}. Allowed: PDF, JPEG, PNG, HEIC, WebP.`,
|
||||
})
|
||||
}
|
||||
|
||||
// Validate matched_transaction_id belongs to this company before we
|
||||
@@ -240,15 +250,18 @@ export const invoiceInboxExtension: Extension = {
|
||||
return NextResponse.json({ error: txErr.message }, { status: 500 })
|
||||
}
|
||||
if (!tx) {
|
||||
return NextResponse.json(
|
||||
{ error: 'matched_transaction_id refers to a transaction outside this company.' },
|
||||
{ status: 400 },
|
||||
)
|
||||
return errorResponseFromCode('INBOX_UPLOAD_TX_NOT_IN_COMPANY', ctx.log)
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const buffer = await file.arrayBuffer()
|
||||
// Staged upload: the row lands as 'processing' and the response
|
||||
// returns before Bedrock runs; the deferred worker (or the sweep
|
||||
// cron) flips it to 'received'. Only this web route defers: email
|
||||
// and WhatsApp ingestion already run in background contexts and
|
||||
// consume the returned item synchronously. When skipExtraction is
|
||||
// set the call stays synchronous (see uploadAndExtract).
|
||||
const result = await uploadAndExtract(
|
||||
ctx.supabase,
|
||||
ctx.userId,
|
||||
@@ -257,13 +270,24 @@ export const invoiceInboxExtension: Extension = {
|
||||
'upload',
|
||||
undefined,
|
||||
matchedTransactionId,
|
||||
{ skipExtraction },
|
||||
{ skipExtraction, deferExtraction: true },
|
||||
)
|
||||
return NextResponse.json({ data: result })
|
||||
} catch (error) {
|
||||
console.error('[invoice-inbox/upload] Failed:', error)
|
||||
// Pass the thrown message through: the document service raises its
|
||||
// user-facing failures (magic-byte mismatch, corrupt content) in
|
||||
// Swedish, and getErrorMessage substitutes the registry copy for
|
||||
// this code when the message is internal English.
|
||||
const message = error instanceof Error && error.message.trim() ? error.message : null
|
||||
return NextResponse.json(
|
||||
{ error: error instanceof Error ? error.message : 'Upload failed' },
|
||||
{
|
||||
error: {
|
||||
code: 'INBOX_UPLOAD_FAILED',
|
||||
message: message ?? 'Uppladdningen misslyckades. Försök igen.',
|
||||
message_en: message ?? 'Upload failed.',
|
||||
},
|
||||
},
|
||||
{ status: 500 }
|
||||
)
|
||||
}
|
||||
@@ -651,15 +675,18 @@ export const invoiceInboxExtension: Extension = {
|
||||
|
||||
const formData = await request.formData()
|
||||
const file = formData.get('file') as File | null
|
||||
if (!file) return NextResponse.json({ error: 'No file provided' }, { status: 400 })
|
||||
if (!file) return errorResponseFromCode('INBOX_UPLOAD_NO_FILE', ctx.log)
|
||||
if (file.size > MAX_FILE_SIZE) {
|
||||
return NextResponse.json({ error: `File too large (max ${MAX_FILE_SIZE / 1024 / 1024} MB)` }, { status: 400 })
|
||||
return errorResponseFromCode('INBOX_UPLOAD_TOO_LARGE', ctx.log, {
|
||||
messageSv: `Filen är för stor. Maxstorlek är ${MAX_FILE_SIZE / 1024 / 1024} MB.`,
|
||||
messageEn: `File exceeds the ${MAX_FILE_SIZE / 1024 / 1024} MB size limit.`,
|
||||
})
|
||||
}
|
||||
if (!UPLOAD_ALLOWED_MIME_TYPES.has(file.type)) {
|
||||
return NextResponse.json(
|
||||
{ error: `Unsupported file type: ${file.type}. Allowed: PDF, JPEG, PNG, HEIC, WebP` },
|
||||
{ status: 400 }
|
||||
)
|
||||
return errorResponseFromCode('INBOX_UPLOAD_UNSUPPORTED_TYPE', ctx.log, {
|
||||
messageSv: `Filtypen stöds inte: ${file.type || 'okänd'}. Tillåtna format: PDF, JPEG, PNG, HEIC och WebP.`,
|
||||
messageEn: `Unsupported file type: ${file.type || 'unknown'}. Allowed: PDF, JPEG, PNG, HEIC, WebP.`,
|
||||
})
|
||||
}
|
||||
|
||||
const { data: item } = await ctx.supabase
|
||||
@@ -774,8 +801,16 @@ export const invoiceInboxExtension: Extension = {
|
||||
})
|
||||
} catch (error) {
|
||||
console.error('[invoice-inbox/attach-document] Failed:', error)
|
||||
// Same pass-through rationale as the /upload catch above.
|
||||
const message = error instanceof Error && error.message.trim() ? error.message : null
|
||||
return NextResponse.json(
|
||||
{ error: error instanceof Error ? error.message : 'Attach failed' },
|
||||
{
|
||||
error: {
|
||||
code: 'INBOX_ATTACH_FAILED',
|
||||
message: message ?? 'Bilagan kunde inte kopplas. Försök igen.',
|
||||
message_en: message ?? 'Failed to attach the document.',
|
||||
},
|
||||
},
|
||||
{ status: 500 }
|
||||
)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
/**
|
||||
* Crash recovery for the staged upload.
|
||||
*
|
||||
* The web upload route inserts the inbox row as 'processing' and defers
|
||||
* Bedrock extraction to an after() worker that can die with the serverless
|
||||
* instance. A row stuck in 'processing' is durable state with no live owner:
|
||||
* this sweep flips it to 'received' with the empty extraction skeleton so it
|
||||
* becomes a normal manually-editable item. Deliberately NO re-extraction
|
||||
* here: the UI retry button covers that, and a cron that silently re-spends
|
||||
* Bedrock tokens on every crash would hide the crashes.
|
||||
*
|
||||
* Overlap with a slow live worker is safe: every mutation is a guarded claim
|
||||
* on status='processing' (and extracted_data still NULL), so the sweep and
|
||||
* the worker never both win one row.
|
||||
*/
|
||||
|
||||
import type { SupabaseClient } from '@supabase/supabase-js'
|
||||
import { createLogger } from '@/lib/logger'
|
||||
import { emptyResult } from './extract-invoice-fields'
|
||||
|
||||
const log = createLogger('invoice-inbox/sweep')
|
||||
|
||||
/**
|
||||
* A deferred extraction is one Bedrock call (the WhatsApp cron budgets
|
||||
* 10-60s for the same call). Two minutes of silence means no live worker
|
||||
* can still deliver a flip that beats the sweep by enough to matter.
|
||||
*/
|
||||
export const PROCESSING_STUCK_MS = 2 * 60 * 1000
|
||||
const BATCH = 50
|
||||
|
||||
export interface InboxSweepSummary {
|
||||
/** Stale 'processing' rows flipped to 'received' with the empty skeleton. */
|
||||
flipped: number
|
||||
}
|
||||
|
||||
/** Run one sweep pass. Never throws. */
|
||||
export async function runInboxSweep(supabase: SupabaseClient): Promise<InboxSweepSummary> {
|
||||
const cutoff = new Date(Date.now() - PROCESSING_STUCK_MS).toISOString()
|
||||
|
||||
const { data: stale, error: selectError } = await supabase
|
||||
.from('invoice_inbox_items')
|
||||
.select('id')
|
||||
.eq('status', 'processing')
|
||||
.lt('created_at', cutoff)
|
||||
.limit(BATCH)
|
||||
if (selectError) {
|
||||
log.error('stale-processing select failed', { error: selectError.message })
|
||||
return { flipped: 0 }
|
||||
}
|
||||
const ids = ((stale ?? []) as Array<{ id: string }>).map((r) => r.id)
|
||||
if (ids.length === 0) return { flipped: 0 }
|
||||
|
||||
// CAS: the status guard keeps a just-finished worker's real result, and
|
||||
// the extracted_data-still-NULL guard keeps any fields a caller PUT onto
|
||||
// the row in the meantime; a row that fails either guard is someone
|
||||
// else's win, not ours.
|
||||
const { data: claimed, error: updateError } = await supabase
|
||||
.from('invoice_inbox_items')
|
||||
.update({
|
||||
status: 'received',
|
||||
extracted_data: emptyResult() as unknown as Record<string, unknown>,
|
||||
extraction_skipped: false,
|
||||
})
|
||||
.in('id', ids)
|
||||
.eq('status', 'processing')
|
||||
.is('extracted_data', null)
|
||||
.select('id')
|
||||
if (updateError) {
|
||||
log.error('stale-processing flip failed', { error: updateError.message })
|
||||
return { flipped: 0 }
|
||||
}
|
||||
|
||||
const flipped = Array.isArray(claimed) ? claimed.length : 0
|
||||
if (flipped > 0) {
|
||||
log.info('flipped stale processing rows to received', { flipped })
|
||||
}
|
||||
return { flipped }
|
||||
}
|
||||
@@ -1,8 +1,11 @@
|
||||
import { after } from 'next/server'
|
||||
import { uploadDocument } from '@/lib/core/documents/document-service'
|
||||
import { extractInvoiceFields, emptyResult } from './extract-invoice-fields'
|
||||
import { hasCapability } from '@/lib/entitlements/has-capability'
|
||||
import { CAPABILITY } from '@/lib/entitlements/keys'
|
||||
import { appendProcessingHistory } from '@/lib/processing-history/append'
|
||||
import { createServiceClientNoCookies } from '@/lib/auth/api-keys'
|
||||
import type { InvoiceExtractionResult } from '@/types'
|
||||
import { PDFDocument } from 'pdf-lib'
|
||||
import path from 'node:path'
|
||||
|
||||
@@ -199,6 +202,20 @@ export async function uploadAndExtract(
|
||||
/** Overrides the system actor id on the DocumentIngested history event.
|
||||
* Omitted = today's behavior (resend-inbound for email, user otherwise). */
|
||||
actorId?: string
|
||||
/**
|
||||
* Staged upload (web route only). When true AND extraction would actually
|
||||
* call Bedrock, the inbox row is inserted first (status 'processing',
|
||||
* extracted_data NULL), the function returns immediately, and extraction
|
||||
* runs after the response via a deferred worker that CAS-flips the row to
|
||||
* 'received'. Verdicts that never reach Bedrock (no AI entitlement,
|
||||
* sandbox, client opt-out) stay on the synchronous path: they are quick
|
||||
* and their response contract (empty skeleton + skip_reason) is
|
||||
* unchanged. skipExtraction in particular MUST stay synchronous: a
|
||||
* BYO-extraction agent PUTs its fields right after upload, and a deferred
|
||||
* flip would overwrite them. Default false = today's synchronous behavior
|
||||
* (email and WhatsApp callers are untouched).
|
||||
*/
|
||||
deferExtraction?: boolean
|
||||
} = {},
|
||||
) {
|
||||
const correlationId = crypto.randomUUID()
|
||||
@@ -319,27 +336,96 @@ export async function uploadAndExtract(
|
||||
// extraction path, so the document is still stored and can be filled in
|
||||
// manually. Highest priority (a hard paywall rule, not a heuristic).
|
||||
const hasAiEntitlement = await hasCapability(supabase, companyId, CAPABILITY.ai)
|
||||
// Long PDFs are sliced to their first pages instead of skipped, but only
|
||||
// when extraction would actually run: slicing after an entitlement/sandbox/
|
||||
// opt-out verdict would be wasted CPU.
|
||||
const slicedBuffer =
|
||||
gatedByPageCount && hasAiEntitlement && !sandbox && !opts.skipExtraction
|
||||
? await slicePdfForExtraction(file.buffer, MAX_PAGES_FOR_AUTO_EXTRACT)
|
||||
: null
|
||||
// Skip-reason priority: no-AI-entitlement > sandbox > client opt-out >
|
||||
// page-count. Opt-out now outranks the page gate (an opted-out caller never
|
||||
// extracts regardless of length), and too_many_pages only fires when the
|
||||
// slice fallback also failed (encrypted/malformed PDF).
|
||||
const skipReason: 'no_ai_entitlement' | 'too_many_pages' | 'client_opt_out' | 'sandbox' | null =
|
||||
// The verdicts that avoid Bedrock entirely, in the established priority
|
||||
// order: no-AI-entitlement > sandbox > client opt-out. All three are
|
||||
// decidable without touching the PDF; too_many_pages is not (it only fires
|
||||
// when the slice fallback fails) and is resolved below, on whichever path
|
||||
// (sync or deferred) actually attempts the slice.
|
||||
const syncSkipReason: 'no_ai_entitlement' | 'client_opt_out' | 'sandbox' | null =
|
||||
!hasAiEntitlement
|
||||
? 'no_ai_entitlement'
|
||||
: sandbox
|
||||
? 'sandbox'
|
||||
: opts.skipExtraction
|
||||
? 'client_opt_out'
|
||||
: gatedByPageCount && slicedBuffer == null
|
||||
? 'too_many_pages'
|
||||
: null
|
||||
: null
|
||||
|
||||
if (opts.deferExtraction && syncSkipReason === null) {
|
||||
// Staged path: extraction WILL call Bedrock, so create the row now and
|
||||
// let the response go. The deferred worker (or, after a crash, the sweep
|
||||
// cron at /api/extensions/invoice-inbox/sweep/cron) flips the row to
|
||||
// 'received'. extracted_data stays NULL until then: that is what keeps
|
||||
// the MCP create-from-inbox guard ("re-run extraction first") holding
|
||||
// for free on in-flight rows.
|
||||
const { data: inbox, error: inboxError } = await supabase
|
||||
.from('invoice_inbox_items')
|
||||
.insert({
|
||||
company_id: companyId,
|
||||
user_id: userId,
|
||||
status: 'processing',
|
||||
source,
|
||||
document_id: doc.id,
|
||||
extracted_data: null,
|
||||
extraction_skipped: false,
|
||||
matched_supplier_id: null,
|
||||
email_from: emailMeta?.from || null,
|
||||
email_subject: emailMeta?.subject || null,
|
||||
email_received_at: emailMeta?.receivedAt || null,
|
||||
email_body_text: emailMeta?.bodyText || null,
|
||||
resend_email_id: emailMeta?.resendEmailId || null,
|
||||
resend_attachment_id: emailMeta?.resendAttachmentId || null,
|
||||
raw_email_payload: emailMeta?.messageId
|
||||
? { messageId: emailMeta.messageId, filename: file.name }
|
||||
: null,
|
||||
correlation_id: correlationId,
|
||||
matched_transaction_id: matchedTransactionId ?? null,
|
||||
whatsapp_message_id: opts.channelMeta?.whatsappMessageId ?? null,
|
||||
channel_context: opts.channelMeta
|
||||
? { channel: 'whatsapp', caption: sanitiseCaption(opts.channelMeta.caption) }
|
||||
: null,
|
||||
})
|
||||
.select('*')
|
||||
.single()
|
||||
|
||||
if (inboxError) throw new Error(`Failed to create inbox item: ${inboxError.message}`)
|
||||
|
||||
scheduleDeferredExtraction({
|
||||
itemId: inbox.id,
|
||||
documentId: doc.id,
|
||||
companyId,
|
||||
correlationId,
|
||||
file,
|
||||
pageCount,
|
||||
gatedByPageCount,
|
||||
})
|
||||
|
||||
return {
|
||||
document_id: doc.id,
|
||||
inbox_item_id: inbox.id,
|
||||
status: inbox.status,
|
||||
extracted_data: null,
|
||||
matched_supplier_id: inbox.matched_supplier_id,
|
||||
matched_transaction_id: inbox.matched_transaction_id,
|
||||
extraction_skipped: false,
|
||||
skip_reason: null,
|
||||
page_count: pageCount,
|
||||
}
|
||||
}
|
||||
|
||||
// Long PDFs are sliced to their first pages instead of skipped, but only
|
||||
// when extraction would actually run: slicing after an entitlement/sandbox/
|
||||
// opt-out verdict would be wasted CPU.
|
||||
const slicedBuffer =
|
||||
gatedByPageCount && syncSkipReason === null
|
||||
? await slicePdfForExtraction(file.buffer, MAX_PAGES_FOR_AUTO_EXTRACT)
|
||||
: null
|
||||
// Skip-reason priority: no-AI-entitlement > sandbox > client opt-out >
|
||||
// page-count. Opt-out outranks the page gate (an opted-out caller never
|
||||
// extracts regardless of length), and too_many_pages only fires when the
|
||||
// slice fallback also failed (encrypted/malformed PDF).
|
||||
const skipReason: 'no_ai_entitlement' | 'too_many_pages' | 'client_opt_out' | 'sandbox' | null =
|
||||
syncSkipReason ??
|
||||
(gatedByPageCount && slicedBuffer == null ? 'too_many_pages' : null)
|
||||
const skipExtraction = skipReason !== null
|
||||
|
||||
// Bring-your-own-extraction: skip the Bedrock call entirely and seed an
|
||||
@@ -453,3 +539,155 @@ export async function uploadAndExtract(
|
||||
page_count: pageCount,
|
||||
}
|
||||
}
|
||||
|
||||
// ── Deferred extraction (staged web upload) ──────────────────
|
||||
|
||||
interface DeferredExtractionJob {
|
||||
itemId: string
|
||||
documentId: string
|
||||
companyId: string
|
||||
correlationId: string
|
||||
file: { name: string; buffer: ArrayBuffer; type: string }
|
||||
pageCount: number | null
|
||||
gatedByPageCount: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Run extraction after the upload response is sent, then CAS-flip the
|
||||
* 'processing' row to 'received'. Exact dispatch-kick idiom (whatsapp-inbox
|
||||
* kickInboundProcessing): never awaited, never throws out, falls back to a
|
||||
* microtask outside a request scope (tests). The request-scoped supabase
|
||||
* client may be gone once the response is flushed, so the worker builds its
|
||||
* own cookieless service client.
|
||||
*/
|
||||
function scheduleDeferredExtraction(job: DeferredExtractionJob): void {
|
||||
const run = async (): Promise<void> => {
|
||||
try {
|
||||
const supabase = createServiceClientNoCookies()
|
||||
|
||||
// Mirrors the synchronous path: extraction failures are swallowed into
|
||||
// the empty skeleton (extractInvoiceFields does most of that itself);
|
||||
// an unsliceable long PDF becomes the too_many_pages skip.
|
||||
let extracted: InvoiceExtractionResult = emptyResult()
|
||||
let rawText: string | null = null
|
||||
let extractionSkipped = false
|
||||
let skipReason: 'too_many_pages' | null = null
|
||||
try {
|
||||
const slicedBuffer = job.gatedByPageCount
|
||||
? await slicePdfForExtraction(job.file.buffer, MAX_PAGES_FOR_AUTO_EXTRACT)
|
||||
: null
|
||||
if (job.gatedByPageCount && slicedBuffer == null) {
|
||||
extractionSkipped = true
|
||||
skipReason = 'too_many_pages'
|
||||
} else {
|
||||
const result = await extractInvoiceFields({
|
||||
buffer: Buffer.from(slicedBuffer ?? job.file.buffer),
|
||||
mimeType: job.file.type,
|
||||
fileName: job.file.name,
|
||||
})
|
||||
extracted = result.data
|
||||
rawText = result.rawText
|
||||
if (slicedBuffer != null && job.pageCount != null) {
|
||||
extracted.pages = { total: job.pageCount, analyzed: MAX_PAGES_FOR_AUTO_EXTRACT }
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
// Persist the empty skeleton below, exactly like the sync path does
|
||||
// on a swallowed failure: the row becomes a normal received item with
|
||||
// empty fields, and the manual-edit / retry affordances take over.
|
||||
console.error('[invoice-inbox] Deferred extraction failed:', err)
|
||||
}
|
||||
|
||||
// Supplier match by org-nr, then case-insensitive name (no AI fuzz).
|
||||
let matchedSupplierId: string | null = null
|
||||
if (extracted.supplier.orgNumber) {
|
||||
const { data: s } = await supabase
|
||||
.from('suppliers')
|
||||
.select('id')
|
||||
.eq('company_id', job.companyId)
|
||||
.eq('org_number', extracted.supplier.orgNumber)
|
||||
.limit(1)
|
||||
.maybeSingle()
|
||||
if (s) matchedSupplierId = s.id
|
||||
}
|
||||
if (!matchedSupplierId && extracted.supplier.name) {
|
||||
const { data: s } = await supabase
|
||||
.from('suppliers')
|
||||
.select('id')
|
||||
.eq('company_id', job.companyId)
|
||||
.ilike('name', extracted.supplier.name)
|
||||
.limit(1)
|
||||
.maybeSingle()
|
||||
if (s) matchedSupplierId = s.id
|
||||
}
|
||||
|
||||
// CAS: only the row still waiting on THIS worker flips. A user retry
|
||||
// or the sweep cron may have claimed it first; their result wins.
|
||||
const { data: claimed, error: updateError } = await supabase
|
||||
.from('invoice_inbox_items')
|
||||
.update({
|
||||
status: 'received',
|
||||
extracted_data: extracted as unknown as Record<string, unknown>,
|
||||
extraction_skipped: extractionSkipped,
|
||||
matched_supplier_id: matchedSupplierId,
|
||||
})
|
||||
.eq('id', job.itemId)
|
||||
.eq('status', 'processing')
|
||||
.select('id')
|
||||
if (updateError) {
|
||||
// The sweep cron flips the row to 'received' once it goes stale.
|
||||
console.error('[invoice-inbox] Deferred extraction flip failed:', updateError)
|
||||
return
|
||||
}
|
||||
if (!Array.isArray(claimed) || claimed.length === 0) return
|
||||
|
||||
try {
|
||||
await appendProcessingHistory({
|
||||
companyId: job.companyId,
|
||||
correlationId: job.correlationId,
|
||||
aggregateType: 'Document',
|
||||
aggregateId: job.documentId,
|
||||
eventType: 'DocumentExtractionAttempted',
|
||||
payload: {
|
||||
document_id: job.documentId,
|
||||
inbox_item_id: job.itemId,
|
||||
succeeded: rawText != null && rawText.length > 0,
|
||||
extracted_total: extracted.totals.total,
|
||||
has_org_number: extracted.supplier.orgNumber != null,
|
||||
has_ocr: extracted.invoice.paymentReference != null,
|
||||
skipped: extractionSkipped,
|
||||
skip_reason: skipReason,
|
||||
page_count: job.pageCount,
|
||||
},
|
||||
actor: { type: 'system', id: 'invoice-inbox-extract' },
|
||||
occurredAt: new Date(),
|
||||
})
|
||||
} catch (err) {
|
||||
console.error('[invoice-inbox] Failed to append DocumentExtractionAttempted:', err)
|
||||
}
|
||||
} catch (err) {
|
||||
console.error('[invoice-inbox] Deferred extraction worker crashed:', err)
|
||||
// Best-effort flip so the row does not sit in 'processing' until the
|
||||
// sweep: same CAS, same empty skeleton the sync swallow persists.
|
||||
try {
|
||||
await createServiceClientNoCookies()
|
||||
.from('invoice_inbox_items')
|
||||
.update({
|
||||
status: 'received',
|
||||
extracted_data: emptyResult() as unknown as Record<string, unknown>,
|
||||
extraction_skipped: false,
|
||||
})
|
||||
.eq('id', job.itemId)
|
||||
.eq('status', 'processing')
|
||||
} catch {
|
||||
// The sweep cron is the recovery of last resort.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
after(() => run())
|
||||
} catch {
|
||||
queueMicrotask(() => void run())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -24,7 +24,10 @@ import {
|
||||
askCompanyQuestion,
|
||||
applyCompanyChoice,
|
||||
} from '@/extensions/general/whatsapp-inbox/lib/company-question'
|
||||
import { STAGED_AWAITING_COMPANY } from '@/extensions/general/whatsapp-inbox/lib/conversation'
|
||||
import {
|
||||
NO_COMPANY_OPTIONS,
|
||||
STAGED_AWAITING_COMPANY,
|
||||
} from '@/extensions/general/whatsapp-inbox/lib/conversation'
|
||||
import { TEMPLATE } from '@/extensions/general/whatsapp-inbox/lib/messages'
|
||||
|
||||
const sendTextMock = vi.mocked(sendText)
|
||||
@@ -103,7 +106,7 @@ describe('askCompanyQuestion', () => {
|
||||
stagedCount: 1,
|
||||
})
|
||||
|
||||
expect(asked).toBe(true)
|
||||
expect(asked).toBe('asked')
|
||||
expect(sendButtonsMock).toHaveBeenCalledTimes(1)
|
||||
const args = sendButtonsMock.mock.calls[0][1]
|
||||
expect(args.buttons).toHaveLength(3)
|
||||
@@ -175,11 +178,84 @@ describe('askCompanyQuestion', () => {
|
||||
stagedCount: 1,
|
||||
})
|
||||
|
||||
expect(asked).toBe(false)
|
||||
expect(asked).toBe('not_asked')
|
||||
expect(sendButtonsMock).not.toHaveBeenCalled()
|
||||
expect(sendListMock).not.toHaveBeenCalled()
|
||||
expect(sendTextMock).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('treats a failed membership query as transient: nothing sent, nothing marked', async () => {
|
||||
const { supabase, enqueue, findCalls } = createQueuedMockSupabase()
|
||||
enqueue({ error: { message: 'connection reset' } }) // memberships query fails
|
||||
|
||||
const asked = await askCompanyQuestion(supabase as unknown as SupabaseClient, {
|
||||
conversation: makeConversation(),
|
||||
link: makeLink(),
|
||||
to: '46701234567',
|
||||
replyBase,
|
||||
stagedCount: 1,
|
||||
})
|
||||
|
||||
expect(asked).toBe('transient_error')
|
||||
expect(sendButtonsMock).not.toHaveBeenCalled()
|
||||
expect(sendListMock).not.toHaveBeenCalled()
|
||||
expect(sendTextMock).not.toHaveBeenCalled()
|
||||
// Neither the question state nor the staged rows were touched: the
|
||||
// CALLER releases its row for retry.
|
||||
expect(findCalls('whatsapp_conversations', 'update')).toHaveLength(0)
|
||||
expect(findCalls('whatsapp_messages', 'update')).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('genuinely zero options: re-marks the staged rows and sends ONE actionable M19', async () => {
|
||||
const { supabase, enqueue, findCalls, calls } = createQueuedMockSupabase()
|
||||
enqueue({ data: [] }) // memberships: really none
|
||||
enqueue({ data: null }) // staged rows re-marked no_company_options
|
||||
enqueue({ data: null }) // M19 burst dedupe: none sent yet
|
||||
|
||||
const asked = await askCompanyQuestion(supabase as unknown as SupabaseClient, {
|
||||
conversation: makeConversation(),
|
||||
link: makeLink(),
|
||||
to: '46701234567',
|
||||
replyBase,
|
||||
stagedCount: 1,
|
||||
})
|
||||
|
||||
expect(asked).toBe('no_options')
|
||||
const marker = findCalls('whatsapp_messages', 'update')[0][0] as Record<string, unknown>
|
||||
expect(marker.error_message).toBe(NO_COMPANY_OPTIONS)
|
||||
// The re-mark targets exactly the rows parked behind this episode.
|
||||
expect(
|
||||
calls.some(
|
||||
(c) =>
|
||||
c.table === 'whatsapp_messages' &&
|
||||
c.method === 'eq' &&
|
||||
c.args[0] === 'error_message' &&
|
||||
c.args[1] === STAGED_AWAITING_COMPANY,
|
||||
),
|
||||
).toBe(true)
|
||||
expect(sendTextMock).toHaveBeenCalledTimes(1)
|
||||
expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m19NoCompany)
|
||||
// No question is ever parked: there is nothing to answer.
|
||||
expect(findCalls('whatsapp_conversations', 'update')).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('suppresses a repeat M19 inside the burst window', async () => {
|
||||
const { supabase, enqueue } = createQueuedMockSupabase()
|
||||
enqueue({ data: [] }) // memberships: none
|
||||
enqueue({ data: null }) // staged rows re-mark
|
||||
enqueue({ data: { id: 'm19-earlier' } }) // an M19 already went out
|
||||
|
||||
const asked = await askCompanyQuestion(supabase as unknown as SupabaseClient, {
|
||||
conversation: makeConversation(),
|
||||
link: makeLink(),
|
||||
to: '46701234567',
|
||||
replyBase,
|
||||
stagedCount: 2,
|
||||
})
|
||||
|
||||
expect(asked).toBe('no_options')
|
||||
expect(sendTextMock).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
describe('applyCompanyChoice', () => {
|
||||
|
||||
@@ -161,7 +161,7 @@ describe('company question is not one-shot when the send fails', () => {
|
||||
stagedCount: 3,
|
||||
})
|
||||
|
||||
expect(asked).toBe(false)
|
||||
expect(asked).toBe('not_asked')
|
||||
const updates = mock
|
||||
.findCalls('whatsapp_conversations', 'update')
|
||||
.map((args) => args[0] as { state?: string; context?: Record<string, unknown> })
|
||||
@@ -184,7 +184,7 @@ describe('company question is not one-shot when the send fails', () => {
|
||||
stagedCount: 1,
|
||||
})
|
||||
|
||||
expect(asked).toBe(true)
|
||||
expect(asked).toBe('asked')
|
||||
expect(mock.findCalls('whatsapp_conversations', 'update')).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -0,0 +1,168 @@
|
||||
/**
|
||||
* Route tests for the authenticated /link endpoints: unmute from the app
|
||||
* (the in-chat counterpart is the literal text `start`) and the health
|
||||
* counters GET /link exposes for the settings panel.
|
||||
*/
|
||||
import { describe, it, expect, vi, beforeEach } from 'vitest'
|
||||
import { createQueuedMockSupabase } from '@/tests/helpers'
|
||||
import type { ExtensionContext } from '@/lib/extensions/types'
|
||||
|
||||
vi.mock('@supabase/supabase-js', () => ({
|
||||
createClient: vi.fn(),
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/supabase/server', () => ({
|
||||
createServiceClient: vi.fn(),
|
||||
}))
|
||||
|
||||
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', errorDetail: null }),
|
||||
getDisplayPhoneNumber: vi.fn().mockResolvedValue(null),
|
||||
}
|
||||
})
|
||||
|
||||
vi.mock('@/extensions/general/whatsapp-inbox/lib/process-inbound', () => ({
|
||||
kickInboundProcessing: vi.fn(),
|
||||
}))
|
||||
|
||||
import { createServiceClient } from '@/lib/supabase/server'
|
||||
import { whatsappInboxExtension } from '@/extensions/general/whatsapp-inbox'
|
||||
|
||||
function findRoute(method: string, path: string) {
|
||||
return whatsappInboxExtension.apiRoutes!.find((r) => r.method === method && r.path === path)!
|
||||
}
|
||||
|
||||
function makeCtx(supabase: unknown): ExtensionContext {
|
||||
return {
|
||||
userId: 'user-1',
|
||||
companyId: 'company-1',
|
||||
extensionId: 'whatsapp-inbox',
|
||||
supabase,
|
||||
} as unknown as ExtensionContext
|
||||
}
|
||||
|
||||
function request(path: string, method = 'POST'): Request {
|
||||
return new Request(`http://localhost:3000/api/extensions/ext/whatsapp-inbox${path}`, { method })
|
||||
}
|
||||
|
||||
describe('POST /link/unmute', () => {
|
||||
const route = findRoute('POST', '/link/unmute')
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
})
|
||||
|
||||
it('401s without an authenticated context', async () => {
|
||||
const response = await route.handler(request('/link/unmute'))
|
||||
expect(response.status).toBe(401)
|
||||
})
|
||||
|
||||
it('clears muted_at on the caller own active link', async () => {
|
||||
const { supabase, enqueue, findCall, calls } = createQueuedMockSupabase()
|
||||
enqueue({ data: [{ id: 'link-1' }] }) // guarded update matched the row
|
||||
|
||||
const response = await route.handler(request('/link/unmute'), makeCtx(supabase))
|
||||
expect(response.status).toBe(200)
|
||||
const body = (await response.json()) as { data: { unmuted: boolean } }
|
||||
expect(body.data).toEqual({ unmuted: true })
|
||||
|
||||
const [patch] = findCall('whatsapp_phone_links', 'update') as [Record<string, unknown>]
|
||||
expect(patch).toEqual({ muted_at: null })
|
||||
// Scoped to the caller's own active link.
|
||||
expect(
|
||||
calls.some(
|
||||
(c) =>
|
||||
c.table === 'whatsapp_phone_links' &&
|
||||
c.method === 'eq' &&
|
||||
c.args[0] === 'user_id' &&
|
||||
c.args[1] === 'user-1',
|
||||
),
|
||||
).toBe(true)
|
||||
expect(
|
||||
calls.some(
|
||||
(c) =>
|
||||
c.table === 'whatsapp_phone_links' && c.method === 'is' && c.args[0] === 'revoked_at',
|
||||
),
|
||||
).toBe(true)
|
||||
})
|
||||
|
||||
it('404s when there is no active link', async () => {
|
||||
const { supabase, enqueue } = createQueuedMockSupabase()
|
||||
enqueue({ data: [] }) // no row matched
|
||||
|
||||
const response = await route.handler(request('/link/unmute'), makeCtx(supabase))
|
||||
expect(response.status).toBe(404)
|
||||
})
|
||||
})
|
||||
|
||||
describe('GET /link', () => {
|
||||
const route = findRoute('GET', '/link')
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
})
|
||||
|
||||
it('401s without an authenticated context', async () => {
|
||||
const response = await route.handler(request('/link', 'GET'))
|
||||
expect(response.status).toBe(401)
|
||||
})
|
||||
|
||||
it('includes the 7-day health counters for the caller link only', async () => {
|
||||
const userMock = createQueuedMockSupabase()
|
||||
userMock.enqueue({
|
||||
data: {
|
||||
id: 'link-1',
|
||||
phone_masked: '+46 70 *** ** 67',
|
||||
default_company_id: null,
|
||||
muted_at: null,
|
||||
verified_at: '2026-08-01T09:00:00Z',
|
||||
},
|
||||
})
|
||||
const serviceMock = createQueuedMockSupabase()
|
||||
serviceMock.enqueue({
|
||||
data: {
|
||||
created_at: '2026-08-12T10:00:00Z',
|
||||
processing_status: 'done',
|
||||
error_message: null,
|
||||
inbox_item_id: 'item-1',
|
||||
},
|
||||
}) // last inbound
|
||||
serviceMock.enqueue({ data: { delivery_status: 'sent' } }) // last outbound
|
||||
serviceMock.enqueue({ count: 2 }) // outboundFailed7d
|
||||
serviceMock.enqueue({ count: 1 }) // parkedInbound7d
|
||||
vi.mocked(createServiceClient).mockReturnValue(serviceMock.supabase as never)
|
||||
|
||||
const response = await route.handler(request('/link', 'GET'), makeCtx(userMock.supabase))
|
||||
expect(response.status).toBe(200)
|
||||
const { data } = (await response.json()) as {
|
||||
data: { linked: boolean; health: { outboundFailed7d: number; parkedInbound7d: number } }
|
||||
}
|
||||
expect(data.linked).toBe(true)
|
||||
expect(data.health).toEqual({ outboundFailed7d: 2, parkedInbound7d: 1 })
|
||||
// Both head counts are keyed strictly by the link row RLS just proved
|
||||
// the caller owns.
|
||||
const linkScoped = serviceMock.calls.filter(
|
||||
(c) =>
|
||||
c.table === 'whatsapp_messages' &&
|
||||
c.method === 'eq' &&
|
||||
c.args[0] === 'phone_link_id' &&
|
||||
c.args[1] === 'link-1',
|
||||
)
|
||||
expect(linkScoped.length).toBeGreaterThanOrEqual(4)
|
||||
})
|
||||
|
||||
it('reports linked: false without touching the service client', async () => {
|
||||
const userMock = createQueuedMockSupabase()
|
||||
userMock.enqueue({ data: null }) // no active link row
|
||||
|
||||
const response = await route.handler(request('/link', 'GET'), makeCtx(userMock.supabase))
|
||||
const { data } = (await response.json()) as { data: { linked: boolean } }
|
||||
expect(data).toEqual({ linked: false })
|
||||
expect(vi.mocked(createServiceClient)).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@@ -20,7 +20,7 @@ vi.mock('@/extensions/general/whatsapp-inbox/lib/company-question', async () =>
|
||||
>('@/extensions/general/whatsapp-inbox/lib/company-question')
|
||||
return {
|
||||
...actual,
|
||||
askCompanyQuestion: vi.fn().mockResolvedValue(true),
|
||||
askCompanyQuestion: vi.fn().mockResolvedValue('asked'),
|
||||
}
|
||||
})
|
||||
|
||||
@@ -142,7 +142,7 @@ describe('processInboundMessage (media intake)', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
sendTextMock.mockResolvedValue({ ok: true, wamid: 'wamid.OUT', errorDetail: null })
|
||||
askCompanyQuestionMock.mockResolvedValue(true)
|
||||
askCompanyQuestionMock.mockResolvedValue('asked')
|
||||
rateLimitMock.mockResolvedValue({ ok: true })
|
||||
downloadMediaMock.mockResolvedValue({
|
||||
buffer: new Uint8Array([1, 2, 3, 4]).buffer,
|
||||
@@ -272,6 +272,48 @@ describe('processInboundMessage (media intake)', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('releases the row for retry when the membership query fails (transient, never parked)', async () => {
|
||||
const { supabase, enqueue, findCalls } = createQueuedMockSupabase()
|
||||
enqueue({ data: makeRow() }) // load row
|
||||
enqueue({ data: { id: 'msg-1' } }) // claim
|
||||
enqueue({ data: makeLink() }) // load link
|
||||
enqueue({ data: makeConversation() }) // load conversation
|
||||
enqueue({ error: { message: 'connection reset' } }) // memberships query fails
|
||||
enqueue({ data: null }) // release back to received
|
||||
|
||||
const outcome = await processInboundMessage(supabase as unknown as SupabaseClient, 'msg-1')
|
||||
|
||||
expect(outcome).toEqual({ kind: 'none' })
|
||||
expect(askCompanyQuestionMock).not.toHaveBeenCalled()
|
||||
expect(sendTextMock).not.toHaveBeenCalled()
|
||||
expect(downloadMediaMock).not.toHaveBeenCalled()
|
||||
expect(uploadAndExtractMock).not.toHaveBeenCalled()
|
||||
// The claim was released untouched: the sweep re-runs it within a minute
|
||||
// and gives up through the normal MAX_ATTEMPTS -> M18 path.
|
||||
const release = lastUpdate(findCalls)
|
||||
expect(release.processing_status).toBe('received')
|
||||
})
|
||||
|
||||
it('un-parks the staged row when the company question hits a transient options failure', async () => {
|
||||
askCompanyQuestionMock.mockResolvedValue('transient_error')
|
||||
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' }, { company_id: 'company-2' }] }) // ask path
|
||||
enqueue({ data: null }) // markStatus skipped (staged)
|
||||
enqueue({ data: null, count: 1 }) // staged count
|
||||
enqueue({ data: null }) // un-park release
|
||||
|
||||
const outcome = await processInboundMessage(supabase as unknown as SupabaseClient, 'msg-1')
|
||||
|
||||
expect(outcome).toEqual({ kind: 'none' })
|
||||
const release = lastUpdate(findCalls)
|
||||
expect(release.processing_status).toBe('received')
|
||||
expect(release.error_message).toBeNull()
|
||||
})
|
||||
|
||||
it('uses a live conversation pin over everything and stamps via=pin', async () => {
|
||||
const pinExpiry = new Date(Date.now() + 2 * 60 * 60 * 1000).toISOString()
|
||||
const { supabase, enqueue, findCalls } = createQueuedMockSupabase()
|
||||
|
||||
@@ -75,6 +75,7 @@ describe('runSweep', () => {
|
||||
enqueue({ data: [] }) // unacked re-arm
|
||||
enqueue({ data: [] }) // TTL scan
|
||||
enqueue({ data: [] }) // pin scan
|
||||
enqueue({ count: 2 }) // outbound failures, last 24h
|
||||
|
||||
const summary = await runSweep(supabase as unknown as SupabaseClient)
|
||||
|
||||
@@ -84,6 +85,7 @@ describe('runSweep', () => {
|
||||
expect(summary.reclaimedReceived).toBe(1)
|
||||
expect(summary.erroredMaxAttempts).toBe(1)
|
||||
expect(summary.finalizedAcks).toBe(1)
|
||||
expect(summary.outboundFailed24h).toBe(2)
|
||||
|
||||
const errorPatch = findCalls('whatsapp_messages', 'update')[0][0] as Record<string, unknown>
|
||||
expect(errorPatch.processing_status).toBe('error')
|
||||
|
||||
@@ -284,6 +284,69 @@ describe('POST /webhook', () => {
|
||||
expect(row.body_text).toBeUndefined()
|
||||
})
|
||||
|
||||
it('media inside the hour but outside the 10 min burst window still gets M1', async () => {
|
||||
const { enqueue } = mockSupabase()
|
||||
enqueue({ data: null }) // no active link
|
||||
enqueue({ data: { ok: true } }) // sender quota RPC
|
||||
enqueue({
|
||||
data: [{ created_at: new Date(Date.now() - 30 * 60 * 1000).toISOString() }],
|
||||
}) // one greeting 30 min ago: inside the text hour, outside the burst
|
||||
enqueue({ data: null, error: null }) // trace row insert ('done')
|
||||
|
||||
await route.handler(signedRequest(envelope({ messages: [imageMessage()] })))
|
||||
|
||||
expect(sendTextMock).toHaveBeenCalledTimes(1)
|
||||
expect(sendTextMock.mock.calls[0][1].template).toBe(TEMPLATE.m1Unlinked)
|
||||
// The privacy invariant holds: still no media touch for unlinked senders.
|
||||
expect(vi.mocked(downloadMedia)).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('media inside the 10 min burst window stays silent (one M1 per burst)', async () => {
|
||||
const { enqueue } = mockSupabase()
|
||||
enqueue({ data: null })
|
||||
enqueue({ data: { ok: true } })
|
||||
enqueue({
|
||||
data: [{ created_at: new Date(Date.now() - 2 * 60 * 1000).toISOString() }],
|
||||
}) // greeted 2 min ago: same burst
|
||||
enqueue({ count: 0 }) // decline-trace day cap
|
||||
enqueue({ data: null, error: null }) // declined trace insert
|
||||
|
||||
await route.handler(signedRequest(envelope({ messages: [imageMessage()] })))
|
||||
expect(sendTextMock).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('a text message inside the hour stays silent (hour rule unchanged for text)', async () => {
|
||||
const { enqueue } = mockSupabase()
|
||||
enqueue({ data: null })
|
||||
enqueue({ data: { ok: true } })
|
||||
enqueue({
|
||||
data: [{ created_at: new Date(Date.now() - 30 * 60 * 1000).toISOString() }],
|
||||
}) // greeted 30 min ago: text keeps the 1/hour rule
|
||||
enqueue({ count: 0 }) // decline-trace day cap
|
||||
enqueue({ data: null, error: null }) // declined trace insert
|
||||
|
||||
await route.handler(signedRequest(envelope({ messages: [textMessage('hej')] })))
|
||||
expect(sendTextMock).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('the fourth media greeting of the day stays silent (daily cap kept)', async () => {
|
||||
const { enqueue } = mockSupabase()
|
||||
enqueue({ data: null })
|
||||
enqueue({ data: { ok: true } })
|
||||
enqueue({
|
||||
data: [
|
||||
{ created_at: new Date(Date.now() - 2 * 60 * 60 * 1000).toISOString() },
|
||||
{ created_at: new Date(Date.now() - 5 * 60 * 60 * 1000).toISOString() },
|
||||
{ created_at: new Date(Date.now() - 9 * 60 * 60 * 1000).toISOString() },
|
||||
],
|
||||
}) // GREETING_DAY_MAX greetings already sent today
|
||||
enqueue({ count: 0 }) // decline-trace day cap
|
||||
enqueue({ data: null, error: null }) // declined trace insert
|
||||
|
||||
await route.handler(signedRequest(envelope({ messages: [imageMessage()] })))
|
||||
expect(sendTextMock).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('stays silent but records the decline when the pre-binding quota is exhausted', async () => {
|
||||
const mock = mockSupabase()
|
||||
mock.enqueue({ data: null })
|
||||
|
||||
@@ -262,7 +262,11 @@ async function handleUnknownSender(
|
||||
|
||||
// Anything else from an unknown number: the AI-disclosure greeting, hard
|
||||
// throttled per phone hash (EU AI Act Art 50 disclosure lives in M1).
|
||||
if (await greetingThrottled(supabase, phoneHash)) {
|
||||
// Media bypasses the hour rule (10-minute burst window instead, daily cap
|
||||
// kept): a photo is a receipt someone expects to be handled, and pure
|
||||
// silence taught senders their receipts were being filed when nothing was.
|
||||
const carriesMedia = msg.type === 'image' || msg.type === 'document'
|
||||
if (await greetingThrottled(supabase, phoneHash, { media: carriesMedia })) {
|
||||
await recordUnknownSenderMessage(
|
||||
supabase, msg, phoneHash, 'skipped', 'Unknown sender: greeting throttled, declined',
|
||||
)
|
||||
@@ -859,6 +863,27 @@ export const whatsappInboxExtension: Extension = {
|
||||
.limit(1)
|
||||
.maybeSingle()
|
||||
|
||||
// Channel health for THIS link, 7-day window, head counts only: how
|
||||
// many replies never reached the sender and how many inbound rows
|
||||
// parked with a reason. The panel turns nonzero counts into one
|
||||
// attention line; nothing else in the app reads delivery_status.
|
||||
const weekAgo = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString()
|
||||
const { count: outboundFailed7d } = await serviceClient
|
||||
.from('whatsapp_messages')
|
||||
.select('id', { count: 'exact', head: true })
|
||||
.eq('phone_link_id', row.id)
|
||||
.eq('direction', 'outbound')
|
||||
.eq('delivery_status', 'failed')
|
||||
.gte('created_at', weekAgo)
|
||||
const { count: parkedInbound7d } = await serviceClient
|
||||
.from('whatsapp_messages')
|
||||
.select('id', { count: 'exact', head: true })
|
||||
.eq('phone_link_id', row.id)
|
||||
.eq('direction', 'inbound')
|
||||
.in('processing_status', ['skipped', 'error'])
|
||||
.not('error_message', 'is', null)
|
||||
.gte('created_at', weekAgo)
|
||||
|
||||
const inboundRow = lastInbound as {
|
||||
created_at: string
|
||||
processing_status: string
|
||||
@@ -877,6 +902,10 @@ export const whatsappInboxExtension: Extension = {
|
||||
lastReplyFailed:
|
||||
(lastOutbound as { delivery_status: string | null } | null)?.delivery_status ===
|
||||
'failed',
|
||||
health: {
|
||||
outboundFailed7d: outboundFailed7d ?? 0,
|
||||
parkedInbound7d: parkedInbound7d ?? 0,
|
||||
},
|
||||
},
|
||||
})
|
||||
},
|
||||
@@ -896,6 +925,31 @@ export const whatsappInboxExtension: Extension = {
|
||||
},
|
||||
},
|
||||
|
||||
{
|
||||
method: 'POST',
|
||||
path: '/link/unmute',
|
||||
handler: async (_request: Request, ctx?: ExtensionContext) => {
|
||||
if (!ctx) return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
|
||||
// Clears a stopp pause from the app (the in-chat counterpart is the
|
||||
// literal text `start`). The user-scoped client is deliberate: RLS
|
||||
// (whatsapp_phone_links_update_own) proves the caller owns the row.
|
||||
// Idempotent on an unpaused link; only a missing active link 404s.
|
||||
const { data } = await ctx.supabase
|
||||
.from('whatsapp_phone_links')
|
||||
.update({ muted_at: null })
|
||||
.eq('user_id', ctx.userId)
|
||||
.is('revoked_at', null)
|
||||
.select('id')
|
||||
if (!Array.isArray(data) || data.length === 0) {
|
||||
return NextResponse.json(
|
||||
{ error: 'Ingen aktiv WhatsApp-koppling hittades.' },
|
||||
{ status: 404 },
|
||||
)
|
||||
}
|
||||
return NextResponse.json({ data: { unmuted: true } })
|
||||
},
|
||||
},
|
||||
|
||||
{
|
||||
method: 'POST',
|
||||
path: '/link/default-company',
|
||||
|
||||
@@ -25,6 +25,7 @@ import {
|
||||
import { botCopy, TEMPLATE } from './messages'
|
||||
import {
|
||||
COMPANY_PIN_TTL_MS,
|
||||
NO_COMPANY_OPTIONS,
|
||||
STAGED_AWAITING_COMPANY,
|
||||
getContext,
|
||||
updateConversation,
|
||||
@@ -36,6 +37,11 @@ const log = createLogger('whatsapp-inbox/company-question')
|
||||
/** Defensive ceiling on the numbered text list (>10 companies). */
|
||||
const MAX_NUMBERED_OPTIONS = 30
|
||||
|
||||
/** M19 (no company to file into) is sent at most once per this window per
|
||||
* conversation: every parked row of a burst walks through the no-options
|
||||
* branch, and five photos must not earn five identical replies. */
|
||||
const NO_OPTIONS_NOTICE_WINDOW_MS = 10 * 60 * 1000
|
||||
|
||||
export type CompanyChoiceVia = 'button' | 'list' | 'numbered'
|
||||
|
||||
export interface CompanyOption {
|
||||
@@ -45,22 +51,33 @@ export interface CompanyOption {
|
||||
|
||||
type ReplyBase = Omit<SendMessageBase, 'to' | 'template'>
|
||||
|
||||
/** All companies the linked user belongs to, alphabetical (stable digits). */
|
||||
/** All companies the linked user belongs to, alphabetical (stable digits).
|
||||
* Returns null when a query FAILED: a transient DB error must read as
|
||||
* "unknown", never as "no memberships", or the caller parks receipts behind
|
||||
* a question that can never be asked. */
|
||||
export async function loadCompanyOptions(
|
||||
supabase: SupabaseClient,
|
||||
userId: string,
|
||||
): Promise<CompanyOption[]> {
|
||||
const { data: memberships } = await supabase
|
||||
): Promise<CompanyOption[] | null> {
|
||||
const { data: memberships, error: membershipsError } = await supabase
|
||||
.from('company_members')
|
||||
.select('company_id')
|
||||
.eq('user_id', userId)
|
||||
if (membershipsError) {
|
||||
log.warn('company options membership query failed', { error: membershipsError.message })
|
||||
return null
|
||||
}
|
||||
const companyIds = [...new Set((memberships ?? []).map((m) => m.company_id as string))]
|
||||
if (companyIds.length === 0) return []
|
||||
|
||||
const { data: companies } = await supabase
|
||||
const { data: companies, error: companiesError } = await supabase
|
||||
.from('companies')
|
||||
.select('id, name')
|
||||
.in('id', companyIds)
|
||||
if (companiesError) {
|
||||
log.warn('company options company query failed', { error: companiesError.message })
|
||||
return null
|
||||
}
|
||||
const options = ((companies ?? []) as { id: string; name: string | null }[]).map((c) => ({
|
||||
id: c.id,
|
||||
name: c.name ?? 'Företag',
|
||||
@@ -69,6 +86,17 @@ export async function loadCompanyOptions(
|
||||
return options.slice(0, MAX_NUMBERED_OPTIONS)
|
||||
}
|
||||
|
||||
/** How an ask attempt ended.
|
||||
* - asked: this caller sent the question; rows stay staged for the answer.
|
||||
* - not_asked: lost the guarded transition or the send failed; the episode
|
||||
* is owned elsewhere (or re-triggers on the next receipt). Rows stay staged.
|
||||
* - no_options: fewer than 2 companies genuinely exist, so no question can
|
||||
* help; the staged rows were re-marked no_company_options and M19 told the
|
||||
* sender to fix the linking in the app. Terminal by design.
|
||||
* - transient_error: the options could not be LOADED (DB error): the caller
|
||||
* must release its row back to 'received' so the sweep retries it. */
|
||||
export type AskCompanyQuestionOutcome = 'asked' | 'not_asked' | 'no_options' | 'transient_error'
|
||||
|
||||
/**
|
||||
* Ask the company question ONCE per open episode. The guarded state
|
||||
* transition (WHERE state <> 'awaiting_company') makes concurrent workers of
|
||||
@@ -79,8 +107,6 @@ export async function loadCompanyOptions(
|
||||
* 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(
|
||||
supabase: SupabaseClient,
|
||||
@@ -92,15 +118,44 @@ export async function askCompanyQuestion(
|
||||
/** How many receipts wait on the answer (question body wording). */
|
||||
stagedCount: number
|
||||
},
|
||||
): Promise<boolean> {
|
||||
): Promise<AskCompanyQuestionOutcome> {
|
||||
const options = await loadCompanyOptions(supabase, args.link.user_id)
|
||||
if (options === null) return 'transient_error'
|
||||
if (options.length < 2) {
|
||||
// Degenerate: resolution should have succeeded. Leave the rows staged;
|
||||
// the sweep TTL cleans up if this persists.
|
||||
// Genuinely degenerate (a real empty/1-length result, not a query error):
|
||||
// the sender has nothing to choose between, so no question can help and
|
||||
// nothing will change until they act in the app. The old behavior parked
|
||||
// the rows silently forever; instead mark them with a terminal, greppable
|
||||
// marker and say so in the chat.
|
||||
log.warn('company question requested with fewer than 2 options', {
|
||||
conversationId: args.conversation.id,
|
||||
})
|
||||
return false
|
||||
await supabase
|
||||
.from('whatsapp_messages')
|
||||
.update({ error_message: NO_COMPANY_OPTIONS })
|
||||
.eq('conversation_id', args.conversation.id)
|
||||
.eq('processing_status', 'skipped')
|
||||
.eq('error_message', STAGED_AWAITING_COMPANY)
|
||||
// One M19 per burst window (same idiom as the M17 rate-limit notice).
|
||||
const since = new Date(Date.now() - NO_OPTIONS_NOTICE_WINDOW_MS).toISOString()
|
||||
const { data: noticeSent } = await supabase
|
||||
.from('whatsapp_messages')
|
||||
.select('id')
|
||||
.eq('direction', 'outbound')
|
||||
.eq('conversation_id', args.conversation.id)
|
||||
.eq('raw_payload->>template', TEMPLATE.m19NoCompany)
|
||||
.gte('created_at', since)
|
||||
.limit(1)
|
||||
.maybeSingle()
|
||||
if (!noticeSent) {
|
||||
await sendText(supabase, {
|
||||
to: args.to,
|
||||
body: botCopy('sv').m19NoCompany(),
|
||||
template: TEMPLATE.m19NoCompany,
|
||||
...args.replyBase,
|
||||
})
|
||||
}
|
||||
return 'no_options'
|
||||
}
|
||||
|
||||
const now = new Date()
|
||||
@@ -119,7 +174,7 @@ export async function askCompanyQuestion(
|
||||
.eq('id', args.conversation.id)
|
||||
.neq('state', 'awaiting_company')
|
||||
.select('*')
|
||||
if (!Array.isArray(won) || won.length === 0) return false
|
||||
if (!Array.isArray(won) || won.length === 0) return 'not_asked'
|
||||
// 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 = {
|
||||
@@ -175,9 +230,9 @@ export async function askCompanyQuestion(
|
||||
log.warn('company question send failed; question rolled back', {
|
||||
conversationId: args.conversation.id,
|
||||
})
|
||||
return false
|
||||
return 'not_asked'
|
||||
}
|
||||
return true
|
||||
return 'asked'
|
||||
}
|
||||
|
||||
export interface AppliedCompanyChoice {
|
||||
|
||||
@@ -30,19 +30,28 @@ import { TEMPLATE } from './messages'
|
||||
|
||||
const log = createLogger('whatsapp-inbox/conversation')
|
||||
|
||||
// The M1 "not linked" greeting is hard throttled per phone hash: 1/hour,
|
||||
// 3/day, then silence. Shared by the unknown-sender path (webhook) and the
|
||||
// revoked-mid-processing path (process-inbound), so it lives here.
|
||||
// The M1 "not linked" greeting is hard throttled per phone hash: text gets
|
||||
// 1/hour, media rides a 10-minute burst window instead (see below); both
|
||||
// share the 3/day cap, then silence. Shared by the unknown-sender path
|
||||
// (webhook) and the revoked-mid-processing path (process-inbound), so it
|
||||
// lives here.
|
||||
const GREETING_HOUR_MS = 60 * 60 * 1000
|
||||
const GREETING_BURST_MS = 10 * 60 * 1000
|
||||
const GREETING_DAY_MS = 24 * 60 * 60 * 1000
|
||||
const GREETING_DAY_MAX = 3
|
||||
|
||||
/** True when another M1 greeting to this phone hash would exceed the cap.
|
||||
* Fails CLOSED: if the throttle window cannot be read, no greeting goes
|
||||
* out, matching the unknown-sender quota's stance. */
|
||||
* out, matching the unknown-sender quota's stance.
|
||||
*
|
||||
* `media: true` marks a message that carried an image or document: the
|
||||
* sender is handing over a receipt they expect to be handled, so repeats are
|
||||
* suppressed only inside a 10-minute burst window (one burst of photos earns
|
||||
* exactly one M1) instead of the full hour. The daily cap applies to both. */
|
||||
export async function greetingThrottled(
|
||||
supabase: SupabaseClient,
|
||||
phoneHash: string,
|
||||
opts: { media?: boolean } = {},
|
||||
): Promise<boolean> {
|
||||
const since = new Date(Date.now() - GREETING_DAY_MS).toISOString()
|
||||
const { data, error } = await supabase
|
||||
@@ -60,8 +69,8 @@ export async function greetingThrottled(
|
||||
}
|
||||
const rows = (data ?? []) as Array<{ created_at: string }>
|
||||
if (rows.length >= GREETING_DAY_MAX) return true
|
||||
const hourAgo = Date.now() - GREETING_HOUR_MS
|
||||
return rows.some((r) => new Date(r.created_at).getTime() > hourAgo)
|
||||
const windowStart = Date.now() - (opts.media ? GREETING_BURST_MS : GREETING_HOUR_MS)
|
||||
return rows.some((r) => new Date(r.created_at).getTime() > windowStart)
|
||||
}
|
||||
|
||||
export const COMPANY_PIN_TTL_MS = 8 * 60 * 60 * 1000
|
||||
@@ -76,6 +85,11 @@ export const MAX_QUESTIONS_PER_DAY = 6
|
||||
export const STAGED_AWAITING_COMPANY = 'staged_awaiting_company'
|
||||
/** Marker after the 48h TTL expired: excluded from any later re-open. */
|
||||
export const COMPANY_CHOICE_EXPIRED = 'company_choice_expired'
|
||||
/** Terminal marker when the company question could not be asked at all:
|
||||
* the sender has fewer than 2 companies to choose between, so nothing will
|
||||
* change until they fix the linking in the app. The rows are never retried;
|
||||
* M19 tells the sender why. */
|
||||
export const NO_COMPANY_OPTIONS = 'no_company_options'
|
||||
|
||||
export type QuestionType = 'representation' | 'context' | 'resend'
|
||||
export type ConversationQuestionType = QuestionType | 'company'
|
||||
|
||||
@@ -56,6 +56,7 @@ export const TEMPLATE = {
|
||||
m17RateLimited: 'm17_rate_limited',
|
||||
m17RateLimitedDay: 'm17_rate_limited_day',
|
||||
m18Error: 'm18_error',
|
||||
m19NoCompany: 'm19_no_company',
|
||||
} as const
|
||||
|
||||
export type TemplateId = (typeof TEMPLATE)[keyof typeof TEMPLATE]
|
||||
@@ -194,6 +195,13 @@ const SV = {
|
||||
|
||||
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.',
|
||||
|
||||
// The company question could not even be asked: the linked user has fewer
|
||||
// than 2 companies to choose between, so the receipt cannot be filed and
|
||||
// nothing changes until they act in the app. Honest and actionable beats
|
||||
// the old silent parking.
|
||||
m19NoCompany: () =>
|
||||
'Jag kunde inte koppla kvittot till något företag. Öppna Accounted och kontrollera WhatsApp-kopplingen under *Inställningar -> WhatsApp*, och skicka sedan kvittot igen.',
|
||||
}
|
||||
|
||||
const EN: typeof SV = {
|
||||
@@ -318,6 +326,9 @@ const EN: typeof SV = {
|
||||
|
||||
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.',
|
||||
|
||||
m19NoCompany: () =>
|
||||
'I could not assign the receipt to any company. Open Accounted and check the WhatsApp linking under *Settings -> WhatsApp*, then send the receipt again.',
|
||||
}
|
||||
|
||||
const COPY: Record<BotLocale, typeof SV> = { sv: SV, en: EN }
|
||||
|
||||
@@ -285,13 +285,18 @@ interface ResolvedCompany {
|
||||
/**
|
||||
* Conversation pin (live + still a member, sliding 8h) -> default company
|
||||
* (still a member) -> sole membership -> null (ask).
|
||||
*
|
||||
* Returns 'transient_error' when the membership query FAILED: a DB blip must
|
||||
* not read as "no memberships", which used to park the rows behind an
|
||||
* unanswerable company question forever. The caller releases the row so the
|
||||
* sweep retries it.
|
||||
*/
|
||||
async function resolveCompanyTarget(
|
||||
supabase: SupabaseClient,
|
||||
link: WhatsAppPhoneLink,
|
||||
conversation: WhatsAppConversation | null,
|
||||
selectedViaOverride: CompanyChoiceVia | undefined,
|
||||
): Promise<ResolvedCompany | null> {
|
||||
): Promise<ResolvedCompany | null | 'transient_error'> {
|
||||
const now = new Date()
|
||||
|
||||
if (conversation && hasLivePin(conversation, now) && conversation.company_id) {
|
||||
@@ -318,10 +323,16 @@ async function resolveCompanyTarget(
|
||||
return { companyId: link.default_company_id, via: 'default' }
|
||||
}
|
||||
|
||||
const { data: memberships } = await supabase
|
||||
const { data: memberships, error: membershipsError } = await supabase
|
||||
.from('company_members')
|
||||
.select('company_id')
|
||||
.eq('user_id', link.user_id)
|
||||
if (membershipsError) {
|
||||
log.warn('membership query failed during company resolution; will retry', {
|
||||
error: membershipsError.message,
|
||||
})
|
||||
return 'transient_error'
|
||||
}
|
||||
const companyIds = [...new Set((memberships ?? []).map((m) => m.company_id as string))]
|
||||
if (companyIds.length === 1) return { companyId: companyIds[0], via: 'single' }
|
||||
return null
|
||||
@@ -457,6 +468,17 @@ async function processMediaMessage(
|
||||
|
||||
// ── Company resolution ─────────────────────────────────
|
||||
const resolved = await resolveCompanyTarget(supabase, link, conversation, opts.companySelectedVia)
|
||||
if (resolved === 'transient_error') {
|
||||
// Release the claim untouched: the sweep re-runs this row within a
|
||||
// minute and gives up through the normal MAX_ATTEMPTS -> M18 path if
|
||||
// the failure persists. Parking it as skipped would be permanent.
|
||||
await supabase
|
||||
.from('whatsapp_messages')
|
||||
.update({ processing_status: 'received' })
|
||||
.eq('id', row.id)
|
||||
.eq('processing_status', 'processing')
|
||||
return { kind: 'none' }
|
||||
}
|
||||
if (!resolved) {
|
||||
if (!conversation) {
|
||||
await markStatus(supabase, row.id, 'error', {
|
||||
@@ -474,13 +496,25 @@ async function processMediaMessage(
|
||||
.eq('conversation_id', conversation.id)
|
||||
.eq('processing_status', 'skipped')
|
||||
.eq('error_message', STAGED_AWAITING_COMPANY)
|
||||
await askCompanyQuestion(supabase, {
|
||||
const askOutcome = await askCompanyQuestion(supabase, {
|
||||
conversation,
|
||||
link,
|
||||
to,
|
||||
replyBase,
|
||||
stagedCount: count ?? 1,
|
||||
})
|
||||
if (askOutcome === 'transient_error') {
|
||||
// The options could not even be loaded: un-park this row so the
|
||||
// sweep retries it, instead of leaving it staged behind a question
|
||||
// that was never asked (and might never be).
|
||||
await supabase
|
||||
.from('whatsapp_messages')
|
||||
.update({ processing_status: 'received', error_message: null })
|
||||
.eq('id', row.id)
|
||||
.eq('processing_status', 'skipped')
|
||||
.eq('error_message', STAGED_AWAITING_COMPANY)
|
||||
return { kind: 'none' }
|
||||
}
|
||||
return { kind: 'media_staged', conversationId: conversation.id }
|
||||
}
|
||||
const companyId = resolved.companyId
|
||||
|
||||
@@ -18,6 +18,9 @@
|
||||
* 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.
|
||||
* 5. Count outbound sends that failed in the last 24h, so the per-minute
|
||||
* "whatsapp sweep complete" log line surfaces delivery problems nothing
|
||||
* else reads (delivery_status is otherwise write-only).
|
||||
*/
|
||||
|
||||
import type { SupabaseClient } from '@supabase/supabase-js'
|
||||
@@ -64,6 +67,10 @@ export interface SweepSummary {
|
||||
finalizedAcks: number
|
||||
expiredQuestions: number
|
||||
clearedPins: number
|
||||
/** Outbound rows with delivery_status='failed' created in the last 24h
|
||||
* (Graph send failure or a Meta 'failed' status callback). Observability
|
||||
* only: the sweep log line is the consumer. */
|
||||
outboundFailed24h: number
|
||||
}
|
||||
|
||||
interface StuckRow {
|
||||
@@ -133,6 +140,7 @@ export async function runSweep(supabase: SupabaseClient): Promise<SweepSummary>
|
||||
finalizedAcks: 0,
|
||||
expiredQuestions: 0,
|
||||
clearedPins: 0,
|
||||
outboundFailed24h: 0,
|
||||
}
|
||||
const finalizeConversations = new Set<string>()
|
||||
const now = Date.now()
|
||||
@@ -380,5 +388,19 @@ export async function runSweep(supabase: SupabaseClient): Promise<SweepSummary>
|
||||
log.error('sweep: pin expiry pass failed', err)
|
||||
}
|
||||
|
||||
// ── 5. Outbound delivery failures, last 24h (one head count) ──
|
||||
try {
|
||||
const since = new Date(now - 24 * 60 * 60 * 1000).toISOString()
|
||||
const { count } = await supabase
|
||||
.from('whatsapp_messages')
|
||||
.select('id', { count: 'exact', head: true })
|
||||
.eq('direction', 'outbound')
|
||||
.eq('delivery_status', 'failed')
|
||||
.gte('created_at', since)
|
||||
summary.outboundFailed24h = count ?? 0
|
||||
} catch (err) {
|
||||
log.error('sweep: outbound failure count failed', err)
|
||||
}
|
||||
|
||||
return summary
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user