accounted/extensions/general/whatsapp-inbox/lib/process-inbound.ts

/**
 * Deferred conversation worker: inbound WhatsApp rows -> Underlag items,
 * burst-debounced combined acks, clarifying questions and answer handling.
 *
 * The webhook persists rows and 200s fast; this worker runs via the after()
 * idiom (lib/webhooks/dispatch-kick.ts) or the per-minute sweep cron. The
 * whatsapp_messages row IS the durable job record: the atomic claim
 * (UPDATE ... WHERE processing_status='received' RETURNING) makes redelivered
 * webhooks and sweep re-claims safe to race.
 *
 * Reply model (PR4): individual media items ingest immediately, but the
 * RECEIPT ack is debounced per conversation. Every burst invocation sleeps to
 * its own debounce deadline and attempts the atomic pending_ack claim; the
 * single winner sends ONE message: M4 (single) or M5 (numbered list), with at
 * most ONE clarifying question merged in (M7/M9/M10, budgeted). Rejection
 * replies (M15/M17/M18/duplicate) stay immediate and per item.
 *
 * Failure policy: NOTHING here returns a retryable status to Meta. Rejected
 * and rate-limited content acks in chat and lands as 'skipped'; real failures
 * land as 'error' + error_message with a single M18 to the user.
 */

import { after } from 'next/server'
import type { SupabaseClient } from '@supabase/supabase-js'
import { createServiceClientNoCookies } from '@/lib/auth/api-keys'
import { checkInboxUploadRateLimit } from '@/lib/rate-limits/inbox'
import { checkAgentRateLimit } from '@/lib/rate-limits/agent'
import { appendProcessingHistory } from '@/lib/processing-history/append'
import { computeSHA256 } from '@/lib/core/documents/document-service'
import { createLogger } from '@/lib/logger'
import { uploadAndExtract } from '@/extensions/general/invoice-inbox/lib/upload-and-extract'
import type {
  InboxChannelContext,
  InvoiceExtractionResult,
  WhatsAppConversation,
  WhatsAppMessage,
  WhatsAppPhoneLink,
} from '@/types'
import { sendText, markReadWithTyping, downloadMedia, GraphApiError } from './graph-api'
import { botCopy, TEMPLATE, type TemplateId } from './messages'
import {
  COMPANY_PIN_TTL_MS,
  MAX_QUESTIONS_PER_BURST,
  MAX_QUESTIONS_PER_DAY,
  STAGED_AWAITING_COMPANY,
  appendRecentQuestion,
  bumpBudget,
  claimAck,
  extractRecipient,
  getContext,
  getOrCreateConversation,
  hasLivePin,
  loadConversation,
  markRecentQuestion,
  questionsAskedToday,
  resolveAnswerTarget,
  serviceWindowOpen,
  type ConversationContext,
  type QueuedQuestion,
  type QuestionType,
} from './conversation'
import { askCompanyQuestion, type CompanyChoiceVia } from './company-question'
import { evaluateQuestion, QUESTION_PRIORITY } from './questions'
import { interpretChatAnswer } from './interpret-answer'
import { appendQuestionHistory, updateItemContext } from './item-context'

const log = createLogger('whatsapp-inbox/process-inbound')

/** Chat intake accepts what phones actually produce. Narrower than the upload
 *  allowlist on purpose: WhatsApp transcodes photos to JPEG, so HEIC never
 *  arrives, and everything else gets the M15 nudge. */
export const CHAT_ALLOWED_MIME_TYPES: ReadonlySet<string> = new Set([
  'image/jpeg',
  'image/png',
  'image/webp',
  'application/pdf',
])

/** M17 is sent at most once per this window per sender, not once per file. */
const RATE_LIMIT_NOTICE_WINDOW_MS = 10 * 60 * 1000

/** Upper bound on how long a burst invocation waits for its own deadline. */
const MAX_DEBOUNCE_WAIT_MS = 15 * 1000

/** Rows one combined ack covers at most (defensive bound). */
const MAX_BURST_ROWS = 20

const EXTENSION_FOR_MIME: Record<string, string> = {
  'image/jpeg': 'jpg',
  'image/png': 'png',
  'image/webp': 'webp',
  'application/pdf': 'pdf',
}

const STATE_FOR_QUESTION: Record<QuestionType, WhatsAppConversation['state']> = {
  representation: 'awaiting_representation',
  context: 'awaiting_context',
  resend: 'awaiting_resend',
}

