usenet-indexer/tests/functional/seed_api.spec.ts

import { test } from '@japa/runner'
import { gzipSync } from 'node:zlib'
import db from '@adonisjs/lucid/services/db'
import testUtils from '@adonisjs/core/services/test_utils'
import SeedOrchestrator from '#services/SeedOrchestrator'
import QueueService from '#services/QueueService'
import indexerConfig from '#config/indexer'
import SeedFetchPool from '#services/SeedFetchPool'
import { deflateSync } from 'node:zlib'
import { segmentFromHeader } from '#lib/ingest/segment'
import { FileAggregator } from '#lib/aggregate/index'
import { toFragment, toSeedFile, type SeedBatch } from '#lib/seed/wire'
import type { IngestSegment } from '#lib/ingest/segment'

function header(subject: string, n: number) {
  return { 'subject': subject, 'from': 'api seeder <api@example.invalid>', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', 'message-id': `<api-${n}-${subject.length}@test>`, ':bytes': '500000' }
}

function batch(next: number): SeedBatch {
  const segments = [1, 2].map((n) => segmentFromHeader(header(`Api.Test.2026 - "Api.Test.2026.mkv" yEnc (${n}/2)`, n), 'alt.binaries.test', n)!)
  const lone = segmentFromHeader(header('Api.Test.2026 - "Api.Test.2026.nfo" yEnc (1/3)', 9), 'alt.binaries.test', 9)!
  const aggregator = new FileAggregator<IngestSegment>()
  return { next, headers: 3, obfuscated: 0, deferred: 0, files: aggregator.add(segments).map(toSeedFile), fragments: [toFragment(lone)] }
}

test.group('Seed API', (group) => {
  group.each.setup(() => testUtils.db().withGlobalTransaction())
  group.teardown(() => QueueService.closeAll())

  test('a node leases a range, posts a gzip batch, and finishes it', async ({ client, assert }) => {
    const token = await SeedOrchestrator.issueToken('test-node')
    await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 })
    const auth = { authorization: `Bearer ${token}` }

    ;(await client.post('/api/v1/seed/lease')).assertStatus(401)
    const leased = await client.post('/api/v1/seed/lease').headers(auth)
    leased.assertStatus(200)
    const range = leased.body().range
    assert.include(range, { group: 'alt.binaries.test', start: 1, end: 100 })

    // As seed:node sends it: a gzip-compressed JSON body.
    const posted = await fetch(`http://${process.env.HOST}:${process.env.PORT}/api/v1/seed/ranges/${range.id}/batches`, {
      method: 'POST',
      headers: { ...auth, 'content-type': 'application/json', 'content-encoding': 'gzip' },
      body: gzipSync(JSON.stringify(batch(51))),
    })
    assert.equal(posted.status, 200)
    assert.include(((await posted.json()) as any).stored, { files: 1 })
    const file = await db.from('files').where('filename', 'Api.Test.2026.mkv').first()
    assert.equal(file.parts, 2)
    // The lone .nfo segment took the regular path (staged or stored as a binary).
    const fragment = (await db.from('staged_segments').count('* as n').first()).n + (await db.from('binaries').where('name', 'Api.Test.2026.nfo').count('* as n').first()).n
    assert.isAbove(Number(fragment), 0)

    const row = await db.from('seed_ranges').where('id', range.id).first()
    assert.equal(Number(row.next_id), 51)
    assert.equal(Number(row.headers), 3)

    ;(await client.post(`/api/v1/seed/ranges/${range.id}/done`).headers(auth)).assertStatus(200)
    assert.equal((await db.from('seed_ranges').where('id', range.id).first()).status, 'done')
    ;(await client.post('/api/v1/seed/lease').headers(auth)).assertStatus(204)
  })

  test('a quiet lease passes to another node, and the first one gets 409', async ({ client, assert }) => {
    const first = await SeedOrchestrator.issueToken('node-a')
    const second = await SeedOrchestrator.issueToken('node-b')
    await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 })
    const range = (await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${first}` })).body().range
    await db.from('seed_ranges').where('id', range.id).update({ next_id: 40, heartbeat_at: new Date(Date.now() - 60 * 60 * 1000) })

    const taken = (await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${second}` })).body().range
    assert.equal(taken.id, range.id)
    assert.equal(taken.start, 40)
    const late = await client.post(`/api/v1/seed/ranges/${range.id}/batches`).headers({ authorization: `Bearer ${first}` }).json(batch(60))
    late.assertStatus(409)

    ;(await client.post(`/api/v1/seed/ranges/${range.id}/release`).headers({ authorization: `Bearer ${second}` }).json({ error: 'test' })).assertStatus(200)
    assert.equal((await db.from('seed_ranges').where('id', range.id).first()).status, 'queued')
  })

  test('no leases while the database is over the disk limit', async ({ client }) => {
    const token = await SeedOrchestrator.issueToken('full-disk')
    await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 })
    const limit = indexerConfig.backfill.maxDefaultGb
    indexerConfig.backfill.maxDefaultGb = 0
    try {
      ;(await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${token}` })).assertStatus(204)
    } finally {
      indexerConfig.backfill.maxDefaultGb = limit
    }
    ;(await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${token}` })).assertStatus(200)
  })

  test('a range given back before it started does not use up an attempt', async ({ client, assert }) => {
    const token = await SeedOrchestrator.issueToken('no-news-server')
    await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 })
    const auth = { authorization: `Bearer ${token}` }
    const range = (await client.post('/api/v1/seed/lease').headers(auth)).body().range
    ;(await client.post(`/api/v1/seed/ranges/${range.id}/release`).headers(auth).json({ error: 'NNTP login: 502', uncounted: true })).assertStatus(200)
    const row = await db.from('seed_ranges').where('id', range.id).first()
    assert.equal(row.status, 'queued')
    assert.equal(row.attempts, 0)
  })

  test('relay: a leased range gets its overview chunks through the API', async ({ client, assert }) => {
    const token = await SeedOrchestrator.issueToken('relay-node')
    await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 })
    const auth = { authorization: `Bearer ${token}` }
    const range = (await client.post('/api/v1/seed/lease').headers(auth)).body().range
    const fetch = SeedFetchPool.fetch
    const connections = indexerConfig.seed.nntpConnections
    const asked: unknown[] = []
    SeedFetchPool.fetch = async (group, from, to) => {
      asked.push([group, from, to])
      return deflateSync(Buffer.from('5\tsubject\r\n'))
    }
    try {
      indexerConfig.seed.nntpConnections = 0
      ;(await client.get(`/api/v1/seed/ranges/${range.id}/chunk`).qs({ from: 1, to: 50 }).headers(auth)).assertStatus(503)
      indexerConfig.seed.nntpConnections = 1
      const chunk = await client.get(`/api/v1/seed/ranges/${range.id}/chunk`).qs({ from: 50, to: 500 }).headers(auth)
      chunk.assertStatus(200)
      // Clamped to the range.
      assert.deepEqual(asked, [['alt.binaries.test', 50, 100]])
      ;(await client.get(`/api/v1/seed/ranges/${range.id}/chunk`).qs({ from: 200, to: 300 }).headers(auth)).assertStatus(400)
    } finally {
      SeedFetchPool.fetch = fetch
      indexerConfig.seed.nntpConnections = connections
    }
  })

  test('a range reported to its end but never finished is marked done, not leased', async ({ client, assert }) => {
    const token = await SeedOrchestrator.issueToken('late-node')
    const [range] = await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 101 }).returning('id')
    ;(await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${token}` })).assertStatus(204)
    assert.equal((await db.from('seed_ranges').where('id', range.id).first()).status, 'done')
  })
})