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 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 \tMon, 05 Oct 2026 10:00:00 +0000\t\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 \tMon, 05 Oct 2026 10:00:00 +0000\t\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) assert.deepEqual((await db.from('rule_counts').where('range_id', rangeId)).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 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] ) const range = await db.from('ranges').where('id', rangeId).first() assert.equal(range.status, 'planned') assert.isNull(range.node_id) // 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) }) })