import { test } from '@japa/runner'
import db from '@adonisjs/lucid/services/db'
import testUtils from '@adonisjs/core/services/test_utils'
import { deflateSync } from 'node:zlib'
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 coordinatorConfig from '#config/coordinator'
import { decodeFrames, fromWire, toWire, RangeIndexer, type Lease } from '#lib/edge/index'
import { parseOverviews } from '#lib/nntp/index'
import { unzipSync } from 'node:zlib'

const GROUP = 'alt.binaries.test'

/** One overview chunk: a two-part release posted at `at`. */
function chunk(at: number): Buffer {
  const lines = [1, 2].map(
    (n) => `${at + n - 1}\tSome.Show.S01E0${at % 7} [1/1] - "Some.Show.S01E0${at % 7}.1080p.mkv" yEnc (${n}/2)\tp <p@x>\tMon, 05 Oct 2026 10:00:00 +0000\t<a${at}-${n}@test>\t\t700000\t5000\tXref: news.example ${GROUP}:${at + n - 1}`
  )
  // An obfuscated subject: counted under the key '' (which must survive the body parser).
  lines.push(`${at + 2}\tobfuscated\tp <p@x>\tMon, 05 Oct 2026 10:00:00 +0000\t<o${at}@test>\t\t700000\t5000\tXref: news.example ${GROUP}:${at + 2}`)
  return deflateSync(Buffer.from(lines.join('\r\n') + '\r\n'))
}

