/*
 * Does the reply's own arithmetic close.
 *
 * ── Why this exists (2026-09-07) ───────────────────────────────────────────
 * The contract now declares the response a sender reads, and the reason it had to is that a
 * sender guessed instead: operato-plant read `rejected` as a list of envelope id strings, the
 * receiver sent `{record, errors}`, `Array.isArray` passed on both, and a batch of three was
 * stamped delivered while one envelope never landed (mes_outbox_events seq 178 — SENT,
 * attempts 0, no error).
 *
 * `offered` is the field that makes that catchable without reading the rejection list at all:
 *
 *   offered === applied + duplicates + rejected.length + vocabulary
 *
 * ── What this file is really guarding ──────────────────────────────────────
 * The first version of that sum left `vocabulary` out, and it was wrong. Vocabulary elements are
 * counted in `offered` and deliberately kept out of the other three — they set state and must
 * never reach the journal, because they carry no time axis. So every batch carrying one would
 * have failed a sender's check, the warning would have been ignored within a week, and the real
 * loss would have been ignored with it.
 *
 * The integration lane found that by looking for a counterexample rather than a confirmation,
 * and had a measured one: 32 distinct facts reconciled as 29 journal + 2 rejected + 1 vocabulary.
 *
 * The count already existed — `liveIngest` has returned `vocabulary` all along — and this
 * boundary dropped it, because the callback was typed as `IngestResult` and the field was not on
 * it. A declared field nobody fills is the same defect one layer over, so what is asserted here
 * is that a number handed to `ingest` arrives in the reply, and that the sum closes on a batch
 * where it is not zero.
 */
import { test } from 'node:test'
import assert from 'node:assert/strict'

import { WEBHOOK_STATUS } from '@operato/ops-contract'

import { handleHook, type HookStore } from '../server/service/reference/reference-hook.js'
import { registerAdapterType, getAdapter } from '../server/service/reference/reference-adapter.js'

const DOMAIN = 'd-1'
const SOURCE = 'sum-source'
const TWIN = 'twin-a'
const SECRET = 'shared-secret'

/** Hands every item straight through — this file is about the counts, not about translation. */
const ADAPTER_TYPE = 'test-sum-push'
if (!getAdapter(ADAPTER_TYPE)) {
  registerAdapterType({
    type: ADAPTER_TYPE,
    label: 'sum push (test)',
    handleInbound: (_cfg: any, _site: any, body: any) => ({ items: body?.items ?? [] })
  } as any)
}

function makeStore(): HookStore {
  const row = {
    id: 'ref-1',
    source: SOURCE,
    adapterType: ADAPTER_TYPE,
    connectionConfig: { hookSecret: SECRET },
    scopeSpec: { produced: [{ siteId: 'site-a', instanceId: TWIN }] }
  }
  return {
    load: async (domainId: string, source: string) => (domainId === DOMAIN && source === SOURCE ? row : null),
    saveCursor: async () => {}
  } as any
}

/** `count` unnumbered items — no `scope`, so sequence checking stays out of the way. */
function items(count: number) {
  return Array.from({ length: count }, (_v, i) => ({ record: { moverId: `eq-${i}`, kind: 'press', status: 'idle' } }))
}

function push(count: number, ingest: any) {
  return handleHook({
    domainId: DOMAIN,
    source: SOURCE,
    instanceId: TWIN,
    headers: { 'x-twin-hook-secret': SECRET },
    body: { items: items(count) },
    nowMs: Date.parse('2026-09-07T02:00:00.000Z'),
    ingest,
    store: makeStore()
  })
}

/** What the sender is told to check. Written once so every case below reads the same way. */
function closes(body: any) {
  return body.offered === body.applied + body.duplicates + (body.rejected?.length ?? 0) + body.vocabulary
}

test('the sum closes when everything was applied', async () => {
  const out = await push(3, (_id: string, r: unknown[]) => ({ applied: r.length, duplicates: 0, rejected: [] }))

  assert.equal(out.status, WEBHOOK_STATUS.ok)
  assert.equal(out.body.offered, 3)
  assert.equal(out.body.vocabulary, 0)
  assert.ok(closes(out.body), JSON.stringify(out.body))
})

