import { test } from '@japa/runner'
import db from '@adonisjs/lucid/services/db'
import testUtils from '@adonisjs/core/services/test_utils'
import BatchStore, { StaleBatch } from '#services/batch_store'
import { RangeIndexer, type EdgeBatch, type ReleaseChunk } from '#lib/edge/index'
import { decodeChunk, encodeChunk } from '#lib/chunks/index'
import { DEFAULT_RULES } from '#lib/rules/index'
import type { Overview } from '#lib/nntp/index'

const GROUP = 'alt.binaries.test'
let id = 0

function post(
  at: number,
  name: string,
  parts: number,
  counter: string,
  tool: 'ngPost' | 'plain' = 'plain'
): Overview[] {
  return Array.from({ length: parts }, (_, i) => {
    id++
    const mid = tool === 'ngPost' ? `${id.toString(16).padStart(32, '0')}@ngPost` : `art${id}@test`
    return [
      at + i,
      {
        'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`,
        'from': 'p <p@x>',
        'date': 'Mon, 05 Oct 2026 10:00:00 +0000',
        'message-id': `<${mid}>`,
        ':bytes': '700000',
        'xref': `news.example ${GROUP}:${at + i}`,
      },
    ] as Overview
  })
}

async function leasedRange() {
  const [group] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
  const [node] = await db.table('nodes').insert({ name: 'test-node' }).returning('id')
  const [range] = await db
    .table('ranges')
    .insert({
      group_id: group.id,
      start_id: 1,
      end_id: 1_000_000,
      next_id: 1,
      status: 'leased',
      node_id: node.id,
      rule_version: 1,
    })
    .returning('id')
  return { rangeId: Number(range.id), nodeId: Number(node.id) }
}

function indexer() {
  return new RangeIndexer({
    group: GROUP,
    start: 1,
    end: 1_000_000,
    rules: {
      ...DEFAULT_RULES,
      samplePercent: 0,
      droppedSignatures: [],
      priorities: { [GROUP]: 1 },
    },
    edgeArticles: 1000,
  })
}

/** One file's articles, posted `minutes` apart from `start` (an ISO time), in `group`. */
function upload(
  at: number,
  name: string,
  parts: number,
  counter: string,
  start: string,
  group = GROUP,
  minutes = 1
): Overview[] {
  return Array.from({ length: parts }, (_, i) => {
    id++
    const date = new Date(Date.parse(start) + i * minutes * 60_000)
      .toUTCString()
      .replace('GMT', '+0000')
    return [
      at + i,
      {
        'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`,
        'from': 'p <p@x>',
        'date': date,
        'message-id': `<up${id}@test>`,
        ':bytes': '700000',
        'xref': `news.example ${group}:${at + i}`,
      },
    ] as Overview
  })
}

async function rangeIn(groupName: string, start: number, nodeId?: number) {
  const g =
    (await db.from('groups').where('name', groupName).first()) ??
    (
      await db
        .table('groups')
        .insert({ name: groupName, priority: groupName === GROUP ? 1 : 2 })
        .returning('*')
    )[0]
  const node =
    nodeId ??
    Number(
      (
        await db
          .table('nodes')
          .insert({ name: `n${groupName}${start}` })
          .returning('id')
      )[0].id
    )
  const [range] = await db
    .table('ranges')
    .insert({
      group_id: g.id,
      start_id: start,
      end_id: start + 99_999,
      next_id: start,
      status: 'leased',
      node_id: node,
      rule_version: 1,
    })
    .returning('id')
  return { rangeId: Number(range.id), nodeId: node }
}

function batchOf(groupName: string, start: number, overviews: Overview[]): EdgeBatch {
  const ix = new RangeIndexer({
    group: groupName,
    start,
    end: start + 99_999,
    rules: {
      ...DEFAULT_RULES,
      samplePercent: 0,
      droppedSignatures: [],
      priorities: { [GROUP]: 1, 'alt.binaries.other': 2 },
    },
    edgeArticles: 1,
  })
  ix.add(overviews, start + 100_000)
  ix.finish()
  return ix.take()
}

async function storeIn(groupName: string, start: number, overviews: Overview[]) {
  const { rangeId, nodeId } = await rangeIn(groupName, start)
  return BatchStore.store(rangeId, nodeId, batchOf(groupName, start, overviews))
}