// Injectable sleep so tests never wait on real debounce windows.
let sleepFn: (ms: number) => Promise<void> = (ms) =>
  new Promise((resolve) => setTimeout(resolve, ms))

/** Test-only: replace (or restore) the debounce sleep implementation. */
export function __setSleepForTests(fn: ((ms: number) => Promise<void>) | null): void {
  sleepFn = fn ?? ((ms) => new Promise((resolve) => setTimeout(resolve, ms)))
}

function fallbackFilename(mime: string): string {
  const ext = EXTENSION_FOR_MIME[mime] ?? 'bin'
  return `whatsapp-${new Date().toISOString().slice(0, 10)}.${ext}`
}

function formatSek(amount: number): string {
  return `${new Intl.NumberFormat('sv-SE', {
    minimumFractionDigits: 0,
    maximumFractionDigits: 2,
  }).format(amount)} kr`
}

async function loadRow(
  supabase: SupabaseClient,
  messageId: string,
): Promise<WhatsAppMessage | null> {
  const { data } = await supabase
    .from('whatsapp_messages')
    .select('*')
    .eq('id', messageId)
    .maybeSingle()
  return (data as WhatsAppMessage | null) ?? null
}

async function loadLink(
  supabase: SupabaseClient,
  phoneLinkId: string,
): Promise<WhatsAppPhoneLink | null> {
  const { data } = await supabase
    .from('whatsapp_phone_links')
    .select('*')
    .eq('id', phoneLinkId)
    .maybeSingle()
  return (data as WhatsAppPhoneLink | null) ?? null
}

async function isMember(
  supabase: SupabaseClient,
  userId: string,
  companyId: string,
): Promise<boolean> {
  const { data } = await supabase
    .from('company_members')
    .select('company_id')
    .eq('user_id', userId)
    .eq('company_id', companyId)
    .limit(1)
    .maybeSingle()
  return data != null
}

/** True when an M17 notice already went to this sender inside the window. */
async function rateLimitNoticeAlreadySent(
  supabase: SupabaseClient,
  senderPhoneHash: string,
): Promise<boolean> {
  const since = new Date(Date.now() - RATE_LIMIT_NOTICE_WINDOW_MS).toISOString()
  const { data } = await supabase
    .from('whatsapp_messages')
    .select('id')
    .eq('direction', 'outbound')
    .eq('sender_phone_hash', senderPhoneHash)
    .eq('raw_payload->>template', TEMPLATE.m17RateLimited)
    .gte('created_at', since)
    .limit(1)
    .maybeSingle()
  return data != null
}

async function markStatus(
  supabase: SupabaseClient,
  messageId: string,
  status: 'skipped' | 'error' | 'done',
  extra: { errorMessage?: string | null; inboxItemId?: string | null } = {},
): Promise<void> {
  // Literal payload (no spread) so the phantom-column scanner can verify it.
  await supabase
    .from('whatsapp_messages')
    .update({
      processing_status: status,
      error_message: extra.errorMessage ?? null,
      inbox_item_id: extra.inboxItemId ?? null,
    })
    .eq('id', messageId)
}

// ── Company resolution (PR4 ladder) ──────────────────────────

interface ResolvedCompany {
  companyId: string
  via: NonNullable<InboxChannelContext['company_selected_via']>
}

/**
 * Conversation pin (live + still a member, sliding 8h) -> default company
 * (still a member) -> sole membership -> null (ask).
 */
async function resolveCompanyTarget(
  supabase: SupabaseClient,
  link: WhatsAppPhoneLink,
  conversation: WhatsAppConversation | null,
  selectedViaOverride: CompanyChoiceVia | undefined,
): Promise<ResolvedCompany | null> {
  const now = new Date()

  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.
      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<string, unknown>,
          })
          .eq('id', conversation.id)
      }
      return { companyId: conversation.company_id, via: selectedViaOverride ?? 'pin' }
    }
  }

  if (link.default_company_id && (await isMember(supabase, link.user_id, link.default_company_id))) {
    return { companyId: link.default_company_id, via: 'default' }
  }

  const { data: memberships } = await supabase
    .from('company_members')
    .select('company_id')
    .eq('user_id', link.user_id)
  const companyIds = [...new Set((memberships ?? []).map((m) => m.company_id as string))]
  if (companyIds.length === 1) return { companyId: companyIds[0], via: 'single' }
  return null
}

// ── Outcomes ─────────────────────────────────────────────────

export type ProcessOutcome =
  | { kind: 'media_processed'; conversationId: string | null }
  | { kind: 'media_staged'; conversationId: string | null }
  | { kind: 'answer'; conversationId: string | null }
  | { kind: 'none' }

