import { describe, it, expect, beforeEach, afterEach } from 'vitest'
import { mkdtempSync, mkdirSync, writeFileSync, rmSync, unlinkSync, utimesSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { walkExposed, pushAccount } from '../../bin/portal-index-push.mjs'

let dir: string

function file(rel: string, body = 'x') {
  const abs = join(dir, rel)
  mkdirSync(join(abs, '..'), { recursive: true })
  writeFileSync(abs, body)
}

function emptyDir(rel: string) {
  mkdirSync(join(dir, rel), { recursive: true })
}

const SCHEMA = [
  '```allowed-top-level',
  'output',
  'quotes',
  'documents',
  'uploads',
  '```',
  '',
  '<!-- ontology-buckets:start -->',
  '- `quotes/` - one folder per Quotation record.',
  '<!-- ontology-buckets:end -->',
].join('\n')

beforeEach(() => {
  dir = mkdtempSync(join(tmpdir(), 'portal-push-'))
  writeFileSync(join(dir, 'SCHEMA.md'), SCHEMA)
  file('output/report.pdf')
  file('quotes/CR2969/quote.pdf')
  file('quotes/CR2974/quote.pdf')
  file('documents/private.pdf')
  file('uploads/client-sent.pdf')
  file('.claude/settings.json')
  file('output/.uploads-tmp/partial.bin')
})
afterEach(() => rmSync(dir, { recursive: true, force: true }))

function fakeClient() {
  const calls: { sql: string; params: unknown[] }[] = []
  const rows = new Set<string>()
  // The pointer is modelled because the flip is read back (Task 1926): a client
  // that always answers "no pointer" would make every push read as superseded.
  let pointer: number | null = null
  return {
    calls,
    async query(sql: string, params: unknown[] = []) {
      calls.push({ sql, params })
      if (/SELECT\s+currentGeneration/i.test(sql)) {
        return pointer === null ? [] : [{ currentGeneration: pointer }]
      }
      if (/INSERT INTO directory_state/i.test(sql)) {
        pointer = Number(params[1])
        return []
      }
      // The open paren is load-bearing: `INSERT INTO directory_state` would
      // otherwise register as a row insert and inflate every count.
      if (/INSERT INTO directory \(/i.test(sql)) {
        rows.add(String(params[1]))
        return []
      }
      if (/DELETE FROM directory\b/i.test(sql)) {
        return []
      }
      if (/SELECT COUNT/i.test(sql)) return [{ n: rows.size }]
      return []
    },
  }
}

const push = (client: unknown, overrides: Record<string, unknown> = {}) =>
  pushAccount({
    accountDir: dir,
    accountId: 'acc-1',
    dbName: 'db',
    client: client as never,
    log: () => {},
    nowIso: '2026-07-20T00:00:00.000Z',
    ...overrides,
  })

describe('walkExposed', () => {
  it('returns files from exposed dirs, at any depth', async () => {
    const paths = (await walkExposed(dir, ['output', 'quotes']))
      .filter((r) => !r.isDir)
      .map((r) => r.relPath)
      .sort()
    expect(paths).toEqual(['output/report.pdf', 'quotes/CR2969/quote.pdf', 'quotes/CR2974/quote.pdf'])
  })

  it('never returns operator data, uploads, dot dirs or upload temps', async () => {
    const paths = (await walkExposed(dir, ['output', 'quotes'])).map((r) => r.relPath)
    expect(paths.some((p) => p.startsWith('documents/'))).toBe(false)
    expect(paths.some((p) => p.startsWith('uploads/'))).toBe(false)
    expect(paths.some((p) => p.includes('.uploads-tmp'))).toBe(false)
    expect(paths.some((p) => p.includes('.claude'))).toBe(false)
  })

  it('returns nothing for a dir that does not exist', async () => {
    expect(await walkExposed(dir, ['nope'])).toEqual([])
  })

  it('carries size and an iso modified time', async () => {
    const row = (await walkExposed(dir, ['output'])).find((r) => !r.isDir)!
    expect(row.sizeBytes).toBe(1)
    expect(row.modifiedAt).toMatch(/^\d{4}-\d{2}-\d{2}T/)
  })

  // An exposed folder holding no files at any depth is the case the client
  // could not see at all: nothing derived a folder name from a file prefix.
  it('emits exactly one row for an exposed dir holding no files, marked isDir', async () => {
    emptyDir('jobs')
    const rows = await walkExposed(dir, ['jobs'])
    expect(rows.length).toBe(1)
    expect(rows[0].relPath).toBe('jobs')
    expect(rows[0].isDir).toBe(true)
    expect(rows[0].sizeBytes).toBe(0)
    expect(rows[0].modifiedAt).toMatch(/^\d{4}-\d{2}-\d{2}T/)
  })

  it('emits the dir row and the file row for a dir holding a nested file', async () => {
    const rows = await walkExposed(dir, ['quotes'])
    const byPath = new Map(rows.map((r) => [r.relPath, r]))
    expect(byPath.get('quotes')?.isDir).toBe(true)
    expect(byPath.get('quotes/CR2969')?.isDir).toBe(true)
    expect(byPath.get('quotes/CR2969/quote.pdf')?.isDir).toBe(false)
  })

  // The dir row's time is the directory's own, not the walk's: a folder whose
  // contents changed is what the operator reads, and `new Date()` here would
  // make every row look freshly touched every cycle.
  it('takes the dir mtime rather than the time of the walk', async () => {
    emptyDir('jobs')
    const past = new Date('2020-03-04T05:06:07.000Z')
    utimesSync(join(dir, 'jobs'), past, past)
    const rows = await walkExposed(dir, ['jobs'])
    expect(rows[0].modifiedAt).toBe('2020-03-04T05:06:07.000Z')
  })

  it('emits no row for a dot-prefixed dir at any level', async () => {
    const paths = (await walkExposed(dir, ['output'])).map((r) => r.relPath)
    expect(paths).not.toContain('output/.uploads-tmp')
  })
})

describe('pushAccount', () => {
  // 7 rows: the 3 files plus output/, quotes/, quotes/CR2969/, quotes/CR2974/.
  it('stages the current walk, then flips the pointer to it', async () => {
    // Task 1910 replaced delete-then-insert with stage-flip-sweep. The first
    // statement now READS the pointer rather than emptying the table, which is
    // what removes the window where a client saw nothing.
    const c = fakeClient()
    const r = await push(c)
    expect(r.exposed).toEqual(['output', 'quotes'])
    expect(r.rows).toBe(7)
    expect(/SELECT currentGeneration/i.test(c.calls[0].sql)).toBe(true)
    expect(c.calls[0].params).toEqual(['acc-1'])
    expect(c.calls.filter((x) => /INSERT INTO directory \(/i.test(x.sql)).length).toBe(7)
  })

  // The folder itself survives its last file: emptied is not the same as gone,
  // and the client still needs somewhere to put the next one.
  it('converges on a delete: a removed file leaves no stale row', async () => {
    const c = fakeClient()
    await push(c)
    unlinkSync(join(dir, 'quotes/CR2974/quote.pdf'))
    // The fake accumulates calls across pushes; only the second one is the
    // converged state.
    const before = c.calls.length
    const second = await push(c)
    expect(second.rows).toBe(6)
    const inserted = c.calls
      .slice(before)
      .filter((x) => /INSERT INTO directory/i.test(x.sql))
      .map((x) => String(x.params[1]))
    expect(inserted).toContain('quotes/CR2974')
    expect(inserted).not.toContain('quotes/CR2974/quote.pdf')
  })

  it('marks each inserted row as a directory or not', async () => {
    const c = fakeClient()
    await push(c)
    const byPath = new Map(
      c.calls
        .filter((x) => /INSERT INTO directory/i.test(x.sql))
        .map((x) => [String(x.params[1]), x.params[5]]),
    )
    expect(byPath.get('quotes')).toBe(1)
    expect(byPath.get('quotes/CR2969')).toBe(1)
    expect(byPath.get('quotes/CR2969/quote.pdf')).toBe(0)
  })

  it('binds every value as a param rather than interpolating it', async () => {
    const c = fakeClient()
    await push(c)
    const insert = c.calls.find((x) => /INSERT INTO directory/i.test(x.sql))!
    expect(insert.sql).not.toContain('output/')
    expect(insert.params[0]).toBe('acc-1')
  })

  // Fail closed, but do NOT delete: an empty publish and a missing schema must
  // not look the same in the table.
  it('exposes nothing and writes nothing when the schema is missing', async () => {
    rmSync(join(dir, 'SCHEMA.md'))
    const c = fakeClient()
    const r = await push(c)
    expect(r.schemaPresent).toBe(false)
    expect(r.exposed).toEqual([])
    expect(c.calls.length).toBe(0)
  })

  it('logs resolve, stage, flip and verify lines', async () => {
    const lines: string[] = []
    const c = fakeClient()
    await push(c, { log: (l: string) => lines.push(l) })
    expect(lines.some((l) => l.includes('op=resolve') && l.includes('exposed=output,quotes'))).toBe(true)
    expect(lines.some((l) => l.includes('op=stage') && l.includes('inserted=7'))).toBe(true)
    expect(lines.some((l) => l.includes('op=flip') && l.includes('generation=1'))).toBe(true)
    expect(lines.some((l) => l.includes('op=verify') && l.includes('rowsAfter=7') && l.includes('expected=7'))).toBe(true)
  })

  it('reports an exposed-set change only when it actually changes', async () => {
    const lines: string[] = []
    // Same set as the fixture resolves to: silent.
    await push(fakeClient(), { log: (l: string) => lines.push(l), previousExposed: ['output', 'quotes'] })
    expect(lines.some((l) => l.includes('op=exposed-changed'))).toBe(false)

    // A bucket the client used to see is gone — the client-visible regression.
    lines.length = 0
    await push(fakeClient(), { log: (l: string) => lines.push(l), previousExposed: ['output', 'quotes', 'invoices'] })
    const line = lines.find((l) => l.includes('op=exposed-changed'))!
    expect(line).toContain('removed=invoices')
    expect(line).toContain('added=none')

    // A newly projected bucket.
    lines.length = 0
    await push(fakeClient(), { log: (l: string) => lines.push(l), previousExposed: ['output'] })
    expect(lines.find((l) => l.includes('op=exposed-changed'))).toContain('added=quotes')
  })

  it('stays silent about changes on a first run, when there is no previous set', async () => {
    const lines: string[] = []
    await push(fakeClient(), { log: (l: string) => lines.push(l) })
    expect(lines.some((l) => l.includes('op=exposed-changed'))).toBe(false)
  })

  it('logs schemaPresent=false so a fail-closed account is distinguishable', async () => {
    rmSync(join(dir, 'SCHEMA.md'))
    const lines: string[] = []
    await push(fakeClient(), { log: (l: string) => lines.push(l) })
    expect(lines.some((l) => l.includes('schemaPresent=false') && l.includes('exposed=none'))).toBe(true)
  })

  it('op=resolve carries extraFolders and the walk includes an operator folder', async () => {
    // inbound-invoices is NOT in the fixture allowed block, so this also proves
    // an extra bypasses the ontology/allowed intersection.
    file('inbound-invoices/INV-42.pdf')
    const lines: string[] = []
    const c = fakeClient()
    await push(c, { log: (l: string) => lines.push(l), exposeFolders: ['inbound-invoices'] })
    expect(lines.some((l) => /op=resolve .*extraFolders=1/.test(l))).toBe(true)
    expect(lines.some((l) => /exposed=.*inbound-invoices/.test(l))).toBe(true)
    const inserted = c.calls.filter((x) => /INSERT INTO directory/i.test(x.sql)).map((x) => String(x.params[1]))
    expect(inserted).toContain('inbound-invoices/INV-42.pdf')
  })

  it('with no exposeFolders the operator folder is not walked and extraFolders=0', async () => {
    file('inbound-invoices/INV-42.pdf')
    const lines: string[] = []
    const c = fakeClient()
    await push(c, { log: (l: string) => lines.push(l) })
    const inserted = c.calls.filter((x) => /INSERT INTO directory/i.test(x.sql)).map((x) => String(x.params[1]))
    expect(inserted).not.toContain('inbound-invoices/INV-42.pdf')
    expect(lines.some((l) => /op=resolve .*extraFolders=0/.test(l))).toBe(true)
  })
})

// The atomic replace (Task 1910), superseding backlog Task 1842.
//
// A generation-aware client: rows carry a generation, and a reader sees only
// the generation the pointer names. That is what makes "never a half-built
// tree" a measurable property rather than an assertion.
function generationClient() {
  const calls: { sql: string; params: unknown[] }[] = []
  /** @type {{accountId: string, relPath: string, generation: number}[]} */
  const rows: { accountId: string; relPath: string; generation: number }[] = []
  let pointer: number | null = null
  return {
    calls,
    rows,
    /** What a client listing right now would see. */
    visible() {
      return pointer === null ? [] : rows.filter((r) => r.generation === pointer)
    },
    pointerValue() {
      return pointer
    },
    async query(sql: string, params: unknown[] = []) {
      calls.push({ sql, params })
      if (/SELECT\s+currentGeneration/i.test(sql)) {
        return pointer === null ? [] : [{ currentGeneration: pointer }]
      }
      if (/INSERT INTO directory_state|UPDATE directory_state/i.test(sql)) {
        pointer = Number(params[1] ?? params[0])
        return []
      }
      if (/INSERT INTO directory\b/i.test(sql)) {
        rows.push({
          accountId: String(params[0]),
          relPath: String(params[1]),
          generation: Number(params[6]),
        })
        return []
      }
      if (/SELECT\s+MAX\(generation\)/i.test(sql)) {
        const mine = rows.filter((r) => r.accountId === String(params[0]))
        return [{ g: mine.length ? Math.max(...mine.map((r) => r.generation)) : null }]
      }
      if (/DELETE FROM directory\b/i.test(sql)) {
        const g = Number(params[1])
        // The post-flip sweep removes strictly OLDER generations
        // (`generation < ?`, Task 1926). `generation != ?` is the shape it
        // replaced and `generation = ?` was the Task 1923 pre-stage clear; both
        // stay modelled so a regression to either shape is visible here.
        for (let i = rows.length - 1; i >= 0; i--) {
          const n = rows[i].generation
          const hit = /generation\s*<\s*\?/i.test(sql)
            ? n < g
            : /generation\s*!=\s*\?/i.test(sql)
              ? n !== g
              : n === g
          if (hit) rows.splice(i, 1)
        }
        return []
      }
      if (/SELECT COUNT/i.test(sql)) {
        // The pre-stage orphan count (Task 1923) filters `generation = ?`; the
        // verify count joins the pointer. Only the former binds a generation.
        if (/generation\s*=\s*\?/i.test(sql)) {
          const g = Number(params[1])
          return [{ n: rows.filter((r) => r.generation === g).length }]
        }
        return [{ n: rows.filter((r) => r.generation === pointer).length }]
      }
      return []
    },
  }
}

// The wedge Task 1923 fixes: a stage that half-completed on a D1 500 leaves
// orphan rows at the NEXT generation while the pointer stays at current. This
// client models the one property that turns that into a permanent freeze — the
// D1 UNIQUE(accountId, relPath, generation) constraint — by throwing on a
// colliding INSERT, exactly as every later cycle's first re-stage did.
function wedgeableClient(seed: { pointer: number; current?: string[]; orphans: string[] }) {
  const calls: { sql: string; params: unknown[] }[] = []
  const rows: { accountId: string; relPath: string; generation: number }[] = []
  let pointer: number | null = seed.pointer
  const orphanGen = seed.pointer + 1
  // The last good tree the pointer names, present in a real wedge alongside the
  // orphans. Seeding it proves the pre-stage clear scopes to `next` and leaves
  // the reader-visible current generation untouched.
  for (const relPath of seed.current ?? []) rows.push({ accountId: 'acc-1', relPath, generation: seed.pointer })
  for (const relPath of seed.orphans) rows.push({ accountId: 'acc-1', relPath, generation: orphanGen })
  return {
    calls,
    rows,
    pointerValue: () => pointer,
    async query(sql: string, params: unknown[] = []) {
      calls.push({ sql, params })
      if (/SELECT\s+currentGeneration/i.test(sql)) {
        return pointer === null ? [] : [{ currentGeneration: pointer }]
      }
      if (/INSERT INTO directory_state|UPDATE directory_state/i.test(sql)) {
        pointer = Number(params[1] ?? params[0])
        return []
      }
      if (/INSERT INTO directory \(/i.test(sql)) {
        const accountId = String(params[0])
        const relPath = String(params[1])
        const generation = Number(params[6])
        if (rows.some((r) => r.accountId === accountId && r.relPath === relPath && r.generation === generation)) {
          throw new Error(
            'd1 query failed: HTTP 500 — UNIQUE constraint failed: directory.accountId, directory.relPath, directory.generation',
          )
        }
        rows.push({ accountId, relPath, generation })
        return []
      }
      if (/SELECT\s+MAX\(generation\)/i.test(sql)) {
        const mine = rows.filter((r) => r.accountId === String(params[0]))
        return [{ g: mine.length ? Math.max(...mine.map((r) => r.generation)) : null }]
      }
      if (/DELETE FROM directory\b/i.test(sql)) {
        const g = Number(params[1])
        for (let i = rows.length - 1; i >= 0; i--) {
          const n = rows[i].generation
          const hit = /generation\s*<\s*\?/i.test(sql)
            ? n < g
            : /generation\s*!=\s*\?/i.test(sql)
              ? n !== g
              : n === g
          if (hit) rows.splice(i, 1)
        }
        return []
      }
      if (/SELECT COUNT/i.test(sql)) {
        if (/generation\s*=\s*\?/i.test(sql)) {
          const g = Number(params[1])
          return [{ n: rows.filter((r) => r.generation === g).length }]
        }
        return [{ n: rows.filter((r) => r.generation === pointer).length }]
      }
      return []
    },
  }
}

describe('a partial stage self-heals (Task 1923, by the Task 1926 generation claim)', () => {
  // The wedge: a stage that died mid-batch left rows at current + 1 while the
  // pointer stayed at current, so every later run recomputed the SAME target
  // and its first INSERT hit UNIQUE(accountId, relPath, generation). The index
  // froze at the stale generation until an operator hand-cleared the orphans,
  // which is what held gls-data for ~6 h on 2026-07-22.
  //
  // Task 1923 healed that with a DELETE before the stage. Task 1926 heals it by
  // claiming above the orphans instead, which needs no DELETE before the flip
  // at all — and that pre-flip DELETE was the statement a concurrent peer could
  // aim at the live tree.
  it('claims above the orphans rather than colliding with them, and sweeps them after the flip', async () => {
    const c = wedgeableClient({
      pointer: 5,
      current: ['output', 'output/report.pdf', 'quotes', 'quotes/CR2969'],
      orphans: ['output', 'output/report.pdf', 'quotes'],
    })
    const lines: string[] = []
    const r = await push(c, { log: (l: string) => lines.push(l) })

    expect(r.rows).toBe(7)
    // 7, not 6: the orphans sit at 6, so this run claims the one above them.
    expect(c.pointerValue()).toBe(7)
    // Nothing survives at a generation the pointer does not name: the sweep
    // took both the stale tree at 5 and the orphans at 6.
    expect(c.rows.every((row) => row.generation === 7)).toBe(true)
    // No statement deletes anything before the flip any more.
    expect(lines.some((l) => l.includes('op=stage-clear'))).toBe(false)
  })

  it('re-stages onto a fresh generation with no orphans present', async () => {
    const c = wedgeableClient({ pointer: 5, orphans: [] })
    const lines: string[] = []
    await push(c, { log: (l: string) => lines.push(l) })
    expect(c.pointerValue()).toBe(6)
    expect(lines.some((l) => l.includes('op=stage-clear'))).toBe(false)
  })
})

describe('the index replace is atomic', () => {
  it('never issues a delete before the pointer flip', async () => {
    // The delete-then-insert window this replaces was Task 1842: a client
    // listing mid-push saw an empty or half-built tree. At 60s cadence that
    // window would be hit sixty times more often than it was hourly.
    const c = generationClient()
    await push(c)
    const flipAt = c.calls.findIndex((x) => /INSERT INTO directory_state/i.test(x.sql))
    const deleteAt = c.calls.findIndex((x) => /^DELETE FROM directory\b/i.test(x.sql.trim()))
    expect(flipAt).toBeGreaterThanOrEqual(0)
    expect(deleteAt).toBeGreaterThan(flipAt)
  })

  it('writes every row before the flip', async () => {
    const c = generationClient()
    await push(c)
    const flipAt = c.calls.findIndex((x) => /INSERT INTO directory_state/i.test(x.sql))
    const lastInsert = c.calls.map((x) => /INSERT INTO directory\b/i.test(x.sql)).lastIndexOf(true)
    expect(lastInsert).toBeLessThan(flipAt)
  })

  it('a reader mid-push sees the old tree, never a partial one', async () => {
    // First publish establishes generation 1.
    const c = generationClient()
    await push(c)
    const first = c.visible().map((r) => r.relPath).sort()
    expect(first.length).toBeGreaterThan(0)

    // A second publish stages generation 2. Until the flip, `visible()` must
    // still be exactly the first tree — not a mixture, not an empty set.
    const seen: string[][] = []
    const watched = {
      ...c,
      async query(sql: string, params: unknown[] = []) {
        const out = await c.query(sql, params)
        seen.push(c.visible().map((r) => r.relPath).sort())
        return out
      },
    }
    await push(watched)

    const flipIdx = c.calls.findIndex(
      (x, i) => i > first.length && /INSERT INTO directory_state/i.test(x.sql),
    )
    for (let i = 0; i < flipIdx; i++) {
      expect(seen[i]).toEqual(first)
    }
  })

  it('advances the generation on each publish', async () => {
    const c = generationClient()
    await push(c)
    expect(c.pointerValue()).toBe(1)
    await push(c)
    expect(c.pointerValue()).toBe(2)
  })

  it('deletes superseded generations after the flip', async () => {
    const c = generationClient()
    await push(c)
    await push(c)
    expect(c.rows.every((r) => r.generation === 2)).toBe(true)
  })

  it('writes nothing at all when the exposed set is empty', async () => {
    // Fail closed must not delete the account rows: that state would be
    // indistinguishable from a successful publish of an account with no
    // deliverables.
    rmSync(join(dir, 'SCHEMA.md'))
    const c = generationClient()
    await push(c)
    expect(c.calls.filter((x) => /INSERT|DELETE|directory_state/i.test(x.sql))).toEqual([])
  })
})

// Task 1926. Two pushes of ONE account are reachable: the admin server's 60s
// loop (pushAllAccounts) and a hand-run `--account <id>` are separate OS
// processes calling the same pushAccount, so no in-process lock can see both.
// These tests drive both writers against ONE store and snapshot the reader's
// view after every statement, so "never a half-built tree" stays a measured
// property rather than a claim.
//
// Statement-level atomicity matches SQLite: each exec() applies whole.
function sharedStore() {
  const rows: { accountId: string; relPath: string; generation: number }[] = []
  let pointer: number | null = null
  const pointerHistory: number[] = []
  const snapshots: string[][] = []
  const deletesAtPointer: number[] = []

  const visible = () =>
    pointer === null
      ? []
      : rows.filter((r) => r.generation === pointer).map((r) => r.relPath).sort()

  // The three DELETE shapes this file can issue. Order matters: `!=` and `<`
  // must be tested before `=`, or `generation != ?` would read as an equality.
  const deleteMatcher = (sql: string, g: number) => {
    if (/generation\s*<\s*\?/i.test(sql)) return (n: number) => n < g
    if (/generation\s*!=\s*\?/i.test(sql)) return (n: number) => n !== g
    return (n: number) => n === g
  }

  return {
    rows,
    snapshots,
    deletesAtPointer,
    pointerHistory,
    visible,
    pointerValue: () => pointer,
    seed(generation: number, relPaths: string[]) {
      for (const relPath of relPaths) rows.push({ accountId: 'acc-1', relPath, generation })
      pointer = generation
      pointerHistory.push(generation)
    },
    async exec(sql: string, params: unknown[] = []) {
      const out = await (async () => {
        if (/SELECT\s+currentGeneration/i.test(sql)) {
          return pointer === null ? [] : [{ currentGeneration: pointer }]
        }
        if (/SELECT\s+MAX\(generation\)/i.test(sql)) {
          const mine = rows.filter((r) => r.accountId === String(params[0]))
          return [{ g: mine.length ? Math.max(...mine.map((r) => r.generation)) : null }]
        }
        if (/INSERT INTO directory_state/i.test(sql)) {
          const g = Number(params[1])
          // The upsert's WHERE, modelled: with the guard present the pointer
          // moves forward only. Without it, every flip lands.
          const forwardOnly =
            /excluded\.currentGeneration\s*>\s*directory_state\.currentGeneration/i.test(sql)
          if (pointer === null || !forwardOnly || g > pointer) {
            pointer = g
            pointerHistory.push(g)
          }
          return []
        }
        if (/INSERT INTO directory \(/i.test(sql)) {
          const accountId = String(params[0])
          const relPath = String(params[1])
          const generation = Number(params[6])
          if (
            rows.some(
              (r) => r.accountId === accountId && r.relPath === relPath && r.generation === generation,
            )
          ) {
            throw new Error(
              'd1 query failed: HTTP 500 — UNIQUE constraint failed: directory.accountId, directory.relPath, directory.generation',
            )
          }
          rows.push({ accountId, relPath, generation })
          return []
        }
        if (/DELETE FROM directory\b/i.test(sql)) {
          const g = Number(params[1])
          const matches = deleteMatcher(sql, g)
          // The bug's signature: a DELETE that removes rows at the generation
          // the pointer names is a reader-visible deletion of the live tree.
          if (pointer !== null && matches(pointer) && rows.some((r) => r.generation === pointer)) {
            deletesAtPointer.push(pointer)
          }
          for (let i = rows.length - 1; i >= 0; i--) if (matches(rows[i].generation)) rows.splice(i, 1)
          return []
        }
        if (/SELECT COUNT/i.test(sql)) {
          if (/generation\s*=\s*\?/i.test(sql)) {
            const g = Number(params[1])
            return [{ n: rows.filter((r) => r.generation === g).length }]
          }
          return [{ n: rows.filter((r) => r.generation === pointer).length }]
        }
        return []
      })()
      snapshots.push(visible())
      return out
    },
  }
}

/** A one-shot rendezvous: `reached` settles when the writer parks, `open()` frees it. */
function gate() {
  let hit!: () => void
  let open!: () => void
  const reached = new Promise<void>((r) => (hit = r))
  const opened = new Promise<void>((r) => (open = r))
  return { reached, opened, hit: () => hit(), open: () => open() }
}

/** A client that parks once, either before or after the matching statement. */
function pausingClient(
  store: ReturnType<typeof sharedStore>,
  g: ReturnType<typeof gate>,
  at: { before?: (sql: string) => boolean; after?: (sql: string) => boolean },
) {
  let armed = true
  const calls: { sql: string; params: unknown[] }[] = []
  return {
    calls,
    async query(sql: string, params: unknown[] = []) {
      calls.push({ sql, params })
      if (armed && at.before?.(sql)) {
        armed = false
        g.hit()
        await g.opened
      }
      const out = await store.exec(sql, params)
      if (armed && at.after?.(sql)) {
        armed = false
        g.hit()
        await g.opened
      }
      return out
    },
  }
}

const plainClient = (store: ReturnType<typeof sharedStore>) => ({
  query: (sql: string, params: unknown[] = []) => store.exec(sql, params),
})

describe('two concurrent pushes of one account (Task 1926)', () => {
  // The task file's own interleave. B reads the pointer, then stalls. A stages,
  // flips and sweeps a whole publish. B resumes holding a stale `current`.
  //
  // Before the guard: B recomputed next = current + 1 = the generation the
  // pointer now names, its pre-stage COUNT returned A's live rows, and its
  // DELETE emptied the tree a client was reading.
  it('a peer holding a stale pointer never empties or halves the live tree', async () => {
    const store = sharedStore()
    store.seed(5, ['output', 'output/report.pdf', 'quotes', 'quotes/CR2969'])
    const g = gate()
    const b = pausingClient(store, g, { after: (sql) => /SELECT\s+currentGeneration/i.test(sql) })

    // Captured rather than awaited bare: B throwing for an unrelated reason
    // would otherwise surface as a passing test.
    const bRun = push(b, { log: () => {} }).catch((e: unknown) => e)
    await g.reached
    await push(plainClient(store), { log: () => {} })
    g.open()
    const bResult = await bRun

    expect(bResult).not.toBeInstanceOf(Error)
    // 4 is the seeded tree, 7 is a complete walk of the fixture. Any other
    // count is a tree a client could see half-built.
    expect(store.snapshots.every((s) => s.length === 4 || s.length === 7)).toBe(true)
    expect(store.deletesAtPointer).toEqual([])
    expect(store.visible().length).toBe(7)
    // B resumed holding a stale pointer of 5, saw A's rows at 6 in the MAX, and
    // claimed 7. Nothing was staged over and nothing was deleted at a live one.
    expect(store.pointerValue()).toBe(7)
  })

  // The other order: A stages a whole generation, then stalls before its flip.
  // B starts fresh, sees A's staged rows in the MAX, claims the generation
  // above, and publishes. A wakes holding a generation that is now behind.
  it('a writer overtaken while staging leaves the winner live and says so', async () => {
    const store = sharedStore()
    store.seed(5, ['output', 'output/report.pdf', 'quotes', 'quotes/CR2969'])
    const g = gate()
    const a = pausingClient(store, g, { before: (sql) => /INSERT INTO directory_state/i.test(sql) })
    const lines: string[] = []

    const aRun = push(a, { log: (l: string) => lines.push(l) })
    await g.reached
    await push(plainClient(store), { log: () => {} })
    g.open()
    const aResult = await aRun

    expect(store.snapshots.every((s) => s.length === 4 || s.length === 7)).toBe(true)
    expect(store.deletesAtPointer).toEqual([])
    // The pointer never goes backward, which is what keeps the loser harmless.
    expect(store.pointerHistory).toEqual([...store.pointerHistory].sort((x, y) => x - y))
    expect(store.pointerValue()).toBe(7)

    const contended = lines.find((l) => l.includes('op=stage-contended'))!
    expect(contended).toContain('account=acc-1')
    expect(contended).toContain('generation=6')
    expect(contended).toContain('pointer=7')
    expect(contended).toContain('action=superseded')
    expect(lines.some((l) => l.includes('op=flip'))).toBe(false)
    // No sweep and no verify from the loser: both would act on, and measure,
    // the WINNER's tree and report a failure where nothing failed.
    expect(a.calls.some((c) => /DELETE FROM directory/i.test(c.sql))).toBe(false)
    expect(lines.some((l) => l.includes('op=verify'))).toBe(false)
    expect(aResult.superseded).toBe(true)
  })
})