import { test } from '@japa/runner'
import db from '@adonisjs/lucid/services/db'
import testUtils from '@adonisjs/core/services/test_utils'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import Node from '#models/node'
import Spool from '#services/spool'
import Corpus from '#services/corpus'
import GroupLeases from '#services/group_leases'
import coordinatorConfig from '#config/coordinator'
import { GROUP, overviewChunk, runEdge, startEdge, waitFor } from '#tests/e2e/helpers/edge'

/**
 * The Rust edge worker against the real node API, database, spool and corpus (REBUILD.md, "Voyager 3: window
 * indexing", 1 and 5): what broke in production on 2026-10-27 (LOG.md (118)) — a stream waiting on a chunk the
 * corpus fill stored, a failed stream, a restart with and without the snapshot, a takeover.
 */
test.group('Edge worker end to end', (group) => {
  const saved = {
    spool: coordinatorConfig.spool.dir,
    corpus: coordinatorConfig.corpus.dir,
    snapshots: coordinatorConfig.snapshotDir,
    chunk: coordinatorConfig.chunkArticles,
    wait: coordinatorConfig.streamChunkWaitMs,
  }
  let dir = ''
  group.each.setup(async () => {
    dir = await mkdtemp(join(tmpdir(), 'e2e-'))
    coordinatorConfig.spool.dir = join(dir, 'spool')
    coordinatorConfig.corpus.dir = join(dir, 'corpus')
    coordinatorConfig.snapshotDir = join(dir, 'snapshots')
    coordinatorConfig.chunkArticles = 100
    return testUtils.db().truncate()
  })
  group.each.teardown(async () => {
    Object.assign(coordinatorConfig.spool, { dir: saved.spool })
    coordinatorConfig.corpus.dir = saved.corpus
    coordinatorConfig.snapshotDir = saved.snapshots
    coordinatorConfig.chunkArticles = saved.chunk
    coordinatorConfig.streamChunkWaitMs = saved.wait
    await rm(dir, { recursive: true, force: true })
  })

  /** A group with one range of three chunks and a node with a token. */
  async function setup(opts: { capture?: boolean } = {}) {
    const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
    const [r] = await db
      .table('ranges')
      .insert({
        group_id: g.id,
        start_id: 1,
        end_id: 300,
        next_id: 1,
        priority: 0,
        capture: opts.capture ?? false,
      })
      .returning('id')
    const node = await Node.create({ name: 'e2e-node' })
    const token = (await Node.accessTokens.create(node)).value!.release()
    return { groupId: Number(g.id), rangeId: Number(r.id), nodeId: Number(node.id), token }
  }
  const range = async (id: number) => (await db.from('ranges').where('id', id).firstOrFail()) as any
  const lease = async (groupId: number) =>
    (await db
      .from('group_leases')
      .where('group_id', groupId)
      .where('lane', 'history')
      .first()) as any
  /** The release whose first article is `at` (a release's name isn't the file's: the identity strips it). */
  const releaseAt = async (at: number) =>
    (await db.from('releases').where('first_article', at).first()) as any
  const names = async () =>
    JSON.stringify(
      await db
        .from('releases')
        .select('name', 'first_article', 'last_article', 'segments', 'complete')
        .orderBy('first_article')
    )
  const has = (lines: string[], text: string) => lines.some((l) => l.includes(text))

  test('a range streamed from the spool and the corpus, with a chunk the corpus fill stores while the stream waits', async ({
    assert,
  }) => {
    const { groupId, rangeId, token } = await setup({ capture: true })
    await Spool.write(rangeId, { from: 1, to: 100 }, overviewChunk(1, 100))
    await Spool.write(rangeId, { from: 201, to: 300 }, overviewChunk(201, 300))
    const run = await startEdge(token)
    // The worker holds the group and streams: chunk 1 goes out, chunk 2 is waited for — the corpus fill stores it.
    await waitFor(
      'the group to be held',
      async () =>
        (await lease(groupId))?.range_id !== null && (await lease(groupId))?.range_id !== undefined
    )
    await new Promise((r) => setTimeout(r, 700))
    await Corpus.write(
      { group: GROUP, start: 1, end: 300 },
      { from: 101, to: 200 },
      overviewChunk(101, 200),
      3
    )
    const { code, lines } = { ...(await run.exit), lines: run.lines }
    assert.equal(code, 0, lines.join('\n'))
    assert.isTrue(has(lines, 'caught up after 1 ranges'), lines.join('\n'))
    assert.isTrue(has(lines, 'snapshot at 301'), lines.join('\n'))
    assert.include(await range(rangeId), { status: 'done' })
    assert.equal(Number((await range(rangeId)).headers), 300)
    const l = await lease(groupId)
    assert.isNull(l.node_id)
    assert.equal(Number(l.snapshot_next), 301)
    assert.equal(Number(l.ranges_done), 1)
    // The window closed on the early files (the frontier is a day past them); every closed release is settled.
    const releases = await db
      .from('releases')
      .select('name', 'settled_at', 'first_article', 'last_article', 'segments')
    assert.isAtLeast(releases.length, 10)
    assert.isTrue(releases.every((r) => r.settled_at !== null))
    const spanning = releases.find((r) => Number(r.first_article) === 100)
    assert.isDefined(spanning, await names())
    assert.deepEqual(
      [Number(spanning.first_article), Number(spanning.last_article), Number(spanning.segments)],
      [100, 101, 2],
      'the file over the chunk boundary came through the carried window'
    )
  })

  test('a stream that fails: the batch so far is posted, the window snapshotted, the group given back; the next run restores it', async ({
    assert,
  }) => {
    coordinatorConfig.streamChunkWaitMs = 1500
    const { groupId, rangeId, token } = await setup()
    await Spool.write(rangeId, { from: 1, to: 100 }, overviewChunk(1, 100))
    const first = await runEdge(token)
    assert.equal(first.code, 0, first.lines.join('\n'))
    assert.isTrue(has(first.lines, 'the group goes back'), first.lines.join('\n'))
    assert.isTrue(has(first.lines, 'snapshot at 101'), first.lines.join('\n'))
    let r = await range(rangeId)
    assert.include(r, { status: 'planned', node_id: null })
    assert.equal(
      Number(r.next_id),
      101,
      'the batch for chunk 1 was stored before the group went back'
    )
    let l = await lease(groupId)
    assert.isNull(l.node_id)
    assert.equal(Number(l.snapshot_next), 101)
    const before = await db.from('releases').count('* as n').first()
    assert.isAtLeast(Number(before!.n), 10)
    assert.isNull(await releaseAt(100), 'the file over the boundary is still open in the window')

    // The next run: the window restored, the rest of the range streamed, the open file completed.
    coordinatorConfig.streamChunkWaitMs = saved.wait
    await Spool.write(rangeId, { from: 101, to: 200 }, overviewChunk(101, 200))
    await Spool.write(rangeId, { from: 201, to: 300 }, overviewChunk(201, 300))
    const second = await runEdge(token)
    assert.equal(second.code, 0, second.lines.join('\n'))
    assert.isTrue(
      has(second.lines, 'window restored from the snapshot at 101'),
      second.lines.join('\n')
    )
    r = await range(rangeId)
    assert.equal(r.status, 'done')
    assert.equal(Number(r.headers), 300)
    l = await lease(groupId)
    assert.equal(Number(l.snapshot_next), 301)
    const spanning = await releaseAt(100)
    assert.isNotNull(spanning, await names())
    assert.deepEqual(
      [Number(spanning.first_article), Number(spanning.last_article), Number(spanning.segments)],
      [100, 101, 2]
    )
  })

  test('a restart whose snapshot file is gone: the window starts empty and the file over the seam ends incomplete', async ({
    assert,
  }) => {
    coordinatorConfig.streamChunkWaitMs = 1500
    const { groupId, rangeId, token } = await setup()
    await Spool.write(rangeId, { from: 1, to: 100 }, overviewChunk(1, 100))
    const first = await runEdge(token)
    assert.isTrue(has(first.lines, 'snapshot at 101'), first.lines.join('\n'))
    await rm(GroupLeases.snapshotPath(groupId), { force: true })

    coordinatorConfig.streamChunkWaitMs = saved.wait
    await Spool.write(rangeId, { from: 101, to: 200 }, overviewChunk(101, 200))
    await Spool.write(rangeId, { from: 201, to: 300 }, overviewChunk(201, 300))
    const second = await runEdge(token)
    assert.equal(second.code, 0, second.lines.join('\n'))
    assert.isTrue(
      has(second.lines, 'but there is none; the window starts empty'),
      second.lines.join('\n')
    )
    assert.equal((await range(rangeId)).status, 'done')
    // Part 1 was in the lost window: one seam. Part 2 alone is a file short of a part.
    assert.isNull(await releaseAt(100), await names())
    const half = await releaseAt(101)
    assert.isNotNull(half, await names())
    assert.deepEqual([Number(half.segments), half.complete], [1, false], await names())
  })

  test("a takeover: the stale holder's next batch is refused, its snapshot too, and the group stays the new holder's", async ({
    assert,
  }) => {
    const { groupId, rangeId, token, nodeId } = await setup()
    await Spool.write(rangeId, { from: 1, to: 100 }, overviewChunk(1, 100))
    const run = await startEdge(token)
    // Chunk 1 posted; the stream now waits for chunk 2.
    await waitFor('the first batch', async () => Number((await range(rangeId)).next_id) === 101)
    // The holder looks dead (its group and range heartbeats stale): another node takes the group over (no snapshot
    // yet: the range goes back to its resume point).
    const stale = new Date(Date.now() - 24 * 3600_000)
    await db.from('group_leases').where('group_id', groupId).update({ heartbeat_at: stale })
    await db.from('ranges').where('id', rangeId).update({ heartbeat_at: stale })
    const other = await Node.create({ name: 'e2e-other' })
    const taken = await GroupLeases.lease(Number(other.id))
    assert.equal(taken?.group.id, groupId)
    assert.isNull(taken?.snapshot)
    assert.include(await range(rangeId), { status: 'planned', node_id: null })
    // The old holder gets chunk 2 all the same, posts, and is refused.
    await Spool.write(rangeId, { from: 101, to: 200 }, overviewChunk(101, 200))
    await Spool.write(rangeId, { from: 201, to: 300 }, overviewChunk(201, 300))
    const { code } = await run.exit
    const lines = run.lines
    assert.equal(code, 0, lines.join('\n'))
    assert.isTrue(has(lines, 'snapshot failed'), lines.join('\n'))
    assert.isTrue(
      has(lines, 'not leased to this node') || has(lines, 'no longer this node'),
      lines.join('\n')
    )
    const l = await lease(groupId)
    assert.equal(Number(l.node_id), Number(other.id))
    assert.isNull(l.snapshot_next)
    const r = await range(rangeId)
    assert.equal(r.status, 'planned')
    assert.equal(Number(r.next_id), 101, 'nothing past the first batch was stored')
    assert.notEqual(Number(r.node_id), nodeId)
  })
})