export interface KickOptions {
  /** Set when re-opening rows parked behind a just-answered company question:
   *  their items record the choice mechanism instead of a generic 'pin'. */
  companySelectedVia?: CompanyChoiceVia
}

/**
 * Process one inbound message end to end (media intake OR text answer).
 * Never throws.
 */
export async function processInboundMessage(
  supabase: SupabaseClient,
  messageId: string,
  opts: KickOptions = {},
): Promise<ProcessOutcome> {
  const row = await loadRow(supabase, messageId)
  if (!row) return { kind: 'none' }

  // Atomic claim: only a row still in 'received' can be taken, so a webhook
  // redelivery racing this worker (or the sweep) claims exactly once.
  const { data: claimed } = await supabase
    .from('whatsapp_messages')
    .update({ processing_status: 'processing', attempts: row.attempts + 1 })
    .eq('id', messageId)
    .eq('processing_status', 'received')
    .select('id')
    .maybeSingle()
  if (!claimed) return { kind: 'none' }

  const attempt = row.attempts + 1
  if (row.message_type === 'text') {
    return processAnswerMessage(supabase, row)
  }
  return processMediaMessage(supabase, row, attempt, opts)
}

// ── Media intake ─────────────────────────────────────────────

async function processMediaMessage(
  supabase: SupabaseClient,
  row: WhatsAppMessage,
  attempt: number,
  opts: KickOptions,
): Promise<ProcessOutcome> {
  const copy = botCopy('sv')
  const to = extractRecipient(row)

  const replyBase = {
    senderPhoneHash: row.sender_phone_hash,
    phoneLinkId: row.phone_link_id,
    conversationId: row.conversation_id,
    correlationId: row.correlation_id,
  }

  try {
    if (!row.phone_link_id || !row.media_id || !to) {
      await markStatus(supabase, row.id, 'error', {
        errorMessage: 'Message row is missing link or media reference',
      })
      return { kind: 'none' }
    }

    if (row.wamid) await markReadWithTyping(row.wamid)

    const link = await loadLink(supabase, row.phone_link_id)
    if (!link || link.revoked_at) {
      await markStatus(supabase, row.id, 'skipped', {
        errorMessage: 'Phone link revoked before processing',
      })
      return { kind: 'none' }
    }

    const conversation = row.conversation_id
      ? await loadConversation(supabase, row.conversation_id)
      : await getOrCreateConversation(supabase, link.id)

    // ── MIME allowlist (before staging: never park junk) ───
    const mime = (row.media_mime ?? '').split(';')[0].trim().toLowerCase()
    if (!CHAT_ALLOWED_MIME_TYPES.has(mime)) {
      await sendText(supabase, { to, body: copy.m15Unsupported(), template: TEMPLATE.m15Unsupported, ...replyBase })
      await markStatus(supabase, row.id, 'skipped', {
        errorMessage: `Unsupported media type: ${mime || 'unknown'}`,
      })
      return { kind: 'media_processed', conversationId: conversation?.id ?? null }
    }

    // ── Company resolution ─────────────────────────────────
    const resolved = await resolveCompanyTarget(supabase, link, conversation, opts.companySelectedVia)
    if (!resolved) {
      if (!conversation) {
        await markStatus(supabase, row.id, 'error', {
          errorMessage: 'No conversation available for company question',
        })
        return { kind: 'none' }
      }
      // Park the row (the row itself is the staged media ref: media stays
      // re-downloadable from Meta for days) and ask the company question
      // exactly once per episode.
      await markStatus(supabase, row.id, 'skipped', { errorMessage: STAGED_AWAITING_COMPANY })
      const { count } = 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)
      await askCompanyQuestion(supabase, {
        conversation,
        link,
        to,
        replyBase,
        stagedCount: count ?? 1,
      })
      return { kind: 'media_staged', conversationId: conversation.id }
    }
    const companyId = resolved.companyId

    // ── Per-company intake quota (ack-and-drop, never retryable) ──
    const limit = await checkInboxUploadRateLimit(supabase, companyId)
    if (!limit.ok) {
      try {
        await appendProcessingHistory({
          companyId,
          correlationId: row.correlation_id ?? row.id,
          aggregateType: 'System',
          aggregateId: row.id,
          eventType: 'RateLimitedDropped',
          payload: {
            channel: 'whatsapp',
            scope: limit.scope,
            retry_after_sec: limit.retryAfterSec,
            whatsapp_message_id: row.id,
          },
          actor: { type: 'system', id: 'whatsapp-inbound' },
          occurredAt: new Date(),
        })
      } catch (err) {
        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 })
      }
      await markStatus(supabase, row.id, 'skipped', { errorMessage: 'Rate limited' })
      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)

    // ── Exact duplicate check within the company ───────────
    const sha256 = await computeSHA256(media.buffer)
    const { data: duplicate } = await supabase
      .from('document_attachments')
      .select('id')
      .eq('company_id', companyId)
      .eq('sha256_hash', sha256)
      .limit(1)
      .maybeSingle()
    if (duplicate) {
      await sendText(supabase, { to, body: copy.m4Duplicate(), template: TEMPLATE.m4Duplicate, ...replyBase })
      await markStatus(supabase, row.id, 'skipped', { errorMessage: 'Duplicate document (sha256)' })
      return { kind: 'media_processed', conversationId: conversation?.id ?? null }
    }

    // ── 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',
      },
    )

    // Record how the company was chosen (audit + FieldsRail in PR5).
    await updateItemContext(supabase, result.inbox_item_id, (context) => ({
      ...context,
      company_selected_via: resolved.via,
    }))

    // A new file while a re-send question is open resolves that question:
    // the fresh item supersedes the unreadable one (WORM + anchored-doc
    // invariant forbid swapping the old item's document).
    if (
      conversation?.state === 'awaiting_resend' &&
      getContext(conversation).pending_question?.type === 'resend'
    ) {
      await resolveResendQuestion(supabase, conversation)
    }

    await markStatus(supabase, row.id, 'done', { inboxItemId: result.inbox_item_id })
    // 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 })
    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 })
      }
    } catch (innerErr) {
      log.error('Failed to record WhatsApp processing error', innerErr, { messageId: row.id })
    }
    return { kind: 'media_processed', conversationId: row.conversation_id }
  }
}