test('the sum closes when a batch carried vocabulary — this is the case that failed', async () => {
  /*
   * Three offered: two events seated, one vocabulary element. Without `vocabulary` in the reply
   * the sender sees `3 - 2 - 0 - 0 = 1` and calls it a loss, on traffic that is entirely correct.
   */
  const out = await push(3, () => ({ applied: 2, duplicates: 0, rejected: [], vocabulary: 1 }))

  assert.equal(out.body.offered, 3)
  assert.equal(out.body.applied, 2)
  assert.equal(out.body.vocabulary, 1, 'the count exists in liveIngest; this boundary must carry it')
  assert.ok(closes(out.body), JSON.stringify(out.body))
})

test('a batch that is only vocabulary is not read as a dead twin', async () => {
  /*
   * This is what the sum test found, and it was a live defect rather than a missing field.
   *
   * The 503 branch inferred "the twin is not running live" from applied, duplicates and rejected
   * all being zero. A vocabulary-only batch is exactly that shape and yet everything in it
   * landed, so the receiver answered 503 to correct traffic — and a 503 means hold and retry, so
   * the sender would have re-sent that batch every thirty seconds indefinitely.
   *
   * An absence is not a cause. The fourth term is what tells the two apart.
   */
  const out = await push(2, () => ({ applied: 0, duplicates: 0, rejected: [], vocabulary: 2 }))

  assert.equal(out.status, WEBHOOK_STATUS.ok, 'zero events applied is not a twin that stopped')
  assert.equal(out.body.applied, 0)
  assert.equal(out.body.vocabulary, 2)
  assert.ok(closes(out.body), 'zero applied is not zero received')
})

test('a dead twin is still reported as one — the fourth term did not loosen the check', async () => {
  /*
   * Breaking the case above must not cost the case it was guarding. Nothing applied and no
   * vocabulary either is the real "nothing landed", and it has to keep answering 503: a 200 there
   * would have the sender mark those facts sent and drop them.
   */
  const out = await push(2, () => ({ applied: 0, duplicates: 0, rejected: [], vocabulary: 0 }))

  assert.equal(out.status, WEBHOOK_STATUS.notLive)
  assert.equal(out.body.retryAfterMs, 30_000, 'hold and come back, do not drop')
})

test('the sum closes on a partial — the rejected list is a term in it', async () => {
  const out = await push(3, (_id: string, r: unknown[]) => ({
    applied: 1,
    duplicates: 0,
    rejected: [{ record: r[1], errors: ['no inputQuantityList'] }],
    vocabulary: 1
  }))

  assert.equal(out.status, WEBHOOK_STATUS.partial, '422 is what a sender must not resend')
  assert.equal(out.body.rejected?.length, 1)
  assert.ok(closes(out.body), JSON.stringify(out.body))
})

test('duplicates are a term too — a resend is normal here, not a loss', async () => {
  const out = await push(4, () => ({ applied: 1, duplicates: 3, rejected: [] }))

  assert.equal(out.body.duplicates, 3)
  assert.ok(closes(out.body))
})

test('an ingest that says nothing about vocabulary still reports the term', async () => {
  /*
   * `vocabulary` is optional on `IngestResult` so a caller that does not count it need not claim
   * none. The reply must still carry the term: a field that appears only sometimes leaves the
   * sender unable to tell "none" from "this receiver does not count it", which is the same
   * ambiguity that made `rejected` guessable.
   */
  const out = await push(2, () => ({ applied: 2, duplicates: 0, rejected: [] }))

  assert.equal(out.body.vocabulary, 0)
  assert.ok('vocabulary' in (out.body as any))
})

test('seq 178 as it would read now — three offered, two applied', async () => {
  /*
   * The real batch: one of the three had no `inputQuantityList` and the kernel was right to
   * refuse it. What went wrong was that the sender could not read the refusal, so it marked all
   * three sent. The sum alone says otherwise, whatever shape the rejection list is in.
   */
  const out = await push(3, (_id: string, r: unknown[]) => ({
    applied: 2,
    duplicates: 0,
    rejected: [{ record: r[2], errors: ['transformation: inputQuantityList is empty'] }]
  }))

  assert.equal(out.body.offered, 3)
  assert.equal(out.body.applied, 2)
  assert.equal(out.body.offered - out.body.applied - out.body.duplicates - out.body.vocabulary, 1)
  assert.ok(closes(out.body), 'and the missing one is accounted for by the rejected list')
})