{
return this.timed(async () => {
const reply = await this.command(`XOVER ${from}-${to}`)
diff --git a/start/env.ts b/start/env.ts
index 1262ffb..a4246af 100644
--- a/start/env.ts
+++ b/start/env.ts
@@ -42,4 +42,15 @@ export default await Env.create(new URL('../', import.meta.url), {
NNTP_USER: Env.schema.string.optional(),
NNTP_PASSWORD: Env.schema.string.optional(),
NNTP_SECURE: Env.schema.boolean.optional(),
+ NNTP_CONNECTIONS: Env.schema.number.optional(),
+ NNTP_TIMEOUT_MS: Env.schema.number.optional(),
+
+ // Coordinator
+ RANGE_ARTICLES: Env.schema.number.optional(),
+ CHUNK_ARTICLES: Env.schema.number.optional(),
+ LEASE_TTL_SECONDS: Env.schema.number.optional(),
+ SPOOL_DIR: Env.schema.string.optional(),
+ SPOOL_AHEAD_PER_LEASE: Env.schema.number.optional(),
+ SPOOL_MIN_AHEAD: Env.schema.number.optional(),
+ SPOOL_MAX_GB: Env.schema.number.optional(),
})
diff --git a/start/routes.ts b/start/routes.ts
index 29fc17f..b8bcf5e 100644
--- a/start/routes.ts
+++ b/start/routes.ts
@@ -33,5 +33,19 @@ router
.prefix('account')
.as('profile')
.use(middleware.auth())
+
+ // Edge workers (REBUILD.md, parts 1–2).
+ router
+ .group(() => {
+ router.post('lease', [controllers.Node, 'lease'])
+ router.get('ranges/:id/stream', [controllers.Node, 'stream'])
+ router.post('ranges/:id/batches', [controllers.Node, 'batch'])
+ router.post('ranges/:id/heartbeat', [controllers.Node, 'heartbeat'])
+ router.post('ranges/:id/finish', [controllers.Node, 'finish'])
+ router.post('ranges/:id/release', [controllers.Node, 'release'])
+ })
+ .prefix('node')
+ .as('node')
+ .use(middleware.auth({ guards: ['node'] }))
})
.prefix('/api/v1')
diff --git a/tests/functional/node_api.spec.ts b/tests/functional/node_api.spec.ts
new file mode 100644
index 0000000..d4c5d14
--- /dev/null
+++ b/tests/functional/node_api.spec.ts
@@ -0,0 +1,83 @@
+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}`
+ )
+ 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)
+
+ ;(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 }))
+ })
+})