/** Mark the awaiting_resend question answered: old item superseded. */
async function resolveResendQuestion(
  supabase: SupabaseClient,
  conversation: WhatsAppConversation,
): Promise<void> {
  const context = getContext(conversation)
  const pending = context.pending_question
  if (!pending || pending.type !== 'resend' || !pending.inbox_item_id) return
  const oldItemId = pending.inbox_item_id

  await updateItemContext(supabase, oldItemId, (itemContext) => ({
    ...itemContext,
    quality: {
      resend_requested_at: itemContext.quality?.resend_requested_at ?? pending.asked_at,
      resent: true,
      superseded: true,
    },
    pending_question: itemContext.pending_question
      ? { ...itemContext.pending_question, status: 'answered' }
      : { 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<string, unknown> })
    .eq('id', conversation.id)
  await appendQuestionHistory(supabase, {
    inboxItemId: oldItemId,
    eventType: 'ChannelQuestionAnswered',
    questionType: 'resend',
  })
}

// ── Burst finalize: ONE combined ack + at most one question ──

interface BurstEntry {
  row: WhatsAppMessage
  itemId: string
  companyId: string | null
  extracted: InvoiceExtractionResult | null
  channelContext: InboxChannelContext | null
  fileSizeBytes: number | null
  ordinal: number
}

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'}`
}

/**
 * Claim and send the combined ack for everything processed in the burst.
 * Safe to call speculatively: losers of the pending_ack claim (or claims
 * that find nothing to ack) do nothing.
 */
export async function finalizeBurst(
  supabase: SupabaseClient,
  conversationId: string,
): Promise<void> {
  try {
    if (!(await claimAck(supabase, conversationId))) return
    const conversation = await loadConversation(supabase, conversationId)
    if (!conversation) return

    const { data: rowsData } = await supabase
      .from('whatsapp_messages')
      .select('*')
      .eq('conversation_id', conversationId)
      .eq('direction', 'inbound')
      .eq('processing_status', 'done')
      .is('acked_at', null)
      .not('inbox_item_id', 'is', null)
      .order('created_at', { ascending: true })
      .limit(MAX_BURST_ROWS)
    const rows = (rowsData ?? []) as WhatsAppMessage[]
    if (rows.length === 0) return

    const now = new Date()
    const nowIso = now.toISOString()
    const rowIds = rows.map((r) => r.id)

    const to = rows.map(extractRecipient).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)) {
      await supabase.from('whatsapp_messages').update({ acked_at: nowIso }).in('id', rowIds)
      return
    }

    // ── Load items + document sizes ────────────────────────
    const itemIds = rows.map((r) => r.inbox_item_id as string)
    const { data: itemsData } = await supabase
      .from('invoice_inbox_items')
      .select('id, company_id, extracted_data, channel_context, document_id')
      .in('id', itemIds)
    const itemsById = new Map(
      ((itemsData ?? []) as Array<{
        id: string
        company_id: string
        extracted_data: Record<string, unknown> | null
        channel_context: InboxChannelContext | null
        document_id: string | null
      }>).map((item) => [item.id, item]),
    )
    const documentIds = [...itemsById.values()]
      .map((item) => item.document_id)
      .filter((id): id is string => id != null)
    const sizesByDocId = new Map<string, number>()
    if (documentIds.length > 0) {
      const { data: docsData } = await supabase
        .from('document_attachments')
        .select('id, file_size_bytes')
        .in('id', documentIds)
      for (const doc of (docsData ?? []) as Array<{ id: string; file_size_bytes: number | null }>) {
        if (doc.file_size_bytes != null) sizesByDocId.set(doc.id, doc.file_size_bytes)
      }
    }

    const entries: BurstEntry[] = rows.map((row, index) => {
      const item = itemsById.get(row.inbox_item_id as string)
      return {
        row,
        itemId: row.inbox_item_id as string,
        companyId: item?.company_id ?? null,
        extracted: (item?.extracted_data as InvoiceExtractionResult | null) ?? null,
        channelContext: item?.channel_context ?? null,
        fileSizeBytes: item?.document_id ? (sizesByDocId.get(item.document_id) ?? null) : null,
        ordinal: index + 1,
      }
    })

    // ── Question selection (budgeted, one in flight) ───────
    const context = getContext(conversation)
    const stateIdle = conversation.state === 'idle'
    const askedToday = questionsAskedToday(context, now)
    const queue: QueuedQuestion[] = [...(context.question_queue ?? [])]

    const candidates = entries
      .map((entry) => ({
        entry,
        candidate: evaluateQuestion({
          extracted: entry.extracted,
          caption: entry.channelContext?.caption ?? entry.row.body_text,
          mime: entry.row.media_mime,
          filename: entry.row.media_filename,
          fileSizeBytes: entry.fileSizeBytes,
        }),
      }))
      // A question is asked exactly once: an item that already carries one
      // (from a crashed earlier finalize) is never re-considered.
      .filter((c) => c.candidate != null && c.entry.channelContext?.pending_question == null)
      .sort(
        (a, b) =>
          QUESTION_PRIORITY[a.candidate!.type] - QUESTION_PRIORITY[b.candidate!.type] ||
          a.entry.ordinal - b.entry.ordinal,
      )

    let toAsk: { type: QuestionType; entry: BurstEntry } | null = null
    const movedToApp: { type: QuestionType; entry: BurstEntry }[] = []
    let burstAdmitted = 0
    for (const { entry, candidate } of candidates) {
      const type = candidate!.type
      if (askedToday >= MAX_QUESTIONS_PER_DAY) {
        movedToApp.push({ type, entry })
        continue
      }
      if (!toAsk && stateIdle && burstAdmitted < MAX_QUESTIONS_PER_BURST) {
        toAsk = { type, entry }
        burstAdmitted++
        continue
      }
      if (burstAdmitted < MAX_QUESTIONS_PER_BURST && queue.length < MAX_QUESTIONS_PER_BURST) {
        queue.push({ type, inbox_item_id: entry.itemId })
        burstAdmitted++
        continue
      }
      movedToApp.push({ type, entry })
    }

    // ── Compose the ONE outbound message ───────────────────
    const copy = botCopy('sv')
    let body: string
    let template: TemplateId
    let ackItemId: string | null = null

    if (entries.length === 1) {
      const entry = entries[0]
      ackItemId = entry.itemId
      const total = entry.extracted?.totals?.total ?? null
      if (toAsk?.type === 'resend') {
        body = copy.m9Resend({})
        template = TEMPLATE.m9Resend
      } else if (toAsk?.type === 'context') {
        body = copy.m10Context({})
        template = TEMPLATE.m10Context
      } else {
        body =
          total != null
            ? copy.m4Ack({
                merchant: entry.extracted?.supplier?.name ?? null,
                amount: formatSek(total),
                date: entry.extracted?.invoice?.invoiceDate ?? null,
              })
            : copy.m4AckEmpty()
        template = total != null ? TEMPLATE.m4Ack : TEMPLATE.m4AckEmpty
        if (toAsk?.type === 'representation') {
          body += `\n\n${copy.m7Representation({})}`
          template = TEMPLATE.m7Representation
        }
      }
    } else {
      body = copy.m5BurstAck({ lines: entries.map(ackLine) })
      template = TEMPLATE.m5BurstAck
      if (toAsk) {
        const ordinal = toAsk.entry.ordinal
        if (toAsk.type === 'representation') {
          body += `\n\n${copy.m7Representation({ ref: `Kvitto ${ordinal} ser ut som representation.` })}`
          template = TEMPLATE.m7Representation
        } else if (toAsk.type === 'resend') {
          body += `\n\n${copy.m9Resend({ ordinal })}`
          template = TEMPLATE.m9Resend
        } else {
          body += `\n\n${copy.m10Context({ ordinal })}`
          template = TEMPLATE.m10Context
        }
        ackItemId = toAsk.entry.itemId
      }
    }

    // ── 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<string, unknown>,
        last_outbound_at: nowIso,
      })
      .eq('id', conversationId)

    if (toAsk) {
      await updateItemContext(supabase, toAsk.entry.itemId, (itemContext) => ({
        ...itemContext,
        pending_question: { type: toAsk!.type, asked_at: nowIso, status: 'open' },
        ...(toAsk!.type === 'resend'
          ? { quality: { ...itemContext.quality, resend_requested_at: nowIso } }
          : {}),
      }))
      await appendQuestionHistory(supabase, {
        inboxItemId: toAsk.entry.itemId,
        eventType: 'ChannelQuestionAsked',
        questionType: toAsk.type,
      })
    }
    for (const moved of movedToApp) {
      await updateItemContext(supabase, moved.entry.itemId, (itemContext) => ({
        ...itemContext,
        pending_question: { type: moved.type, asked_at: nowIso, status: 'moved_to_app' },
      }))
    }

    const last = rows[rows.length - 1]
    await sendText(supabase, {
      to,
      body,
      template,
      senderPhoneHash: last.sender_phone_hash,
      phoneLinkId: conversation.phone_link_id,
      conversationId,
      correlationId: last.correlation_id,
      inboxItemId: ackItemId,
    })

    await supabase.from('whatsapp_messages').update({ acked_at: nowIso }).in('id', rowIds)
  } catch (err) {
    log.error('burst finalize failed', err, { conversationId })
  }
}