test.group('Node API', (group) => {
  const config = { ...coordinatorConfig, spool: { ...coordinatorConfig.spool } }
  let dir = ''
  group.each.setup(async () => {
    dir = await mkdtemp(join(tmpdir(), 'spool-'))
    coordinatorConfig.spool.dir = dir
    coordinatorConfig.chunkArticles = 100
    return testUtils.db().withGlobalTransaction()
  })
  group.each.teardown(async () => {
    coordinatorConfig.spool.dir = config.spool.dir
    coordinatorConfig.chunkArticles = config.chunkArticles
    await rm(dir, { recursive: true, force: true })
  })

  test('lease, stream from the spool, post batches, finish', async ({ client, assert }) => {
    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, spooled_chunks: 3 }).returning('id')
    const rangeId = Number(r.id)
    for (const span of Spool.chunksOf({ start: 1, end: 300 })) await Spool.write(rangeId, span, chunk(span.from + 10))
    const node = await Node.create({ name: 'n1' })
    const token = (await Node.accessTokens.create(node)).value!.release()

    // No token, no lease.
    ;(await client.post('/api/v1/node/lease')).assertStatus(401)

    const lease = await client.post('/api/v1/node/lease').bearerToken(token)
    lease.assertStatus(200)
    const leased = lease.body() as Lease
    assert.deepEqual(leased.range, { id: rangeId, group: GROUP, start: 1, end: 300, next: 1 })
    assert.equal(leased.rules.priorities[GROUP], 1)
    // Nothing else spooled: no second lease.
    ;(await client.post('/api/v1/node/lease').bearerToken(token)).assertStatus(204)

    const stream = await client.get(`/api/v1/node/ranges/${rangeId}/stream`).bearerToken(token)
    stream.assertStatus(200)
    const raw = stream.body() as unknown
    const body: Buffer = Buffer.isBuffer(raw) ? raw : Buffer.from(stream.text(), 'binary')
    const frames = []
    for await (const frame of decodeFrames([body])) frames.push(frame)
    assert.deepEqual(frames.map((f) => [f.from, f.to]), [[1, 100], [101, 200], [201, 300]])

    const indexer = new RangeIndexer({ group: GROUP, start: 1, end: 300, rules: leased.rules })
    for (const f of frames) indexer.add(parseOverviews(unzipSync(f.data).toString('utf8')), f.to + 1)
    indexer.finish()
    const batch = toWire(indexer.take())
    assert.deepEqual(fromWire(batch).releases.length, 3)
    const posted = await client.post(`/api/v1/node/ranges/${rangeId}/batches`).bearerToken(token).json(batch)
    posted.assertStatus(200)
    assert.include(posted.body(), { releases: 3, newReleases: 3 })
    // The same batch again: refused.
    ;(await client.post(`/api/v1/node/ranges/${rangeId}/batches`).bearerToken(token).json(batch)).assertStatus(409)

    // Counts by the range's block (a range without one is a block of its own, until blocks are leased).
    const { block_id: blockId } = await db.from('ranges').where('id', rangeId).firstOrFail()
    assert.deepEqual((await db.from('rule_counts').where('block_id', blockId)).map((c) => [c.rule, c.key, Number(c.articles)]), [['obfuscated', '', 3]])

    ;(await client.post(`/api/v1/node/ranges/${rangeId}/finish`).bearerToken(token)).assertStatus(200)
    const range = await db.from('ranges').where('id', rangeId).first()
    assert.equal(range.status, 'done')
    assert.isFalse(await Spool.has(rangeId, { from: 1, to: 100 }))
  })
  test('a group lease: the next range in order, its stream from the corpus, the snapshot through the API', async ({ client, assert }) => {
    const corpus = await mkdtemp(join(tmpdir(), 'corpus-'))
    const snapshots = await mkdtemp(join(tmpdir(), 'snapshots-'))
    const saved = { corpus: coordinatorConfig.corpus.dir, snapshots: coordinatorConfig.snapshotDir }
    coordinatorConfig.corpus.dir = corpus
    coordinatorConfig.snapshotDir = snapshots
    try {
      const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
      // Two ranges of the group: the first stored in the corpus (no spool), the second not.
      const [r1] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 200, next_id: 1, capture: true, stored_at: new Date() }).returning('id')
      const [r2] = await db.table('ranges').insert({ group_id: g.id, start_id: 201, end_id: 400, next_id: 201 }).returning('id')
      const corpusRange = { group: GROUP, start: 1, end: 200 }
      await Corpus.write(corpusRange, { from: 1, to: 100 }, chunk(11), 2)
      await Corpus.write(corpusRange, { from: 101, to: 200 }, chunk(111), 2)
      const node = await Node.create({ name: 'n1' })
      const token = (await Node.accessTokens.create(node)).value!.release()

      const lease = await client.post('/api/v1/node/groups/lease').bearerToken(token)
      lease.assertStatus(200)
      const groupId = Number(lease.body().group.id)
      assert.equal(groupId, Number(g.id))
      assert.isNull(lease.body().snapshot)
      ;(await client.post('/api/v1/node/groups/lease').bearerToken(token)).assertStatus(204)

      const next = await client.post(`/api/v1/node/groups/${groupId}/next`).bearerToken(token)
      next.assertStatus(200)
      assert.deepEqual(next.body().range, { id: Number(r1.id), start: 1, end: 200, next: 1 })

      // The stored range's chunks come from the corpus, nothing spooled.
      const stream = await client.get(`/api/v1/node/ranges/${r1.id}/stream`).bearerToken(token)
      stream.assertStatus(200)
      const raw = stream.body() as unknown
      const body: Buffer = Buffer.isBuffer(raw) ? raw : Buffer.from(stream.text(), 'binary')
      const frames = []
      for await (const frame of decodeFrames([body])) frames.push(frame)
      assert.deepEqual(frames.map((f) => [f.from, f.to]), [[1, 100], [101, 200]])
      assert.equal(parseOverviews(unzipSync(frames[0].data).toString('utf8')).length, 3)

      // The snapshot, standing at the second range's start.
      const putRaw = (url: string, bytes: Buffer) => {
        const request = client.put(url).bearerToken(token)
        request.request.set('Content-Type', 'application/octet-stream').send(bytes)
        return request
      }
      ;(await putRaw(`/api/v1/node/groups/${groupId}/snapshot`, Buffer.from('window'))).assertStatus(400)
      ;(await putRaw(`/api/v1/node/groups/${groupId}/snapshot?next=201`, Buffer.from('window at 201'))).assertStatus(200)
      await db.from('ranges').where('id', r1.id).update({ status: 'done', next_id: 201 })
      const got = await client.get(`/api/v1/node/groups/${groupId}/snapshot`).bearerToken(token)
      got.assertStatus(200)
      assert.equal(got.header('x-snapshot-next'), '201')
      const gotRaw = got.body() as unknown
      assert.equal((Buffer.isBuffer(gotRaw) ? gotRaw : Buffer.from(got.text(), 'binary')).toString(), 'window at 201')
      assert.equal(Number((await client.post(`/api/v1/node/groups/${groupId}/next`).bearerToken(token)).body().range.id), Number(r2.id))

      ;(await client.post(`/api/v1/node/groups/${groupId}/heartbeat`).bearerToken(token)).assertStatus(200)
      ;(await client.post(`/api/v1/node/groups/${groupId}/release`).bearerToken(token)).assertStatus(200)
      ;(await client.post(`/api/v1/node/groups/${groupId}/heartbeat`).bearerToken(token)).assertStatus(409)
      assert.include(await db.from('ranges').where('id', r2.id).firstOrFail(), { status: 'planned', node_id: null })

      // The tail lane (`?lane=tail`): the group's tail range, with a cursor and snapshot of its own.
      const [t] = await db.table('ranges').insert({ group_id: g.id, start_id: 401, end_id: 500, next_id: 401, priority: -3 }).returning('id')
      ;(await client.post('/api/v1/node/groups/lease?lane=sideways').bearerToken(token)).assertStatus(400)
      const tail = await client.post('/api/v1/node/groups/lease?lane=tail').bearerToken(token)
      tail.assertStatus(200)
      assert.equal(tail.body().lane, 'tail')
      assert.equal(Number(tail.body().group.id), groupId)
      assert.deepEqual((await client.post(`/api/v1/node/groups/${groupId}/next?lane=tail`).bearerToken(token)).body().range, { id: Number(t.id), start: 401, end: 500, next: 401 })
      ;(await client.post(`/api/v1/node/groups/${groupId}/heartbeat`).bearerToken(token)).assertStatus(409)
      ;(await client.post(`/api/v1/node/groups/${groupId}/heartbeat?lane=tail`).bearerToken(token)).assertStatus(200)
      ;(await putRaw(`/api/v1/node/groups/${groupId}/snapshot?next=401&lane=tail`, Buffer.from('tail at 401'))).assertStatus(200)
      const gotTail = await client.get(`/api/v1/node/groups/${groupId}/snapshot?lane=tail`).bearerToken(token)
      gotTail.assertStatus(200)
      assert.equal(gotTail.header('x-snapshot-next'), '401')
      ;(await client.post(`/api/v1/node/groups/${groupId}/release?lane=tail`).bearerToken(token)).assertStatus(200)
      assert.include(await db.from('ranges').where('id', t.id).firstOrFail(), { status: 'planned', node_id: null })
    } finally {
      coordinatorConfig.corpus.dir = saved.corpus
      coordinatorConfig.snapshotDir = saved.snapshots
      await rm(corpus, { recursive: true, force: true })
      await rm(snapshots, { recursive: true, force: true })
    }
  })

  test('a chunk the corpus fill stores while the stream waits for it comes from the corpus', async ({ client, assert }) => {
    const corpus = await mkdtemp(join(tmpdir(), 'corpus-'))
    const saved = coordinatorConfig.corpus.dir
    coordinatorConfig.corpus.dir = corpus
    try {
      const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
      // A capture range not in the corpus: its first chunk spooled by the leased run, the second claimed by
      // the corpus fill, which stores it in the corpus only, after the stream started.
      const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 200, next_id: 1, capture: true }).returning('id')
      const rangeId = Number(r.id)
      await Spool.write(rangeId, { from: 1, to: 100 }, chunk(11))
      const node = await Node.create({ name: 'n1' })
      const token = (await Node.accessTokens.create(node)).value!.release()
      await db.from('ranges').where('id', rangeId).update({ status: 'leased', node_id: node.id, leased_at: new Date(), heartbeat_at: new Date() })

      const pending = client.get(`/api/v1/node/ranges/${rangeId}/stream`).bearerToken(token)
      setTimeout(() => Corpus.write({ group: GROUP, start: 1, end: 200 }, { from: 101, to: 200 }, chunk(111), 2), 800)
      const stream = await pending
      stream.assertStatus(200)
      const raw = stream.body() as unknown
      const body: Buffer = Buffer.isBuffer(raw) ? raw : Buffer.from(stream.text(), 'binary')
      const frames = []
      for await (const frame of decodeFrames([body])) frames.push(frame)
      assert.deepEqual(frames.map((f) => [f.from, f.to]), [[1, 100], [101, 200]])
      assert.equal(parseOverviews(unzipSync(frames[1].data).toString('utf8')).length, 3)
      assert.isFalse(await Spool.has(rangeId, { from: 101, to: 200 }))
    } finally {
      coordinatorConfig.corpus.dir = saved
      await rm(corpus, { recursive: true, force: true })
    }
  })

  test('a chunk that does not inflate: reported, removed from the spool, the range goes back', async ({ client, assert }) => {
    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, spooled_chunks: 3 }).returning('id')
    const rangeId = Number(r.id)
    for (const span of Spool.chunksOf({ start: 1, end: 300 })) await Spool.write(rangeId, span, chunk(span.from + 10))
    const node = await Node.create({ name: 'n1' })
    const token = (await Node.accessTokens.create(node)).value!.release()
    ;(await client.post('/api/v1/node/lease').bearerToken(token)).assertStatus(200)

    ;(await client.post(`/api/v1/node/ranges/${rangeId}/corrupt`).bearerToken(token).json({ from: 101 })).assertStatus(200)
    assert.deepEqual(
      await Promise.all(Spool.chunksOf({ start: 1, end: 300 }).map((span) => Spool.has(rangeId, span))),
      [true, false, true]
    )
    let range = await db.from('ranges').where('id', rangeId).first()
    assert.equal(range.status, 'planned')
    assert.isNull(range.node_id)
    // Fetched plain next time (the server's compressed reply was cut short).
    assert.deepEqual(range.plain_chunks.map(Number), [101])
    // No longer this node's: a second report is refused.
    ;(await client.post(`/api/v1/node/ranges/${rangeId}/corrupt`).bearerToken(token).json({ from: 101 })).assertStatus(409)
    // Leased again, the chunk still doesn't inflate: once more (the compressed copy may have been spooled
    // again meanwhile), then the range is set aside, not leased in a loop.
    for (const status of ['planned', 'failed']) {
      await Spool.write(rangeId, { from: 101, to: 200 }, chunk(111))
      ;(await client.post('/api/v1/node/lease').bearerToken(token)).assertStatus(200)
      ;(await client.post(`/api/v1/node/ranges/${rangeId}/corrupt`).bearerToken(token).json({ from: 101 })).assertStatus(200)
      range = await db.from('ranges').where('id', rangeId).first()
      assert.equal(range.status, status)
    }
    assert.include(range.error, 'also fetched plain')
  })
})