const releasesOf = () =>
  db
    .from('releases')
    .orderBy('id')
    .select(
      db.raw(
        `encode(release_key, 'hex') AS release_key, encode(identity_key, 'hex') AS identity_key, files, chunks, posted_at, last_posted_at, settled_at`
      )
    )

test.group('BatchStore', (group) => {
  group.each.setup(() => testUtils.db().withGlobalTransaction())

  test('stores releases by id references, their texts, chunks, probe candidates and rule counts, and moves the resume point', async ({
    assert,
  }) => {
    const { rangeId, nodeId } = await leasedRange()
    const ix = indexer()
    ix.add(
      [
        ...post(100_000, 'Some.Show.S01E01.1080p.WEB.part1.rar', 2, 'Some.Show.S01E01 [1/2]'),
        ...post(100_010, 'Some.Show.S01E01.1080p.WEB.par2', 1, 'Some.Show.S01E01 [2/2]'),
        ...post(100_020, 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 3, '[1/1]', 'ngPost'),
      ],
      500_000
    )
    const first = ix.take()
    const stored = await BatchStore.store(rangeId, nodeId, first)
    assert.deepEqual(stored, { releases: 1, newReleases: 1, finalized: 0, candidates: 2 })

    const release = await db.from('releases').first()
    assert.include(release, {
      files: 2,
      segments: '3',
      chunks: 1,
      readable: true,
      has_par2: true,
      file_total: 2,
      complete: true,
      edge: false,
      source: 1,
    })
    assert.isNotNull(release.settled_at)
    // The poster and the groups by id, the keys 20 bytes, the name searchable.
    assert.equal(
      (await db.from('posters').where('id', release.poster_id).firstOrFail()).poster,
      'p <p@x>'
    )
    assert.deepEqual(
      (await db.from('group_sets').where('id', release.group_set_id).firstOrFail()).groups,
      [GROUP]
    )
    assert.lengthOf(release.release_key, 20)
    assert.lengthOf(release.identity_key, 20)
    assert.deepEqual(await db.from('release_texts').select('release_id', 'kind', 'text'), [
      { release_id: release.id, kind: 1, text: 'Some Show S01E01' },
    ])
    const chunk = await db.from('release_chunks').where('release_id', release.id).first()
    assert.deepEqual(
      decodeChunk(chunk.payload)
        .map((f) => f.name)
        .sort(),
      ['Some.Show.S01E01.1080p.WEB.par2', 'Some.Show.S01E01.1080p.WEB.part1.rar']
    )
    // Claimable at once: the release is finished.
    assert.deepEqual(
      (await db.from('probe_candidates').orderBy('kind')).map((c) => [
        c.kind,
        c.ready,
        c.unreadable,
      ]),
      [
        ['archive', true, false],
        ['par2', true, false],
      ]
    )
    // Keyed by the range's block (a range without one is a block of its own, until blocks are leased).
    const range = await db.from('ranges').where('id', rangeId).first()
    assert.isNotNull(range.block_id)
    assert.equal(Number(chunk.block_id), Number(range.block_id))
    assert.deepEqual(
      (await db.from('rule_counts')).map((c) => [
        Number(c.block_id),
        c.rule,
        c.key,
        Number(c.articles),
      ]),
      [[Number(range.block_id), 'garbage', 'ngPost alt.binaries.test', 3]]
    )
    assert.equal(Number(range.next_id), 500_000)

    // The same batch again (a retry after a lost reply) is refused, and changes nothing.
    await assert.rejects(() => BatchStore.store(rangeId, nodeId, first), StaleBatch)
    assert.equal((await db.from('releases').first()).files, 2)

    // The end of the range.
    ix.finish()
    await BatchStore.store(rangeId, nodeId, ix.take())
    assert.equal(Number((await db.from('ranges').where('id', rangeId).first()).next_id), 1_000_001)
  })

  test('a range leased to another node is refused', async ({ assert }) => {
    const { rangeId, nodeId } = await leasedRange()
    const ix = indexer()
    ix.add(post(100_000, 'A.Release.2020.mkv', 1, 'A [1/1]'), 200_000)
    await assert.rejects(
      () => BatchStore.store(rangeId, nodeId + 1, ix.take()),
      /not leased to this node/
    )
  })

  test('every chunk is a release of its own: a re-upload, and a part read by another range (the block merger joins those)', async ({
    assert,
  }) => {
    // One range: the same file posted Monday and again Thursday are two uploads.
    await storeIn(GROUP, 1, [
      ...upload(
        10,
        'Some.Show.S01E01.1080p.mkv',
        2,
        'Some.Show.S01E01.1080p [1/2]',
        '2026-10-05T10:00:00Z'
      ),
      ...upload(
        20,
        'Some.Show.S01E01.1080p.par2',
        1,
        'Some.Show.S01E01.1080p [2/2]',
        '2026-10-05T10:05:00Z'
      ),
      ...upload(
        500,
        'Some.Show.S01E01.1080p.mkv',
        2,
        'Some.Show.S01E01.1080p [1/2]',
        '2026-10-08T10:00:00Z'
      ),
    ])
    let rows = await releasesOf()
    assert.deepEqual(
      rows.map((r) => r.files),
      [2, 1]
    )
    assert.equal(rows[0].identity_key, rows[1].identity_key)
    assert.notEqual(rows[0].release_key, rows[1].release_key)

    // The next range: Thursday's par2 an hour later is written as a release of its own (its key: the identity, the
    // group and its first article); the window keeps such parts together, and the block merger joins the edges.
    await storeIn(
      GROUP,
      100_001,
      upload(
        100_010,
        'Some.Show.S01E01.1080p.par2',
        1,
        'Some.Show.S01E01.1080p [2/2]',
        '2026-10-08T11:00:00Z'
      )
    )
    rows = await releasesOf()
    assert.deepEqual(
      rows.map((r) => r.files),
      [2, 1, 1]
    )
    assert.lengthOf(new Set(rows.map((r) => r.release_key)), 3)
    assert.lengthOf(new Set(rows.map((r) => r.identity_key)), 1)
    assert.equal(Number((await db.from('release_chunks').count('* as n').first())!.n), 3)
  })

  test('a provisional tail release is finished once by a later chunk of its key, which adds the files not sent before', async ({
    assert,
  }) => {
    const { rangeId, nodeId } = await rangeIn(GROUP, 1)
    const batch = batchOf(
      GROUP,
      1,
      upload(
        10,
        'Some.Show.S01E02.1080p.mkv',
        2,
        'Some.Show.S01E02.1080p [1/2]',
        '2026-10-05T10:00:00Z'
      )
    )
    assert.lengthOf(batch.releases, 1)
    const provisional: ReleaseChunk = { ...batch.releases[0], finished: false }
    const stored = await BatchStore.store(rangeId, nodeId, {
      ...batch,
      next: 50_000,
      releases: [provisional],
    })
    assert.include(stored, { newReleases: 1, finalized: 0 })
    let [row] = await releasesOf()
    assert.isNull(row.settled_at)
    assert.include(row, { files: 1, chunks: 1 })

    // The finish: the par2 that came within the window, under the same key (the identity, the group, the first
    // article), with its probe candidate: claimable at once, the release being finished.
    const second = batchOf(
      GROUP,
      1,
      upload(
        30,
        'Some.Show.S01E02.1080p.par2',
        1,
        'Some.Show.S01E02.1080p [2/2]',
        '2026-10-05T10:20:00Z'
      )
    )
    const [par2] = second.releases
    const finish: ReleaseChunk = {
      ...par2,
      key: provisional.key,
      upload: provisional.upload,
      firstArticle: provisional.firstArticle,
      finished: true,
    }
    const candidates = second.candidates.map((c) => ({
      ...c,
      releaseKey: provisional.key,
      upload: provisional.upload,
    }))
    assert.deepEqual(
      candidates.map((c) => c.kind),
      ['par2']
    )
    const finished = await BatchStore.store(rangeId, nodeId, {
      ...batch,
      from: 50_000,
      next: 60_000,
      releases: [finish],
      candidates,
      counts: [],
      samples: [],
      fragments: [],
    })
    assert.include(finished, { releases: 1, newReleases: 0, finalized: 1, candidates: 1 })
    ;[row] = await releasesOf()
    assert.isNotNull(row.settled_at)
    assert.include(row, { files: 2, chunks: 2 })
    const release = await db.from('releases').firstOrFail()
    assert.include(release, { complete: true, has_par2: true, last_article: '30' })
    assert.deepEqual(await db.from('probe_candidates').select('kind', 'ready'), [
      { kind: 'par2', ready: true },
    ])
    assert.deepEqual(
      (await db.from('release_chunks').orderBy('id')).flatMap((c) =>
        decodeChunk(c.payload).map((f) => f.name)
      ),
      ['Some.Show.S01E02.1080p.mkv', 'Some.Show.S01E02.1080p.par2']
    )

    // Sent once more, finished already: left as it is.
    const again = await BatchStore.store(rangeId, nodeId, {
      ...batch,
      from: 60_000,
      next: 70_000,
      releases: [{ ...finish, payload: encodeChunk([]), files: 0, segments: 0, bytes: 0 }],
      candidates: [],
      counts: [],
      samples: [],
      fragments: [],
    })
    assert.include(again, { releases: 1, newReleases: 0, finalized: 0 })
    assert.include((await releasesOf())[0], { files: 2, chunks: 2 })
    assert.equal(Number((await db.from('release_chunks').count('* as n').first())!.n), 2)
  })

  test('a split 7z keeps the tail of its highest volume only; RAR gets none', async ({
    assert,
  }) => {
    await storeIn(GROUP, 1, [
      ...upload(10, 'TsIgNBewuFbuwWy.7z.001', 2, '[1/4]', '2026-10-23T08:24:00Z'),
      ...upload(20, 'TsIgNBewuFbuwWy.7z.002', 2, '[2/4]', '2026-10-23T08:26:00Z'),
      ...upload(30, 'TsIgNBewuFbuwWy.7z.003', 2, '[3/4]', '2026-10-23T08:28:00Z'),
      ...upload(40, 'TsIgNBewuFbuwWy.7z.004', 2, '[4/4]', '2026-10-23T08:30:00Z'),
      ...upload(
        50,
        'Some.Show.S01E03.part01.rar',
        2,
        'Some.Show.S01E03 [1/2]',
        '2026-10-23T09:00:00Z'
      ),
      ...upload(
        60,
        'Some.Show.S01E03.part02.rar',
        2,
        'Some.Show.S01E03 [2/2]',
        '2026-10-23T09:02:00Z'
      ),
    ])
    const candidates = await db.from('probe_candidates').orderBy('file').select('kind', 'file')
    assert.deepEqual(
      candidates.map((c) => [c.kind, c.file]),
      [
        ['archive', 'Some.Show.S01E03.part01.rar'],
        ['archive', 'TsIgNBewuFbuwWy.7z.001'],
        ['7z-tail', 'TsIgNBewuFbuwWy.7z.004'],
      ]
    )
  })

  test('the same upload in another group (multi-posted) is a release of its own', async ({
    assert,
  }) => {
    await storeIn(
      GROUP,
      1,
      upload(
        10,
        'Some.Show.S01E01.1080p.mkv',
        2,
        'Some.Show.S01E01.1080p [1/1]',
        '2026-10-05T10:00:00Z'
      )
    )
    await storeIn(
      'alt.binaries.other',
      1,
      upload(
        10,
        'Some.Show.S01E01.1080p.mkv',
        2,
        'Some.Show.S01E01.1080p [1/1]',
        '2026-10-05T10:01:00Z',
        'alt.binaries.other'
      )
    )
    const rows = await releasesOf()
    assert.lengthOf(rows, 2)
    assert.equal(rows[0].identity_key, rows[1].identity_key)
    assert.notEqual(rows[0].release_key, rows[1].release_key)
  })

  test('a small release of sidecars only is marked for the sidecar fold', async ({ assert }) => {
    await storeIn(GROUP, 1_000_000, [
      ...upload(1_000_000, 'Murdoch Mysteries S8D1.nzb', 1, '[00/80]', '2026-10-05T10:00:00Z'),
      ...upload(1_000_010, 'Some.Tool.v1.zip', 1, '[01/01]', '2026-10-05T10:00:00Z'),
      ...upload(1_000_020, 'Some.Show.S01E01.part01.rar', 2, '[02/30]', '2026-10-05T10:00:00Z'),
    ])
    const flags = new Map(
      (await db.from('releases').select('name', 'sidecar_only')).map((r) => [
        r.name,
        r.sidecar_only,
      ])
    )
    assert.isTrue(flags.get('Murdoch Mysteries S8D1'))
    assert.isFalse(flags.get('Some.Tool.v1'))
    // Over 1 MB: not looked at.
    assert.isNull(flags.get('Some.Show.S01E01'))
  })
})