// ── Answer handling (the only path that may reach the LLM) ───

function renderParticipants(
  participants: { name: string; company: string | null }[],
): string {
  return participants
    .map((p) => (p.company ? `${p.name} (${p.company})` : p.name))
    .join(', ')
}

async function processAnswerMessage(
  supabase: SupabaseClient,
  row: WhatsAppMessage,
): Promise<ProcessOutcome> {
  const copy = botCopy('sv')
  const to = extractRecipient(row)
  const replyBase = {
    senderPhoneHash: row.sender_phone_hash,
    phoneLinkId: row.phone_link_id,
    conversationId: row.conversation_id,
    correlationId: row.correlation_id,
  }

  try {
    if (!row.phone_link_id || !to) {
      await markStatus(supabase, row.id, 'error', { errorMessage: 'Answer row is missing link' })
      return { kind: 'none' }
    }
    const link = await loadLink(supabase, row.phone_link_id)
    if (!link || link.revoked_at) {
      await markStatus(supabase, row.id, 'skipped', { errorMessage: 'Phone link revoked' })
      return { kind: 'none' }
    }
    const conversation = row.conversation_id
      ? await loadConversation(supabase, row.conversation_id)
      : null
    const text = (row.body_text ?? '').trim()
    if (!conversation || !text) {
      await markStatus(supabase, row.id, 'skipped', { errorMessage: 'Nothing to answer' })
      return { kind: 'none' }
    }

    const raw = row.raw_payload as { context?: { id?: unknown } } | null
    const quotedWamid = typeof raw?.context?.id === 'string' ? raw.context.id : null
    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 })
      await markStatus(supabase, row.id, 'done')
      return { kind: 'answer', conversationId: conversation.id }
    }

    const nowIso = new Date().toISOString()
    let confirmBody: string
    let confirmTemplate: TemplateId

    if (target.type === 'representation' && text.toLowerCase() === 'nej') {
      // Exact 'nej' short-circuits WITHOUT an LLM call.
      await updateItemContext(supabase, target.inboxItemId, (itemContext) => ({
        ...itemContext,
        representation: {
          participants: [],
          purpose: null,
          event_date: null,
          raw_answer: text,
          answered_at: nowIso,
          denied: true,
        },
        pending_question: itemContext.pending_question
          ? { ...itemContext.pending_question, status: 'answered' }
          : undefined,
      }))
      confirmBody = copy.m8RepDenied()
      confirmTemplate = TEMPLATE.m8RepDenied
    } else {
      // Rate-limit gate, then the ONE LLM call. Any failure (gate, API,
      // parse, validation) degrades to storing the raw text: never a retry
      // loop, never a user-visible error.
      let interpretation: Awaited<ReturnType<typeof interpretChatAnswer>> = { ok: false }
      const rate = await checkAgentRateLimit(supabase, link.user_id)
      if (rate.ok) {
        interpretation = await interpretChatAnswer({ text, questionType: target.type })
      }

      if (target.type === 'representation') {
        if (interpretation.ok && interpretation.data.is_denial) {
          await updateItemContext(supabase, target.inboxItemId, (itemContext) => ({
            ...itemContext,
            representation: {
              participants: [],
              purpose: null,
              event_date: null,
              raw_answer: text,
              answered_at: nowIso,
              denied: true,
            },
            pending_question: itemContext.pending_question
              ? { ...itemContext.pending_question, status: 'answered' }
              : undefined,
          }))
          confirmBody = copy.m8RepDenied()
          confirmTemplate = TEMPLATE.m8RepDenied
        } else if (
          interpretation.ok &&
          ((interpretation.data.participants?.length ?? 0) > 0 || interpretation.data.purpose)
        ) {
          const data = interpretation.data
          await updateItemContext(supabase, target.inboxItemId, (itemContext) => ({
            ...itemContext,
            representation: {
              participants: data.participants ?? [],
              purpose: data.purpose,
              event_date: data.event_date,
              raw_answer: text,
              answered_at: nowIso,
            },
            ...(data.note ? { user_note: data.note } : {}),
            pending_question: itemContext.pending_question
              ? { ...itemContext.pending_question, status: 'answered' }
              : undefined,
          }))
          confirmBody = copy.m8RepConfirmed({
            participants: renderParticipants(data.participants ?? []),
            purpose: data.purpose,
          })
          confirmTemplate = TEMPLATE.m8RepConfirmed
        } else {
          await updateItemContext(supabase, target.inboxItemId, (itemContext) => ({
            ...itemContext,
            user_note: text,
            representation: {
              participants: [],
              purpose: null,
              event_date: null,
              raw_answer: text,
              answered_at: nowIso,
            },
            pending_question: itemContext.pending_question
              ? { ...itemContext.pending_question, status: 'answered' }
              : undefined,
          }))
          confirmBody = copy.m8RepPartial()
          confirmTemplate = TEMPLATE.m8RepPartial
        }
      } else {
        const note =
          interpretation.ok && interpretation.data.note ? interpretation.data.note : text
        await updateItemContext(supabase, target.inboxItemId, (itemContext) => ({
          ...itemContext,
          user_note: note,
          pending_question: itemContext.pending_question
            ? { ...itemContext.pending_question, status: 'answered' }
            : undefined,
        }))
        confirmBody = copy.m10ContextConfirm()
        confirmTemplate = TEMPLATE.m10ContextConfirm
      }
    }

    // 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'),
    }
    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<string, unknown> })
      .eq('id', conversation.id)

    await appendQuestionHistory(supabase, {
      inboxItemId: target.inboxItemId,
      eventType: 'ChannelQuestionAnswered',
      questionType: target.type,
      correlationId: row.correlation_id,
    })

    await sendText(supabase, {
      to,
      body: confirmBody,
      template: confirmTemplate,
      ...replyBase,
      inboxItemId: target.inboxItemId,
    })

    if (nextState === 'idle') {
      await askNextQueuedQuestion(supabase, conversation.id, to, replyBase)
    }

    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).
    log.error('WhatsApp answer processing failed', err, { messageId: row.id })
    try {
      await markStatus(supabase, row.id, 'error', {
        errorMessage: (err instanceof Error ? err.message : String(err)).slice(0, 500),
      })
    } catch (innerErr) {
      log.error('Failed to record WhatsApp answer error', innerErr, { messageId: row.id })
    }
    return { kind: 'none' }
  }
}

/** Ask the next queued question, if any survives the daily budget. */
async function askNextQueuedQuestion(
  supabase: SupabaseClient,
  conversationId: string,
  to: string,
  replyBase: {
    senderPhoneHash: string | null
    phoneLinkId: string | null
    conversationId: string | null
    correlationId: string | null
  },
): Promise<void> {
  const conversation = await loadConversation(supabase, conversationId)
  if (!conversation || conversation.state !== 'idle') return
  const context = getContext(conversation)
  const queue: QueuedQuestion[] = [...(context.question_queue ?? [])]
  if (queue.length === 0) return

  const now = new Date()
  const nowIso = now.toISOString()
  const askedToday = questionsAskedToday(context, now)
  const windowOpen = serviceWindowOpen(conversation, now)

  let next: QueuedQuestion | null = null
  const moved: QueuedQuestion[] = []
  while (queue.length > 0) {
    const candidate = queue.shift()!
    if (!windowOpen || askedToday >= MAX_QUESTIONS_PER_DAY) {
      moved.push(candidate)
      continue
    }
    next = candidate
    break
  }
  for (const item of moved) {
    await updateItemContext(supabase, item.inbox_item_id, (itemContext) => ({
      ...itemContext,
      pending_question: { type: item.type, asked_at: nowIso, status: 'moved_to_app' },
    }))
  }
  if (!next) {
    await supabase
      .from('whatsapp_conversations')
      .update({ context: { ...context, question_queue: [] } as Record<string, unknown> })
      .eq('id', conversationId)
    return
  }

  const { data: itemData } = await supabase
    .from('invoice_inbox_items')
    .select('id, extracted_data')
    .eq('id', next.inbox_item_id)
    .maybeSingle()
  const extracted =
    ((itemData as { extracted_data?: Record<string, unknown> | null } | null)?.extracted_data as
      | InvoiceExtractionResult
      | null
      | undefined) ?? null
  const merchant = extracted?.supplier?.name ?? null

  const copy = botCopy('sv')
  let body: string
  let template: TemplateId
  if (next.type === 'representation') {
    body = copy.m7Representation({
      ref: merchant ? `Kvittot från *${merchant}* ser ut som representation.` : null,
    })
    template = TEMPLATE.m7Representation
  } else if (next.type === 'resend') {
    body = copy.m9Resend({})
    template = TEMPLATE.m9Resend
  } else {
    body = copy.m10Context({})
    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<string, unknown>,
      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' },
    ...(next!.type === 'resend'
      ? { quality: { ...itemContext.quality, resend_requested_at: nowIso } }
      : {}),
  }))
  await appendQuestionHistory(supabase, {
    inboxItemId: next.inbox_item_id,
    eventType: 'ChannelQuestionAsked',
    questionType: next.type,
  })

  await sendText(supabase, {
    to,
    body,
    template,
    ...replyBase,
    inboxItemId: next.inbox_item_id,
  })
}

// ── Kick: the after() entry point ────────────────────────────

/**
 * Schedule processing of freshly persisted message rows after the webhook
 * response is sent, then finalize the burst ack for every touched
 * conversation. Exact dispatch-kick idiom: never awaited by the caller,
 * never throws, falls back to a microtask outside a request scope (tests).
 */
export function kickInboundProcessing(messageIds: string[], opts: KickOptions = {}): void {
  if (messageIds.length === 0) return

  const run = async (): Promise<void> => {
    try {
      const supabase = createServiceClientNoCookies()
      const conversationIds = new Set<string>()
      for (const id of messageIds) {
        const outcome = await processInboundMessage(supabase, id, opts)
        if (outcome.kind === 'media_processed' && outcome.conversationId) {
          conversationIds.add(outcome.conversationId)
        }
      }
      for (const conversationId of conversationIds) {
        const conversation = await loadConversation(supabase, conversationId)
        if (!conversation?.pending_ack) continue
        const waitMs = conversation.debounce_until
          ? Math.max(0, new Date(conversation.debounce_until).getTime() - Date.now())
          : 0
        // Sleep to this invocation's own deadline, then attempt the claim.
        // If a newer message pushed debounce_until forward meanwhile, the
        // claim fails here and the newer invocation's later claim wins.
        await sleepFn(Math.min(waitMs, MAX_DEBOUNCE_WAIT_MS))
        await finalizeBurst(supabase, conversationId)
      }
    } catch (err) {
      // The sweep cron re-claims 'received' rows: a failed kick is a
      // latency regression, never a lost message.
      log.warn('deferred WhatsApp processing failed; sweep will retry', {
        error: err instanceof Error ? err.message : String(err),
      })
    }
  }

  try {
    after(() => run())
  } catch {
    queueMicrotask(() => void run())
  }
}