Rebuild: the batch store (catalog upserts, write-once chunks, candidates, rule counts, fragments, resume point)

A batch is stored in one transaction and only at its range's resume
point, so a retried batch can't count twice. Garbage is judged before
per-article: a dropped tool isn't worth reading bodies for either.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.adonisjs/client/manifest.d.ts

@@ -5,10 +5,5 @@
 
 /// <reference path="../../adonisrc.ts" />
 /// <reference path="../../config/auth.ts" />
-/// <reference path="../../config/gc.ts" />
 /// <reference path="../../config/hash.ts" />
-/// <reference path="../../config/indexer.ts" />
 /// <reference path="../../config/logger.ts" />
-/// <reference path="../../config/nntp.ts" />
-/// <reference path="../../config/queue.ts" />
-/// <reference path="../../config/redis.ts" />

.adonisjs/server/controllers.ts

@@ -6,7 +6,5 @@
 export const controllers = {
   AccessTokens: () => import('#controllers/access_tokens_controller'),
   NewAccount: () => import('#controllers/new_account_controller'),
-  Newznab: () => import('#controllers/newznab_controller'),
   Profile: () => import('#controllers/profile_controller'),
-  Releases: () => import('#controllers/releases_controller'),
 }

.adonisjs/server/routes.d.ts

@@ -4,27 +4,15 @@ type ParamValue = string | number | bigint | boolean
 
 export type ScannedRoutes = {
   ALL: {
-    'releases.index': { paramsTuple?: []; params?: {} }
-    'releases.show': { paramsTuple: [ParamValue]; params: {'id': ParamValue} }
-    'releases.nzb': { paramsTuple: [ParamValue]; params: {'id': ParamValue} }
-    'releases.stats': { paramsTuple?: []; params?: {} }
     'auth.new_account.store': { paramsTuple?: []; params?: {} }
     'auth.access_tokens.store': { paramsTuple?: []; params?: {} }
     'profile.profile.show': { paramsTuple?: []; params?: {} }
     'profile.access_tokens.destroy': { paramsTuple?: []; params?: {} }
   }
   GET: {
-    'releases.index': { paramsTuple?: []; params?: {} }
-    'releases.show': { paramsTuple: [ParamValue]; params: {'id': ParamValue} }
-    'releases.nzb': { paramsTuple: [ParamValue]; params: {'id': ParamValue} }
-    'releases.stats': { paramsTuple?: []; params?: {} }
     'profile.profile.show': { paramsTuple?: []; params?: {} }
   }
   HEAD: {
-    'releases.index': { paramsTuple?: []; params?: {} }
-    'releases.show': { paramsTuple: [ParamValue]; params: {'id': ParamValue} }
-    'releases.nzb': { paramsTuple: [ParamValue]; params: {'id': ParamValue} }
-    'releases.stats': { paramsTuple?: []; params?: {} }
     'profile.profile.show': { paramsTuple?: []; params?: {} }
   }
   POST: {

.adonisjs/server/routes.json

@@ -1 +0,0 @@
-{"root":[{"domain":"root","pattern":"/api/v1/releases","matchers":{},"tokens":[{"old":"/api/v1/releases","type":0,"val":"api","end":""},{"old":"/api/v1/releases","type":0,"val":"v1","end":""},{"old":"/api/v1/releases","type":0,"val":"releases","end":""}],"meta":{"params":[]},"name":"releases.index","handler":{"method":"index","reference":[null,"index"],"importExpression":"()=>import('#controllers/releases_controller')","name":"ReleasesController"},"methods":["GET","HEAD"],"middleware":{}},{"domain":"root","pattern":"/api/v1/releases/:id","matchers":{"id":{"match":{}}},"tokens":[{"old":"/api/v1/releases/:id","type":0,"val":"api","end":""},{"old":"/api/v1/releases/:id","type":0,"val":"v1","end":""},{"old":"/api/v1/releases/:id","type":0,"val":"releases","end":""},{"old":"/api/v1/releases/:id","type":1,"val":"id","end":"","matcher":{}}],"meta":{"params":["id"]},"name":"releases.show","handler":{"method":"show","reference":[null,"show"],"importExpression":"()=>import('#controllers/releases_controller')","name":"ReleasesController"},"methods":["GET","HEAD"],"middleware":{}},{"domain":"root","pattern":"/api/v1/releases/:id/nzb","matchers":{"id":{"match":{}}},"tokens":[{"old":"/api/v1/releases/:id/nzb","type":0,"val":"api","end":""},{"old":"/api/v1/releases/:id/nzb","type":0,"val":"v1","end":""},{"old":"/api/v1/releases/:id/nzb","type":0,"val":"releases","end":""},{"old":"/api/v1/releases/:id/nzb","type":1,"val":"id","end":"","matcher":{}},{"old":"/api/v1/releases/:id/nzb","type":0,"val":"nzb","end":""}],"meta":{"params":["id"]},"name":"releases.nzb","handler":{"method":"nzb","reference":[null,"nzb"],"importExpression":"()=>import('#controllers/releases_controller')","name":"ReleasesController"},"methods":["GET","HEAD"],"middleware":{}},{"domain":"root","pattern":"/api/v1/stats","matchers":{},"tokens":[{"old":"/api/v1/stats","type":0,"val":"api","end":""},{"old":"/api/v1/stats","type":0,"val":"v1","end":""},{"old":"/api/v1/stats","type":0,"val":"stats","end":""}],"meta":{"params":[]},"name":"releases.stats","handler":{"method":"stats","reference":[null,"stats"],"importExpression":"()=>import('#controllers/releases_controller')","name":"ReleasesController"},"methods":["GET","HEAD"],"middleware":{}},{"domain":"root","pattern":"/api/v1/auth/signup","matchers":{},"tokens":[{"old":"/api/v1/auth/signup","type":0,"val":"api","end":""},{"old":"/api/v1/auth/signup","type":0,"val":"v1","end":""},{"old":"/api/v1/auth/signup","type":0,"val":"auth","end":""},{"old":"/api/v1/auth/signup","type":0,"val":"signup","end":""}],"meta":{"params":[]},"name":"auth.new_account.store","handler":{"method":"store","reference":[null,"store"],"importExpression":"()=>import('#controllers/new_account_controller')","name":"NewAccount"},"methods":["POST"],"middleware":{}},{"domain":"root","pattern":"/api/v1/auth/login","matchers":{},"tokens":[{"old":"/api/v1/auth/login","type":0,"val":"api","end":""},{"old":"/api/v1/auth/login","type":0,"val":"v1","end":""},{"old":"/api/v1/auth/login","type":0,"val":"auth","end":""},{"old":"/api/v1/auth/login","type":0,"val":"login","end":""}],"meta":{"params":[]},"name":"auth.access_tokens.store","handler":{"method":"store","reference":[null,"store"],"importExpression":"()=>import('#controllers/access_tokens_controller')","name":"AccessTokens"},"methods":["POST"],"middleware":{}},{"domain":"root","pattern":"/api/v1/account/profile","matchers":{},"tokens":[{"old":"/api/v1/account/profile","type":0,"val":"api","end":""},{"old":"/api/v1/account/profile","type":0,"val":"v1","end":""},{"old":"/api/v1/account/profile","type":0,"val":"account","end":""},{"old":"/api/v1/account/profile","type":0,"val":"profile","end":""}],"meta":{"params":[]},"name":"profile.profile.show","handler":{"method":"show","reference":[null,"show"],"importExpression":"()=>import('#controllers/profile_controller')","name":"Profile"},"methods":["GET","HEAD"],"middleware":{}},{"domain":"root","pattern":"/api/v1/account/logout","matchers":{},"tokens":[{"old":"/api/v1/account/logout","type":0,"val":"api","end":""},{"old":"/api/v1/account/logout","type":0,"val":"v1","end":""},{"old":"/api/v1/account/logout","type":0,"val":"account","end":""},{"old":"/api/v1/account/logout","type":0,"val":"logout","end":""}],"meta":{"params":[]},"name":"profile.access_tokens.destroy","handler":{"method":"destroy","reference":[null,"destroy"],"importExpression":"()=>import('#controllers/access_tokens_controller')","name":"AccessTokens"},"methods":["POST"],"middleware":{}}]}

.dockerignore

@@ -1,11 +0,0 @@
-node_modules
-build
-tmp
-debug
-.git
-.idea
-.vscode
-.env
-.env.*
-!.env.example
-*.log

.env.example

@@ -12,9 +12,14 @@ APP_URL=http://${HOST}:${PORT}
 # Session
 SESSION_DRIVER=cookie
 
+# Database (the rebuild's own)
+DB_HOST=127.0.0.1
+DB_PORT=5432
+DB_USER=usenet_indexer
+DB_PASSWORD=
+DB_DATABASE=usenet_rebuild
+
 #--------------------------------------------------------------------
 # CORS (configure allowed origins for API access)
 #--------------------------------------------------------------------
 # CORS_ORIGIN=http://localhost:5173,http://localhost:3000
-# Newznab API key (/api?apikey=...); leave empty to allow access without a key
-NEWZNAB_API_KEY=

.gitea/workflows/deploy.yaml

@@ -0,0 +1,81 @@
+# Tests every push; a push to master that passes is deployed to homelab.
+#
+# Deploying: the job SSHes to homelab with DEPLOY_SSH_KEY. That key may only
+# run ~/bin/deploy-usenet-indexer (a forced command in authorized_keys),
+# which fetches the commit from this repo, builds the image, runs the
+# migrations and recreates the app services (never postgres or redis).
+name: Test and deploy
+
+on:
+  push:
+  workflow_dispatch:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:17
+        env:
+          POSTGRES_USER: indexer
+          POSTGRES_PASSWORD: indexer
+          POSTGRES_DB: usenet_indexer
+        options: >-
+          --health-cmd "pg_isready -U indexer -d usenet_indexer"
+          --health-interval 5s
+          --health-retries 20
+      redis:
+        image: redis:8
+    env:
+      TZ: UTC
+      PORT: 3333
+      HOST: 0.0.0.0
+      NODE_ENV: test
+      LOG_LEVEL: warn
+      APP_KEY: ci-only-app-key-not-a-secret-000000
+      APP_URL: http://localhost:3333
+      SESSION_DRIVER: memory
+      DB_CONNECTION: pg
+      DB_HOST: postgres
+      DB_PORT: 5432
+      DB_USER: indexer
+      DB_PASSWORD: indexer
+      DB_DATABASE: usenet_indexer
+      REDIS_HOST: redis
+      REDIS_PORT: 6379
+      # Tests never connect to a news server.
+      NNTP_HOST: localhost
+      NNTP_PORT: 563
+      NNTP_USER: ci
+      NNTP_PASSWORD: ci
+      NNTP_SECURE: true
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version-file: .nvmrc
+          cache: npm
+      - run: npm ci
+      - name: Type check
+        run: npx tsc --noEmit
+      - name: Migrate
+        run: node ace migration:run --force
+      - name: Test
+        run: node ace test
+
+  deploy:
+    needs: test
+    if: github.ref == 'refs/heads/master'
+    runs-on: ubuntu-latest
+    steps:
+      - name: Deploy to homelab
+        env:
+          DEPLOY_SSH_KEY: ${{ secrets.DEPLOY_SSH_KEY }}
+          DEPLOY_KNOWN_HOSTS: ${{ secrets.DEPLOY_KNOWN_HOSTS }}
+        run: |
+          install -m 700 -d ~/.ssh
+          printf '%s\n' "$DEPLOY_SSH_KEY" > ~/.ssh/deploy_key
+          chmod 600 ~/.ssh/deploy_key
+          printf '%s\n' "$DEPLOY_KNOWN_HOSTS" > ~/.ssh/known_hosts
+          ssh -i ~/.ssh/deploy_key -o IdentitiesOnly=yes -o BatchMode=yes \
+            daan@10.206.0.175 "deploy ${{ github.sha }}"

Dockerfile

@@ -1,29 +0,0 @@
-# syntax=docker/dockerfile:1
-
-# Node is pinned: on 24.21, jsonschema 1.5.0 (used by Ace to validate commands)
-# throws "Invalid URL" on $ref, so no ace command starts. See LOG.md.
-ARG NODE_VERSION=24.16
-
-# Build stage: dev dependencies, TypeScript compile, native addons.
-# The full node image ships python3/make/g++, needed to compile yencode.
-FROM node:${NODE_VERSION}-bookworm AS build
-WORKDIR /app
-COPY package.json package-lock.json ./
-RUN npm ci
-COPY . .
-RUN node ace build \
- && cd build \
- && npm ci --omit=dev
-
-# Runtime stage: compiled app with production dependencies only.
-FROM node:${NODE_VERSION}-bookworm-slim
-ENV NODE_ENV=production
-# bsdtar (libarchive) unpacks compressed RAR entries, e.g. NZBs posted inside RARs.
-RUN apt-get update \
- && apt-get install -y --no-install-recommends libarchive-tools \
- && rm -rf /var/lib/apt/lists/*
-WORKDIR /app
-COPY --from=build /app/build ./
-USER node
-# Every service picks its own command, e.g. `node ace worker:fetch`.
-CMD ["node", "ace", "list"]

REBUILD.md

@@ -0,0 +1,137 @@
+# Rebuild: header-first indexer (spec, draft for approval)
+
+Branch `rebuild`. Phase 1 finds releases from headers and makes their NZBs.
+Everything learned from articles (names from yEnc headers, par2, archive
+listings, encryption, spots) is phase 2 and adds to what phase 1 stored; it
+never rewrites it. Sources: CONCEPT.md, LOG.md 2026-10-08 (13)–(14).
+
+## Target
+
+7,000 days (all of Eweka's retention, back to 2008-08) of the 1000 largest
+`alt.binaries.*` groups. Estimated: 528 B overview lines; 435 B in their
+main group; 140 B kept segments in ~1.25 B files and ~115 M releases after
+dropping garbage; 28 TB of compressed overview traffic. Throughput goal:
+≥ 25 B overview lines/day (~50 connections), so ~3 weeks.
+
+## Parts
+
+1. **Coordinator** (server). Plans each group's history as ranges of 2 M
+   articles, leases them to workers, tracks progress and retries. The
+   group list (owner rule) and the garbage signature table go with every
+   lease. Fetches XOVER for the workers (relay: workers have no NNTP
+   account; the gzip stream is passed on as is).
+
+   **Prefetch**: the coordinator fetches the next ranges ahead of the
+   workers, in chunks of 50 k lines and in order, into a spool on disk
+   (~106 MB per 2 M range at ~53 B per line; on the HDD, written and read
+   sequentially). It keeps about two ranges ahead per active lease, capped
+   by size (default 10 GB), and fetches with all its header connections all
+   the time, independent of how fast workers are. A lease goes to a range
+   that is already (partly) spooled; a worker reads its chunks from the
+   spool, waiting only for chunks still being fetched, and falls back to a
+   direct relay fetch for a range that isn't spooled. A worker gets its
+   whole range in one streaming request (`GET /ranges/:id/stream`): the
+   spooled chunks in order, waiting for those still being fetched. It still
+   parses chunk by chunk (2 M lines are ~560 MB of text) and posts results
+   per batch, moving the range's resume point. Chunks matter on the NNTP
+   side: a compressed XOVER reply can't be resumed, old ranges answer
+   slowly, and chunks of one range can be fetched in parallel. Spooled chunks are
+   deleted when their range's results are stored, and fetched again if a
+   range is leased again later.
+   Order: oldest era first (2008–2018 holds 63% of the readable files on
+   17% of the lines), then forward.
+
+2. **Edge worker** (node, TypeScript reusing `lib/`). Measured (`edge:bench`,
+   2026-10-09): 73 k headers/s per core on an a.b.ath range from 2013,
+   103 k/s on a recent a.b.boneless range (2 M headers each), ~1.2 GB peak
+   per process; the 25 B lines/day goal needs ~3–4 cores over all nodes,
+   so no Rust. Per lease: compressed chunks of 10–50 k lines, in order →
+   parse subjects → skip crossposts outside their owner group (of the
+   indexed groups in the article's Xref, the one with the lowest priority
+   number; see the group list) → group articles into files → files
+   into releases (`lib/release` rules) → drop garbage (random name +
+   dropped signature or tool) → set aside per-article random subjects and
+   obfuscated subjects (counted only; phase 2). Returns per batch: release
+   chunks, edge fragments, probe candidates, counters. Holds no state beyond
+   its lease.
+
+   **Probe candidates**: per release, the articles phase 2 should read, as
+   (release key, kind, message-id, file in its chunk): the first article of
+   the first archive volume (`.part01.rar`, `.rar`, `.7z.001`), of the par2
+   index (smallest `.par2`), and of a lone random-named file; for a 7z set
+   also its last volume (the listing is in the tail). At most a few per
+   release (~115 M releases → ~150–250 M articles), not one per file. Per-
+   article and obfuscated subjects get none (they need every article's body).
+
+3. **Chunk store** (bulk, append-only). One record per release per batch:
+   its files' names, sizes and segment lists, compactly encoded (template
+   codec for message-ids, zstd). Written once, never updated. A batch is
+   stored in one transaction and only when it starts at its range's resume
+   point, so a batch sent twice or late is refused and the catalog's sums
+   stay right (no per-file dedup); a recovery run for a retracted rule only
+   adds what that rule had dropped. ~3 TB.
+
+4. **Release catalog** (hot, Postgres on NVMe). One row per release: key,
+   name, poster, groups, posted_at, file count, size, segments, expected
+   file count (`[i/N]`), has-par2, readable-name flag, state. Updated with
+   aggregates once per batch (one upsert per release per batch), plus a
+   small name index for search. ~115 M rows, ~50–100 GB.
+
+5. **Rules and recovery**. What a worker skips isn't stored (message-ids
+   are the bulk: garbage alone would be ~2.5 TB); a skipped range is read
+   again instead (headers are cheap: ~53 B per overview line compressed).
+   - *Per range, per rule and signature*: how many articles and files the
+     rule dropped (garbage per (tool, group), per-article random subjects,
+     obfuscated subjects, and crossposts per owner group). A few rows per
+     range.
+   - *Rule versions*: every range records the version of the rule set it ran
+     with. Retracting or changing a rule makes the ranges where it matched
+     stale; the coordinator leases them again for that rule only, and the
+     worker returns just what the rule had dropped.
+   - *Sample*: ~1% of what each rule drops (by a hash of the release key) is
+     kept with full file details and probe candidates, so a rule can be
+     checked, and its losses measured in phase 2, before it is retracted.
+
+6. **Probe queue**. The candidates, in the catalog's database: one small row
+   each (release, kind, message-id, chunk reference), priority by the
+   release's state (settled first) and kind (random names before readable
+   ones). Phase 2 takes them from here; phase 1 only fills it.
+
+7. **Fragments**. Files crossing a range edge: kept by range and article
+   number; merged when the neighbouring range is done, then written as a
+   chunk. Never staged per segment.
+
+8. **Settling**. A release is settled when the ranges covering its posting
+   span are done and no chunk arrived for a day: completeness is computed
+   then (expected vs found files and segments). Phase 2 picks its work from
+   settled releases.
+
+9. **NZB and API**. NZB built from a release's chunks; a minimal Newznab
+   search (names) and get. Web UI later.
+
+## Group list
+
+Editable at any time: an added group gets its ranges planned, a removed one
+no new leases. Each group gets a priority number when it is added (the first
+list by size, later additions after it); a crosspost belongs to the indexed
+group in its Xref with the lowest number. So adding a group never takes
+articles from groups already done, and removing one re-reads only the ranges
+of other groups whose counts show crossposts skipped for it (part 5). Start:
+the 1000 largest `alt.binaries.*` groups by LIST ACTIVE (~266 k ranges of
+2 M articles; boneless alone 80 k).
+
+## Not in phase 1
+
+Inspection and archive walks, par2/SFV naming, spots, NZB imports, the
+after-probe garbage deletes, live indexing (the current indexer keeps
+running and does those meanwhile).
+
+## Decisions
+
+a. Chunk payload: a write-once `bytea` table in Postgres, partitioned by
+   month, on the HDD; files on disk only if writes still limit. **Accepted.**
+b. A separate database on the same server. **Accepted.**
+c. Group list: editable, owner by priority number (above). **Accepted.**
+d. Reposts: separate releases in phase 1, merged in phase 2. **Accepted.**
+e. Headers through the coordinator's relay, with prefetch (part 1).
+   **Accepted.**

adonisrc.ts

@@ -1,4 +1,4 @@
-import { indexEntities } from '@adonisjs/core'
+import { indexEntities } from '@adonisjs/core/generators'
 import { defineConfig } from '@adonisjs/core/app'
 import { generateRegistry } from '@tuyau/core/hooks'
 
@@ -53,7 +53,6 @@ export default defineConfig({
     () => import('@adonisjs/cors/cors_provider'),
     () => import('@adonisjs/auth/auth_provider'),
     () => import('#providers/api_provider'),
-    () => import('@adonisjs/static/static_provider')
   ],
 
   /*
@@ -104,10 +103,7 @@ export default defineConfig({
   | the production build.
   |
   */
-  metaFiles: [{
+  metaFiles: [],
-    pattern: 'public/**',
-    reloadServer: false,
-  }],
 
   hooks: {
     init: [

app/controllers/newznab_controller.ts

@@ -1,121 +0,0 @@
-import type { HttpContext } from '@adonisjs/core/http'
-import env from '#start/env'
-import SearchService from '#services/SearchService'
-import { capsXml, errorXml, expandCategories, resultsXml, type NewznabServer } from '#lib/newznab/index'
-
-const LIMITS = { max: 100, default: 50 }
-
-/** Search functions and the category each defaults to when `cat` is absent. */
-const SEARCH_FUNCTIONS: Record<string, number | undefined> = {
-  'search': undefined,
-  'tvsearch': 5000,
-  'tv-search': 5000,
-  'movie': 2000,
-  'movie-search': 2000,
-  'music': 3000,
-  'audio': 3000,
-  'book': 7000,
-}
-
-/** The configured key, if any (a Secret from .env, or a plain string set at runtime). */
-function configuredApiKey(): string | undefined {
-  const value: any = env.get('NEWZNAB_API_KEY')
-  const key = typeof value === 'string' ? value : value?.release()
-  return key || undefined
-}
-
-function int(value: unknown): number | undefined {
-  return typeof value === 'string' && /^\d+$/.test(value) ? Number(value) : undefined
-}
-
-/**
- * Newznab API (CONCEPT.md D23) for Prowlarr, Sonarr, Radarr and the like:
- * GET /api?t=caps|search|tvsearch|movie|music|book|get&apikey=…
- */
-export default class NewznabController {
-  async handle({ request, response }: HttpContext) {
-    const qs = request.qs()
-    const t = String(qs.t ?? '')
-    const server = this.server(request)
-
-    const xml = (body: string, type = 'application/xml') => {
-      response.header('Content-Type', `${type}; charset=utf-8`)
-      return response.send(body)
-    }
-
-    const apiKey = configuredApiKey()
-    if (apiKey && qs.apikey !== apiKey) return xml(errorXml(100, 'Incorrect user credentials'))
-
-    if (t === 'caps') return xml(capsXml(server, LIMITS))
-
-    if (t === 'get') {
-      const id = int(qs.id)
-      if (id === undefined) return xml(errorXml(200, 'Missing parameter (id)'))
-      const nzb = await SearchService.nzb(id)
-      if (!nzb) return xml(errorXml(300, 'No such item'))
-      const filename = `${nzb.name.replace(/[^\w.\-()[\] ]+/g, '_').slice(0, 180) || 'release'}.nzb`
-      response.header('Content-Disposition', `attachment; filename="${filename}"`)
-      return xml(nzb.xml, 'application/x-nzb')
-    }
-
-    if (!(t in SEARCH_FUNCTIONS)) return xml(errorXml(202, `No such function (${t || 'none'})`))
-
-    const requested = String(qs.cat ?? '')
-      .split(',')
-      .map((c) => int(c.trim()))
-      .filter((c): c is number => c !== undefined)
-    const fallback = SEARCH_FUNCTIONS[t]
-    const categories = expandCategories(requested.length ? requested : fallback ? [fallback] : [])
-    const limit = Math.min(int(qs.limit) ?? LIMITS.default, LIMITS.max)
-    const offset = int(qs.offset) ?? 0
-
-    // ID-only lookups (tvdbid, imdbid, …) are not supported: caps says so, but
-    // answer with an empty result rather than everything.
-    const q = typeof qs.q === 'string' ? qs.q.trim() : ''
-    const idOnly = !q && ['tvdbid', 'tvmazeid', 'rid', 'imdbid', 'tmdbid'].some((p) => qs[p])
-
-    const result = idOnly
-      ? { total: 0, releases: [] }
-      : await SearchService.search({
-          q,
-          categories,
-          readable: true,
-          notIncomplete: true,
-          hideContainers: true,
-          maxAgeDays: int(qs.maxage),
-          season: int(qs.season),
-          episode: int(qs.season) !== undefined ? int(qs.ep) : undefined,
-          sort: 'newest',
-          perPage: limit,
-          offset,
-        })
-
-    const items = result.releases.map((r: any) => {
-      const tv = /(?:^|[._ -])S(\d{1,2})E(\d{1,3})/i.exec(r.name)
-      return {
-        id: r.id,
-        title: r.name,
-        category: r.category ?? 8010,
-        size: r.size,
-        files: r.fileCount,
-        postedAt: r.postedAt ? new Date(r.postedAt) : null,
-        poster: r.poster,
-        groups: r.groups ?? [],
-        password: Boolean(r.encrypted),
-        season: tv ? `S${tv[1].padStart(2, '0')}` : undefined,
-        episode: tv ? `E${tv[2].padStart(2, '0')}` : undefined,
-      }
-    })
-    return xml(resultsXml(server, items, offset, result.total), 'application/rss+xml')
-  }
-
-  private server(request: HttpContext['request']): NewznabServer {
-    const proto = request.header('x-forwarded-proto') ?? 'http'
-    const host = request.header('x-forwarded-host') ?? request.header('host') ?? 'localhost'
-    return {
-      title: 'Usenet Indexer',
-      baseUrl: `${proto}://${host}`,
-      apiKey: configuredApiKey(),
-    }
-  }
-}

app/controllers/releases_controller.ts

@@ -1,44 +0,0 @@
-import type { HttpContext } from '@adonisjs/core/http'
-import SearchService, { type SortOrder } from '#services/SearchService'
-
-const flag = (value: unknown) => value === '1' || value === 'true' || value === true
-
-export default class ReleasesController {
-  /** GET /api/v1/releases?q=&readable=1&complete=1&hide_encrypted=1&source=&poster=&sort=&page=&per_page= */
-  async index({ request }: HttpContext) {
-    const qs = request.qs()
-    return SearchService.search({
-      q: typeof qs.q === 'string' ? qs.q : undefined,
-      readable: flag(qs.readable),
-      complete: flag(qs.complete),
-      hideEncrypted: flag(qs.hide_encrypted),
-      includePar2Only: flag(qs.par2_only),
-      source: typeof qs.source === 'string' && qs.source ? qs.source : undefined,
-      poster: typeof qs.poster === 'string' && qs.poster ? qs.poster : undefined,
-      sort: qs.sort as SortOrder | undefined,
-      page: Number(qs.page) || 1,
-      perPage: Number(qs.per_page) || 50,
-    })
-  }
-
-  /** GET /api/v1/releases/:id */
-  async show({ params, response }: HttpContext) {
-    const release = await SearchService.release(Number(params.id))
-    return release ?? response.notFound({ error: 'Release not found' })
-  }
-
-  /** GET /api/v1/releases/:id/nzb */
-  async nzb({ params, response }: HttpContext) {
-    const nzb = await SearchService.nzb(Number(params.id))
-    if (!nzb) return response.notFound({ error: 'Release not found' })
-    const filename = `${nzb.name.replace(/[^\w.\-()[\] ]+/g, '_').slice(0, 180) || 'release'}.nzb`
-    response.header('Content-Type', 'application/x-nzb')
-    response.header('Content-Disposition', `attachment; filename="${filename}"`)
-    return response.send(nzb.xml)
-  }
-
-  /** GET /api/v1/stats */
-  async stats() {
-    return SearchService.stats()
-  }
-}

app/models/base_model.ts

@@ -1,4 +0,0 @@
-import { BaseModel as AdonisBaseModel } from '@adonisjs/lucid/orm'
-
-export default class BaseModel extends AdonisBaseModel {
-}

app/models/file.ts

@@ -1,35 +0,0 @@
-import { DateTime } from 'luxon'
-import { column } from '@adonisjs/lucid/orm'
-import BaseModel from '#models/base_model'
-
-export default class File extends BaseModel {
-  @column({ isPrimary: true })
-  declare id: number
-
-  @column({ columnName: 'file_key' })
-  declare fileKey: string | null
-
-  @column()
-  declare filename: string
-
-  @column()
-  declare poster: string
-
-  @column()
-  declare date: number
-
-  @column()
-  declare parts: number
-
-  @column({ columnName: 'message_ids' })
-  declare messageIds: Record<string, any>
-
-  @column()
-  declare groups: string[]
-
-  @column.dateTime({ autoCreate: true })
-  declare createdAt: DateTime
-
-  @column.dateTime({ autoCreate: true, autoUpdate: true })
-  declare updatedAt: DateTime
-}

app/models/group.ts

@@ -1,32 +0,0 @@
-import { DateTime } from 'luxon'
-import { column } from '@adonisjs/lucid/orm'
-import BaseModel from '#models/base_model'
-
-export default class Group extends BaseModel {
-  @column({ isPrimary: true })
-  declare id: number
-
-  @column()
-  declare name: string
-
-  @column()
-  declare active: boolean
-
-  /** Every article up to here was fetched successfully (see FetchRangeService). */
-  @column({ columnName: 'last_indexed_id' })
-  declare lastIndexedId: bigint
-
-  /** Download bodies of obfuscated articles to identify them (bandwidth-heavy). */
-  @column({ columnName: 'fetch_bodies' })
-  declare fetchBodies: boolean
-
-  /** How far the scheduler has queued fetch jobs. */
-  @column({ columnName: 'last_queued_id' })
-  declare lastQueuedId: bigint
-
-  @column.dateTime({ autoCreate: true })
-  declare createdAt: DateTime
-
-  @column.dateTime({ autoCreate: true, autoUpdate: true })
-  declare updatedAt: DateTime
-}

app/services/ArchiveService.ts

@@ -1,160 +0,0 @@
-import { execFile } from 'node:child_process'
-import { mkdtemp, rm, writeFile } from 'node:fs/promises'
-import { tmpdir } from 'node:os'
-import path from 'node:path'
-import { promisify } from 'node:util'
-import db from '@adonisjs/lucid/services/db'
-import logger from '@adonisjs/core/services/logger'
-import IngestService from '#services/IngestService'
-import { storedEntryData, type RarEntry, type RarInfo } from '#lib/rar/index'
-import { fileStem, rarReleaseIdentity } from '#lib/release/index'
-
-/** Stored per file and release; a listing of thousands of files is not useful. */
-const MAX_ENTRIES = 500
-
-export interface ArchiveResult {
-  info: RarInfo
-  releaseId?: number
-  releaseRenamed?: string
-  grouped?: boolean
-}
-
-/**
- * Records what a RAR volume's headers say: the listing on the file, the
- * release it belongs to (from the shared directory or continued entry, when
- * the file has none yet), and for first volumes the contents and password
- * flag on the release. See CONCEPT.md (D19, D21).
- */
-class ArchiveService {
-  async record(file: any, binary: any, info: RarInfo): Promise<ArchiveResult> {
-    const stored = { ...info, entries: info.entries.slice(0, MAX_ENTRIES).map(({ stored: _s, dataOffset: _d, ...entry }) => entry) }
-    await db.from('files').where('id', file.id).update({ archive: JSON.stringify(stored) })
-
-    let releaseId = file.release_id ?? binary?.release_id ?? null
-    let grouped = false
-    const identity = rarReleaseIdentity(info.entries)
-    if (!releaseId && identity) {
-      const ids = await IngestService.upsertReleases(db.connection().getWriteClient(), [
-        {
-          release: identity,
-          poster: binary?.poster ?? file.poster,
-          weakIdentity: true,
-          groups: binary?.groups ?? file.groups ?? [],
-          postedAt: binary?.posted_at ? new Date(binary.posted_at) : undefined,
-        },
-      ])
-      releaseId = ids.get(identity.key)!
-      grouped = true
-      if (binary) await db.from('binaries').where('id', binary.id).update({ release_id: releaseId })
-      await db.from('files').where('id', file.id).update({ release_id: releaseId })
-    }
-    if (!releaseId) return { info }
-
-    const release = await db.from('releases').where('id', releaseId).first()
-    const update: Record<string, unknown> = {}
-    if (info.firstVolume !== false) {
-      update.encrypted = info.headersEncrypted || info.entries.some((e) => e.encrypted)
-      if (info.entries.length > 0) update.contents = JSON.stringify(stored.entries)
-    } else if (info.headersEncrypted) {
-      update.encrypted = true
-    }
-    const better = releaseNameFromContents(info)
-    const renamed = release && better && (looksRandom(release.name) || release.source === 'post') ? better : undefined
-    if (renamed) update.name = renamed
-    if (Object.keys(update).length > 0) await db.from('releases').where('id', releaseId).update(update)
-
-    if (renamed) logger.info(`Release ${releaseId} renamed "${release.name}" → "${renamed}" from its archive.`)
-    return { info, releaseId, releaseRenamed: renamed, grouped }
-  }
-}
-
-const run = promisify(execFile)
-
-/** Tools that print one archive entry to stdout, tried in order. */
-const EXTRACTORS: [string, (archive: string, entry: string) => string[]][] = [
-  ['bsdtar', (archive, entry) => ['-xOf', archive, entry]],
-  ['unrar', (archive, entry) => ['p', '-inul', '-p-', '--', archive, entry]],
-]
-
-/**
- * The unpacked data of one entry of a complete (single-volume) archive:
- * copied out when stored, else unpacked with bsdtar (libarchive, in the
- * Docker image) or unrar. Null when that is not possible.
- */
-export async function extractEntry(archive: Buffer, entry: RarEntry, maxBytes = 64 * 1024 * 1024): Promise<Buffer | null> {
-  const stored = storedEntryData(archive, entry)
-  if (stored) return stored
-  if (entry.encrypted || entry.directory || entry.splitBefore || entry.splitAfter) return null
-  if (entry.size !== null && entry.size > maxBytes) return null
-
-  const dir = await mkdtemp(path.join(tmpdir(), 'rar-'))
-  try {
-    const file = path.join(dir, 'archive.rar')
-    await writeFile(file, archive)
-    for (const [tool, args] of EXTRACTORS) {
-      try {
-        const { stdout } = await run(tool, args(file, entry.name), { encoding: 'buffer', maxBuffer: maxBytes, timeout: 60_000 })
-        if (stdout.length > 0) return stdout
-      } catch (error: any) {
-        if (error.code !== 'ENOENT') logger.warn(`${tool} could not extract "${entry.name}": ${error.message}`)
-      }
-    }
-    return null
-  } finally {
-    await rm(dir, { recursive: true, force: true })
-  }
-}
-
-/**
- * A release name from an archive listing: the top-level directory shared by
- * all entries when it reads like a release name, otherwise the stem of the
- * largest entry with a readable name (e.g. an archive holding just
- * `Joker.Folie.a.Deux.2024.2160p….nzb`). Samples are skipped.
- */
-export function releaseNameFromContents(info: Pick<RarInfo, 'entries'>): string | undefined {
-  const entries = info.entries.filter((e) => !e.directory)
-  const tops = new Set(info.entries.map((e) => e.name.split(/[\\/]/)[0]))
-  if (tops.size === 1 && info.entries.some((e) => /[\\/]/.test(e.name))) {
-    const [top] = tops
-    if (/[A-Za-z]{2,}.*[._-].*[A-Za-z0-9]{2,}/.test(top) && !looksRandom(top)) return top
-  }
-  const candidates = entries
-    .map((e) => ({ stem: fileStem(e.name.split(/[\\/]/).pop() ?? e.name), size: e.size ?? 0 }))
-    .filter(({ stem }) => !/(^|[._ -])sample([._ -]|$)/i.test(stem))
-    .filter(({ stem }) => /[A-Za-z]{2,}.*[._ -].*[A-Za-z0-9]{2,}/.test(stem) && !looksRandom(stem))
-    .sort((a, b) => b.size - a.size)
-  return candidates[0]?.stem
-}
-
-/**
- * Renames random-named releases from their stored archive listing (no
- * downloads). Used by `release:rename`.
- */
-export async function renameFromStoredContents(): Promise<number> {
-  const releases = await db
-    .from('releases')
-    .whereNotNull('contents')
-    .whereRaw("jsonb_array_length(contents) > 0")
-    .select('id', 'name', 'source', 'contents')
-  let renamed = 0
-  for (const release of releases) {
-    if (!looksRandom(release.name) && release.source !== 'post') continue
-    const better = releaseNameFromContents({ entries: release.contents })
-    if (!better || better === release.name) continue
-    await db.from('releases').where('id', release.id).update({ name: better })
-    renamed++
-  }
-  return renamed
-}
-
-/** Obfuscated names: one long token of mixed letters and digits, no separators. */
-export function looksRandom(name: string): boolean {
-  // An unknown short extension (e.g. ".bin") says nothing about the name.
-  name = name.replace(/\.[A-Za-z0-9]{1,4}$/, '')
-  if (/^[0-9a-f]{16,}$/i.test(name)) return true
-  if (!/^[A-Za-z0-9]{16,}$/.test(name) || !/[a-z]/.test(name) || !/[A-Z]/.test(name)) return false
-  // Digits, or many capitals without separators (e.g. RmISakytFwXSCEzjVEpaKk).
-  return /\d/.test(name) || (name.match(/[A-Z]/g) ?? []).length >= 6
-}
-
-export default new ArchiveService()

app/services/ContentService.ts

@@ -1,187 +0,0 @@
-import { createHash } from 'node:crypto'
-import db from '@adonisjs/lucid/services/db'
-import NntpService from '#services/NntpService'
-import Par2Service, { type Par2Result } from '#services/Par2Service'
-import ArchiveService, { extractEntry, looksRandom, type ArchiveResult } from '#services/ArchiveService'
-import NameService from '#services/NameService'
-import { decodeSegment, YencCrcError, YencError, YencFile } from '#lib/yenc/index'
-import NzbImportService, { type NzbImportResult } from '#services/NzbImportService'
-import { sniffContentType, type ContentType } from '#lib/content/index'
-import { parseRarHeaders, type RarEntry, type RarInfo } from '#lib/rar/index'
-import { fileStem } from '#lib/release/index'
-import { looksLikeFilename } from '#lib/subject/index'
-
-const FIRST_CHUNK = 16 * 1024
-
-/**
- * MD5 of the first 16 KiB of a file, as par2 records it, from the body of
- * the file's first part. Undefined when that part is shorter than 16 KiB but
- * the file is not (the chunk would span two parts).
- */
-export function firstChunkMd5(body: Buffer, fileSize: number): string | undefined {
-  return chunkMd5(decodeSegment(body).data, fileSize)
-}
-
-function chunkMd5(data: Buffer, fileSize: number): string | undefined {
-  if (data.length < Math.min(FIRST_CHUNK, fileSize)) return undefined
-  return createHash('md5').update(data.subarray(0, FIRST_CHUNK)).digest('hex')
-}
-
-export interface ContentResult {
-  par2?: Par2Result
-  archive?: ArchiveResult
-  named: number
-  contentType: ContentType | null
-  nzb?: NzbImportResult
-  /** NZBs read from inside an archive. */
-  archiveNzbs?: NzbImportResult[]
-}
-
-/**
- * Takes names and groupings from file contents (CONCEPT.md D21). For a
- * completed file with a random name, one article (the first) is enough:
- * - its fingerprint (size + first-16-KiB MD5) matches it to a par2 entry,
- * - par2 files are recognised by content and name every file of their set,
- * - RAR headers give the release grouping, name and contents.
- */
-class ContentService {
-  /** Random or missing names, and named first RAR volumes (for contents and password flag). */
-  isCandidate(name: string) {
-    if (!looksLikeFilename(name) || looksRandom(fileStem(name))) return true
-    if (/\.nzb$/i.test(name)) return true
-    return /\.part0*1\.rar$/i.test(name) || (/\.rar$/i.test(name) && !/\.part\d+\.rar$/i.test(name))
-  }
-
-  /** Downloads a whole (small) file whose first segment is already decoded. */
-  private async readWhole(first: Buffer, messageIds: Record<string, { id: string; size: number }>): Promise<Buffer | null> {
-    const parts = Object.entries(messageIds).sort(([a], [b]) => Number(a) - Number(b))
-    const bytes = parts.reduce((sum, [, p]) => sum + (p.size ?? 0), 0)
-    if (parts.length > WHOLE_MAX_PARTS || bytes > WHOLE_MAX_BYTES) return null
-    if (parts.length === 1) return first
-
-    let assembled: YencFile | undefined
-    const conn = await NntpService.acquire()
-    let failed = false
-    try {
-      for (const [, part] of parts) {
-        const segment = decodeSegment(await NntpService.fetchBody(conn, part.id), { lenientSize: true })
-        assembled ??= YencFile.fromSegment(segment)
-        assembled.add(segment)
-      }
-    } catch (error) {
-      failed = true
-      throw error
-    } finally {
-      NntpService.release(conn, failed)
-    }
-    return assembled!.getBuffer()
-  }
-
-  /**
-   * Downloads a small archive that lists NZBs and imports each of them
-   * (CONCEPT.md D28): NZBs are posted inside RARs too, e.g. for obfuscated
-   * uploads by other posters.
-   */
-  private async importArchiveNzbs(file: any, first: Buffer): Promise<NzbImportResult[]> {
-    const archive = await this.readWhole(first, file.message_ids)
-    const info = archive && parseRarHeaders(archive)
-    if (!archive || !info) return []
-    const entries = info.entries.filter(isNzbEntry)
-    const results: NzbImportResult[] = []
-    for (const entry of entries) {
-      const xml = await extractEntry(archive, entry)
-      if (!xml) continue
-      const imported = await NzbImportService.import(file, xml.toString('utf8'), {
-        entryName: entry.name,
-        moveContainer: entries.length === 1,
-      })
-      if (imported) results.push(imported)
-    }
-    return results
-  }
-
-  async inspect(fileId: number): Promise<ContentResult | null> {
-    const file = await db.from('files').where('id', fileId).first()
-    const binary = await db.from('binaries').where('file_id', fileId).first()
-    if (!file || !this.isCandidate(file.filename)) return null
-    const first = file.message_ids?.['1']
-    if (!first?.id) return null
-
-    const conn = await NntpService.acquire()
-    let failed = false
-    let body: Buffer
-    try {
-      body = await NntpService.fetchBody(conn, first.id)
-    } catch (error) {
-      failed = true
-      throw error
-    } finally {
-      NntpService.release(conn, failed)
-    }
-    // Some tools declare a size that leaves out a header (nzbe); the CRC still has to match.
-    let segment
-    try {
-      segment = decodeSegment(body)
-    } catch (error) {
-      if (!(error instanceof YencError) || error instanceof YencCrcError) throw error
-      segment = decodeSegment(body, { lenientSize: true })
-    }
-    const size = segment.meta.begin.size
-    const contentType = sniffContentType(segment.data)
-    await db.from('files').where('id', fileId).update({ content_type: contentType })
-
-    // Fingerprint, so par2 sets (now or later) can name this file exactly.
-    if (binary) {
-      await db
-        .from('binaries')
-        .where('id', binary.id)
-        .update({
-          size: db.raw('coalesce(size, ?)', [size]),
-          md5_16k: db.raw('coalesce(md5_16k, ?)', [chunkMd5(segment.data, size) ?? null]),
-        })
-    }
-
-    const result: ContentResult = { named: 0, contentType }
-    if (contentType === 'nzbe' || contentType === 'rclone') {
-      // An encrypted NZB, or rclone crypt (names and data encrypted): nothing
-      // to read, and useless without the key.
-      const releaseId = file.release_id ?? binary?.release_id
-      if (releaseId) {
-        await db.from('releases').where('id', releaseId).update({ container: contentType, encrypted: true })
-      }
-      return result
-    }
-    if (contentType === 'nzb') {
-      const xml = await this.readWhole(segment.data, file.message_ids)
-      if (xml) result.nzb = (await NzbImportService.import(file, xml.toString('utf8'))) ?? undefined
-      return result
-    }
-    const set = await Par2Service.read(segment.data, file.message_ids)
-    if (set && binary) {
-      result.par2 = await Par2Service.attach(binary, set)
-    } else {
-      const info = parseRarHeaders(segment.data)
-      if (info) {
-        result.archive = await ArchiveService.record(file, binary, info)
-        if (hasNzbEntries(info)) result.archiveNzbs = await this.importArchiveNzbs(file, segment.data)
-      }
-    }
-    if (binary) result.named = await NameService.resolve([binary.id])
-    return result
-  }
-}
-
-/** NZBs that can be read from a single, unencrypted archive. */
-function hasNzbEntries(info: RarInfo): boolean {
-  return !info.volume && !info.headersEncrypted && info.entries.some(isNzbEntry)
-}
-
-function isNzbEntry(entry: RarEntry): boolean {
-  return /\.nzb$/i.test(entry.name) && !entry.encrypted && !entry.directory
-}
-
-/** Small posted files (NZBs) that are read completely. */
-const WHOLE_MAX_PARTS = 40
-const WHOLE_MAX_BYTES = 25 * 1024 * 1024
-
-export default new ContentService()

app/services/DeferredService.ts

@@ -1,264 +0,0 @@
-import { createHash } from 'node:crypto'
-import db from '@adonisjs/lucid/services/db'
-import indexerConfig from '#config/indexer'
-import NntpService from '#services/NntpService'
-import QueueService from '#services/QueueService'
-import { type OverviewHeader } from '#services/IngestService'
-import { decodeSegment, inferTotalParts, parseYencMeta } from '#lib/yenc/index'
-import { sniffContentType } from '#lib/content/index'
-import { parseRarHeaders } from '#lib/rar/index'
-
-/** Expanding re-reads the file's article range with XOVER; beyond this it is too costly. */
-const EXPAND_MAX_RANGE = 200_000
-/** A claimed row whose worker died is taken again after this long. */
-const RECLAIM_AFTER = '1 hour'
-
-export interface DeferredArticle {
-  articleNumber: number
-  header: OverviewHeader
-}
-
-/** File name from a deferred uploader's subject: `[i/M] "name" yEnc`, no part counter. */
-export function deferredName(subject: string | undefined): string | null {
-  return /"([^"]+)"/.exec(subject ?? '')?.[1]?.trim() || null
-}
-
-export function isDeferred(header: OverviewHeader): boolean {
-  const id = header['message-id'] ?? ''
-  return indexerConfig.deferBodyMessageIds.some((re) => re.test(id))
-}
-
-/** A recorded range never spans more articles than this (one XOVER to expand it). */
-const RANGE_MAX_SPAN = 50_000
-
-export function bodyJob(header: OverviewHeader, group: string, articleNumber: number | bigint) {
-  return {
-    name: 'process-body',
-    data: { header, group, articleNumber: Number(articleNumber) },
-    // By message-id: an article crossposted to several indexed groups is downloaded once.
-    opts: { jobId: `body-${createHash('sha1').update(header['message-id'] ?? '').digest('hex')}` },
-  }
-}
-
-/**
- * Deferred uploads (CONCEPT.md D28). Their articles skip the body queue and
- * are recorded here per file, from the headers alone. `worker:deferred`
- * later downloads one body per file for its size, part count and content
- * type, and sends small files (NZBs in RARs, par2 index files) to the body
- * queue completely.
- */
-class DeferredService {
-  /**
-   * Records one group's deferred articles: per file when the subject names
-   * the file, otherwise only as article ranges. Returns the number of files.
-   */
-  async record(groupName: string, articles: DeferredArticle[]): Promise<number> {
-    const named = articles.filter((a) => deferredName(a.header.subject) !== null)
-    await this.recordRanges(groupName, articles.filter((a) => deferredName(a.header.subject) === null))
-    return this.recordFiles(groupName, named)
-  }
-
-  private async recordRanges(groupName: string, articles: DeferredArticle[]) {
-    if (articles.length === 0) return
-    const sorted = [...articles].sort((a, b) => Number(a.articleNumber) - Number(b.articleNumber))
-    const now = new Date()
-    const rows: Record<string, unknown>[] = []
-    for (const article of sorted) {
-      const n = Number(article.articleNumber)
-      const bytes = Number.parseInt(article.header[':bytes'] ?? '0', 10) || 0
-      const last = rows.at(-1)
-      if (last && n - (last.first_article as number) <= RANGE_MAX_SPAN) {
-        last.last_article = n
-        ;(last.article_count as number)++
-        last.bytes = (last.bytes as number) + bytes
-      } else {
-        rows.push({ group_name: groupName, first_article: n, last_article: n, article_count: 1, bytes, created_at: now })
-      }
-    }
-    await db.table('deferred_ranges').multiInsert(rows)
-  }
-
-  private async recordFiles(groupName: string, articles: DeferredArticle[]): Promise<number> {
-    const files = new Map<string, {
-      name: string; subject: string; poster: string; first: number; firstId: string; last: number
-      count: number; bytes: number; postedAt: Date | null
-    }>()
-    for (const { articleNumber, header } of articles) {
-      const name = deferredName(header.subject)!
-      const n = Number(articleNumber)
-      const bytes = Number.parseInt(header[':bytes'] ?? '0', 10) || 0
-      const date = header.date ? new Date(header.date) : null
-      const postedAt = date && !Number.isNaN(date.getTime()) ? date : null
-      const file = files.get(name)
-      if (!file) {
-        files.set(name, {
-          name, subject: header.subject ?? '', poster: header.from ?? '', first: n,
-          firstId: (header['message-id'] ?? '').replace(/^<|>$/g, ''), last: n, count: 1, bytes, postedAt,
-        })
-        continue
-      }
-      if (n < file.first) {
-        file.first = n
-        file.firstId = (header['message-id'] ?? '').replace(/^<|>$/g, '')
-      }
-      file.last = Math.max(file.last, n)
-      file.count++
-      file.bytes += bytes
-      if (postedAt && (!file.postedAt || postedAt < file.postedAt)) file.postedAt = postedAt
-    }
-    if (files.size === 0) return 0
-
-    const now = new Date()
-    const rows = [...files.values()].map((f) => ({
-      file_key: createHash('sha1').update(`deferred\n${f.name}`).digest('hex'),
-      name: f.name,
-      subject: f.subject,
-      poster: f.poster,
-      group_name: groupName,
-      first_article: f.first,
-      first_message_id: f.firstId,
-      last_article: f.last,
-      article_count: f.count,
-      bytes: f.bytes,
-      posted_at: f.postedAt,
-      first_seen_at: now,
-      last_seen_at: now,
-    }))
-    // Counts and the article range only add up within the group the file was first seen in.
-    await db.rawQuery(
-      `INSERT INTO deferred_files AS d (${Object.keys(rows[0]).join(', ')})
-       SELECT * FROM jsonb_to_recordset(?::jsonb) AS r(
-         file_key varchar, name text, subject text, poster text, group_name text, first_article bigint,
-         first_message_id text, last_article bigint, article_count int, bytes bigint, posted_at timestamptz,
-         first_seen_at timestamptz, last_seen_at timestamptz)
-       ON CONFLICT (file_key) DO UPDATE SET
-         last_seen_at = EXCLUDED.last_seen_at,
-         first_message_id = CASE WHEN d.group_name = EXCLUDED.group_name AND EXCLUDED.first_article < d.first_article
-           THEN EXCLUDED.first_message_id ELSE d.first_message_id END,
-         first_article = CASE WHEN d.group_name = EXCLUDED.group_name THEN LEAST(d.first_article, EXCLUDED.first_article) ELSE d.first_article END,
-         last_article = CASE WHEN d.group_name = EXCLUDED.group_name THEN GREATEST(d.last_article, EXCLUDED.last_article) ELSE d.last_article END,
-         article_count = d.article_count + CASE WHEN d.group_name = EXCLUDED.group_name THEN EXCLUDED.article_count ELSE 0 END,
-         bytes = d.bytes + CASE WHEN d.group_name = EXCLUDED.group_name THEN EXCLUDED.bytes ELSE 0 END,
-         posted_at = LEAST(d.posted_at, EXCLUDED.posted_at)`,
-      [JSON.stringify(rows)]
-    )
-    return files.size
-  }
-
-  /** Takes up to `limit` files that stopped growing; they are marked `processing`. */
-  async claim(limit: number): Promise<any[]> {
-    const settled = new Date(Date.now() - indexerConfig.deferredSettleMs)
-    const { rows } = await db.rawQuery(
-      `UPDATE deferred_files SET status = 'processing', processed_at = now()
-       WHERE id IN (
-         SELECT id FROM deferred_files
-         WHERE (status = 'pending' AND last_seen_at < ?)
-            OR (status = 'processing' AND processed_at < now() - interval '${RECLAIM_AFTER}')
-         ORDER BY id LIMIT ? FOR UPDATE SKIP LOCKED)
-       RETURNING *`,
-      [settled, limit]
-    )
-    return rows
-  }
-
-  /**
-   * Queues the bodies of up to `limit` recorded ranges, oldest first: the
-   * articles are found again with XOVER. Returns ranges and articles queued.
-   */
-  async expandRanges(limit: number): Promise<{ ranges: number; articles: number }> {
-    const ranges = await db.from('deferred_ranges').whereNull('expanded_at').orderBy('id').limit(limit)
-    let articles = 0
-    for (const range of ranges) {
-      const conn = await NntpService.acquire()
-      let failed = false
-      let overviews: [number, OverviewHeader][]
-      try {
-        await conn.group(range.group_name)
-        overviews = (await conn.xover(String(range.first_article), String(range.last_article))).overviews
-      } catch (error) {
-        failed = true
-        throw error
-      } finally {
-        NntpService.release(conn, failed)
-      }
-      const jobs = overviews
-        .filter(([, header]) => header && isDeferred(header) && deferredName(header.subject) === null)
-        .map(([articleNumber, header]) => bodyJob(header, range.group_name, articleNumber))
-      for (let i = 0; i < jobs.length; i += 5000) await QueueService.bodyQueue.addBulk(jobs.slice(i, i + 5000))
-      await db.from('deferred_ranges').where('id', range.id).update({ expanded_at: new Date() })
-      articles += jobs.length
-    }
-    return { ranges: ranges.length, articles }
-  }
-
-  /** Reads one body of the file; expands small files into the body queue. Returns the new status. */
-  async process(file: any): Promise<string> {
-    const update: Record<string, unknown> = { processed_at: new Date(), error: null }
-    try {
-      const conn = await NntpService.acquire()
-      let failed = false
-      let body: Buffer
-      try {
-        body = await NntpService.fetchBody(conn, file.first_message_id)
-      } catch (error) {
-        failed = true
-        throw error
-      } finally {
-        NntpService.release(conn, failed)
-      }
-
-      const meta = parseYencMeta(body)
-      const totalParts = inferTotalParts(meta)
-      Object.assign(update, {
-        yenc_name: meta.begin.name,
-        size: meta.begin.size,
-        total_parts: totalParts,
-        sampled_part: meta.begin.part ?? 1,
-      })
-      if ((meta.begin.part ?? 1) === 1) {
-        const data = decodeSegment(body, { lenientSize: true, verifyCrc: false }).data
-        update.content_type = sniffContentType(data)
-        const rar = parseRarHeaders(data)
-        if (rar) update.encrypted = rar.headersEncrypted || rar.entries.some((e) => e.encrypted)
-      }
-
-      update.status = 'sampled'
-      if (totalParts <= indexerConfig.deferredExpandMaxParts && update.encrypted !== true) {
-        const queued = await this.expand(file)
-        if (queued > 0) update.status = 'expanded'
-      }
-    } catch (error: any) {
-      update.status = 'failed'
-      update.error = String(error.message).slice(0, 500)
-    }
-    await db.from('deferred_files').where('id', file.id).update(update)
-    return update.status as string
-  }
-
-  /** Finds the file's articles again (XOVER over its range) and queues their bodies. */
-  private async expand(file: any): Promise<number> {
-    const first = Number(file.first_article)
-    const last = Number(file.last_article)
-    if (last - first > EXPAND_MAX_RANGE) return 0
-
-    const conn = await NntpService.acquire()
-    let failed = false
-    let overviews: [number, OverviewHeader][]
-    try {
-      await conn.group(file.group_name)
-      overviews = (await conn.xover(String(first), String(last))).overviews
-    } catch (error) {
-      failed = true
-      throw error
-    } finally {
-      NntpService.release(conn, failed)
-    }
-    const jobs = overviews
-      .filter(([, header]) => header && deferredName(header.subject) === file.name)
-      .map(([articleNumber, header]) => bodyJob(header, file.group_name, articleNumber))
-    if (jobs.length > 0) await QueueService.bodyQueue.addBulk(jobs)
-    return jobs.length
-  }
-}
-
-export default new DeferredService()

app/services/FetchRangeService.ts

@@ -1,115 +0,0 @@
-import db from '@adonisjs/lucid/services/db'
-import logger from '@adonisjs/core/services/logger'
-import QueueService from '#services/QueueService'
-
-/** A failed range is retried after this long... */
-const REQUEUE_AFTER_MS = 5 * 60 * 1000
-/** ...at most this many times; after that it stays visible as a gap. */
-const MAX_REQUEUES = 5
-/** A queued range without a job in Redis (e.g. Redis was flushed) is re-queued after this long. */
-const LOST_AFTER_MS = 60 * 60 * 1000
-
-interface Group {
-  id: number
-  name: string
-}
-
-/**
- * Tracks the article ranges given to worker:fetch. A range only counts as
- * indexed once the worker reports it done, so failed or lost fetch jobs are
- * retried instead of silently skipped. See CONCEPT.md (D18).
- */
-class FetchRangeService {
-  async queue(group: Group, startId: bigint, endId: bigint) {
-    const now = new Date()
-    const [row] = await db
-      .table('fetch_ranges')
-      .insert({ group_id: group.id, start_id: startId.toString(), end_id: endId.toString(), created_at: now, updated_at: now })
-      .returning(['id', 'requeues'])
-    await this.addJob(group.name, Number(row.id), 0, startId, endId)
-  }
-
-  private async addJob(groupName: string, rangeId: number, requeues: number, startId: bigint, endId: bigint) {
-    await QueueService.nntpFetchQueue.add(
-      'fetch-headers',
-      { groupName, startId: startId.toString(), endId: endId.toString(), rangeId },
-      // The job id changes per attempt: BullMQ ignores an add() for an id it still knows.
-      { jobId: this.jobId(rangeId, requeues) }
-    )
-  }
-
-  private jobId(rangeId: number, requeues: number) {
-    return `range-${rangeId}-${requeues}`
-  }
-
-  async markDone(rangeId: number) {
-    await db.from('fetch_ranges').where('id', rangeId).update({ status: 'done', error: null, updated_at: new Date() })
-  }
-
-  async markFailed(rangeId: number, error: string) {
-    await db.from('fetch_ranges').where('id', rangeId).update({ status: 'failed', error, updated_at: new Date() })
-  }
-
-  /** Re-queues failed ranges, and queued ranges whose job no longer exists. */
-  async requeue(group: Group): Promise<number> {
-    const now = Date.now()
-    const candidates = await db
-      .from('fetch_ranges')
-      .where('group_id', group.id)
-      .where('requeues', '<', MAX_REQUEUES)
-      .where((q) =>
-        q
-          .where((f) => f.where('status', 'failed').where('updated_at', '<', new Date(now - REQUEUE_AFTER_MS)))
-          .orWhere((l) => l.where('status', 'queued').where('updated_at', '<', new Date(now - LOST_AFTER_MS)))
-      )
-      .orderBy('start_id')
-
-    let requeued = 0
-    for (const range of candidates) {
-      if (range.status === 'queued') {
-        const job = await QueueService.nntpFetchQueue.getJob(this.jobId(Number(range.id), range.requeues))
-        const state = job ? await job.getState() : 'missing'
-        if (state !== 'missing' && state !== 'failed' && state !== 'unknown') continue
-      }
-      await db
-        .from('fetch_ranges')
-        .where('id', range.id)
-        .update({ status: 'queued', requeues: range.requeues + 1, updated_at: new Date() })
-      await this.addJob(group.name, Number(range.id), range.requeues + 1, BigInt(range.start_id), BigInt(range.end_id))
-      requeued++
-      logger.warn(
-        `Re-queued ${group.name} ${range.start_id}-${range.end_id} (attempt ${range.requeues + 2}): ${range.error ?? 'job lost'}`
-      )
-    }
-    return requeued
-  }
-
-  /**
-   * Moves `groups.last_indexed_id` up to just below the first range that is
-   * not done yet, and forgets done ranges. Returns the new watermark.
-   */
-  async advanceWatermark(group: Group): Promise<bigint> {
-    const { rows } = await db.rawQuery(
-      `UPDATE groups g SET last_indexed_id = greatest(g.last_indexed_id, coalesce(
-         (SELECT min(start_id) - 1 FROM fetch_ranges WHERE group_id = g.id AND status <> 'done'),
-         g.last_queued_id))
-       WHERE g.id = ? RETURNING last_indexed_id`,
-      [group.id]
-    )
-    await db.from('fetch_ranges').where('group_id', group.id).where('status', 'done').delete()
-    return BigInt(rows[0].last_indexed_id)
-  }
-
-  /** Ranges that failed for good (for monitoring). */
-  async abandoned(group: Group) {
-    return db
-      .from('fetch_ranges')
-      .where('group_id', group.id)
-      .where('status', 'failed')
-      .where('requeues', '>=', MAX_REQUEUES)
-      .count('* as count')
-      .first()
-  }
-}
-
-export default new FetchRangeService()

app/services/IngestService.ts

@@ -1,471 +0,0 @@
-import db from '@adonisjs/lucid/services/db'
-import logger from '@adonisjs/core/services/logger'
-import type { QueryClientContract } from '@adonisjs/lucid/types/database'
-import type { Knex } from 'knex'
-import QueueService from '#services/QueueService'
-import NameService from '#services/NameService'
-import { parseXrefGroups } from '#lib/nntp/xref'
-import { fileKey, parseSubject, weakFileKey } from '#lib/subject/index'
-import { releaseIdentity, type ReleaseIdentity } from '#lib/release/index'
-import { inferTotalParts, type YencMeta } from '#lib/yenc/index'
-
-/** One segment, with the identity of the file it belongs to already resolved. */
-export interface IngestSegment {
-  fileKey: string
-  poster: string
-  /** Subject without segment counter. */
-  subject: string
-  name: string
-  totalParts: number
-  number: number
-  /** Without angle brackets. */
-  messageId: string
-  bytes: number
-  articleNumber?: bigint | number
-  postedAt?: Date
-  groups: string[]
-  /**
-   * The identity comes from the yEnc name only (obfuscated posts). Such names
-   * can be shared by several equal-sized files of one post, so a second
-   * article for the same part number marks the binary as conflicted.
-   */
-  weakIdentity?: boolean
-  /** Release (download) this file belongs to, when it can be determined. */
-  release?: ReleaseIdentity | null
-  /** Exact file size from =ybegin (body-derived segments only). */
-  size?: number
-  /** yEnc CRCs (body-derived segments only), to verify split volumes. */
-  pcrc32?: string
-  partSize?: number
-  fileCrc32?: string
-  /** Part 1 only: MD5 of the first 16 KiB of the file, as par2 records it. */
-  md5_16k?: string
-}
-
-export interface CompletedBinary {
-  id: number
-  fileKey: string
-}
-
-const BINARY_CHUNK = 1000
-/** last_seen_at is only rewritten when it is older than this (it drives GC at 48 h). */
-const TOUCH_MS = 10 * 60 * 1000
-
-type Merge = 'least' | 'coalesce' | 'touch'
-
-/**
- * Inserts rows keyed by `key` and returns their ids. Existing rows are not
- * locked unless something in them changes (`least`: an earlier value,
- * `coalesce`: a value where there was none, `touch`: a timestamp more than
- * TOUCH_MS newer). Parallel fetch jobs see mostly the same rows, and
- * `INSERT … ON CONFLICT DO UPDATE` made them wait on each other for every one.
- */
-async function upsertKeyed(
-  knex: Knex,
-  table: 'binaries' | 'releases',
-  key: string,
-  rows: Record<string, unknown>[],
-  merge: Record<string, [Merge, string]>
-): Promise<Map<string, number>> {
-  const ids = new Map<string, number>()
-  if (rows.length === 0) return ids
-  const inserted: Record<string, any>[] = await knex(table).insert(rows).onConflict(key).ignore().returning(['id', key])
-  for (const row of inserted) ids.set(row[key], Number(row.id))
-
-  const incoming = new Map(rows.filter((r) => !ids.has(r[key] as string)).map((r) => [r[key] as string, r]))
-  if (incoming.size === 0) return ids
-  const columns = Object.keys(merge)
-  const existing: Record<string, any>[] = await knex(table).whereIn(key, [...incoming.keys()]).select('id', key, ...columns)
-
-  const changes: Record<string, unknown>[] = []
-  for (const row of existing) {
-    ids.set(row[key], Number(row.id))
-    const next = incoming.get(row[key])!
-    const changed = columns.some((column) => {
-      const [kind] = merge[column]
-      const current = row[column]
-      const value = next[column]
-      if (value === null || value === undefined) return false
-      if (kind === 'coalesce') return current === null
-      if (kind === 'least') return current === null || value < current
-      return current === null || +(value as Date) - +current > TOUCH_MS
-    })
-    if (changed) changes.push({ id: Number(row.id), ...Object.fromEntries(columns.map((c) => [c, next[c] ?? null])) })
-  }
-  if (changes.length === 0) return ids
-
-  const set = columns
-    .map((c) => {
-      const [kind] = merge[c]
-      const fn = kind === 'least' ? 'least' : kind === 'touch' ? 'greatest' : 'coalesce'
-      return `${c} = ${fn}(t.${c}, v.${c})`
-    })
-    .join(', ')
-  const types = columns.map((c) => `${c} ${merge[c][1]}`).join(', ')
-  changes.sort((a, b) => (a.id as number) - (b.id as number))
-  for (let attempt = 1; ; attempt++) {
-    try {
-      await knex.raw(
-        `UPDATE ${table} t SET ${set}
-         FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, ${types})
-         WHERE t.id = v.id`,
-        [JSON.stringify(changes)]
-      )
-      return ids
-    } catch (error: any) {
-      // Rare now, but two jobs can still update overlapping rows in a different order.
-      if (error.code !== '40P01' || attempt >= 3) throw error
-    }
-  }
-}
-/** Size-only matching below this is too ambiguous (nfo, sfv, small volumes). */
-export const PAR2_SIZE_ONLY_MIN = 1024 * 1024
-/** A par2 set and its files are posted together; don't match across days. */
-export const PAR2_MATCH_WINDOW_SECONDS = 12 * 60 * 60
-const PART_CHUNK = 2000
-
-function chunks<T>(items: T[], size: number): T[][] {
-  const result: T[][] = []
-  for (let i = 0; i < items.length; i += size) result.push(items.slice(i, i + size))
-  return result
-}
-
-function validDate(date?: Date) {
-  return date && !Number.isNaN(date.getTime()) ? date : null
-}
-
-/** An XOVER overview entry as returned by nntp-js. */
-export type OverviewHeader = Record<string, string>
-
-function common(header: OverviewHeader, groupName: string, articleNumber?: bigint | number) {
-  const xrefGroups = parseXrefGroups(header.xref)
-  return {
-    poster: header.from ?? '',
-    messageId: (header['message-id'] ?? '').replace(/^<|>$/g, ''),
-    bytes: Number.parseInt(header[':bytes'] ?? '0', 10) || 0,
-    articleNumber,
-    postedAt: header.date ? new Date(header.date) : undefined,
-    groups: xrefGroups.length > 0 ? xrefGroups : [groupName],
-  }
-}
-
-/** Builds a segment from the subject line; null when the subject is obfuscated. */
-export function segmentFromHeader(
-  header: OverviewHeader,
-  groupName: string,
-  articleNumber?: bigint | number
-): IngestSegment | null {
-  const parsed = header.subject ? parseSubject(header.subject) : null
-  if (!parsed || !header['message-id']) return null
-  const base = common(header, groupName, articleNumber)
-  return {
-    ...base,
-    fileKey: fileKey(base.poster, parsed.base, parsed.totalSegments),
-    subject: parsed.base,
-    name: parsed.name,
-    totalParts: parsed.totalSegments,
-    number: parsed.segment,
-    release: releaseIdentity({ poster: base.poster, subject: parsed.base, name: parsed.name, weak: false }),
-  }
-}
-
-/**
- * Builds a segment for an obfuscated post from the yEnc header in its body.
- * Subject and poster can be random per article, so the identity comes from
- * the yEnc header only (see weakFileKey).
- */
-export function segmentFromYenc(
-  header: OverviewHeader,
-  groupName: string,
-  meta: Pick<YencMeta, 'begin' | 'part'> & Partial<Pick<YencMeta, 'end'>>,
-  articleNumber?: bigint | number
-): IngestSegment | null {
-  const base = common(header, groupName, articleNumber)
-  const totalParts = inferTotalParts(meta)
-  const number = meta.begin.part ?? 1
-  if (!base.messageId || number < 1 || number > totalParts) return null
-  const subject = `"${meta.begin.name}" yEnc`
-  return {
-    ...base,
-    fileKey: weakFileKey(meta.begin.name, meta.begin.size, totalParts),
-    subject,
-    name: meta.begin.name,
-    totalParts,
-    number,
-    weakIdentity: true,
-    release: releaseIdentity({ poster: base.poster, subject, name: meta.begin.name, weak: true }),
-    size: meta.begin.size,
-    // Single-part posts only carry crc32, which is then also the part's CRC.
-    pcrc32: meta.end?.pcrc32 ?? (meta.part ? undefined : meta.end?.crc32),
-    partSize: meta.part ? meta.part.end - meta.part.begin + 1 : meta.begin.size,
-    fileCrc32: meta.end?.crc32,
-  }
-}
-
-/**
- * Stores segments in `binaries`/`parts` and queues every binary that became
- * complete exactly once. Idempotent: ingesting the same segments again is a
- * no-op. See CONCEPT.md (D12).
- */
-class IngestService {
-  async ingest(segments: IngestSegment[]): Promise<CompletedBinary[]> {
-    if (segments.length === 0) return []
-
-    // Short statements, no enclosing transaction: every step is idempotent, and
-    // row locks held for a whole batch made parallel fetch jobs wait on each other.
-    const binaryIds = await (async () => {
-      const knex = db.connection().getWriteClient()
-      const now = new Date()
-      const releaseIds = await this.upsertReleases(knex, segments, now)
-
-      // One row per file_key (ON CONFLICT cannot touch a row twice per statement),
-      // sorted so concurrent workers lock rows in the same order and cannot deadlock.
-      const binaries = new Map<string, Record<string, unknown>>()
-      for (const segment of segments) {
-        const postedAt = validDate(segment.postedAt)
-        const existing = binaries.get(segment.fileKey)
-        if (existing) {
-          if (postedAt && (!existing.posted_at || postedAt < (existing.posted_at as Date))) {
-            existing.posted_at = postedAt
-          }
-          continue
-        }
-        binaries.set(segment.fileKey, {
-          file_key: segment.fileKey,
-          poster: segment.poster,
-          subject: segment.subject,
-          name: segment.name,
-          total_parts: segment.totalParts,
-          groups: JSON.stringify(segment.groups),
-          posted_at: postedAt,
-          first_seen_at: now,
-          last_seen_at: now,
-          release_id: segment.release ? releaseIds.get(segment.release.key) : null,
-          size: segment.size ?? null,
-          md5_16k: segment.md5_16k ?? null,
-        })
-      }
-      const binaryRows = [...binaries.values()].sort((a, b) =>
-        (a.file_key as string).localeCompare(b.file_key as string)
-      )
-
-      const idsByKey = new Map<string, number>()
-      for (const chunk of chunks(binaryRows, BINARY_CHUNK)) {
-        const ids = await upsertKeyed(knex, 'binaries', 'file_key', chunk, {
-          last_seen_at: ['touch', 'timestamptz'],
-          posted_at: ['least', 'timestamptz'],
-          release_id: ['coalesce', 'bigint'],
-          md5_16k: ['coalesce', 'varchar'],
-        })
-        for (const [key, id] of ids) idsByKey.set(key, id)
-      }
-
-      const partRows = segments
-        .map((segment) => ({
-          binary_id: idsByKey.get(segment.fileKey)!,
-          number: segment.number,
-          message_id: segment.messageId,
-          bytes: segment.bytes,
-          article_number: segment.articleNumber?.toString() ?? null,
-          pcrc32: segment.pcrc32 ?? null,
-          part_size: segment.partSize ?? null,
-          file_crc32: segment.fileCrc32 ?? null,
-          md5_16k: segment.md5_16k ?? null,
-        }))
-      const weakRows = partRows.filter((_, i) => segments[i].weakIdentity)
-
-      partRows.sort((a, b) => a.binary_id - b.binary_id || a.number - b.number)
-      for (const chunk of chunks(partRows, PART_CHUNK)) {
-        await knex('parts').insert(chunk).onConflict(['binary_id', 'number']).ignore()
-      }
-
-      if (weakRows.length > 0) await this.flagConflicts(db.connection(), weakRows)
-
-      return [...idsByKey.values()]
-    })()
-
-    // par2 naming needs a size, which only body-derived segments have; for
-    // header batches these queries were most of Postgres' work.
-    const sized = [...new Set(segments.filter((s) => s.size != null).map((s) => s.fileKey))]
-    if (sized.length > 0) {
-      const sizedIds = (await db.from('binaries').whereIn('file_key', sized).select('id')).map((r) => Number(r.id))
-      await this.matchPar2Releases(sizedIds)
-      await NameService.resolve(sizedIds)
-    }
-
-    // After commit, so the last writer of a binary always sees all of its parts.
-    const completed = await this.claimCompleted(binaryIds)
-    await this.enqueue(completed)
-    return completed
-  }
-
-  /**
-   * Creates or touches the releases referenced by `segments` and returns
-   * their ids by release key. Also used by `release:rebuild`.
-   */
-  async upsertReleases(
-    knex: Knex,
-    segments: Pick<IngestSegment, 'release' | 'poster' | 'groups' | 'postedAt' | 'weakIdentity'>[],
-    now = new Date()
-  ): Promise<Map<string, number>> {
-    const releases = new Map<string, Record<string, unknown>>()
-    for (const segment of segments) {
-      const release = segment.release
-      if (!release) continue
-      const postedAt = validDate(segment.postedAt)
-      const existing = releases.get(release.key)
-      if (existing) {
-        if (postedAt && (!existing.posted_at || postedAt < (existing.posted_at as Date))) {
-          existing.posted_at = postedAt
-        }
-        // Not every file carries the [i/M] counter (e.g. a par2 index): take it from any that does.
-        if (existing.file_total === null && release.fileTotal) existing.file_total = release.fileTotal
-        continue
-      }
-      releases.set(release.key, {
-        release_key: release.key,
-        name: release.name,
-        // Obfuscating posters randomise the poster, so it says nothing about the release.
-        poster: segment.weakIdentity ? null : segment.poster,
-        file_total: release.fileTotal ?? null,
-        source: release.source,
-        groups: JSON.stringify(segment.groups),
-        posted_at: postedAt,
-        first_seen_at: now,
-        last_seen_at: now,
-      })
-    }
-
-    const ids = new Map<string, number>()
-    const rows = [...releases.values()].sort((a, b) =>
-      (a.release_key as string).localeCompare(b.release_key as string)
-    )
-    for (const chunk of chunks(rows, BINARY_CHUNK)) {
-      const result = await upsertKeyed(knex, 'releases', 'release_key', chunk, {
-        last_seen_at: ['touch', 'timestamptz'],
-        posted_at: ['least', 'timestamptz'],
-        file_total: ['coalesce', 'integer'],
-        source: ['coalesce', 'varchar'],
-      })
-      for (const [key, id] of result) ids.set(key, id)
-    }
-    return ids
-  }
-
-  /**
-   * Assigns body-derived binaries without a release and without a
-   * fingerprint to a known par2 set by size: only when that size is listed
-   * once in the set, the file is large, and exactly one set nearby lists it.
-   * Binaries with a fingerprint are matched exactly by NameService instead.
-   */
-  async matchPar2Releases(binaryIds: number[]) {
-    if (binaryIds.length === 0) return
-    await db.rawQuery(
-      `UPDATE binaries b SET release_id = m.release_id
-       FROM (
-         SELECT b2.id, min(rf.release_id) AS release_id
-         FROM binaries b2
-         JOIN release_files rf ON rf.size = b2.size
-         JOIN releases r ON r.id = rf.release_id
-         WHERE b2.id = ANY(?) AND b2.release_id IS NULL AND b2.md5_16k IS NULL AND b2.size >= ?
-           AND abs(extract(epoch FROM (r.posted_at - b2.posted_at))) < ?
-           AND (SELECT count(*) FROM release_files x WHERE x.release_id = rf.release_id AND x.size = rf.size) = 1
-         GROUP BY b2.id
-         HAVING count(DISTINCT rf.release_id) = 1
-       ) m
-       WHERE b.id = m.id`,
-      [binaryIds, PAR2_SIZE_ONLY_MIN, PAR2_MATCH_WINDOW_SECONDS]
-    )
-  }
-
-  /**
-   * Flags binaries where a stored part has a different message-id and a
-   * different part CRC than the incoming one: two files were merged under
-   * one identity. Identical reposts (same CRC) are ignored.
-   */
-  private async flagConflicts(
-    trx: QueryClientContract,
-    rows: {
-      binary_id: number
-      number: number
-      message_id: string
-      bytes: number
-      article_number: string | null
-      pcrc32: string | null
-      part_size: number | null
-      file_crc32: string | null
-      md5_16k: string | null
-    }[]
-  ) {
-    const bindings = [
-      rows.map((r) => r.binary_id),
-      rows.map((r) => r.number),
-      rows.map((r) => r.message_id),
-      rows.map((r) => r.bytes),
-      rows.map((r) => r.article_number),
-      rows.map((r) => r.pcrc32),
-      rows.map((r) => r.part_size),
-      rows.map((r) => r.file_crc32),
-      rows.map((r) => r.md5_16k),
-    ]
-    // Keep the extra articles: SplitService separates the volumes from them.
-    const conflicts = await trx.rawQuery(
-      `INSERT INTO part_conflicts (binary_id, number, message_id, bytes, article_number, pcrc32, part_size, file_crc32, md5_16k)
-       SELECT v.binary_id, v.number, v.message_id, v.bytes, v.article_number, v.pcrc32, v.part_size, v.file_crc32, v.md5_16k
-       FROM unnest(?::bigint[], ?::int[], ?::text[], ?::int[], ?::bigint[], ?::text[], ?::int[], ?::text[], ?::text[])
-         AS v(binary_id, number, message_id, bytes, article_number, pcrc32, part_size, file_crc32, md5_16k)
-       JOIN parts p ON p.binary_id = v.binary_id AND p.number = v.number
-       WHERE p.message_id <> v.message_id
-         -- Same part CRC: the same data posted again (a repost), not another file.
-         AND (p.pcrc32 IS NULL OR v.pcrc32 IS NULL OR p.pcrc32 <> v.pcrc32)
-       ON CONFLICT DO NOTHING
-       RETURNING binary_id`,
-      bindings as any
-    )
-    const binaryIds: number[] = [
-      ...new Set<number>(conflicts.rows.map((r: { binary_id: string }) => Number(r.binary_id))),
-    ]
-    if (binaryIds.length === 0) return
-    const result = await trx.rawQuery(
-      `UPDATE binaries SET conflicted_at = now()
-       WHERE conflicted_at IS NULL AND id = ANY(?) RETURNING name`,
-      [binaryIds]
-    )
-    for (const row of result.rows) {
-      logger.warn(`Binary "${row.name}" has conflicting parts; it will be split into volumes.`)
-    }
-  }
-
-  /**
-   * Marks complete binaries among `binaryIds` as claimed. The single UPDATE
-   * locks each row, so concurrent callers never claim the same binary twice.
-   */
-  async claimCompleted(binaryIds: number[]): Promise<CompletedBinary[]> {
-    if (binaryIds.length === 0) return []
-    const result = await db.rawQuery(
-      `UPDATE binaries b SET completed_at = now()
-       WHERE b.id = ANY(?) AND b.completed_at IS NULL AND b.conflicted_at IS NULL
-         AND (SELECT count(*) FROM parts p WHERE p.binary_id = b.id) >= b.total_parts
-       RETURNING b.id, b.file_key`,
-      [binaryIds]
-    )
-    return result.rows.map((row: { id: string; file_key: string }) => ({
-      id: Number(row.id),
-      fileKey: row.file_key,
-    }))
-  }
-
-  async enqueue(completed: CompletedBinary[]) {
-    if (completed.length === 0) return
-    // jobId = file_key: BullMQ ignores an add() for an id that already exists.
-    await QueueService.fileQueue.addBulk(
-      completed.map(({ id, fileKey }) => ({
-        name: 'process-file',
-        data: { binaryId: id },
-        opts: { jobId: fileKey },
-      }))
-    )
-  }
-}
-
-export default new IngestService()

app/services/NameService.ts

@@ -1,128 +0,0 @@
-import db from '@adonisjs/lucid/services/db'
-import { looksRandom } from '#services/ArchiveService'
-
-/**
- * A release that is nothing more than one randomly named file (its own
- * filename stem). Such a file moves into the par2 set that names it exactly;
- * real groupings (par2, RAR, NZB, spot, subject, post id) are left alone.
- */
-export const WEAK_RELEASE_SQL = (column: string) =>
-  `EXISTS (SELECT 1 FROM releases wr WHERE wr.id = ${column} AND wr.source IN ('stem', 'weak-stem')
-     AND wr.file_count <= 1 AND (wr.name !~ '[._ -]' OR NOT is_readable_name(wr.name)))`
-
-/**
- * Whether `candidate` (e.g. from a par2 listing) should replace `current`:
- * the current name is random, or a single token without separators
- * (mOzyyxCyYUj, F8lv3w5agMDI3; too short for looksRandom), while the
- * candidate reads like a release name.
- */
-export function isBetterName(current: string, candidate: string): boolean {
-  if (looksRandom(candidate) || !/[._ -]/.test(candidate)) return false
-  return looksRandom(current) || !/[._ -]/.test(current)
-}
-
-/** Smaller files (nfo, sfv) collide too easily on size alone. */
-const MIN_SIZE_ONLY = 64 * 1024
-
-/**
- * Gives binaries (and their files) the real filename that a par2 set lists
- * for them. See CONCEPT.md (D21).
- * - Exact: size and MD5 of the first 16 KiB both match a par2 entry. This
- *   also puts the binary in that par2 set's release.
- * - Size only, when the fingerprint is unknown: within the binary's own
- *   release, only when exactly one listed file has that size.
- */
-class NameService {
-  async resolve(binaryIds: number[]): Promise<number> {
-    if (binaryIds.length === 0) return 0
-    const exact = await db.rawQuery(
-      `UPDATE binaries b SET real_name = m.name,
-         release_id = CASE WHEN b.release_id IS NULL OR ${WEAK_RELEASE_SQL('b.release_id')} THEN m.release_id ELSE b.release_id END
-       FROM (
-         SELECT DISTINCT ON (b2.id) b2.id, rf.name, rf.release_id
-         FROM binaries b2
-         JOIN release_files rf ON rf.size = b2.size AND rf.md5_16k = b2.md5_16k
-         WHERE b2.id = ANY(?)
-         ORDER BY b2.id, rf.release_id DESC
-       ) m
-       WHERE b.id = m.id AND (b.real_name IS DISTINCT FROM m.name OR b.release_id IS NULL
-         OR (b.release_id <> m.release_id AND ${WEAK_RELEASE_SQL('b.release_id')}))`,
-      [binaryIds]
-    )
-    const bySize = await db.rawQuery(
-      `UPDATE binaries b SET real_name = m.name
-       FROM (
-         SELECT b2.id, min(rf.name) AS name
-         FROM binaries b2
-         JOIN release_files rf ON rf.release_id = b2.release_id AND rf.size = b2.size
-         WHERE b2.id = ANY(?) AND b2.real_name IS NULL AND b2.md5_16k IS NULL AND b2.size >= ?
-         GROUP BY b2.id
-         HAVING count(DISTINCT rf.name) = 1
-       ) m
-       WHERE b.id = m.id`,
-      [binaryIds, MIN_SIZE_ONLY]
-    )
-    await db.rawQuery(
-      `UPDATE files f SET real_name = b.real_name,
-         release_id = CASE WHEN f.release_id IS NULL OR ${WEAK_RELEASE_SQL('f.release_id')} THEN coalesce(b.release_id, f.release_id) ELSE f.release_id END
-       FROM binaries b
-       WHERE b.id = ANY(?) AND b.file_id = f.id
-         AND (f.real_name IS DISTINCT FROM b.real_name OR (f.release_id IS NULL AND b.release_id IS NOT NULL)
-           OR (f.release_id <> b.release_id AND ${WEAK_RELEASE_SQL('f.release_id')}))`,
-      [binaryIds]
-    )
-    return (exact.rowCount ?? 0) + (bySize.rowCount ?? 0)
-  }
-
-  /**
-   * When a par2 set arrives: the binaries it can name, i.e. those matching
-   * one of its entries exactly (size + first-16-KiB MD5) and those in its
-   * release. Selecting every binary of the listed sizes instead picked up
-   * hundreds of thousands of equal-sized volumes per set (LOG.md 2026-09-28 (18)).
-   */
-  async resolveForRelease(releaseId: number): Promise<number> {
-    const { rows } = await db.rawQuery(
-      `SELECT b.id FROM release_files rf
-       JOIN binaries b ON b.md5_16k = rf.md5_16k AND b.size = rf.size
-       WHERE rf.release_id = ?
-       UNION
-       SELECT id FROM binaries WHERE release_id = ? AND size IS NOT NULL`,
-      [releaseId, releaseId]
-    )
-    return this.resolve(rows.map((r: { id: string }) => Number(r.id)))
-  }
-}
-
-/**
- * Repair for data from before files moved into the par2 set that names them
- * (used by `release:rename`): regroups exact matches out of random-named
- * single-file releases, then renames par2 releases that still carry a
- * random name after their listing. No downloads.
- */
-export async function regroupAndRenameFromPar2(): Promise<{ regrouped: number; renamed: number }> {
-  const { rows } = await db.rawQuery(
-    `SELECT DISTINCT b.id FROM binaries b
-     JOIN release_files rf ON rf.md5_16k = b.md5_16k AND rf.size = b.size
-     WHERE b.release_id IS DISTINCT FROM rf.release_id
-       AND (b.release_id IS NULL OR ${WEAK_RELEASE_SQL('b.release_id')})`
-  )
-  const ids = rows.map((r: { id: string }) => Number(r.id))
-  for (let i = 0; i < ids.length; i += 1000) await nameService.resolve(ids.slice(i, i + 1000))
-
-  const { par2ReleaseName } = await import('#lib/release/index')
-  const listings = await db.rawQuery(
-    `SELECT r.id, r.name, array_agg(rf.name) AS names FROM releases r
-     JOIN release_files rf ON rf.release_id = r.id GROUP BY r.id, r.name`
-  )
-  let renamed = 0
-  for (const release of listings.rows) {
-    const better = par2ReleaseName(release.names)
-    if (!better || better === release.name || !isBetterName(release.name, better)) continue
-    await db.from('releases').where('id', release.id).update({ name: better })
-    renamed++
-  }
-  return { regrouped: ids.length, renamed }
-}
-
-const nameService = new NameService()
-export default nameService

app/services/NntpService.ts

@@ -1,119 +0,0 @@
-import nntpConfig from '#config/nntp'
-import { NNTP } from "nntp-js";
-import { parseGroupResponse, type GroupInfo } from '#lib/nntp/group'
-
-class NntpService {
-  private readonly poolSize: number;
-  private allConnections: Set<any>;
-  private idleConnections: any[];
-  private waiters: ((conn: any) => void)[];
-  private createdCount: number;
-
-  constructor(poolSize = 10) {
-    this.poolSize = poolSize;
-    this.allConnections = new Set();
-    this.idleConnections = [];
-    this.waiters = [];
-    this.createdCount = 0;
-  }
-
-  private async _createConnection() {
-    const conn = new NNTP(nntpConfig.host, nntpConfig.port);
-    try {
-      await conn.connect();
-      if (nntpConfig.user) {
-        await conn.login(nntpConfig.user, nntpConfig.password?.release());
-      }
-    } catch (error) {
-      // Free the slot, otherwise a failed connect shrinks the pool for good.
-      this.createdCount--;
-      this._destroy(conn);
-      throw error;
-    }
-    this.allConnections.add(conn);
-    return conn;
-  }
-
-  public async acquire() {
-    if (this.idleConnections.length > 0) {
-      return this.idleConnections.pop();
-    }
-
-    if (this.createdCount < this.poolSize) {
-      this.createdCount++;
-      console.log(`Creating new connection (${this.createdCount}/${this.poolSize}).`);
-      return this._createConnection();
-    }
-
-    return new Promise(resolve => this.waiters.push(resolve));
-  }
-
-  /**
-   * Returns a connection to the pool. Pass `failed` when a command on it
-   * threw: the connection may be dead (e.g. closed by the server while idle),
-   * so it is discarded instead of being handed to the next job.
-   */
-  public release(conn: any, failed = false) {
-    if (failed) {
-      this.discard(conn);
-      return;
-    }
-    const resolve = this.waiters.shift();
-    if (resolve) {
-      resolve(conn);
-    } else {
-      this.idleConnections.push(conn);
-    }
-  }
-
-  /** Closes a broken connection and frees its slot for a new one. */
-  public discard(conn: any) {
-    this.allConnections.delete(conn);
-    this.createdCount--;
-    this._destroy(conn);
-
-    // Someone is waiting for a slot: open a fresh connection for them.
-    const resolve = this.waiters.shift();
-    if (resolve) {
-      this.createdCount++;
-      this._createConnection().then(resolve, () => this.waiters.unshift(resolve));
-    }
-  }
-
-  private _destroy(conn: any) {
-    try {
-      conn.sock?.destroy();
-    } catch {
-      // already closed
-    }
-  }
-
-  /** Selects a group and returns its correctly parsed article range. */
-  public async selectGroup(conn: any, name: string): Promise<GroupInfo> {
-    const { resp } = await conn.group(name)
-    return parseGroupResponse(resp)
-  }
-
-  /** Fetches a raw (still dot-stuffed) article body as a Buffer. */
-  public async fetchBody(conn: any, messageId: string): Promise<Buffer> {
-    const id = messageId.startsWith('<') ? messageId : `<${messageId}>`
-    const { data } = await conn.body(id)
-    return Buffer.from(data.buffer, data.byteOffset, data.byteLength)
-  }
-
-  public async shutdown() {
-    console.log('Shutting down all connections in the pool.');
-    // A connection the server already closed may never answer QUIT.
-    const quit = (conn: any) =>
-      Promise.race([conn.quit(), new Promise((resolve) => setTimeout(resolve, 2000).unref())])
-        .catch(() => {})
-        .finally(() => this._destroy(conn));
-    await Promise.all([...this.allConnections].map(quit));
-    this.allConnections.clear();
-    this.idleConnections.length = 0;
-    this.waiters.length = 0;
-    this.createdCount = 0;
-  }
-}
-
-export default new NntpService(nntpConfig.poolSize);

app/services/NzbImportService.ts

@@ -1,168 +0,0 @@
-import { createHash } from 'node:crypto'
-import db from '@adonisjs/lucid/services/db'
-import logger from '@adonisjs/core/services/logger'
-import IngestService from '#services/IngestService'
-import { type ReleaseIdentity } from '#lib/release/index'
-import { parseNzb, type ParsedNzb } from '#lib/nzb/index'
-import { parseSubject } from '#lib/subject/index'
-import { fileStem } from '#lib/release/index'
-import { looksRandom } from '#services/ArchiveService'
-
-export interface NzbImportResult {
-  releaseId: number
-  releaseName: string
-  listed: number
-  known: number
-  added: number
-  password: boolean
-}
-
-/** Filename of an NZB entry, from its subject (tolerating odd quoting). */
-export function nzbFileName(subject: string): string {
-  const parsed = parseSubject(subject)
-  const name = parsed?.name ?? /"([^"]+)"/.exec(subject)?.[1] ?? subject
-  return name.trim().replace(/^"+|"+$/g, '').trim() || subject.trim()
-}
-
-export interface NzbImportOptions {
-  /**
-   * For NZBs inside an archive: the entry's name. It keeps several NZBs of
-   * one archive apart and names their release when the NZB has no name.
-   */
-  entryName?: string
-  /** Move the posted file into the release the NZB describes (default true). */
-  moveContainer?: boolean
-  /** A name given with the NZB (e.g. a Spotnet title); wins over the NZB's own. */
-  name?: string
-  /** Release source for new releases (default `nzb`). */
-  source?: ReleaseIdentity['source']
-}
-
-/**
- * Reads an NZB that was posted to Usenet into the index (CONCEPT.md D26).
- * Files it lists that are already known (by first segment) keep their
- * release, which the NZB then joins; the others are added as files, so posts
- * spread over groups we do not index still become complete.
- */
-class NzbImportService {
-  async import(nzbFile: any, xml: string, options: NzbImportOptions = {}): Promise<NzbImportResult | null> {
-    const nzb = parseNzb(xml)
-    if (!nzb || nzb.files.length === 0) return null
-
-    const entries = nzb.files.map((f) => ({
-      file: f,
-      name: nzbFileName(f.subject),
-      firstId: f.segments[0].id,
-    }))
-
-    return db.transaction(async (trx) => {
-      const known = await trx
-        .from('files')
-        .whereRaw(`(message_ids->'1'->>'id') = ANY(?)`, [entries.map((e) => e.firstId)])
-        .select('id', 'release_id', `message_ids`)
-      const knownIds = new Set(known.map((f) => f.message_ids?.['1']?.id))
-
-      // Join the release the listed files are already in, else the posted NZB's own, else a new one.
-      const counts = new Map<number, number>()
-      for (const f of known) if (f.release_id) counts.set(Number(f.release_id), (counts.get(Number(f.release_id)) ?? 0) + 1)
-      let releaseId: number | undefined = [...counts].sort((a, b) => b[1] - a[1])[0]?.[0]
-      const title = options.name?.trim() || this.title(nzb, nzbFile, entries.map((e) => e.name), options.entryName)
-      if (!releaseId) {
-        const origin = `nzb\n${nzbFile.message_ids?.['1']?.id ?? nzbFile.id}`
-        const key = createHash('sha1').update(options.entryName ? `${origin}\n${options.entryName}` : origin).digest('hex')
-        const ids = await IngestService.upsertReleases(trx.knexClient, [
-          {
-            release: { key, name: title, source: options.source ?? 'nzb' },
-            poster: nzbFile.poster,
-            groups: nzbFile.groups ?? [],
-            postedAt: nzbFile.date ? new Date(Number(nzbFile.date)) : undefined,
-          },
-        ])
-        releaseId = ids.get(key)!
-      }
-
-      const now = new Date()
-      const missing = entries.filter((e) => !knownIds.has(e.firstId))
-      for (let i = 0; i < missing.length; i += 500) {
-        await trx
-          .knexClient('files')
-          .insert(
-            missing.slice(i, i + 500).map(({ file, name, firstId }) => ({
-              file_key: createHash('sha1').update(`nzb-file\n${firstId}`).digest('hex'),
-              filename: name,
-              poster: file.poster || nzbFile.poster,
-              date: (file.date || Math.floor(Number(nzbFile.date) / 1000)) * 1000,
-              parts: file.segments.length,
-              size: file.segments.reduce((sum, s) => sum + s.bytes, 0),
-              message_ids: JSON.stringify(Object.fromEntries(file.segments.map((s) => [s.number, { id: s.id, size: s.bytes }]))),
-              groups: JSON.stringify(file.groups),
-              release_id: releaseId,
-              created_at: now,
-              updated_at: now,
-            }))
-          )
-          .onConflict('file_key')
-          .ignore()
-      }
-      // The poster's NZB says which files form the post: known files join this
-      // release, also when name rules had split them over several.
-      const previous = [...new Set(known.map((f) => f.release_id).filter((id) => id && Number(id) !== releaseId))]
-      await trx
-        .from('files')
-        .whereIn('id', known.filter((f) => Number(f.release_id) !== releaseId).map((f) => f.id))
-        .update({ release_id: releaseId })
-      if (previous.length) {
-        await trx.rawQuery(
-          `UPDATE binaries b SET release_id = ? FROM files f WHERE b.file_id = f.id AND f.release_id = ? AND b.release_id IS DISTINCT FROM ?`,
-          [releaseId, releaseId, releaseId]
-        )
-        await trx.from('releases').whereIn('id', previous).where('file_count', 0).delete()
-      }
-      // The NZB file joins the release it describes, unless it is part of a real release already.
-      const own = nzbFile.release_id ? Number(nzbFile.release_id) : null
-      if (own !== releaseId && options.moveContainer !== false) {
-        const ownRelease = own ? await trx.from('releases').where('id', own).first() : null
-        if (!ownRelease || ownRelease.file_count <= 1) {
-          await trx.from('files').where('id', nzbFile.id).update({ release_id: releaseId })
-          if (own) await trx.from('releases').where('id', own).where('file_count', 0).delete()
-        }
-      }
-
-      const release = await trx.from('releases').where('id', releaseId).first()
-      const update: Record<string, unknown> = {}
-      if (nzb.meta.password) update.password = nzb.meta.password
-      if (!release.file_total) update.file_total = entries.length
-      if ((looksRandom(release.name) || release.source === 'post') && !looksRandom(title)) update.name = title
-      if (Object.keys(update).length) await trx.from('releases').where('id', releaseId).update(update)
-
-      const result = {
-        releaseId,
-        releaseName: (update.name as string) ?? release.name,
-        listed: entries.length,
-        known: known.length,
-        added: missing.length,
-        password: Boolean(nzb.meta.password),
-      }
-      logger.info(
-        `NZB "${options.entryName ? `${nzbFile.filename}:${options.entryName}` : nzbFile.filename}" → release ${releaseId} "${result.releaseName}": ${result.listed} files listed, ` +
-          `${result.known} known, ${result.added} added${result.password ? ', with password' : ''}.`
-      )
-      return result
-    })
-  }
-
-  /** Release name: the NZB's meta name, else the NZB's (entry or file) name, else its largest file's stem. */
-  private title(nzb: ParsedNzb, nzbFile: any, names: string[], entryName?: string): string {
-    if (nzb.meta.name) return nzb.meta.name
-    const entry = entryName ? fileStem(entryName.split(/[\\/]/).pop() ?? entryName) : ''
-    if (entry && !looksRandom(entry)) return entry
-    const own = fileStem(String(nzbFile.real_name ?? nzbFile.filename ?? ''))
-    if (own && !looksRandom(own)) return own
-    const largest = nzb.files
-      .map((f, i) => ({ name: names[i], bytes: f.segments.reduce((s, x) => s + x.bytes, 0) }))
-      .sort((a, b) => b.bytes - a.bytes)[0]
-    return fileStem(largest?.name ?? own)
-  }
-}
-
-export default new NzbImportService()

app/services/Par2Service.ts

@@ -1,165 +0,0 @@
-import db from '@adonisjs/lucid/services/db'
-import logger from '@adonisjs/core/services/logger'
-import NntpService from '#services/NntpService'
-import IngestService, { PAR2_MATCH_WINDOW_SECONDS } from '#services/IngestService'
-import { decodeSegment, YencFile } from '#lib/yenc/index'
-import { isPar2, parsePar2, type Par2Set } from '#lib/par2/index'
-import { par2ReleaseIdentity, par2ReleaseName } from '#lib/release/index'
-import NameService, { WEAK_RELEASE_SQL, isBetterName } from '#services/NameService'
-
-/** Par2 index files are small; recovery volumes can be large and are not needed. */
-const MAX_PARTS = 20
-/** Size-only matching below this is too ambiguous. */
-const SIZE_ONLY_MIN = 1024 * 1024
-
-function countBy<T>(values: T[]): Map<T, number> {
-  const counts = new Map<T, number>()
-  for (const value of values) counts.set(value, (counts.get(value) ?? 0) + 1)
-  return counts
-}
-
-export interface Par2Result {
-  releaseId: number
-  releaseName: string
-  listed: number
-  matched: number
-}
-
-/**
- * Recognises releases of obfuscated posts from their par2 files: the par2
- * lists the real name and exact size of every file in the set, and
- * body-derived binaries carry their exact size. See CONCEPT.md (D16).
- */
-class Par2Service {
-  /**
-   * Reads a par2 file whose first segment is already decoded, downloading the
-   * remaining parts if there are any. Returns null when it is not par2.
-   */
-  async read(first: Buffer, messageIds: Record<string, { id: string }>): Promise<Par2Set | null> {
-    if (!isPar2(first)) return null
-    const ids = Object.entries(messageIds)
-      .sort(([a], [b]) => Number(a) - Number(b))
-      .map(([, part]) => part.id)
-    if (ids.length > MAX_PARTS) return null
-    if (ids.length === 1) return parsePar2(first)
-
-    let assembled: YencFile | undefined
-    const conn = await NntpService.acquire()
-    let failed = false
-    try {
-      for (const id of ids) {
-        const segment = decodeSegment(await NntpService.fetchBody(conn, id))
-        assembled ??= YencFile.fromSegment(segment)
-        assembled.add(segment)
-      }
-    } catch (error) {
-      failed = true
-      throw error
-    } finally {
-      NntpService.release(conn, failed)
-    }
-    return parsePar2(assembled!.getBuffer())
-  }
-
-  async attach(binary: any, set: Par2Set): Promise<Par2Result> {
-    const result = await db.transaction(async (trx) => {
-      // Exact: size and first-16-KiB MD5 of a listed file. Only these may pull
-      // the set into a release the files already belong to; size alone is
-      // ambiguous across posts (standard volume sizes recur).
-      const exact = await trx
-        .from('binaries')
-        .whereIn(['size', 'md5_16k'] as any, set.files.map((f) => [f.length, f.md5_16k]) as any)
-        .select('id', 'release_id')
-
-      // Size only, for files without a fingerprint: the size must be listed
-      // once in this set, be large, and belong to exactly one binary nearby.
-      const listedOnce = [...countBy(set.files.map((f) => f.length))]
-        .filter(([size, n]) => n === 1 && size >= SIZE_ONLY_MIN)
-        .map(([size]) => size)
-      const bySize = listedOnce.length
-        ? await trx
-            .from('binaries')
-            .whereIn('size', listedOnce)
-            .whereNull('md5_16k')
-            .whereRaw('abs(extract(epoch FROM (posted_at - ?::timestamptz))) < ?', [
-              binary.posted_at,
-              PAR2_MATCH_WINDOW_SECONDS,
-            ])
-            .select('id', 'release_id', 'size')
-        : []
-      const sizeCounts = countBy(bySize.map((b) => Number(b.size)))
-      const fuzzy = bySize.filter((b) => sizeCounts.get(Number(b.size)) === 1)
-      const candidates = [...exact, ...fuzzy]
-
-      // Prefer a release the exactly matched files are already in (e.g.
-      // obfuscated volumes grouped by their stem); otherwise the set's own.
-      const counts = countBy(exact.filter((c) => c.release_id).map((c) => Number(c.release_id)))
-      let releaseId = [...counts].sort((a, b) => b[1] - a[1])[0]?.[0]
-      const identity = par2ReleaseIdentity(
-        set.recoverySetId,
-        set.files.map((f) => f.name)
-      )
-      if (!releaseId) {
-        const ids = await IngestService.upsertReleases(trx.knexClient, [
-          {
-            release: identity,
-            poster: binary.poster,
-            weakIdentity: true,
-            groups: binary.groups,
-            postedAt: binary.posted_at ? new Date(binary.posted_at) : undefined,
-          },
-        ])
-        releaseId = ids.get(identity.key)!
-      }
-
-      await trx
-        .knexClient('release_files')
-        .insert(
-          set.files.map((f) => ({
-            release_id: releaseId,
-            par2_file_id: f.fileId,
-            name: f.name,
-            size: f.length,
-            md5_16k: f.md5_16k,
-            md5: f.md5,
-            set_id: set.recoverySetId,
-          }))
-        )
-        .onConflict(['release_id', 'par2_file_id'])
-        .ignore()
-
-      const matchIds = candidates.filter((c) => !c.release_id).map((c) => Number(c.id))
-      await trx.from('binaries').whereIn('id', [binary.id, ...matchIds]).whereNull('release_id').update({ release_id: releaseId })
-      // Exact matches sitting alone in a random-named release (one file of a
-      // season, each posted under its own random name) join the set's release.
-      await trx.rawQuery(
-        `UPDATE binaries b SET release_id = ? WHERE b.id = ANY(?) AND b.release_id <> ? AND ${WEAK_RELEASE_SQL('b.release_id')}`,
-        [releaseId, exact.map((c) => Number(c.id)), releaseId]
-      )
-      await trx.rawQuery(
-        `UPDATE files f SET release_id = b.release_id FROM binaries b
-         WHERE b.file_id = f.id AND b.release_id = ? AND f.release_id IS DISTINCT FROM b.release_id`,
-        [releaseId]
-      )
-
-      // The set may have joined a release named after one random filename.
-      let releaseName = (await trx.from('releases').where('id', releaseId).first())?.name
-      const better = par2ReleaseName(set.files.map((f) => f.name))
-      if (releaseName && better && isBetterName(releaseName, better)) {
-        await trx.from('releases').where('id', releaseId).update({ name: better })
-        releaseName = better
-      }
-      logger.info(
-        `Par2 set ${set.recoverySetId.slice(0, 8)} → release ${releaseId} "${releaseName}": ` +
-          `${set.files.length} files listed, ${matchIds.length} binaries matched by size.`
-      )
-      return { releaseId, releaseName, listed: set.files.length, matched: matchIds.length }
-    })
-
-    // Real names for every known file of the set (exact fingerprint, or size within the release).
-    await NameService.resolveForRelease(result.releaseId)
-    return result
-  }
-}
-
-export default new Par2Service()

app/services/QueueService.ts

@@ -1,61 +0,0 @@
-import { Queue, type DefaultJobOptions } from 'bullmq'
-import queueConfig from '#config/queue'
-import gcConfig from '#config/gc'
-
-const defaultJobOptions: DefaultJobOptions = {
-  // Transient NNTP errors (e.g. a pooled connection the server closed) are retried.
-  attempts: 3,
-  backoff: { type: 'exponential', delay: 2000 },
-  removeOnComplete: {
-    age: gcConfig.jobs.completedAgeSeconds,
-    count: gcConfig.jobs.completedMaxCount,
-  },
-  removeOnFail: { age: gcConfig.jobs.failedAgeSeconds },
-}
-
-/**
- * Queues are created (and connect to Redis) on first use. Ace imports every
- * command at boot, so eager queues would keep one-off commands from exiting.
- */
-class QueueService {
-  private readonly queues = new Map<string, Queue>()
-
-  private queue(name: string): Queue {
-    let queue = this.queues.get(name)
-    if (!queue) {
-      queue = new Queue(name, { connection: queueConfig.connection, defaultJobOptions })
-      this.queues.set(name, queue)
-    }
-    return queue
-  }
-
-  get nntpFetchQueue() {
-    return this.queue('nntp-fetch-queue')
-  }
-  get fileQueue() {
-    return this.queue('file-queue')
-  }
-  get bodyQueue() {
-    return this.queue('body-queue')
-  }
-  get collectionQueue() {
-    return this.queue('collection-queue')
-  }
-
-  /** Queues no longer used by the pipeline, kept here so GC can empty them. */
-  get legacy(): Queue[] {
-    return [this.queue('header-queue')]
-  }
-
-  get all(): Queue[] {
-    return [this.nntpFetchQueue, this.fileQueue, this.bodyQueue, this.collectionQueue]
-  }
-
-  /** Closes the queues that were opened. */
-  async closeAll() {
-    await Promise.all([...this.queues.values()].map((queue) => queue.close()))
-    this.queues.clear()
-  }
-}
-
-export default new QueueService()

app/services/RedisService.ts

@@ -1,25 +0,0 @@
-import { createRequire } from 'module'
-import redisConfig from '#config/redis'
-
-const require = createRequire(import.meta.url)
-const IORedis = require('ioredis')
-
-/**
- * Connects on first use. Ace imports every command at boot, so an eager
- * connection would keep even `node ace list` from exiting.
- */
-class RedisService {
-  private instance?: any
-
-  public get client(): any {
-    this.instance ??= new IORedis(redisConfig.connections.main)
-    return this.instance
-  }
-
-  async close() {
-    if (this.instance) await this.instance.quit()
-    this.instance = undefined
-  }
-}
-
-export default new RedisService()

app/services/SearchService.ts

@@ -1,284 +0,0 @@
-import db from '@adonisjs/lucid/services/db'
-import { buildNzb } from '#lib/nzb/index'
-import SpotService from '#services/SpotService'
-
-export type SortOrder = 'newest' | 'oldest' | 'largest' | 'files' | 'name'
-
-export interface SearchParams {
-  q?: string
-  /** Only releases whose name is readable (not an obfuscated token). */
-  readable?: boolean
-  /** Only releases that have all their files. */
-  complete?: boolean
-  /** Hide password-protected releases. */
-  hideEncrypted?: boolean
-  source?: string
-  poster?: string
-  sort?: SortOrder
-  page?: number
-  perPage?: number
-  /** Newznab: exact category ids (already expanded from parents). */
-  categories?: number[]
-  /** Newznab: leave out releases known to be missing files. */
-  notIncomplete?: boolean
-  maxAgeDays?: number
-  /** TV: season and episode numbers, matched as S01E02 or 1x02 in the name. */
-  season?: number
-  episode?: number
-  /** Overrides `page` (Newznab uses offsets). */
-  offset?: number
-  /** Leave out releases that are only an encrypted container (nzbe). */
-  hideContainers?: boolean
-  /** Include incomplete releases that consist of par2 files only (hidden by default). */
-  includePar2Only?: boolean
-}
-
-const SORTS: Record<SortOrder, string> = {
-  newest: 'r.posted_at DESC NULLS LAST, r.id DESC',
-  oldest: 'r.posted_at ASC NULLS LAST, r.id ASC',
-  largest: 'r.size DESC, r.id DESC',
-  files: 'r.file_count DESC, r.id DESC',
-  name: 'r.name ASC, r.id ASC',
-}
-
-/**
- * Shown in search: releases with files, and Spotnet listings whose NZB is
- * fetched on first download (CONCEPT.md D29). Matches the partial indexes.
- */
-const VISIBLE_SQL = `(r.file_count > 0 OR r.source = 'spot')`
-
-/** Incomplete, and every file is a par2 file: nothing to download but repair data. */
-const PAR2_ONLY_SQL = `(r.file_count > 0 AND NOT r.complete AND NOT EXISTS (
-  SELECT 1 FROM files f WHERE f.release_id = r.id
-    AND f.content_type IS DISTINCT FROM 'par2' AND f.filename !~* '\\.par2$'))`
-
-/** A search word as an ILIKE pattern; dots, underscores, dashes and spaces are interchangeable. */
-function pattern(word: string): string {
-  const escaped = word.replace(/[\\%]/g, (c) => `\\${c}`)
-  return `%${escaped.replace(/[._\- ]+/g, '_')}%`
-}
-
-/**
- * Searching and browsing releases for the web interface (CONCEPT.md D22).
- */
-class SearchService {
-  async search(params: SearchParams) {
-    const where: string[] = [
-      VISIBLE_SQL,
-      // Retracted by its poster or a Spotnet moderator (CONCEPT.md D29).
-      `NOT EXISTS (SELECT 1 FROM spots s WHERE s.release_id = r.id AND s.status = 'removed')`,
-    ]
-    const bindings: unknown[] = []
-
-    const words = (params.q ?? '').trim().split(/\s+/).filter(Boolean).slice(0, 8)
-    for (const word of words) {
-      const p = pattern(word)
-      // Every word must match the release name, a file name, or the archive contents.
-      // One trigram-indexed lookup per table, not a subquery per release.
-      where.push(`(r.name ILIKE ? OR r.contents::text ILIKE ?
-        OR r.id IN (SELECT f.release_id FROM files f WHERE f.filename ILIKE ? OR f.real_name ILIKE ?))`)
-      bindings.push(p, p, p, p)
-    }
-    if (params.readable) where.push('is_readable_name(r.name)')
-    if (params.hideEncrypted) where.push('r.encrypted IS NOT TRUE')
-    if (params.hideContainers) where.push('r.container IS NULL')
-    // par2 files whose data never arrived (e.g. uploads spread over groups we can't fetch).
-    if (!params.includePar2Only) where.push(`(${PAR2_ONLY_SQL}) IS NOT TRUE`)
-    // Kept current by triggers (migration …860).
-    if (params.complete) where.push('r.complete')
-    if (params.source) {
-      where.push('r.source = ?')
-      bindings.push(params.source)
-    }
-    if (params.poster) {
-      where.push('r.poster ILIKE ?')
-      bindings.push(pattern(params.poster))
-    }
-    if (params.categories?.length) {
-      where.push('r.category = ANY(?)')
-      bindings.push(params.categories)
-    }
-    if (params.notIncomplete) {
-      where.push(`(r.complete OR (r.file_total IS NULL AND NOT EXISTS (SELECT 1 FROM release_files rf WHERE rf.release_id = r.id)))`)
-    }
-    if (params.maxAgeDays) {
-      where.push(`r.posted_at >= now() - make_interval(days => ?)`)
-      bindings.push(params.maxAgeDays)
-    }
-    if (params.season !== undefined) {
-      const s = params.season
-      const e = params.episode
-      where.push('r.name ~* ?')
-      bindings.push(
-        e !== undefined
-          ? `(^|[._ -])(S0*${s}E0*${e}|0*${s}x0*${e})([^0-9]|$)`
-          : `(^|[._ -])(S0*${s}([._ -]|E\\d|$)|0*${s}x\\d)`
-      )
-    }
-
-    const perPage = Math.min(Math.max(params.perPage ?? 50, 1), 200)
-    const page = Math.max(params.page ?? 1, 1)
-    const offset = params.offset ?? (page - 1) * perPage
-    const order = SORTS[params.sort ?? 'newest'] ?? SORTS.newest
-    const whereSql = where.join(' AND ')
-
-    const [{ rows }, { rows: countRows }] = await Promise.all([
-      db.rawQuery(
-        `SELECT r.id, r.name, r.source, r.poster, r.groups, r.posted_at, r.file_total, r.file_count, r.category, r.container,
-                r.size, r.encrypted, is_readable_name(r.name) AS readable,
-                (SELECT count(*) FROM release_files rf WHERE rf.release_id = r.id) AS listed_files,
-                (SELECT count(*) FROM release_files rf WHERE rf.release_id = r.id AND EXISTS (
-                   SELECT 1 FROM files f JOIN binaries b ON b.file_id = f.id
-                   WHERE f.release_id = r.id AND (b.size = rf.size OR f.real_name = rf.name))) AS listed_present,
-                jsonb_array_length(coalesce(r.contents, '[]'::jsonb)) AS archive_entries,
-                r.complete, ${PAR2_ONLY_SQL} AS par2_only
-         FROM releases r WHERE ${whereSql}
-         ORDER BY ${order} LIMIT ? OFFSET ?`,
-        [...bindings, perPage, offset]
-      ),
-      db.rawQuery(`SELECT count(*) AS total FROM releases r WHERE ${whereSql}`, bindings),
-    ])
-
-    return {
-      page,
-      perPage,
-      offset,
-      total: Number(countRows[0].total),
-      releases: rows.map((r: any) => ({
-        id: Number(r.id),
-        name: r.name,
-        source: r.source,
-        poster: r.poster,
-        groups: r.groups,
-        postedAt: r.posted_at,
-        fileTotal: r.file_total,
-        fileCount: r.file_count,
-        size: Number(r.size),
-        encrypted: r.encrypted,
-        readable: r.readable,
-        listedFiles: Number(r.listed_files),
-        listedPresent: Number(r.listed_present),
-        archiveEntries: Number(r.archive_entries),
-        complete: r.complete,
-        category: r.category,
-        container: r.container,
-        par2Only: r.par2_only,
-      })),
-    }
-  }
-
-  async release(id: number) {
-    const release = await db.from('releases').where('id', id).first()
-    if (!release) return null
-
-    const [files, listed, pending] = await Promise.all([
-      db
-        .from('files as f')
-        .leftJoin('binaries as b', 'b.file_id', 'f.id')
-        .where('f.release_id', id)
-        .orderByRaw('coalesce(f.real_name, f.filename)')
-        .select('f.id', 'f.filename', 'f.real_name', 'f.size', 'f.parts', 'f.date', 'f.poster', 'b.size as exact_size'),
-      db.rawQuery(
-        `SELECT rf.name, rf.size,
-                EXISTS (SELECT 1 FROM files f JOIN binaries b ON b.file_id = f.id
-                        WHERE f.release_id = rf.release_id AND (f.real_name = rf.name OR b.size = rf.size)) AS present
-         FROM release_files rf WHERE rf.release_id = ? ORDER BY rf.name`,
-        [id]
-      ),
-      db.rawQuery(
-        `SELECT count(*) AS binaries, coalesce(sum(b.total_parts), 0) AS parts_total,
-                coalesce(sum((SELECT count(*) FROM parts p WHERE p.binary_id = b.id)), 0) AS parts_have
-         FROM binaries b WHERE b.release_id = ? AND b.completed_at IS NULL AND b.conflicted_at IS NULL`,
-        [id]
-      ),
-    ])
-
-    const spot = release.source === 'spot'
-      ? await db.from('spots').where('release_id', id).select('tag', 'poster', 'category', 'website').first()
-      : null
-    return {
-      id: Number(release.id),
-      name: release.name,
-      source: release.source,
-      spot: spot ? { tag: spot.tag, poster: spot.poster, category: spot.category, website: spot.website } : null,
-      listingOnly: release.source === 'spot' && Number(release.file_count) === 0,
-      poster: release.poster,
-      groups: release.groups,
-      postedAt: release.posted_at,
-      fileTotal: release.file_total,
-      fileCount: release.file_count,
-      size: Number(release.size),
-      encrypted: release.encrypted,
-      container: release.container,
-      contents: release.contents ?? [],
-      files: files.map((f) => ({
-        id: Number(f.id),
-        name: f.real_name ?? f.filename,
-        postedName: f.filename,
-        realName: f.real_name,
-        size: Number(f.exact_size ?? f.size),
-        parts: f.parts,
-        postedAt: f.date ? new Date(Number(f.date)) : null,
-        poster: f.poster,
-      })),
-      par2Files: listed.rows.map((r: any) => ({ name: r.name, size: Number(r.size), present: r.present })),
-      pending: {
-        binaries: Number(pending.rows[0].binaries),
-        partsHave: Number(pending.rows[0].parts_have),
-        partsTotal: Number(pending.rows[0].parts_total),
-      },
-    }
-  }
-
-  async nzb(id: number): Promise<{ name: string; xml: string } | null> {
-    let release = await db.from('releases').where('id', id).first()
-    if (!release) return null
-    if (release.source === 'spot' && Number(release.file_count) === 0) {
-      // A Spotnet listing: read its NZB from Usenet now (the import may join known files elsewhere).
-      const imported = await SpotService.importListing(id)
-      if (!imported) return null
-      id = imported
-      release = await db.from('releases').where('id', id).first()
-    }
-    const files = await db
-      .from('files')
-      .where('release_id', id)
-      .orderByRaw('coalesce(real_name, filename)')
-      .select('filename', 'real_name', 'poster', 'date', 'groups', 'message_ids')
-    const xml = buildNzb(
-      files.map((f) => ({
-        name: f.real_name ?? f.filename,
-        poster: f.poster,
-        date: Number(f.date),
-        groups: f.groups?.length ? f.groups : ['alt.binaries.test'],
-        segments: f.message_ids,
-      })),
-      { title: release.name, password: release.password ?? undefined }
-    )
-    return { name: release.name, xml }
-  }
-
-  async stats() {
-    const { rows } = await db.rawQuery(`
-      SELECT (SELECT count(*) FROM releases r WHERE ${VISIBLE_SQL}) AS releases,
-             (SELECT count(*) FROM releases r WHERE ${VISIBLE_SQL} AND is_readable_name(name)) AS readable_releases,
-             (SELECT count(*) FROM files) AS files,
-             (SELECT coalesce(sum(size), 0) FROM files) AS size,
-             (SELECT count(*) FROM releases WHERE encrypted) AS encrypted_releases,
-             (SELECT max(updated_at) FROM groups) AS last_indexed_at,
-             (SELECT json_agg(json_build_object('source', source, 'releases', n) ORDER BY n DESC)
-                FROM (SELECT coalesce(source, 'unknown') AS source, count(*) AS n FROM releases r WHERE ${VISIBLE_SQL} GROUP BY 1) s) AS sources`)
-    const r = rows[0]
-    return {
-      releases: Number(r.releases),
-      readableReleases: Number(r.readable_releases),
-      files: Number(r.files),
-      size: Number(r.size),
-      encryptedReleases: Number(r.encrypted_releases),
-      lastIndexedAt: r.last_indexed_at,
-      sources: r.sources ?? [],
-    }
-  }
-}
-
-export default new SearchService()

app/services/Shutdown.ts

@@ -1,45 +0,0 @@
-import { setTimeout as sleep } from 'node:timers/promises'
-
-/** After a signal, give cleanup this long before exiting anyway. */
-const FORCE_EXIT_MS = 25_000
-
-/**
- * Graceful shutdown for the long-running commands. SIGTERM (docker stop) and
- * SIGINT wake up `wait()` and `sleep()`, so workers can finish their active
- * jobs and close their connections before the process exits.
- */
-class Shutdown {
-  private readonly controller = new AbortController()
-
-  constructor() {
-    for (const signal of ['SIGTERM', 'SIGINT'] as const) {
-      process.once(signal, () => {
-        this.controller.abort()
-        setTimeout(() => process.exit(0), FORCE_EXIT_MS).unref()
-      })
-    }
-  }
-
-  get requested(): boolean {
-    return this.controller.signal.aborted
-  }
-
-  /** Resolves when a shutdown signal arrives. */
-  wait(): Promise<void> {
-    if (this.requested) return Promise.resolve()
-    return new Promise((resolve) =>
-      this.controller.signal.addEventListener('abort', () => resolve(), { once: true })
-    )
-  }
-
-  /** Sleeps, but returns early on shutdown. */
-  async sleep(ms: number): Promise<void> {
-    try {
-      await sleep(ms, undefined, { signal: this.controller.signal })
-    } catch {
-      // aborted by a shutdown signal
-    }
-  }
-}
-
-export default new Shutdown()

app/services/SplitService.ts

@@ -1,146 +0,0 @@
-import { createHash } from 'node:crypto'
-import db from '@adonisjs/lucid/services/db'
-import IngestService from '#services/IngestService'
-import NameService from '#services/NameService'
-import { splitVerified } from '#lib/split/index'
-
-/** Conflicted binaries handled per run; the rest follow on the next tick. */
-const BATCH = 200
-
-export interface SplitResult {
-  parents: number
-  volumes: number
-  completed: number
-}
-
-/**
- * Splits conflicted obfuscated binaries (several equal-sized files under one
- * size+parts identity) into one binary per volume. Only volumes whose
- * combined part CRCs match the file CRC of their last part are created, so a
- * split can never produce a corrupt file; the rest waits for more articles.
- * See CONCEPT.md (D20).
- */
-class SplitService {
-  async splitPending(): Promise<SplitResult> {
-    const parents = await db
-      .from('binaries')
-      .whereNotNull('conflicted_at')
-      .whereNull('parent_binary_id')
-      // split_at = 'infinity' marks binaries that cannot be split (see migration …810).
-      .where((q) => q.whereNull('split_at').orWhereRaw('last_seen_at > split_at'))
-      // Most recently active first: those are the posts still coming in.
-      .orderBy('last_seen_at', 'desc')
-      .limit(BATCH)
-
-    const volumeIds: number[] = []
-    for (const parent of parents) {
-      volumeIds.push(...(await this.split(parent)))
-    }
-
-    // Former conflicts that turned out to be reposts (unflagged, keep their
-    // split_at): no new article may arrive to trigger their completion.
-    const unflagged = await db
-      .from('binaries')
-      .whereNull('conflicted_at')
-      .whereNull('completed_at')
-      .whereNull('parent_binary_id')
-      .whereNotNull('split_at')
-      .whereRaw("split_at <> 'infinity'")
-      .limit(BATCH)
-      .select('id')
-    volumeIds.push(...unflagged.map((b) => Number(b.id)))
-
-    await IngestService.matchPar2Releases(volumeIds)
-    await NameService.resolve(volumeIds)
-    const completed = await IngestService.claimCompleted(volumeIds)
-    await IngestService.enqueue(completed)
-    return { parents: parents.length, volumes: volumeIds.length, completed: completed.length }
-  }
-
-  /** Creates the newly verified volumes of one conflicted binary; returns their ids. */
-  private async split(parent: any): Promise<number[]> {
-    return db.transaction(async (trx) => {
-      // Unverified volumes from before verification existed can never complete.
-      await trx.from('binaries').where('parent_binary_id', parent.id).whereNull('completed_at').delete()
-
-      const { rows } = await trx.rawQuery(
-        `WITH taken AS (
-           SELECT p.message_id FROM parts p JOIN binaries v ON v.id = p.binary_id
-           WHERE v.parent_binary_id = ?
-         ), candidates AS (
-           SELECT number, message_id, bytes, article_number, pcrc32, part_size, file_crc32 FROM parts WHERE binary_id = ?
-           UNION ALL
-           SELECT number, message_id, bytes, article_number, pcrc32, part_size, file_crc32 FROM part_conflicts WHERE binary_id = ?
-         )
-         SELECT * FROM candidates WHERE message_id NOT IN (SELECT message_id FROM taken)`,
-        [parent.id, parent.id, parent.id]
-      )
-      const volumes = splitVerified(
-        rows.map((r: any) => ({
-          number: r.number,
-          messageId: r.message_id,
-          bytes: r.bytes,
-          articleNumber: r.article_number === null ? null : Number(r.article_number),
-          pcrc32: r.pcrc32,
-          partSize: r.part_size,
-          fileCrc32: r.file_crc32,
-        })),
-        { totalParts: parent.total_parts }
-      )
-
-      const now = new Date()
-      const ids: number[] = []
-      for (const articles of volumes) {
-        // A verified volume is fixed, so its first article identifies it.
-        const fileKey = createHash('sha1').update(`${parent.file_key}#${articles[0].messageId}`).digest('hex')
-        const inserted = await trx
-          .knexClient('binaries')
-          .insert({
-            file_key: fileKey,
-            poster: parent.poster,
-            subject: parent.subject,
-            name: parent.name,
-            total_parts: parent.total_parts,
-            groups: JSON.stringify(parent.groups),
-            posted_at: parent.posted_at,
-            size: parent.size,
-            release_id: parent.release_id,
-            parent_binary_id: parent.id,
-            first_seen_at: now,
-            last_seen_at: now,
-          })
-          .onConflict('file_key')
-          .ignore()
-          .returning('id')
-        if (inserted.length === 0) continue
-        const id = Number(inserted[0].id)
-        await trx.knexClient('parts').insert(
-          articles.map((a) => ({
-            binary_id: id,
-            number: a.number,
-            message_id: a.messageId,
-            bytes: a.bytes,
-            article_number: a.articleNumber,
-            pcrc32: a.pcrc32,
-            part_size: a.partSize,
-            file_crc32: a.fileCrc32,
-          }))
-        )
-        // The fingerprint of a volume is that of its part 1 article.
-        await trx.rawQuery(
-          `UPDATE binaries SET md5_16k = (
-             SELECT md5_16k FROM part_conflicts WHERE binary_id = ? AND message_id = ?
-             UNION ALL SELECT md5_16k FROM parts WHERE binary_id = ? AND message_id = ? LIMIT 1)
-           WHERE id = ?`,
-          [parent.id, articles[0].messageId, parent.id, articles[0].messageId, id]
-        )
-        ids.push(id)
-      }
-
-      await trx.from('binaries').where('id', parent.id).update({ split_at: now })
-      return ids
-    })
-  }
-}
-
-export default new SplitService()

app/services/SpotService.ts

@@ -1,172 +0,0 @@
-import db from '@adonisjs/lucid/services/db'
-import logger from '@adonisjs/core/services/logger'
-import indexerConfig from '#config/indexer'
-import NntpService from '#services/NntpService'
-import NzbImportService from '#services/NzbImportService'
-import { type OverviewHeader } from '#services/IngestService'
-import { parseNzb } from '#lib/nzb/index'
-import {
-  moderationTarget,
-  spotterId,
-  decodeSpotBinary,
-  parseSpotArticleHeaders,
-  parseSpotHeader,
-  parseSpotXml,
-  verifySpotHeader,
-  type SpotHeader,
-} from '#lib/spotnet/index'
-
-/** Spots whose NZB is larger than this are skipped (Spotweb skips >50 KB of XML too). */
-const MAX_NZB_SEGMENTS = 20
-/** A personal dispose only counts within this long after the spot (Spotweb: 5 days). */
-const DISPOSE_WINDOW_SECONDS = 5 * 24 * 60 * 60
-/** A claimed spot whose worker died is taken again after this long. */
-const RECLAIM_AFTER = '1 hour'
-
-export function isSpotGroup(name: string): boolean {
-  return indexerConfig.spotGroups.includes(name)
-}
-
-/**
- * Spotnet (CONCEPT.md D29). Spots are signed posts in free.pt; their full
- * header holds XML with the message ids of a compressed NZB, which is
- * imported like a posted NZB (D26). Spot NZBs name many obfuscated posts
- * (nyuu, ngPost, the skipped multi-group uploader).
- */
-class SpotService {
-  /** Queues the spots among a batch of one spot group's headers. Returns how many. */
-  async recordHeaders(groupName: string, overviews: [number, OverviewHeader][]): Promise<number> {
-    const now = new Date()
-    const rows = []
-    for (const [articleNumber, header] of overviews) {
-      if (!header?.from || !header['message-id']) continue
-      const spot = parseSpotHeader(header.from, header.subject ?? '', header['message-id'], header.date)
-      if (!spot?.title) continue
-      rows.push({
-        message_id: spot.messageId,
-        group_name: groupName,
-        article_number: Number(articleNumber),
-        title: spot.title,
-        tag: spot.tag || null,
-        poster: spot.poster,
-        category: spot.category,
-        subcats: JSON.stringify(spot.subcats),
-        size: spot.size || null,
-        posted_at: spot.stamp ? new Date(spot.stamp * 1000) : null,
-        key_id: spot.keyId,
-        spotter_id: spotterId(spot.selfSignedPubKey) || null,
-        created_at: now,
-      })
-    }
-    if (rows.length === 0) return 0
-    await db.knexQuery().table('spots').insert(rows).onConflict('message_id').ignore()
-    return rows.length
-  }
-
-  /** Takes up to `limit` pending spots, oldest first; they are marked `processing`. */
-  async claim(limit: number): Promise<any[]> {
-    const { rows } = await db.rawQuery(
-      `UPDATE spots SET status = 'processing', processed_at = now()
-       WHERE message_id IN (
-         SELECT message_id FROM spots
-         WHERE status = 'pending' OR (status = 'processing' AND processed_at < now() - interval '${RECLAIM_AFTER}')
-         ORDER BY created_at, article_number LIMIT ? FOR UPDATE SKIP LOCKED)
-       RETURNING *`,
-      [limit]
-    )
-    return rows
-  }
-
-  /** Verifies the spot on its raw header, then reads and imports its NZB. Returns the new status. */
-  async process(spot: any): Promise<string> {
-    const update: Record<string, unknown> = { processed_at: new Date(), error: null }
-    try {
-      const raw = await this.fetch(`HEAD <${spot.message_id}>`)
-      const { headers, xml } = parseSpotArticleHeaders(raw.toString('latin1'))
-      // The signature covers the ISO-8859-1 bytes, which the overview (read as UTF-8) may have mangled.
-      const header = parseSpotHeader(headers.from ?? '', headers.subject ?? '', spot.message_id, headers.date)
-      if (!header || !verifySpotHeader(header)) {
-        update.status = 'unverified'
-      } else if (moderationTarget(header)) {
-        update.status = 'moderation'
-        await this.moderate(header, moderationTarget(header)!)
-      } else {
-        const parsed = parseSpotXml(xml)
-        update.title = header.title
-        update.poster = header.poster
-        update.spotter_id = spotterId(header.selfSignedPubKey) || null
-        if (parsed?.website) update.website = parsed.website
-        const segments = parsed?.nzbSegments ?? []
-        if (segments.length === 0 || segments.length > MAX_NZB_SEGMENTS) {
-          update.status = 'no-nzb'
-        } else {
-          const bodies = []
-          for (const id of segments) bodies.push(await this.fetch(`BODY <${id.replace(/^<|>$/g, '')}>`))
-          const nzbXml = decodeSpotBinary(bodies).toString('utf8')
-          const nzb = parseNzb(nzbXml)
-          if (!nzb || nzb.files.length === 0) {
-            update.status = 'no-nzb'
-          } else {
-            const groups = [...new Set(nzb.files.flatMap((f) => f.groups))]
-            const result = await NzbImportService.import(
-              {
-                id: null,
-                filename: `${header.title}.nzb`,
-                poster: header.poster,
-                groups,
-                date: header.stamp * 1000,
-                message_ids: { '1': { id: spot.message_id } },
-              },
-              nzbXml,
-              { name: header.title, source: 'spot', moveContainer: false }
-            )
-            update.status = result ? 'imported' : 'no-nzb'
-            if (result) update.release_id = result.releaseId
-          }
-        }
-      }
-    } catch (error: any) {
-      update.status = 'failed'
-      update.error = String(error.message).slice(0, 500)
-      logger.warn(`Spot ${spot.message_id} "${spot.title}": ${error.message}`)
-    }
-    await db.from('spots').where('message_id', spot.message_id).update(update)
-    return update.status as string
-  }
-
-  /**
-   * Applies a delete/dispose/remove: the target spot is marked `removed`
-   * (its release is then hidden from search). As in Spotweb, a personal
-   * dispose must come from the same spotter within 5 days; a moderator's
-   * (signed with Spotnet's key, no personal key) always counts.
-   */
-  private async moderate(header: SpotHeader, targetId: string) {
-    const target = await db.from('spots').where('message_id', targetId).first()
-    if (!target) return
-    const personal = header.selfSignedPubKey.length > 50 ? spotterId(header.selfSignedPubKey) : ''
-    if (personal) {
-      const age = header.stamp - (target.posted_at ? new Date(target.posted_at).getTime() / 1000 : 0)
-      if (target.spotter_id !== personal || age > DISPOSE_WINDOW_SECONDS) return
-    }
-    await db.from('spots').where('message_id', targetId).update({ status: 'removed' })
-    logger.info(`Spot "${target.title}" removed by ${personal ? 'its poster' : 'a moderator'}.`)
-  }
-
-  /** One HEAD or BODY, as raw bytes. */
-  private async fetch(command: string): Promise<Buffer> {
-    const conn = await NntpService.acquire()
-    let failed = false
-    try {
-      const { data } = await conn._artcmd(command)
-      return Buffer.from(data.buffer, data.byteOffset, data.byteLength)
-    } catch (error: any) {
-      // 430 (no such article) is an answer, not a broken connection.
-      failed = !/^4\d\d/.test(String(error.message))
-      throw error
-    } finally {
-      NntpService.release(conn, failed)
-    }
-  }
-}
-
-export default new SpotService()

app/services/batch_store.ts

@@ -0,0 +1,116 @@
+import db from '@adonisjs/lucid/services/db'
+import type { EdgeBatch } from '#lib/edge/index'
+import ChunkPartitions from '#services/chunk_partitions'
+
+export class StaleBatch extends Error {}
+
+export interface StoredBatch {
+  releases: number
+  newReleases: number
+  candidates: number
+  fragments: number
+}
+
+/**
+ * Stores one batch of a leased range (REBUILD.md, parts 3–7), in one
+ * transaction: release aggregates in the catalog, one chunk per release,
+ * probe candidates, rule counts and samples, edge fragments, and the
+ * range's resume point. A batch that doesn't start at the resume point was
+ * sent twice or too late, and is refused (StaleBatch), so the sums stay right.
+ */
+class BatchStore {
+  async store(rangeId: number, nodeId: number, batch: EdgeBatch): Promise<StoredBatch> {
+    await ChunkPartitions.ensure()
+    return db.transaction(async (trx) => {
+      const range = await trx.from('ranges').where('id', rangeId).forUpdate().first()
+      if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new StaleBatch(`Range ${rangeId} is not leased to this node`)
+      if (Number(range.next_id) !== batch.from) throw new StaleBatch(`Range ${rangeId} is at ${range.next_id}, the batch starts at ${batch.from}`)
+      const result: StoredBatch = { releases: batch.releases.length, newReleases: 0, candidates: 0, fragments: batch.fragments.length }
+
+      // Releases, sorted by key so concurrent batches lock them in the same order.
+      const releases = [...batch.releases].sort((a, b) => a.key.localeCompare(b.key))
+      const ids = new Map<string, number>()
+      for (let i = 0; i < releases.length; i += 1000) {
+        const chunk = releases.slice(i, i + 1000)
+        const { rows } = await trx.rawQuery(
+          `INSERT INTO releases (release_key, name, source, poster, groups, posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2)
+           SELECT v.key, v.name, v.source, v.poster, v.groups, to_timestamp(v.posted_at / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, v.first_article, v.last_article, v.readable, v.has_par2
+           FROM jsonb_to_recordset(?::jsonb) AS v(key text, name text, source text, poster text, groups jsonb, posted_at bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean)
+           ON CONFLICT (release_key) DO UPDATE SET
+             files = releases.files + excluded.files,
+             segments = releases.segments + excluded.segments,
+             bytes = releases.bytes + excluded.bytes,
+             chunks = releases.chunks + 1,
+             posted_at = least(releases.posted_at, excluded.posted_at),
+             file_total = coalesce(releases.file_total, excluded.file_total),
+             first_article = least(releases.first_article, excluded.first_article),
+             last_article = greatest(releases.last_article, excluded.last_article),
+             readable = releases.readable OR excluded.readable,
+             has_par2 = releases.has_par2 OR excluded.has_par2,
+             last_seen_at = now()
+           RETURNING id, release_key, (xmax = 0) AS inserted`,
+          [
+            JSON.stringify(
+              chunk.map((r) => ({
+                key: r.key, name: r.name, source: r.source, poster: r.poster, groups: r.groups, posted_at: r.postedAt, file_total: r.fileTotal,
+                files: r.files, segments: r.segments, bytes: r.bytes, first_article: r.firstArticle, last_article: r.lastArticle, readable: r.readable, has_par2: r.hasPar2,
+              }))
+            ),
+          ]
+        )
+        for (const row of rows) {
+          ids.set(row.release_key, Number(row.id))
+          if (row.inserted) result.newReleases++
+        }
+      }
+
+      // One write-once chunk per release.
+      for (let i = 0; i < releases.length; i += 500) {
+        const rows = releases.slice(i, i + 500).map((r) => ({
+          release_id: ids.get(r.key)!,
+          range_id: rangeId,
+          batch_seq: batch.from,
+          files: r.files,
+          segments: r.segments,
+          payload: r.payload,
+        }))
+        if (rows.length > 0) await trx.insertQuery().table('release_chunks').multiInsert(rows)
+      }
+
+      const candidates = batch.candidates.filter((c) => ids.has(c.releaseKey))
+      for (let i = 0; i < candidates.length; i += 2000) {
+        const rows = candidates.slice(i, i + 2000).map((c) => ({ release_id: ids.get(c.releaseKey)!, kind: c.kind, message_id: c.messageId, file: c.file }))
+        const added = await trx.knexQuery().table('probe_candidates').insert(rows).onConflict(['release_id', 'kind', 'message_id']).ignore().returning('id')
+        result.candidates += added.length
+      }
+
+      if (batch.counts.length > 0) {
+        await trx.rawQuery(
+          `INSERT INTO rule_counts (range_id, rule, key, articles, files)
+           SELECT ?, v.rule, v.key, v.articles, v.files FROM jsonb_to_recordset(?::jsonb) AS v(rule text, key text, articles bigint, files bigint)
+           ON CONFLICT (range_id, rule, key) DO UPDATE SET articles = rule_counts.articles + excluded.articles, files = rule_counts.files + excluded.files`,
+          [rangeId, JSON.stringify(batch.counts)]
+        )
+      }
+
+      for (const s of batch.samples) {
+        await trx.insertQuery().table('rule_samples').insert({
+          range_id: rangeId, rule: s.rule, rule_key: s.ruleKey, release_key: s.key, name: s.name, files: s.files, segments: s.segments, payload: s.payload,
+        })
+      }
+
+      for (let i = 0; i < batch.fragments.length; i += 2000) {
+        const rows = batch.fragments.slice(i, i + 2000).map((f) => ({ range_id: rangeId, file_key: f.fileKey, article_number: f.articleNumber, segment: JSON.stringify(f) }))
+        await trx.insertQuery().table('fragments').multiInsert(rows)
+      }
+
+      await trx
+        .from('ranges')
+        .where('id', rangeId)
+        .update({ next_id: batch.next, headers: db.raw('headers + ?', [batch.headers]), heartbeat_at: new Date() })
+      return result
+    })
+  }
+}
+
+export default new BatchStore()

app/services/chunk_partitions.ts

@@ -0,0 +1,32 @@
+import db from '@adonisjs/lucid/services/db'
+import env from '#start/env'
+
+/** Months ahead that always have a release_chunks partition. */
+const AHEAD = 2
+
+/**
+ * release_chunks is partitioned by month of writing (REBUILD.md, decision a).
+ * Inserts fail without a partition for now(), so the current and the next
+ * months' partitions are created before they are needed; checked once per
+ * process and month.
+ */
+class ChunkPartitions {
+  private ensuredFor: string | null = null
+
+  async ensure(now = new Date()) {
+    const month = now.toISOString().slice(0, 7)
+    if (this.ensuredFor === month) return
+    const tablespace = env.get('CHUNK_TABLESPACE')
+    for (let i = 0; i <= AHEAD; i++) {
+      const from = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + i, 1))
+      const to = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + i + 1, 1))
+      const name = `release_chunks_${from.toISOString().slice(0, 7).replace('-', '')}`
+      await db.rawQuery(
+        `CREATE TABLE IF NOT EXISTS ${name} PARTITION OF release_chunks FOR VALUES FROM ('${from.toISOString()}') TO ('${to.toISOString()}')${tablespace ? ` TABLESPACE ${tablespace}` : ''}`
+      )
+    }
+    this.ensuredFor = month
+  }
+}
+
+export default new ChunkPartitions()

bin/console.ts

@@ -12,7 +12,7 @@
 */
 
 await import('reflect-metadata')
-const { Ignitor, prettyPrintError } = await import('@adonisjs/core')
+const { Ignitor, prettyPrintError } = await import('@adonisjs/core/ignitor')
 
 /**
  * URL to the application root. AdonisJS need it to resolve

bin/server.ts

@@ -10,7 +10,7 @@
 */
 
 await import('reflect-metadata')
-const { Ignitor, prettyPrintError } = await import('@adonisjs/core')
+const { Ignitor, prettyPrintError } = await import('@adonisjs/core/ignitor')
 
 /**
  * URL to the application root. AdonisJS need it to resolve

bin/test.ts

@@ -13,7 +13,7 @@
 process.env.NODE_ENV = 'test'
 
 import 'reflect-metadata'
-import { Ignitor, prettyPrintError } from '@adonisjs/core'
+import { Ignitor, prettyPrintError } from '@adonisjs/core/ignitor'
 import { configure, processCLIArgs, run } from '@japa/runner'
 
 /**

commands/BodyQueuePrune.ts

@@ -1,97 +0,0 @@
-import { BaseCommand, flags } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import indexerConfig from '#config/indexer'
-import QueueService from '#services/QueueService'
-import RedisService from '#services/RedisService'
-import DeferredService, { isDeferred } from '#services/DeferredService'
-
-const BATCH = 2000
-
-/** Rebuilds the wait list (KEYS[1]) without the ids in the set KEYS[2], in one atomic pass. */
-const FILTER_LIST = `
-local ids = redis.call('LRANGE', KEYS[1], 0, -1)
-local keep = {}
-for _, id in ipairs(ids) do
-  if redis.call('SISMEMBER', KEYS[2], id) == 0 then keep[#keep + 1] = id end
-end
-redis.call('DEL', KEYS[1])
-for i = 1, #keep, 5000 do
-  redis.call('RPUSH', KEYS[1], unpack(keep, i, math.min(i + 4999, #keep)))
-end
-return #ids - #keep`
-
-/**
- * Removes waiting body jobs whose message-id is on the skip or defer list,
- * e.g. after adding a pattern (CONCEPT.md D24, D28). Deferred ones are
- * recorded in deferred_files first, so they can still be processed later.
- */
-export default class BodyQueuePrune extends BaseCommand {
-  public static commandName = 'body-queue:prune'
-  public static description = 'Drops waiting body jobs that are now skipped or deferred.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  @flags.boolean({ description: 'Only count' })
-  declare dryRun: boolean
-
-  @flags.boolean({ description: 'Drop deferred jobs without recording them (already recorded by an earlier run)' })
-  declare skipRecord: boolean
-
-  public async run() {
-    const redis = RedisService.client
-    const queue = QueueService.bodyQueue
-    const waitKey = queue.toKey('wait')
-    let skipped = 0
-    let deferred = 0
-    let scanned = 0
-    const dropped: string[] = []
-    try {
-      const ids: string[] = await redis.lrange(waitKey, 0, -1)
-      for (let i = 0; i < ids.length; i += BATCH) {
-        const batch = ids.slice(i, i + BATCH)
-        const data: [Error | null, string | null][] = await batch
-          .reduce((pipeline, id) => pipeline.hget(queue.toKey(id), 'data'), redis.pipeline())
-          .exec()
-        const drop: string[] = []
-        const toDefer = new Map<string, { articleNumber: number; header: Record<string, string> }[]>()
-        batch.forEach((id, j) => {
-          const job = data[j][1] ? JSON.parse(data[j][1]!) : null
-          if (!job?.header) return
-          const messageId = job.header['message-id'] ?? ''
-          if (isDeferred(job.header)) {
-            const list = toDefer.get(job.group) ?? []
-            list.push({ articleNumber: Number(job.articleNumber), header: job.header })
-            toDefer.set(job.group, list)
-            drop.push(id)
-            deferred++
-          } else if (indexerConfig.skipBodyMessageIds.some((re) => re.test(messageId))) {
-            drop.push(id)
-            skipped++
-          }
-        })
-        scanned += batch.length
-        if (this.dryRun || drop.length === 0) continue
-        if (!this.skipRecord) for (const [group, articles] of toDefer) await DeferredService.record(group, articles)
-        dropped.push(...drop)
-      }
-      if (dropped.length > 0) {
-        const dropKey = queue.toKey('prune-ids')
-        await redis.del(dropKey)
-        for (let i = 0; i < dropped.length; i += BATCH) await redis.sadd(dropKey, ...dropped.slice(i, i + BATCH))
-        await redis.eval(FILTER_LIST, 2, waitKey, dropKey)
-        await redis.del(dropKey)
-        for (let i = 0; i < dropped.length; i += BATCH) {
-          await redis.unlink(...dropped.slice(i, i + BATCH).map((id) => queue.toKey(id)))
-        }
-      }
-      this.logger.info(
-        `Scanned ${scanned} waiting body jobs: ${this.dryRun ? 'would drop' : 'dropped'} ${deferred} deferred and ${skipped} skipped.`
-      )
-    } finally {
-      await QueueService.closeAll()
-      await RedisService.close()
-    }
-  }
-}

commands/ContentBackfill.ts

@@ -1,68 +0,0 @@
-import { BaseCommand, flags } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import db from '@adonisjs/lucid/services/db'
-import QueueService from '#services/QueueService'
-import ContentService from '#services/ContentService'
-
-const BATCH = 1000
-
-/**
- * Queues completed files for content inspection (names and groupings from
- * par2 and RAR headers, CONCEPT.md D21) that were completed before it
- * existed. Costs one article download per file, done by worker:collection.
- */
-export default class ContentBackfill extends BaseCommand {
-  public static commandName = 'content:backfill'
-  public static description = 'Queues existing files with random names for content inspection.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  @flags.number({ description: 'Queue at most this many files' })
-  declare limit?: number
-
-  @flags.boolean({ description: 'Also files that were inspected before (after release:rebuild --reset-content)' })
-  declare all: boolean
-
-  @flags.boolean({ description: 'Only archives whose listing contains NZBs, to import those (CONCEPT.md D28)' })
-  declare archiveNzbs: boolean
-
-  public async run() {
-    let lastId = 0
-    let queued = 0
-    try {
-      while (!this.limit || queued < this.limit) {
-        // Files never inspected: no fingerprint on their binary and no archive listing.
-        const query = db
-          .from('files as f')
-          .join('binaries as b', 'b.file_id', 'f.id')
-          .where('f.id', '>', lastId)
-        if (this.archiveNzbs) query.whereRaw(`f.archive::text ~* '\\.nzb"'`)
-        else if (!this.all) query.whereNull('b.md5_16k').whereNull('f.archive')
-        const files = await query
-          .orderBy('f.id')
-          .limit(BATCH)
-          .select('f.id', 'f.filename')
-        if (files.length === 0) break
-        lastId = Number(files.at(-1).id)
-
-        const jobs = files
-          .filter((f) => ContentService.isCandidate(f.filename))
-          .slice(0, this.limit ? this.limit - queued : undefined)
-          .map((f) => ({
-            name: 'process-collection',
-            data: { fileId: Number(f.id) },
-            // A new id per run: BullMQ ignores an add() for an id it still knows.
-            // Targeted backfills go ahead of whatever is waiting.
-            opts: { jobId: `content-${f.id}-${Date.now()}`, lifo: this.archiveNzbs },
-          }))
-        if (jobs.length > 0) await QueueService.collectionQueue.addBulk(jobs)
-        queued += jobs.length
-      }
-      this.logger.info(`Queued ${queued} files for content inspection.`)
-    } finally {
-      await QueueService.closeAll()
-    }
-  }
-}

commands/DeferredExpandRanges.ts

@@ -1,32 +0,0 @@
-import { BaseCommand, flags } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import NntpService from '#services/NntpService'
-import QueueService from '#services/QueueService'
-import DeferredService from '#services/DeferredService'
-
-/**
- * Sends recorded deferred ranges (fully obfuscated ngPost articles, CONCEPT.md
- * D28) to the body queue, oldest first. Every article in them is downloaded:
- * their headers can't be grouped into files.
- */
-export default class DeferredExpandRanges extends BaseCommand {
-  public static commandName = 'deferred:expand-ranges'
-  public static description = 'Queues the bodies of deferred article ranges (costly: every article is downloaded).'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  @flags.number({ description: 'Ranges to expand', default: 10 })
-  declare limit: number
-
-  public async run() {
-    try {
-      const { ranges, articles } = await DeferredService.expandRanges(this.limit)
-      this.logger.info(`Expanded ${ranges} ranges: ${articles} articles sent to the body queue.`)
-    } finally {
-      await NntpService.shutdown()
-      await QueueService.closeAll()
-    }
-  }
-}

commands/GcScheduler.ts

@@ -1,282 +0,0 @@
-import { BaseCommand, flags } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import type { Queue } from 'bullmq'
-import db from '@adonisjs/lucid/services/db'
-import gcConfig from '#config/gc'
-import QueueService from '#services/QueueService'
-import RedisService from '#services/RedisService'
-import Shutdown from '#services/Shutdown'
-
-const CLEAN_BATCH = 1000
-const SCAN_COUNT = 1000
-
-export default class GcScheduler extends BaseCommand {
-  public static commandName = 'gc:scheduler'
-  public static description = 'Periodically removes old BullMQ jobs and stale binaries.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  @flags.boolean({ description: 'Run one sweep and exit' })
-  declare once: boolean
-
-  @flags.boolean({ description: 'Report what would be removed without changing anything' })
-  declare dryRun: boolean
-
-  public async run() {
-    this.logger.info(
-      `GC started (interval ${gcConfig.intervalMs / 1000}s${this.dryRun ? ', dry run' : ''}).`
-    )
-
-    try {
-      do {
-        try {
-          await this.sweep()
-        } catch (err: any) {
-          this.logger.error(`GC sweep failed: ${err.message}`)
-        }
-        if (!this.once) await Shutdown.sleep(gcConfig.intervalMs)
-      } while (!this.once && !Shutdown.requested)
-    } finally {
-      await QueueService.closeAll()
-      await RedisService.close()
-    }
-  }
-
-  private async sweep() {
-    const { completedAgeSeconds, failedAgeSeconds } = gcConfig.jobs
-    for (const queue of [...QueueService.all, ...QueueService.legacy]) {
-      const completed = await this.cleanJobs(queue, 'completed', completedAgeSeconds)
-      const failed = await this.cleanJobs(queue, 'failed', failedAgeSeconds)
-      if (completed || failed) {
-        this.logger.info(
-          `${queue.name}: ${this.verb('remove', 'removed')} ${completed} completed and ${failed} failed jobs.`
-        )
-      }
-    }
-
-    const trimmed = await this.trimBodyQueue()
-    if (trimmed) this.logger.info(`Body queue: ${this.verb('drop', 'dropped')} ${trimmed} oldest waiting jobs.`)
-
-    const binaries = await this.deleteStaleBinaries()
-    this.logger.info(`Binaries: ${this.verb('delete', 'deleted')} ${binaries} stale.`)
-
-    if (!this.dryRun) {
-      const budget = Date.now() + gcConfig.deleteBudgetMs
-      const parts = await this.dropCopiedParts(budget)
-      const single = await this.deleteSingleArticleBinaries(budget)
-      const releases = await this.deleteEmptyReleases(budget)
-      if (parts || single || releases) {
-        this.logger.info(
-          `Space: ${parts} copied parts, ${single} single-article binaries, ${releases} empty releases deleted.`
-        )
-      }
-    }
-
-    const deferred = await this.deleteOldDeferredFiles()
-    if (deferred) this.logger.info(`Deferred files: ${this.verb('delete', 'deleted')} ${deferred} past retention.`)
-
-    const { expired, scanned } = await this.expireLegacyPendingFiles()
-    if (scanned > 0) {
-      this.logger.info(
-        `Legacy pending files: scanned ${scanned}, ${this.verb('set a TTL on', 'set a TTL on')} ${expired} without expiry.`
-      )
-    }
-  }
-
-  /**
-   * Drops the oldest waiting body jobs above gcConfig.bodyQueueMax. BullMQ
-   * workers take jobs from the right end of the wait list, so the oldest are
-   * there.
-   */
-  private async trimBodyQueue(): Promise<number> {
-    const redis = RedisService.client
-    const key = QueueService.bodyQueue.toKey('wait')
-    const excess = (await redis.llen(key)) - gcConfig.bodyQueueMax
-    if (excess <= 0 || this.dryRun) return Math.max(excess, 0)
-    let dropped = 0
-    while (dropped < excess) {
-      const ids: string[] = (await redis.rpop(key, Math.min(5000, excess - dropped))) ?? []
-      if (ids.length === 0) break
-      await redis.unlink(...ids.map((id) => QueueService.bodyQueue.toKey(id)))
-      dropped += ids.length
-    }
-    return dropped
-  }
-
-  /**
-   * Deletes binaries (and, by cascade, their parts) that received no segment
-   * for `staleAfterSeconds`: incomplete ones are given up on; completed ones
-   * already live on in `files`; conflicted ones will never become a file.
-   * Completed ones still waiting for worker:file are kept.
-   */
-  private async deleteStaleBinaries(): Promise<number> {
-    const query = db
-      .from('binaries')
-      .where('last_seen_at', '<', new Date(Date.now() - gcConfig.binaries.staleAfterSeconds * 1000))
-      .where((q) =>
-        q.whereNull('completed_at').orWhereNotNull('file_id').orWhereNotNull('conflicted_at')
-      )
-
-    if (this.dryRun) {
-      const [{ count }] = await query.count('* as count')
-      return Number(count)
-    }
-    return Number(await query.delete())
-  }
-
-  private async deleteOldDeferredFiles(): Promise<number> {
-    const cutoff = new Date(Date.now() - gcConfig.deferredRetentionDays * 24 * 60 * 60 * 1000)
-    let total = 0
-    for (const [table, column] of [['deferred_files', 'first_seen_at'], ['deferred_ranges', 'created_at']]) {
-      const query = db.from(table).where(column, '<', cutoff)
-      if (this.dryRun) {
-        const [{ count }] = await query.count('* as count')
-        total += Number(count)
-      } else {
-        total += Number(await query.delete())
-      }
-    }
-    return total
-  }
-
-  // Keyset cursors over the tables, kept between sweeps so one pass over
-  // millions of rows is spread over several sweeps.
-  private singleCursor = 0
-  private releaseCursor = 0
-
-  /** Parts of header-derived binaries that are a file already (worker:file now drops them itself). */
-  private async dropCopiedParts(budget: number): Promise<number> {
-    let total = 0
-    while (Date.now() < budget) {
-      const result = await db.rawQuery(
-        `DELETE FROM parts WHERE binary_id IN (
-           SELECT b.id FROM binaries b
-           WHERE b.file_id IS NOT NULL AND b.size IS NULL
-             AND EXISTS (SELECT 1 FROM parts p WHERE p.binary_id = b.id)
-           LIMIT ?)`,
-        [500]
-      )
-      total += result.rowCount ?? 0
-      if (!result.rowCount) break
-    }
-    return total
-  }
-
-  /** Header-derived binaries that never got a second article (gcConfig.binaries.singleArticleAfterSeconds). */
-  private async deleteSingleArticleBinaries(budget: number): Promise<number> {
-    const cutoff = new Date(Date.now() - gcConfig.binaries.singleArticleAfterSeconds * 1000)
-    let total = 0
-    while (Date.now() < budget) {
-      const { rows } = await db.rawQuery(
-        `WITH batch AS (
-           SELECT id FROM binaries WHERE id > ? ORDER BY id LIMIT ?
-         ), gone AS (
-           DELETE FROM binaries b USING batch
-           WHERE b.id = batch.id AND b.completed_at IS NULL AND b.size IS NULL AND b.total_parts > 1
-             AND b.last_seen_at < ? AND (SELECT count(*) FROM parts p WHERE p.binary_id = b.id) <= 1
-           RETURNING 1)
-         SELECT (SELECT max(id) FROM batch) AS last, (SELECT count(*) FROM gone) AS deleted`,
-        [this.singleCursor, gcConfig.deleteBatch, cutoff]
-      )
-      total += Number(rows[0].deleted)
-      if (rows[0].last === null) {
-        this.singleCursor = 0
-        break
-      }
-      this.singleCursor = Number(rows[0].last)
-    }
-    return total
-  }
-
-  /** Releases nothing refers to any more: no files, binaries, par2 listing or spot. */
-  private async deleteEmptyReleases(budget: number): Promise<number> {
-    const cutoff = new Date(Date.now() - gcConfig.emptyReleaseAfterSeconds * 1000)
-    let total = 0
-    while (Date.now() < budget) {
-      const { rows } = await db.rawQuery(
-        `WITH batch AS (
-           SELECT id FROM releases WHERE id > ? ORDER BY id LIMIT ?
-         ), gone AS (
-           DELETE FROM releases r USING batch
-           WHERE r.id = batch.id AND r.file_count = 0 AND r.first_seen_at < ?
-             AND NOT EXISTS (SELECT 1 FROM binaries b WHERE b.release_id = r.id)
-             AND NOT EXISTS (SELECT 1 FROM files f WHERE f.release_id = r.id)
-             AND NOT EXISTS (SELECT 1 FROM release_files rf WHERE rf.release_id = r.id)
-             AND NOT EXISTS (SELECT 1 FROM spots s WHERE s.release_id = r.id)
-           RETURNING 1)
-         SELECT (SELECT max(id) FROM batch) AS last, (SELECT count(*) FROM gone) AS deleted`,
-        [this.releaseCursor, gcConfig.deleteBatch, cutoff]
-      )
-      total += Number(rows[0].deleted)
-      if (rows[0].last === null) {
-        this.releaseCursor = 0
-        break
-      }
-      this.releaseCursor = Number(rows[0].last)
-    }
-    return total
-  }
-
-  /** Removes jobs in `state` that finished more than `ageSeconds` ago. */
-  private async cleanJobs(queue: Queue, state: 'completed' | 'failed', ageSeconds: number) {
-    const graceMs = ageSeconds * 1000
-    if (this.dryRun) {
-      const cutoff = Date.now() - graceMs
-      // Finished-job sets are sorted by finishedOn, so a score range counts them cheaply.
-      return RedisService.client.zcount(queue.toKey(state), '-inf', cutoff)
-    }
-
-    let total = 0
-    let removed: string[]
-    do {
-      removed = await queue.clean(graceMs, CLEAN_BATCH, state)
-      total += removed.length
-    } while (removed.length === CLEAN_BATCH)
-    return total
-  }
-
-  /**
-   * Pending-file hashes of the old header worker (and the pre-AdonisJS app)
-   * never expire. Give them a TTL rather than deleting them outright.
-   */
-  private async expireLegacyPendingFiles() {
-    const redis = RedisService.client
-    const { prefix, ttlSeconds } = gcConfig.legacyPendingFiles
-    let cursor = '0'
-    let scanned = 0
-    let expired = 0
-
-    do {
-      const [next, keys]: [string, string[]] = await redis.scan(
-        cursor,
-        'MATCH',
-        `${prefix}*`,
-        'COUNT',
-        SCAN_COUNT
-      )
-      cursor = next
-      if (keys.length === 0) continue
-      scanned += keys.length
-
-      const ttls: [Error | null, number][] = await keys
-        .reduce((pipeline, key) => pipeline.ttl(key), redis.pipeline())
-        .exec()
-      const orphaned = keys.filter((_, i) => ttls[i][1] === -1)
-      expired += orphaned.length
-
-      if (!this.dryRun && orphaned.length > 0) {
-        await orphaned
-          .reduce((pipeline, key) => pipeline.expire(key, ttlSeconds, 'NX'), redis.pipeline())
-          .exec()
-      }
-    } while (cursor !== '0')
-
-    return { scanned, expired }
-  }
-
-  private verb(present: string, past: string) {
-    return this.dryRun ? `would ${present}` : past
-  }
-}

commands/GroupAdd.ts

@@ -1,72 +0,0 @@
-import { BaseCommand, args, flags } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import Group from '#models/group'
-import NntpService from '#services/NntpService'
-import { parseGroupResponse } from '#lib/nntp/group'
-
-/**
- * Adds newsgroups to index (or reactivates them). The scheduler picks them
- * up on its next tick, starting INDEX_INITIAL_BACKFILL articles back.
- */
-export default class GroupAdd extends BaseCommand {
-  public static commandName = 'group:add'
-  public static description = 'Adds newsgroups to index.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  @args.spread({ description: 'Group names, e.g. alt.binaries.hdtv' })
-  declare names: string[]
-
-  @flags.boolean({ description: 'Deactivate the groups instead' })
-  declare disable: boolean
-
-  @flags.boolean({
-    description: 'Also download bodies to identify obfuscated posts (one body per article; bandwidth-heavy)',
-  })
-  declare bodies: boolean
-
-  @flags.number({ description: 'Start this many articles back instead of INDEX_INITIAL_BACKFILL (new groups only)' })
-  declare backfill?: number
-
-  @flags.boolean({ description: 'With --backfill: also for groups indexed before, skipping what was missed since' })
-  declare reset: boolean
-
-  public async run() {
-    for (const name of this.names) {
-      if (!/^[a-z0-9][a-z0-9.+_-]*$/i.test(name)) {
-        this.logger.error(`Not a valid group name: ${name}`)
-        continue
-      }
-      const group = await Group.firstOrCreate({ name }, { name, active: !this.disable, fetchBodies: this.bodies })
-      group.active = !this.disable
-      if (this.bodies) group.fetchBodies = true
-      if (this.backfill !== undefined && (this.reset || BigInt(group.lastQueuedId ?? 0) === 0n)) {
-        // The scheduler continues after last_queued_id.
-        const last = await this.lastArticle(name)
-        group.lastQueuedId = (last > BigInt(this.backfill) ? last - BigInt(this.backfill) : 0n)
-        group.lastIndexedId = group.lastQueuedId
-      }
-      await group.save()
-      this.logger.info(
-        `${name}: ${group.active ? 'active' : 'inactive'}, bodies ${group.fetchBodies ? 'on' : 'off'}` +
-          (this.backfill !== undefined ? `, starting after article ${group.lastQueuedId}` : '')
-      )
-    }
-    if (this.backfill !== undefined) await NntpService.shutdown()
-  }
-
-  private async lastArticle(name: string): Promise<bigint> {
-    const conn = await NntpService.acquire()
-    let failed = false
-    try {
-      return parseGroupResponse((await conn.group(name)).resp).last
-    } catch (error) {
-      failed = true
-      throw error
-    } finally {
-      NntpService.release(conn, failed)
-    }
-  }
-}

commands/GroupsTop.ts

@@ -1,176 +0,0 @@
-import { writeFile } from 'node:fs/promises'
-import { setTimeout as sleep } from 'node:timers/promises'
-import { BaseCommand, flags } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import NntpService from '#services/NntpService'
-import indexerConfig from '#config/indexer'
-import { parseGroupResponse } from '#lib/nntp/group'
-import { parseSubject } from '#lib/subject/index'
-
-interface GroupRate {
-  name: string
-  perMinute: number
-  sampled?: number
-  normal?: number
-  obfuscated?: number
-  skipped?: number
-  avgArticleBytes?: number
-  avgHeaderBytes?: number
-}
-
-/**
- * Lists the busiest newsgroups (by articles posted per minute, from two
- * LIST ACTIVE snapshots) and optionally samples their newest headers to
- * estimate what indexing them would cost. See LOG.md (2026-09-28 (14)).
- *
- *   node ace groups:top --minutes 5 --limit 1000 --sample 300 --out /data/top-groups.txt
- */
-export default class GroupsTop extends BaseCommand {
-  public static commandName = 'groups:top'
-  public static description = 'Lists the busiest binary groups and estimates the bandwidth to index them.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  @flags.string({ description: 'Group pattern for LIST ACTIVE', default: 'alt.binaries.*' })
-  declare pattern: string
-
-  @flags.number({ description: 'Minutes between the two snapshots', default: 5 })
-  declare minutes: number
-
-  @flags.number({ description: 'Number of groups to list', default: 1000 })
-  declare limit: number
-
-  @flags.number({ description: 'Headers to sample per listed group (0 = no sampling)', default: 0 })
-  declare sample: number
-
-  @flags.string({ description: 'Write the list to this file (default: print only)' })
-  declare out?: string
-
-  public async run() {
-    const conn = await NntpService.acquire()
-    let failed = false
-    try {
-      this.logger.info(`Snapshot 1 of ${this.pattern}…`)
-      const first = await this.snapshot(conn)
-      this.logger.info(`${first.size} groups. Waiting ${this.minutes} min for snapshot 2…`)
-      const started = Date.now()
-      await sleep(this.minutes * 60_000)
-      const second = await this.snapshot(conn)
-      const minutes = (Date.now() - started) / 60_000
-
-      const groups: GroupRate[] = []
-      for (const [name, high] of second) {
-        const before = first.get(name)
-        if (before === undefined || high <= before) continue
-        groups.push({ name, perMinute: Number(high - before) / minutes })
-      }
-      groups.sort((a, b) => b.perMinute - a.perMinute)
-      const top = groups.slice(0, this.limit)
-      this.logger.info(`${groups.length} active groups; listing ${top.length}.`)
-
-      if (this.sample > 0) {
-        for (const [i, group] of top.entries()) {
-          try {
-            Object.assign(group, await this.sampleGroup(conn, group.name))
-          } catch (error: any) {
-            this.logger.warning(`${group.name}: ${error.message}`)
-          }
-          if ((i + 1) % 50 === 0) this.logger.info(`Sampled ${i + 1}/${top.length} groups…`)
-        }
-      }
-
-      const text = this.format(top, minutes)
-      if (this.out) {
-        await writeFile(this.out, text)
-        this.logger.info(`Written to ${this.out}`)
-      } else {
-        console.log(text)
-      }
-    } catch (error) {
-      failed = true
-      throw error
-    } finally {
-      NntpService.release(conn, failed)
-      await NntpService.shutdown()
-    }
-  }
-
-  /** Group name → highest article number, from LIST ACTIVE. */
-  private async snapshot(conn: any): Promise<Map<string, bigint>> {
-    const { data } = await conn.list(this.pattern)
-    const result = new Map<string, bigint>()
-    for (const line of Buffer.from(data).toString('latin1').split('\r\n')) {
-      const [name, high] = line.split(' ')
-      if (name && /^\d+$/.test(high ?? '')) result.set(name, BigInt(high))
-    }
-    return result
-  }
-
-  /** Classifies the newest headers the way worker:fetch does. */
-  private async sampleGroup(conn: any, name: string) {
-    const { last } = parseGroupResponse((await conn.group(name)).resp)
-    const start = last - BigInt(this.sample) + 1n
-    const { overviews } = await conn.xover((start > 0n ? start : 1n).toString(), last.toString())
-    let normal = 0
-    let skipped = 0
-    let obfuscated = 0
-    let obfuscatedBytes = 0
-    let headerBytes = 0
-    for (const [, header] of overviews as [number, Record<string, string>][]) {
-      if (!header) continue
-      headerBytes += Object.values(header).reduce((sum, v) => sum + String(v).length + 1, 0)
-      if (header.subject && parseSubject(header.subject)) normal++
-      else if (indexerConfig.skipBodyMessageIds.some((re) => re.test(header['message-id'] ?? ''))) skipped++
-      else {
-        obfuscated++
-        obfuscatedBytes += Number(header[':bytes']) || 0
-      }
-    }
-    const sampled = normal + skipped + obfuscated
-    return {
-      sampled,
-      normal,
-      skipped,
-      obfuscated,
-      avgArticleBytes: obfuscated ? obfuscatedBytes / obfuscated : 0,
-      avgHeaderBytes: sampled ? headerBytes / sampled : 0,
-    }
-  }
-
-  private format(groups: GroupRate[], minutes: number): string {
-    const sampled = groups.some((g) => g.sampled)
-    const lines = [
-      `# Busiest ${this.pattern} groups by articles/min, measured over ${minutes.toFixed(1)} min on ${new Date().toISOString()}`,
-      sampled
-        ? '# rank\tgroup\tarticles/min\tobfuscated%\tskipped%\tbody Mbit/s\theader Mbit/s'
-        : '# rank\tgroup\tarticles/min',
-    ]
-    let bodyTotal = 0
-    let headerTotal = 0
-    let articlesTotal = 0
-    for (const [i, g] of groups.entries()) {
-      articlesTotal += g.perMinute
-      if (!g.sampled) {
-        lines.push(`${i + 1}\t${g.name}\t${Math.round(g.perMinute)}`)
-        continue
-      }
-      const obfuscatedShare = g.obfuscated! / g.sampled
-      // Bodies: every obfuscated article is downloaded once. Headers: every article.
-      const bodyMbit = (g.perMinute * obfuscatedShare * g.avgArticleBytes! * 8) / 60 / 1e6
-      const headerMbit = (g.perMinute * g.avgHeaderBytes! * 8) / 60 / 1e6
-      bodyTotal += bodyMbit
-      headerTotal += headerMbit
-      lines.push(
-        [i + 1, g.name, Math.round(g.perMinute), (100 * obfuscatedShare).toFixed(1), ((100 * g.skipped!) / g.sampled).toFixed(1), bodyMbit.toFixed(1), headerMbit.toFixed(2)].join('\t')
-      )
-    }
-    lines.push(`# total: ${Math.round(articlesTotal)} articles/min`)
-    if (sampled) {
-      lines.push(`# bandwidth to index like alt.binaries.test (headers + obfuscated bodies): ${(bodyTotal + headerTotal).toFixed(0)} Mbit/s`)
-      lines.push(`#   of which bodies ${bodyTotal.toFixed(0)} Mbit/s, headers ${headerTotal.toFixed(1)} Mbit/s (skip list applied)`)
-    }
-    return lines.join('\n') + '\n'
-  }
-}

commands/IndexScheduler.ts

@@ -1,119 +0,0 @@
-import { BaseCommand } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import Group from '#models/group'
-import NntpService from '#services/NntpService'
-import QueueService from '#services/QueueService'
-import indexerConfig from '#config/indexer'
-import Shutdown from '#services/Shutdown'
-import FetchRangeService from '#services/FetchRangeService'
-import SplitService from '#services/SplitService'
-
-export default class IndexScheduler extends BaseCommand {
-  public static commandName = 'index:scheduler'
-  public static description = 'Periodically checks for new articles and schedules them for fetching.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  private pool = NntpService
-
-  public async run() {
-    this.logger.info('Scheduler started. Awaiting tasks...')
-
-    const schedule = async () => {
-      this.logger.info('Checking for new headers...')
-      const groups = await Group.query().where('active', true)
-
-      if (groups.length === 0) {
-        this.logger.info('No active groups to index. Add some via `node ace db:seed` or manually.')
-        return
-      }
-
-      let conn
-      let failed = false
-      try {
-        conn = await this.pool.acquire()
-
-        for (const group of groups) {
-          try {
-            const { first: firstArticle, last: lastArticle } = await this.pool.selectGroup(
-              conn,
-              group.name
-            )
-
-            await FetchRangeService.requeue(group)
-
-            // Postgres returns bigint columns as strings.
-            const lastQueued = BigInt(group.lastQueuedId ?? 0)
-            const { batchSize, maxBatchesPerTick, initialBackfill } = indexerConfig
-            let startId =
-              lastQueued === 0n ? lastArticle - initialBackfill + 1n : lastQueued + 1n
-            if (startId < firstArticle) startId = firstArticle
-
-            if (startId > lastArticle) {
-              await FetchRangeService.advanceWatermark(group)
-              this.logger.info(`No new headers for group ${group.name}.`)
-              continue
-            }
-
-            let batches = 0
-            let endId = startId - 1n
-            while (endId < lastArticle && batches < maxBatchesPerTick) {
-              const batchStart = endId + 1n
-              endId = batchStart + batchSize - 1n < lastArticle ? batchStart + batchSize - 1n : lastArticle
-              await FetchRangeService.queue(group, batchStart, endId)
-              batches++
-            }
-
-            this.logger.info(
-              `Queued ${batches} fetch job(s) for ${group.name}: articles ${startId} to ${endId}` +
-                (endId < lastArticle ? ` (${lastArticle - endId} left for next tick)` : '')
-            )
-
-            group.lastQueuedId = endId
-            await group.save()
-            const watermark = await FetchRangeService.advanceWatermark(group)
-            if (endId - watermark > batchSize * BigInt(maxBatchesPerTick) * 3n) {
-              this.logger.warning(
-                `${group.name}: indexed up to ${watermark}, ${endId - watermark} articles behind the queue.`
-              )
-            }
-          } catch (err: any) {
-            failed = true
-            this.logger.error(`Error processing group ${group.name}: ${err.message}`)
-          }
-        }
-      } catch (err: any) {
-        failed = true
-        this.logger.error(`Error in scheduler main loop: ${err.message}`)
-      } finally {
-        if (conn) {
-          this.pool.release(conn, failed)
-        }
-      }
-    }
-
-    const split = async () => {
-      try {
-        const result = await SplitService.splitPending()
-        if (result.parents > 0) {
-          this.logger.info(
-            `Split ${result.parents} conflicted binaries: ${result.volumes} verified volumes, ${result.completed} completed.`
-          )
-        }
-      } catch (err: any) {
-        this.logger.error(`Error splitting conflicted binaries: ${err.message}`)
-      }
-    }
-
-    // Run immediately, then wait a full interval after each run so runs never overlap.
-    while (!Shutdown.requested) {
-      await schedule()
-      await split()
-      await Shutdown.sleep(indexerConfig.intervalMs)
-    }
-    await this.pool.shutdown()
-    await QueueService.closeAll()
-  }
-}

commands/ReleaseRebuild.ts

@@ -1,163 +0,0 @@
-import { BaseCommand, flags } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import db from '@adonisjs/lucid/services/db'
-import IngestService from '#services/IngestService'
-import NameService from '#services/NameService'
-import QueueService from '#services/QueueService'
-import { rarReleaseIdentity, releaseIdentity } from '#lib/release/index'
-
-const BATCH = 5000
-
-/**
- * Recomputes the release of every binary (and its file) with the current
- * rules in lib/release. Used after changing those rules.
- */
-export default class ReleaseRebuild extends BaseCommand {
-  public static commandName = 'release:rebuild'
-  public static description = 'Reassigns all binaries and files to releases using the current grouping rules.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  @flags.boolean({
-    description:
-      'Also drop everything derived from file contents (par2 lists, par2/RAR groupings, real names), ' +
-      'to rebuild it with content:backfill --all',
-  })
-  declare resetContent: boolean
-
-  public async run() {
-    if (this.resetContent) await this.dropContentDerived()
-
-    let lastId = 0
-    let assigned = 0
-    let unassigned = 0
-    try {
-      while (true) {
-        const binaries = await db
-          .from('binaries')
-          .where('id', '>', lastId)
-          .orderBy('id')
-          .limit(BATCH)
-          .select('id', 'poster', 'subject', 'name', 'groups', 'posted_at')
-        if (binaries.length === 0) break
-        lastId = Number(binaries.at(-1).id)
-
-        const segments = binaries.map((binary) => {
-          // Body-derived binaries store a synthetic subject, see segmentFromYenc.
-          const weak = /^"[^"]*" yEnc$/.test(binary.subject)
-          return {
-            id: Number(binary.id),
-            poster: binary.poster,
-            groups: binary.groups,
-            postedAt: binary.posted_at ? new Date(binary.posted_at) : undefined,
-            weakIdentity: weak,
-            release: releaseIdentity({ poster: binary.poster, subject: binary.subject, name: binary.name, weak }),
-          }
-        })
-
-        await db.transaction(async (trx) => {
-          const releaseIds = await IngestService.upsertReleases(trx.knexClient, segments)
-          const ids = segments.map((s) => s.id)
-          const releaseIdsPerBinary = segments.map((s) =>
-            s.release ? releaseIds.get(s.release.key)! : null
-          )
-          await trx.rawQuery(
-            `UPDATE binaries b SET release_id = v.release_id
-             FROM unnest(?::bigint[], ?::bigint[]) AS v(id, release_id)
-             WHERE b.id = v.id AND b.release_id IS DISTINCT FROM v.release_id
-               AND NOT (v.release_id IS NULL
-                        AND b.release_id IN (SELECT id FROM releases WHERE source IN ('par2', 'rar')))
-               -- A posted NZB decided this grouping (NzbImportService); name rules don't override it.
-               AND NOT EXISTS (SELECT 1 FROM files n WHERE n.release_id = b.release_id AND n.content_type = 'nzb')`,
-            [ids, releaseIdsPerBinary] as any
-          )
-        })
-        assigned += segments.filter((s) => s.release).length
-        unassigned += segments.filter((s) => !s.release).length
-      }
-
-      // Files grouped by their RAR headers: regroup from the stored listing (no downloads).
-      let rarGrouped = 0
-      let rarCursor = 0
-      while (true) {
-        const rows = await db
-          .from('files as f')
-          .join('binaries as b', 'b.file_id', 'f.id')
-          .whereNotNull('f.archive')
-          .whereNull('b.release_id')
-          .where('f.id', '>', rarCursor)
-          .orderBy('f.id')
-          .limit(BATCH)
-          .select('f.id as file_id', 'f.archive', 'b.id as binary_id', 'b.poster', 'b.groups', 'b.posted_at')
-        if (rows.length === 0) break
-        rarCursor = Number(rows.at(-1).file_id)
-        const segments = rows.map((r) => ({
-          fileId: Number(r.file_id),
-          binaryId: Number(r.binary_id),
-          poster: r.poster,
-          groups: r.groups,
-          postedAt: r.posted_at ? new Date(r.posted_at) : undefined,
-          weakIdentity: true,
-          release: rarReleaseIdentity(r.archive?.entries ?? []),
-        }))
-        const ids = await IngestService.upsertReleases(db.connection().getWriteClient(), segments)
-        for (const s of segments.filter((x) => x.release)) {
-          const releaseId = ids.get(s.release!.key)
-          await db.from('binaries').where('id', s.binaryId).update({ release_id: releaseId })
-          await db.from('files').where('id', s.fileId).update({ release_id: releaseId })
-          rarGrouped++
-        }
-      }
-
-      // Real names from par2 for binaries with a known size (no downloads).
-      let named = 0
-      let nameCursor = 0
-      while (true) {
-        const rows = await db.from('binaries').whereNotNull('size').where('id', '>', nameCursor).orderBy('id').limit(BATCH).select('id')
-        if (rows.length === 0) break
-        nameCursor = Number(rows.at(-1).id)
-        named += await NameService.resolve(rows.map((r) => Number(r.id)))
-      }
-
-      const files = await db.rawQuery(
-        `UPDATE files f SET release_id = b.release_id FROM binaries b
-         WHERE b.file_id = f.id AND f.release_id IS DISTINCT FROM b.release_id`
-      )
-      const orphans = await db.rawQuery(
-        `DELETE FROM releases r WHERE NOT EXISTS (SELECT 1 FROM binaries b WHERE b.release_id = r.id)
-           AND NOT EXISTS (SELECT 1 FROM files f WHERE f.release_id = r.id)`
-      )
-      if (this.resetContent) {
-        this.logger.info('Content-derived data dropped; run `node ace content:backfill --all` to rebuild it.')
-      }
-      this.logger.info(
-        `Binaries: ${assigned} in a release, ${unassigned} without, ${rarGrouped} regrouped from RAR headers, ` +
-          `${named} named from par2. ` +
-          `Files updated: ${files.rowCount}. Empty releases removed: ${orphans.rowCount}.`
-      )
-    } finally {
-      await QueueService.closeAll()
-    }
-  }
-
-  /** par2 lists, par2/RAR release assignments and recovered names. */
-  private async dropContentDerived() {
-    await db.transaction(async (trx) => {
-      await trx.rawQuery('DELETE FROM release_files')
-      await trx.rawQuery(`UPDATE binaries SET real_name = NULL WHERE real_name IS NOT NULL`)
-      await trx.rawQuery(`UPDATE files SET real_name = NULL, archive = NULL WHERE real_name IS NOT NULL OR archive IS NOT NULL`)
-      await trx.rawQuery(`UPDATE releases SET contents = NULL, encrypted = NULL WHERE contents IS NOT NULL OR encrypted IS NOT NULL`)
-      // Binaries whose name rules find no release were grouped from contents (or wrongly by size).
-      await trx.rawQuery(`
-        UPDATE binaries SET release_id = NULL
-        WHERE subject ~ '^"[^"]*" yEnc$' AND release_id IS NOT NULL
-          AND (name !~ '\\.[A-Za-z0-9]{1,5}$' OR release_id IN (SELECT id FROM releases WHERE source IN ('par2', 'rar')))`)
-      await trx.rawQuery(`
-        UPDATE files f SET release_id = b.release_id FROM binaries b
-        WHERE b.file_id = f.id AND f.release_id IS DISTINCT FROM b.release_id`)
-    })
-    this.logger.info('Dropped par2 lists, content-derived groupings and recovered names.')
-  }
-}

commands/ReleaseRename.ts

@@ -1,20 +0,0 @@
-import { BaseCommand } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import { renameFromStoredContents } from '#services/ArchiveService'
-import { regroupAndRenameFromPar2 } from '#services/NameService'
-
-/** Renames random-named releases from their stored archive listings (no downloads). */
-export default class ReleaseRename extends BaseCommand {
-  public static commandName = 'release:rename'
-  public static description = 'Renames random-named releases from their par2 listings and archive contents.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  public async run() {
-    const par2 = await regroupAndRenameFromPar2()
-    this.logger.info(`par2: ${par2.regrouped} files moved into the set that names them, ${par2.renamed} releases renamed.`)
-    this.logger.info(`Renamed ${await renameFromStoredContents()} releases from their archive contents.`)
-  }
-}

commands/ReleaseStats.ts

@@ -1,57 +0,0 @@
-import { BaseCommand } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import db from '@adonisjs/lucid/services/db'
-
-/** Prints how well files are being grouped into releases. */
-export default class ReleaseStats extends BaseCommand {
-  public static commandName = 'release:stats'
-  public static description = 'Shows release recognition statistics.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  public async run() {
-    const { rows } = await db.rawQuery(`
-      WITH r AS (
-        SELECT r.id, r.file_total, r.poster IS NULL AS weak, r.source,
-               count(b.id) AS binaries,
-               count(b.id) FILTER (WHERE b.completed_at IS NOT NULL) AS completed
-        FROM releases r LEFT JOIN binaries b ON b.release_id = r.id
-        GROUP BY r.id
-      )
-      SELECT
-        count(*) AS releases,
-        count(*) FILTER (WHERE completed > 0) AS with_files,
-        count(*) FILTER (WHERE completed >= 2) AS multi_file,
-        count(*) FILTER (WHERE file_total IS NOT NULL AND completed >= file_total) AS complete_by_counter,
-        count(*) FILTER (WHERE weak) AS from_obfuscated,
-        count(*) FILTER (WHERE weak AND completed > 0) AS from_obfuscated_with_files
-      FROM r`)
-    const { rows: coverage } = await db.rawQuery(`
-      SELECT count(*) AS completed_files,
-             count(*) FILTER (WHERE release_id IS NOT NULL) AS in_release
-      FROM binaries WHERE completed_at IS NOT NULL`)
-    const { rows: bySource } = await db.rawQuery(`
-      SELECT coalesce(r.source, 'unknown') AS source, count(DISTINCT r.id) AS releases,
-             count(b.id) FILTER (WHERE b.completed_at IS NOT NULL) AS completed_files
-      FROM releases r JOIN binaries b ON b.release_id = r.id GROUP BY 1 ORDER BY 2 DESC`)
-    // Readable names: descriptive as posted, or recovered from contents.
-    const { rows: naming } = await db.rawQuery(`
-      WITH f AS (
-        SELECT real_name IS NOT NULL AS recovered,
-               is_readable_name(coalesce(real_name, filename)) AS readable_name,
-               regexp_replace(coalesce(real_name, filename),
-                 '(\\.(part\\d+\\.rar|r\\d{2,3}|rar|vol\\d+[+-]\\d+\\.par2|par2|7z(\\.\\d{3})?|zip|\\d{3}|nfo|sfv|nzb|tar|zst|mkv|mp4|avi|flac|mp3|jpg))+$', '', 'i') AS stem,
-               coalesce(real_name, filename) AS name
-        FROM files
-      )
-      SELECT count(*) AS files,
-             count(*) FILTER (WHERE recovered) AS names_recovered,
-             count(*) FILTER (WHERE name ~ '\\.[A-Za-z0-9]{1,5}$' AND readable_name) AS readable
-      FROM f`)
-    this.logger.info(JSON.stringify({ ...rows[0], ...coverage[0] }))
-    this.logger.info(`naming: ${JSON.stringify(naming[0])}`)
-    this.logger.info(`by source: ${JSON.stringify(bySource)}`)
-  }
-}

commands/edge_bench.ts

@@ -0,0 +1,130 @@
+import { BaseCommand, flags } from '@adonisjs/core/ace'
+import type { CommandOptions } from '@adonisjs/core/types/ace'
+import { mkdir, readFile, readdir, writeFile } from 'node:fs/promises'
+import { join } from 'node:path'
+import { unzipSync } from 'node:zlib'
+import env from '#start/env'
+import { XoverClient, parseOverviews } from '#lib/nntp/index'
+import { RangeIndexer } from '#lib/edge/index'
+import { DEFAULT_RULES, type Rules } from '#lib/rules/index'
+
+/**
+ * Measures the edge worker on a real range (REBUILD.md: TypeScript or Rust):
+ * --capture saves a range's compressed XOVER chunks to a directory once;
+ * without it the saved chunks are replayed through the range indexer, timing
+ * inflate, parse and indexing per chunk, with memory and output sizes.
+ */
+export default class EdgeBench extends BaseCommand {
+  static commandName = 'edge:bench'
+  static description = 'Measures the edge worker on a captured range'
+  static options: CommandOptions = { startApp: true }
+
+  @flags.string({ description: 'Directory of the captured chunks', required: true })
+  declare dir: string
+
+  @flags.boolean({ description: 'Capture the range first (needs NNTP_* in the environment)' })
+  declare capture: boolean
+
+  @flags.string({ description: 'Group to capture' })
+  declare group?: string
+
+  @flags.number({ description: 'First article to capture' })
+  declare start?: number
+
+  @flags.number({ description: 'Articles to capture', default: 2_000_000 })
+  declare articles: number
+
+  @flags.number({ description: 'Articles per XOVER', default: 50_000 })
+  declare chunk: number
+
+  @flags.number({ description: 'Chunks without a new part before an open file is given up', default: 5 })
+  declare idle: number
+
+  @flags.string({ description: 'Indexed groups in priority order, comma-separated (owner rule)' })
+  declare priorities?: string
+
+  async run() {
+    if (this.capture) await this.captureRange()
+    await this.replay()
+  }
+
+  private async captureRange() {
+    if (!this.group || this.start === undefined) throw new Error('--group and --start are needed to capture')
+    await mkdir(this.dir, { recursive: true })
+    const client = await XoverClient.connect({
+      host: env.get('NNTP_HOST')!,
+      port: env.get('NNTP_PORT')!,
+      user: env.get('NNTP_USER'),
+      password: env.get('NNTP_PASSWORD'),
+      secure: env.get('NNTP_SECURE', true),
+      compress: true,
+      timeoutMs: 300_000,
+    })
+    await client.group(this.group)
+    const end = this.start + this.articles - 1
+    let index = 0
+    for (let from = this.start; from <= end; from += this.chunk, index++) {
+      const to = Math.min(from + this.chunk - 1, end)
+      const started = Date.now()
+      const raw = await client.xoverRaw(from, to)
+      await writeFile(join(this.dir, `${String(index).padStart(4, '0')}-${from}-${to}.gz`), raw)
+      this.logger.info(`chunk ${index}: ${from}–${to}, ${(raw.length / 1e6).toFixed(1)} MB in ${((Date.now() - started) / 1000).toFixed(1)} s`)
+    }
+    await writeFile(join(this.dir, 'range.json'), JSON.stringify({ group: this.group, start: this.start, end }))
+    client.close()
+  }
+
+  private async replay() {
+    const { group, start, end } = JSON.parse(await readFile(join(this.dir, 'range.json'), 'utf8')) as { group: string; start: number; end: number }
+    const priorities = Object.fromEntries((this.priorities ?? group).split(',').map((g, i) => [g.trim(), i + 1]))
+    if (!(group in priorities)) priorities[group] = Object.keys(priorities).length + 1
+    const rules: Rules = { ...DEFAULT_RULES, priorities, droppedSignatures: [] }
+    const files = (await readdir(this.dir)).filter((f) => f.endsWith('.gz')).sort()
+    const indexer = new RangeIndexer({ group, start, end, rules, idleChunks: this.idle })
+    const t = { inflate: 0, parse: 0, index: 0 }
+    let compressed = 0
+    let text = 0
+    let peakRss = 0
+    const out = { headers: 0, releases: 0, files: 0, segments: 0, payload: 0, candidates: 0, fragments: 0, samples: 0 }
+    const counts = new Map<string, number>()
+    const cpuStart = process.cpuUsage()
+    const wall = Date.now()
+    for (const file of files) {
+      const raw = await readFile(join(this.dir, file))
+      compressed += raw.length
+      let at = performance.now()
+      const data = unzipSync(raw).toString('utf8')
+      t.inflate += performance.now() - at
+      text += data.length
+      at = performance.now()
+      const overviews = parseOverviews(data)
+      t.parse += performance.now() - at
+      at = performance.now()
+      const next = Number(file.split('-')[2].replace('.gz', '')) + 1
+      indexer.add(overviews, next)
+      if (file === files.at(-1)) indexer.finish()
+      const batch = indexer.take()
+      t.index += performance.now() - at
+      out.headers += batch.headers
+      out.releases += batch.releases.length
+      for (const r of batch.releases) {
+        out.files += r.files
+        out.segments += r.segments
+        out.payload += r.payload.length
+      }
+      out.candidates += batch.candidates.length
+      out.fragments += batch.fragments.length
+      out.samples += batch.samples.length
+      for (const c of batch.counts) counts.set(`${c.rule} ${c.key}`.trim(), (counts.get(`${c.rule} ${c.key}`.trim()) ?? 0) + c.articles)
+      peakRss = Math.max(peakRss, process.memoryUsage().rss)
+    }
+    const cpu = process.cpuUsage(cpuStart)
+    const cpuS = (cpu.user + cpu.system) / 1e6
+    const total = t.inflate + t.parse + t.index
+    this.logger.info(`${group} ${start}–${end}: ${files.length} chunks, ${(compressed / 1e6).toFixed(0)} MB compressed, ${(text / 1e6).toFixed(0)} MB text`)
+    this.logger.info(`${out.headers.toLocaleString('en')} headers in ${(total / 1000).toFixed(1)} s (${Math.round(out.headers / (total / 1000)).toLocaleString('en')}/s on one core; CPU ${cpuS.toFixed(1)} s, wall ${((Date.now() - wall) / 1000).toFixed(1)} s)`)
+    this.logger.info(`  inflate ${(t.inflate / 1000).toFixed(1)} s, parse ${(t.parse / 1000).toFixed(1)} s, index ${(t.index / 1000).toFixed(1)} s; peak RSS ${(peakRss / 1e6).toFixed(0)} MB`)
+    this.logger.info(`  kept: ${out.releases} release chunks, ${out.files} files, ${out.segments} segments, payload ${(out.payload / 1e6).toFixed(1)} MB (${(out.payload / Math.max(1, out.segments)).toFixed(1)} B/segment); ${out.candidates} candidates, ${out.fragments} fragments, ${out.samples} samples`)
+    for (const [k, v] of [...counts].sort((a, b) => b[1] - a[1]).slice(0, 8)) this.logger.info(`  dropped ${k}: ${v.toLocaleString('en')} articles`)
+  }
+}

commands/workers/Body.ts

@@ -1,97 +0,0 @@
-import { BaseCommand } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import { Worker } from 'bullmq'
-import fs from 'node:fs/promises'
-import path from 'node:path'
-import queueConfig from '#config/queue'
-import Shutdown from '#services/Shutdown'
-import QueueService from '#services/QueueService'
-import env from '#start/env'
-import NntpService from '#services/NntpService'
-import IngestService, { segmentFromYenc } from '#services/IngestService'
-import { parseYencMeta, YencError } from '#lib/yenc/index'
-import { firstChunkMd5 } from '#services/ContentService'
-
-export default class BodyWorker extends BaseCommand {
-  public static commandName = 'worker:body'
-  public static description = 'Starts a worker to process article bodies for yEnc metadata.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  public async run() {
-    this.logger.info('Starting body worker...')
-
-    const pool = NntpService
-
-    const worker = new Worker('body-queue', async (job) => {
-      const { header, group, articleNumber } = job.data
-      const messageId = header['message-id']
-      this.logger.debug(`Processing header with unparsable subject: ${header.subject}`)
-
-      let conn
-      let failed = false
-      try {
-        conn = await pool.acquire()
-        const bodyBuffer = await pool.fetchBody(conn, messageId)
-
-        let meta
-        try {
-          meta = parseYencMeta(bodyBuffer)
-        } catch (parseError: any) {
-          if (parseError instanceof YencError && bodyBuffer.indexOf('=ybegin ') === -1) {
-            this.logger.debug(`Not a yEnc post, skipping: ${messageId}`)
-            return
-          }
-          this.logger.error(`Failed to parse yEnc data for message ID ${messageId}. Dumping buffer.`)
-          const debugDir = path.join(this.app.appRoot.pathname, 'debug')
-          await fs.mkdir(debugDir, { recursive: true })
-          const timestamp = new Date().toISOString().replace(/:/g, '-')
-          const dumpFile = path.join(debugDir, `body-error-${timestamp}-${messageId.replace(/[<>]/g, '')}.bin`)
-          await fs.writeFile(dumpFile, bodyBuffer)
-          this.logger.error(`Problematic body buffer saved to: ${dumpFile}`)
-          throw parseError
-        }
-
-        pool.release(conn)
-        conn = undefined
-
-        const segment = segmentFromYenc(header, group, meta, articleNumber)
-        if (segment?.number === 1) segment.md5_16k = firstChunkMd5(bodyBuffer, meta.begin.size)
-        if (!segment) {
-          this.logger.warning(`Inconsistent yEnc part numbering in ${messageId}, skipping.`)
-          return
-        }
-        const completed = await IngestService.ingest([segment])
-        this.logger.debug(
-          `Stored part ${segment.number}/${segment.totalParts} of "${segment.name}"` +
-            (completed.length > 0 ? ' (file complete)' : '')
-        )
-      } catch (error: any) {
-        failed = true
-        this.logger.error(`Error in body worker for message ID ${messageId}: ${error.message}`)
-        throw error
-      } finally {
-        if (conn) {
-          pool.release(conn, failed)
-        }
-      }
-    }, {
-      connection: queueConfig.connection,
-      // Each job holds one NNTP connection while downloading, so keep this <= the pool size.
-      concurrency: env.get('BODY_CONCURRENCY', 1),
-    })
-
-    worker.on('failed', (job, err) => {
-      this.logger.error(`Body job ${job?.id} failed: ${err.message}`)
-    })
-
-    this.logger.info('Body worker started and listening for jobs.')
-    await Shutdown.wait()
-    this.logger.info('Shutting down: finishing active jobs...')
-    await worker.close()
-    await NntpService.shutdown()
-    await QueueService.closeAll()
-  }
-}

commands/workers/Collection.ts

@@ -1,76 +0,0 @@
-import { BaseCommand } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import { Worker } from 'bullmq'
-import queueConfig from '#config/queue'
-import env from '#start/env'
-import Shutdown from '#services/Shutdown'
-import QueueService from '#services/QueueService'
-import NntpService from '#services/NntpService'
-import ContentService from '#services/ContentService'
-import File from '#models/file'
-
-export default class CollectionWorker extends BaseCommand {
-  public static commandName = 'worker:collection'
-  public static description = 'Starts a worker to process file collections (e.g., RAR archives).'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  public async run() {
-    this.logger.info('Starting collection worker...')
-
-    const worker = new Worker('collection-queue', async (job) => {
-      const { fileId } = job.data
-      this.logger.debug(`Processing file ID ${fileId} for collection.`)
-
-      const file = await File.find(fileId)
-      if (!file) {
-        this.logger.error(`File with ID ${fileId} not found in the database.`)
-        return
-      }
-
-      try {
-        const content = await ContentService.inspect(fileId)
-        if (content?.par2) {
-          const { par2 } = content
-          this.logger.info(`Par2 "${file.filename}": release "${par2.releaseName}", ${par2.listed} files listed.`)
-        } else if (content?.archive) {
-          const { info, releaseRenamed, grouped } = content.archive
-          const flags = [info.headersEncrypted && 'encrypted headers', info.entries.some((e) => e.encrypted) && 'encrypted files']
-            .filter(Boolean)
-            .join(', ')
-          this.logger.info(
-            `Archive "${file.filename}": ${info.format}, ${info.entries.length} entries${flags ? ` (${flags})` : ''}` +
-              (grouped ? ', grouped into a release' : '') +
-              (releaseRenamed ? `, release renamed to "${releaseRenamed}"` : '')
-          )
-        }
-        if (content?.named) this.logger.info(`"${file.filename}" got its real name from par2.`)
-        if (content?.contentType === 'nzbe') this.logger.info(`"${file.filename}" is an encrypted NZB container.`)
-        if (content?.contentType === 'rclone') this.logger.info(`"${file.filename}" is rclone-encrypted.`)
-        if (content?.nzb) {
-          const n = content.nzb
-          this.logger.info(`NZB "${file.filename}": ${n.listed} files (${n.known} known, ${n.added} added) → "${n.releaseName}".`)
-        }
-      } catch (error: any) {
-        this.logger.warning(`Content inspection of "${file.filename}" failed: ${error.message}`)
-      }
-    }, {
-      connection: queueConfig.connection,
-      // Each job holds one NNTP connection while downloading; keep <= the pool size.
-      concurrency: env.get('COLLECTION_CONCURRENCY', 1),
-    })
-
-    worker.on('failed', (job, err) => {
-      this.logger.error(`Collection job ${job?.id} failed: ${err.message}`)
-    })
-
-    this.logger.info('Collection worker started and listening for jobs.')
-    await Shutdown.wait()
-    this.logger.info('Shutting down: finishing active jobs...')
-    await worker.close()
-    await NntpService.shutdown()
-    await QueueService.closeAll()
-  }
-}

commands/workers/Deferred.ts

@@ -1,57 +0,0 @@
-import { BaseCommand, flags } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import env from '#start/env'
-import Shutdown from '#services/Shutdown'
-import NntpService from '#services/NntpService'
-import QueueService from '#services/QueueService'
-import DeferredService from '#services/DeferredService'
-
-const IDLE_MS = 60_000
-
-/**
- * Works through `deferred_files` (CONCEPT.md D28): one body per file, and
- * small files completely. Not started by default (compose profile
- * `deferred`); run it when there is bandwidth to spare.
- */
-export default class DeferredWorker extends BaseCommand {
-  public static commandName = 'worker:deferred'
-  public static description = 'Samples deferred files (e.g. ngPost uploads): one article body per file.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  @flags.number({ description: 'Stop after this many files (default: keep running)' })
-  declare limit?: number
-
-  public async run() {
-    const concurrency = env.get('DEFERRED_CONCURRENCY', 4)
-    this.logger.info(`Deferred worker started (concurrency ${concurrency}).`)
-    const counts: Record<string, number> = {}
-    let done = 0
-    try {
-      while (!Shutdown.requested && (!this.limit || done < this.limit)) {
-        const batch = await DeferredService.claim(Math.min(concurrency * 4, this.limit ? this.limit - done : Infinity))
-        if (batch.length === 0) {
-          if (this.limit) break
-          await Shutdown.sleep(IDLE_MS)
-          continue
-        }
-        let next = 0
-        await Promise.all(
-          Array.from({ length: concurrency }, async () => {
-            while (next < batch.length) {
-              const status = await DeferredService.process(batch[next++])
-              counts[status] = (counts[status] ?? 0) + 1
-            }
-          })
-        )
-        done += batch.length
-        this.logger.info(`${done} deferred files processed: ${Object.entries(counts).map(([k, v]) => `${v} ${k}`).join(', ')}.`)
-      }
-    } finally {
-      await NntpService.shutdown()
-      await QueueService.closeAll()
-    }
-  }
-}

commands/workers/Fetch.ts

@@ -1,110 +0,0 @@
-import { BaseCommand } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import { Worker } from 'bullmq'
-import queueConfig from '#config/queue'
-import env from '#start/env'
-import db from '@adonisjs/lucid/services/db'
-import indexerConfig from '#config/indexer'
-import Shutdown from '#services/Shutdown'
-import QueueService from '#services/QueueService'
-import NntpService from '#services/NntpService'
-import FetchRangeService from '#services/FetchRangeService'
-import IngestService, { segmentFromHeader, type IngestSegment } from '#services/IngestService'
-import DeferredService, { bodyJob, isDeferred, type DeferredArticle } from '#services/DeferredService'
-import SpotService, { isSpotGroup } from '#services/SpotService'
-
-export default class FetchWorker extends BaseCommand {
-  public static commandName = 'worker:fetch'
-  public static description = 'Starts a worker that fetches headers (XOVER) and stores them as binaries/parts.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  public async run() {
-    this.logger.info('Starting fetch worker...')
-
-    const pool = NntpService
-    const bodyQueue = QueueService.bodyQueue
-
-    const worker = new Worker('nntp-fetch-queue', async (job) => {
-      const { groupName, startId, endId } = job.data
-      this.logger.info(`Processing fetch job for ${groupName}, articles ${startId}-${endId}`)
-
-      let overviews: [number, Record<string, string>][]
-      let conn
-      let failed = false
-      try {
-        conn = await pool.acquire()
-        await conn.group(groupName)
-        overviews = (await conn.xover(startId, endId)).overviews
-      } catch (error: any) {
-        failed = true
-        this.logger.error(`Error fetching headers for ${groupName}: ${error.message}`)
-        throw error
-      } finally {
-        if (conn) {
-          pool.release(conn, failed)
-        }
-      }
-
-      if (isSpotGroup(groupName)) {
-        const spots = await SpotService.recordHeaders(groupName, overviews)
-        if (job.data.rangeId) await FetchRangeService.markDone(job.data.rangeId)
-        this.logger.info(`${groupName} ${startId}-${endId}: ${overviews.length} headers, ${spots} spots queued.`)
-        return
-      }
-
-      const group = await db.from('groups').where('name', groupName).first()
-      const fetchBodies = Boolean(group?.fetch_bodies)
-
-      const segments: IngestSegment[] = []
-      const obfuscated: { articleNumber: number; header: Record<string, string> }[] = []
-      const deferred: DeferredArticle[] = []
-      let skipped = 0
-      for (const [articleNumber, header] of overviews) {
-        if (!header) continue
-        const segment = segmentFromHeader(header, groupName, articleNumber)
-        if (segment) segments.push(segment)
-        // Recorded per file (or as article ranges) in every group, bodies or not; processed later if at all.
-        else if (isDeferred(header)) deferred.push({ articleNumber, header })
-        else if (!fetchBodies || indexerConfig.skipBodyMessageIds.some((re) => re.test(header['message-id'] ?? ''))) skipped++
-        else obfuscated.push({ articleNumber, header })
-      }
-
-      const completed = await IngestService.ingest(segments)
-      const deferredFiles = await DeferredService.record(groupName, deferred)
-
-      if (obfuscated.length > 0) {
-        await bodyQueue.addBulk(obfuscated.map(({ articleNumber, header }) => bodyJob(header, groupName, articleNumber)))
-      }
-
-      if (job.data.rangeId) await FetchRangeService.markDone(job.data.rangeId)
-
-      this.logger.info(
-        `${groupName} ${startId}-${endId}: ${overviews.length} headers, ${segments.length} stored, ` +
-          `${obfuscated.length} sent to body-queue${skipped ? ` (${skipped} skipped)` : ''}` +
-          `${deferred.length ? `, ${deferred.length} deferred (${deferredFiles} files)` : ''}, ${completed.length} files completed.`
-      )
-    }, {
-      connection: queueConfig.connection,
-      // XOVER is quick; most of a job is storing. Each concurrent job uses one DB connection (DB_POOL_MAX).
-      concurrency: env.get('FETCH_CONCURRENCY', 5),
-    })
-
-    worker.on('failed', async (job, err) => {
-      this.logger.error(`Fetch job ${job?.id} failed for group ${job?.data.groupName}: ${err.message}`)
-      // Only after the last attempt; the scheduler re-queues the range later.
-      if (job?.data.rangeId && job.attemptsMade >= (job.opts.attempts ?? 1)) {
-        await FetchRangeService.markFailed(job.data.rangeId, err.message).catch(() => {})
-      }
-    })
-
-    this.logger.info('Fetch worker started and listening for jobs.')
-    await Shutdown.wait()
-    this.logger.info('Shutting down: finishing active jobs...')
-    await worker.close()
-    await NntpService.shutdown()
-    await QueueService.closeAll()
-  }
-}

commands/workers/File.ts

@@ -1,117 +0,0 @@
-import { BaseCommand } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import { Worker } from 'bullmq'
-import db from '@adonisjs/lucid/services/db'
-import queueConfig from '#config/queue'
-import Shutdown from '#services/Shutdown'
-import QueueService from '#services/QueueService'
-
-export default class FileWorker extends BaseCommand {
-  public static commandName = 'worker:file'
-  public static description = 'Starts a worker that turns completed binaries into files.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  public async run() {
-    this.logger.info('Starting file worker...')
-
-    const collectionQueue = QueueService.collectionQueue
-
-    const worker = new Worker('file-queue', async (job) => {
-      const { binaryId } = job.data
-
-      const binary = await db.from('binaries').where('id', binaryId).first()
-      if (!binary) {
-        this.logger.warning(`Binary ${binaryId} no longer exists (garbage collected?).`)
-        return
-      }
-
-      if (binary.conflicted_at) {
-        this.logger.warning(`Binary ${binaryId} ("${binary.name}") has conflicting parts, skipping.`)
-        return
-      }
-
-      const parts = await db
-        .from('parts')
-        .where('binary_id', binaryId)
-        .orderBy('number')
-        .select('number', 'message_id', 'bytes')
-      if (parts.length < binary.total_parts) {
-        throw new Error(`Binary ${binaryId} has ${parts.length}/${binary.total_parts} parts.`)
-      }
-
-      const messageIds = Object.fromEntries(
-        parts.map((part) => [part.number, { id: part.message_id, size: part.bytes }])
-      )
-
-      // Already known as a file, e.g. added from a posted NZB before its own articles completed.
-      const existing = await db
-        .from('files')
-        .whereRaw(`(message_ids->'1'->>'id') = ?`, [messageIds['1']?.id ?? ''])
-        .first()
-      if (existing) {
-        await db.from('binaries').where('id', binaryId).update({ file_id: existing.id })
-        await dropCopiedParts(binary)
-        this.logger.debug(`File "${binary.name}" already exists as ID ${existing.id} (same first segment).`)
-        return
-      }
-
-      // file_key is unique: a second completion of the same upload inserts nothing.
-      const knex = db.connection().getWriteClient()
-      const now = new Date()
-      const [inserted] = await knex('files')
-        .insert({
-          file_key: binary.file_key,
-          filename: binary.name,
-          poster: binary.poster,
-          date: binary.posted_at ? new Date(binary.posted_at).getTime() : now.getTime(),
-          parts: parts.length,
-          message_ids: JSON.stringify(messageIds),
-          groups: JSON.stringify(binary.groups),
-          release_id: binary.release_id,
-          real_name: binary.real_name,
-          size: parts.reduce((sum, part) => sum + Number(part.bytes), 0),
-          created_at: now,
-          updated_at: now,
-        })
-        .onConflict('file_key')
-        .ignore()
-        .returning('id')
-
-      const fileId =
-        inserted?.id ?? (await db.from('files').where('file_key', binary.file_key).first())?.id
-      await db.from('binaries').where('id', binaryId).update({ file_id: fileId })
-      await dropCopiedParts(binary)
-
-      if (!inserted) {
-        this.logger.debug(`File "${binary.name}" (${binary.file_key}) already exists as ID ${fileId}.`)
-        return
-      }
-
-      this.logger.info(`Saved file "${binary.name}" (${parts.length} parts) as ID ${fileId}.`)
-      await collectionQueue.add('process-collection', { fileId }, { jobId: `file-${fileId}` })
-    }, { connection: queueConfig.connection })
-
-    worker.on('failed', (job, err) => {
-      this.logger.error(`File job ${job?.id} failed: ${err.message}`)
-    })
-
-    this.logger.info('File worker started and listening for jobs.')
-    await Shutdown.wait()
-    this.logger.info('Shutting down: finishing active jobs...')
-    await worker.close()
-    await QueueService.closeAll()
-  }
-}
-
-/**
- * The file holds every message id now. Binaries identified by their subject
- * (no size: header-derived) don't need their parts any more; body-derived
- * ones keep them, so articles of an equal-sized next volume are recognised
- * as a conflict and split off (CONCEPT.md D20). LOG.md 2026-09-28 (17).
- */
-async function dropCopiedParts(binary: { id: number; size: number | null }) {
-  if (binary.size === null) await db.from('parts').where('binary_id', binary.id).delete()
-}

commands/workers/Spots.ts

@@ -1,54 +0,0 @@
-import { BaseCommand, flags } from '@adonisjs/core/ace'
-import { CommandOptions } from '@adonisjs/core/types/ace'
-import env from '#start/env'
-import Shutdown from '#services/Shutdown'
-import NntpService from '#services/NntpService'
-import SpotService from '#services/SpotService'
-
-const IDLE_MS = 30_000
-
-/**
- * Works through queued Spotnet spots (CONCEPT.md D29): verifies each one,
- * downloads its compressed NZB (a few KB to a few MB) and imports it.
- */
-export default class SpotsWorker extends BaseCommand {
-  public static commandName = 'worker:spots'
-  public static description = 'Reads Spotnet spots: verifies them and imports their NZBs.'
-
-  public static options: CommandOptions = {
-    startApp: true,
-  }
-
-  @flags.number({ description: 'Stop after this many spots (default: keep running)' })
-  declare limit?: number
-
-  public async run() {
-    const concurrency = env.get('SPOT_CONCURRENCY', 4)
-    this.logger.info(`Spot worker started (concurrency ${concurrency}).`)
-    const counts: Record<string, number> = {}
-    let done = 0
-    try {
-      while (!Shutdown.requested && (!this.limit || done < this.limit)) {
-        const batch = await SpotService.claim(Math.min(concurrency * 4, this.limit ? this.limit - done : Infinity))
-        if (batch.length === 0) {
-          if (this.limit) break
-          await Shutdown.sleep(IDLE_MS)
-          continue
-        }
-        let next = 0
-        await Promise.all(
-          Array.from({ length: concurrency }, async () => {
-            while (next < batch.length) {
-              const status = await SpotService.process(batch[next++])
-              counts[status] = (counts[status] ?? 0) + 1
-            }
-          })
-        )
-        done += batch.length
-        this.logger.info(`${done} spots: ${Object.entries(counts).map(([k, v]) => `${v} ${k}`).join(', ')}.`)
-      }
-    } finally {
-      await NntpService.shutdown()
-    }
-  }
-}

config/app.ts

@@ -32,6 +32,10 @@ export const http = defineConfig({
    */
   useAsyncLocalStorage: false,
 
+  router: {
+    matcher: 'tree',
+  },
+
   /**
    * Redirect configuration controls the behavior of
    * response.redirect().back() and query string forwarding.

config/database.ts

@@ -1,137 +1,34 @@
+import env from '#start/env'
 import app from '@adonisjs/core/services/app'
 import { defineConfig } from '@adonisjs/lucid'
-import env from '#start/env'
 
+/**
+ * The rebuild's own database (REBUILD.md, decision b), on the same server
+ * as the current indexer's.
+ */
 const dbConfig = defineConfig({
-  /**
+  connection: 'pg',
-   * Default connection used for all queries.
-   */
-  connection: env.get('DB_CONNECTION', 'sqlite'),
-
   connections: {
-    /**
-     * SQLite connection (default).
-     */
-    sqlite: {
-      client: 'better-sqlite3',
-
-      connection: {
-        filename: app.tmpPath('db.sqlite3'),
-      },
-
-      /**
-       * Required by Knex for SQLite defaults.
-       */
-      useNullAsDefault: true,
-
-      migrations: {
-        /**
-         * Sort migration files naturally by filename.
-         */
-        naturalSort: true,
-
-        /**
-         * Paths containing migration files.
-         */
-        paths: ['database/migrations'],
-      },
-
-      schemaGeneration: {
-        /**
-         * Enable schema generation from Lucid models.
-         */
-        enabled: true,
-
-        /**
-         * Custom schema rules file paths.
-         */
-        rulesPaths: ['./database/schema_rules.js'],
-      },
-    },
-
-    /**
-     * PostgreSQL connection.
-     * Install package to switch: npm install pg
-     */
     pg: {
       client: 'pg',
       connection: {
         host: env.get('DB_HOST'),
         port: env.get('DB_PORT'),
         user: env.get('DB_USER'),
-        password: env.get('DB_PASSWORD')?.release(),
+        password: env.get('DB_PASSWORD'),
         database: env.get('DB_DATABASE'),
-        // For idempotent bulk writers (worker:fetch): don't wait for the WAL
-        // flush on every statement. A crash loses the last moments of writes,
-        // which the retried job writes again.
-        ...(env.get('DB_SYNCHRONOUS_COMMIT') === 'off' ? { options: '-c synchronous_commit=off' } : {}),
       },
-      // One connection per concurrent job; knex defaults to 10.
       pool: { min: 0, max: env.get('DB_POOL_MAX', 10) },
       migrations: {
         naturalSort: true,
         paths: ['database/migrations'],
       },
-      debug: app.inDev,
+      schemaGeneration: {
+        enabled: true,
+        rulesPaths: ['./database/schema_rules.js'],
+      },
+      debug: app.inDev && env.get('DB_DEBUG', false),
     },
-
-    /**
-     * MySQL / MariaDB connection.
-     * Install package to switch: npm install mysql2
-     */
-    // mysql: {
-    //   client: 'mysql2',
-    //   connection: {
-    //     host: env.get('DB_HOST'),
-    //     port: env.get('DB_PORT'),
-    //     user: env.get('DB_USER'),
-    //     password: env.get('DB_PASSWORD'),
-    //     database: env.get('DB_DATABASE'),
-    //   },
-    //   migrations: {
-    //     naturalSort: true,
-    //     paths: ['database/migrations'],
-    //   },
-    //   debug: app.inDev,
-    // },
-
-    /**
-     * Microsoft SQL Server connection.
-     * Install package to switch: npm install tedious
-     */
-    // mssql: {
-    //   client: 'mssql',
-    //   connection: {
-    //     server: env.get('DB_HOST'),
-    //     port: env.get('DB_PORT'),
-    //     user: env.get('DB_USER'),
-    //     password: env.get('DB_PASSWORD'),
-    //     database: env.get('DB_DATABASE'),
-    //   },
-    //   migrations: {
-    //     naturalSort: true,
-    //     paths: ['database/migrations'],
-    //   },
-    //   debug: app.inDev,
-    // },
-
-    /**
-     * libSQL (Turso) connection.
-     * Install package to switch: npm install @libsql/client
-     */
-    // libsql: {
-    //   client: 'libsql',
-    //   connection: {
-    //     url: env.get('LIBSQL_URL'),
-    //     authToken: env.get('LIBSQL_AUTH_TOKEN'),
-    //   },
-    //   useNullAsDefault: true,
-    //   migrations: {
-    //     naturalSort: true,
-    //     paths: ['database/migrations'],
-    //   },
-    //   debug: app.inDev,
-    // },
   },
 })
 

config/gc.ts

@@ -1,57 +0,0 @@
-import env from '#start/env'
-
-/**
- * Retention for transient pipeline state in Redis. See CONCEPT.md (D11).
- */
-export default {
-  /** How often `gc:scheduler` sweeps. */
-  intervalMs: env.get('GC_INTERVAL_MS', 10 * 60 * 1000),
-
-  jobs: {
-    /** Completed jobs are only kept for inspection; keep them briefly. */
-    completedAgeSeconds: env.get('GC_COMPLETED_AGE_SECONDS', 60 * 60),
-    completedMaxCount: 1000,
-    /** Failed jobs are kept longer so they can be inspected and retried. */
-    failedAgeSeconds: env.get('GC_FAILED_AGE_SECONDS', 7 * 24 * 60 * 60),
-  },
-
-  /**
-   * Upper bound on waiting body jobs. Obfuscated traffic can exceed what the
-   * line downloads for hours; beyond this, the oldest waiting jobs are
-   * dropped so Redis stays bounded (~1 KB per job).
-   */
-  bodyQueueMax: env.get('GC_BODY_QUEUE_MAX', 500_000),
-
-  binaries: {
-    /**
-     * A binary that gets no new segments for this long is deleted: incomplete
-     * ones are given up on, completed ones have been copied into `files`.
-     */
-    staleAfterSeconds: env.get('GC_BINARY_STALE_SECONDS', 48 * 60 * 60),
-    /**
-     * Header-derived binaries that still have one article after this long:
-     * uploads that give every article its own random name (and poster) never
-     * complete from headers. A real file gets its next articles in minutes.
-     */
-    singleArticleAfterSeconds: env.get('GC_SINGLE_ARTICLE_SECONDS', 2 * 60 * 60),
-  },
-
-  /** Releases without files or binaries (their binaries were collected) are deleted after this long. */
-  emptyReleaseAfterSeconds: env.get('GC_EMPTY_RELEASE_SECONDS', 2 * 60 * 60),
-
-  /** Bulk deletes run in batches of this size, for at most this long per sweep. */
-  deleteBatch: 20_000,
-  deleteBudgetMs: env.get('GC_DELETE_BUDGET_MS', 3 * 60 * 1000),
-
-  /**
-   * Deferred files (CONCEPT.md D28) are deleted this long after they were
-   * first seen, processed or not. The articles stay on the server.
-   */
-  deferredRetentionDays: env.get('GC_DEFERRED_RETENTION_DAYS', 30),
-
-  /** Redis hashes from the old header worker (before D12); they only need to expire. */
-  legacyPendingFiles: {
-    prefix: 'file:',
-    ttlSeconds: 48 * 60 * 60,
-  },
-}

config/indexer.ts

@@ -1,47 +0,0 @@
-import env from '#start/env'
-
-export default {
-  /** How often `index:scheduler` checks the groups for new articles. */
-  intervalMs: env.get('INDEX_INTERVAL_MS', 60 * 1000),
-  /** Articles per fetch job (one XOVER call). */
-  batchSize: BigInt(env.get('INDEX_BATCH_SIZE', 20_000)),
-  /** Upper bound on fetch jobs queued per group per tick, so catching up is spread out. */
-  maxBatchesPerTick: env.get('INDEX_MAX_BATCHES_PER_TICK', 5),
-  /**
-   * Where a group that was never indexed starts: this many articles before
-   * the newest one. The server's low-water mark can be years back.
-   */
-  initialBackfill: BigInt(env.get('INDEX_INITIAL_BACKFILL', 50_000)),
-
-  /**
-   * Obfuscated articles whose message-id matches one of these are not sent
-   * to the body queue: their uploader spreads every file over many groups,
-   * so identifying them would take far more bandwidth than there is
-   * (see LOG.md, 2026-09-28 (9)). Normal-subject articles are unaffected.
-   */
-  skipBodyMessageIds: [
-    // hex7$hex7$hex7@hex10.hex5, ~60k articles/min over 10+ groups
-    /^<?[0-9a-f]{7}\$[0-9a-f]{7}\$[0-9a-f]{7}@[0-9a-f]{10}\.[0-9a-f]{5}>?$/,
-  ],
-
-  /**
-   * Obfuscated articles whose message-id matches one of these don't go to
-   * the body queue. Instead they're recorded per file in `deferred_files`
-   * (their subject keeps the file name, only the part counter is missing).
-   * `worker:deferred` later fetches one body per file (CONCEPT.md D28).
-   */
-  deferBodyMessageIds: [
-    // ngPost: <hex32@ngPost>. Mostly RAR5 with encrypted headers (LOG.md 2026-09-28 (14)).
-    /^<?[0-9a-f]{32}@ngPost>?$/,
-  ],
-  /**
-   * Spotnet header groups (CONCEPT.md D29): their headers are spots, not
-   * binaries. Add one with `group:add free.pt --backfill 5000`.
-   */
-  spotGroups: ['free.pt'],
-
-  /** worker:deferred leaves a file alone until it has had no new articles for this long. */
-  deferredSettleMs: env.get('DEFERRED_SETTLE_MS', 15 * 60 * 1000),
-  /** Files of at most this many parts are downloaded completely (small archives, NZBs, par2 index files). */
-  deferredExpandMaxParts: env.get('DEFERRED_EXPAND_MAX_PARTS', 8),
-}

config/nntp.ts

@@ -1,11 +0,0 @@
-import env from '#start/env'
-
-export default {
-  host: env.get('NNTP_HOST'),
-  port: env.get('NNTP_PORT', 119),
-  user: env.get('NNTP_USER'),
-  password: env.get('NNTP_PASSWORD'),
-  secure: env.get('NNTP_SECURE', false),
-  /** Connections per process. Mind the provider's limit across all processes. */
-  poolSize: env.get('NNTP_POOL_SIZE', 10),
-}

config/queue.ts

@@ -1,8 +0,0 @@
-import env from '#start/env'
-
-export default {
-  connection: {
-    host: env.get('REDIS_HOST', '127.0.0.1'),
-    port: env.get('REDIS_PORT', 6379),
-  },
-}

config/redis.ts

@@ -1,12 +0,0 @@
-import env from '#start/env'
-
-export default {
-  connection: 'main',
-  connections: {
-    main: {
-      host: env.get('REDIS_HOST', '127.0.0.1'),
-      port: env.get('REDIS_PORT'),
-            password: env.get('REDIS_PASSWORD')?.release(),
-          },
-  },
-}

config/static.ts

@@ -1,17 +0,0 @@
-import { defineConfig } from '@adonisjs/static'
-
-/**
- * Configuration options to tweak the static files middleware.
- * The complete set of options are documented on the
- * official documentation website.
- *
- * https://docs.adonisjs.com/guides/static-assets
- */
-const staticServerConfig = defineConfig({
-  enabled: true,
-  etag: true,
-  lastModified: true,
-  dotFiles: 'ignore',
-})
-
-export default staticServerConfig

database/migrations/1768620764700_create_groups_table.ts

@@ -1,21 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-export default class extends BaseSchema {
-  protected tableName = 'groups'
-
-  async up() {
-    this.schema.createTable(this.tableName, (table) => {
-      table.increments('id')
-      table.string('name').notNullable().unique()
-      table.boolean('active').notNullable().defaultTo(true)
-      table.bigInteger('last_indexed_id').defaultTo(0)
-
-      table.timestamp('created_at', { useTz: true })
-      table.timestamp('updated_at', { useTz: true })
-    })
-  }
-
-  async down() {
-    this.schema.dropTable(this.tableName)
-  }
-}

database/migrations/1768620764710_create_files_table.ts

@@ -1,24 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-export default class extends BaseSchema {
-  protected tableName = 'files'
-
-  async up() {
-    this.schema.createTable(this.tableName, (table) => {
-      table.increments('id').primary()
-      table.string('filename').notNullable()
-      table.string('poster').notNullable()
-      table.bigInteger('date').notNullable()
-      table.integer('parts').notNullable()
-      table.jsonb('message_ids').notNullable()
-      table.jsonb('groups')
-
-      table.timestamp('created_at', { useTz: true })
-      table.timestamp('updated_at', { useTz: true })
-    })
-  }
-
-  async down() {
-    this.schema.dropTable(this.tableName)
-  }
-}

database/migrations/1768620764720_create_binaries_table.ts

@@ -1,32 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Files that are still being collected from XOVER. One row per uploaded file,
- * identified by file_key (see lib/subject/file_key.ts).
- */
-export default class extends BaseSchema {
-  protected tableName = 'binaries'
-
-  async up() {
-    this.schema.createTable(this.tableName, (table) => {
-      table.bigIncrements('id')
-      table.string('file_key', 40).notNullable().unique()
-      table.text('poster').notNullable()
-      table.text('subject').notNullable()
-      table.text('name').notNullable()
-      table.integer('total_parts').notNullable()
-      table.jsonb('groups').notNullable()
-      table.timestamp('posted_at', { useTz: true })
-      table.timestamp('first_seen_at', { useTz: true }).notNullable()
-      table.timestamp('last_seen_at', { useTz: true }).notNullable().index()
-      table.timestamp('completed_at', { useTz: true })
-      // Set when two different articles claim the same part number (see IngestService).
-      table.timestamp('conflicted_at', { useTz: true })
-      table.integer('file_id').unsigned().references('files.id').onDelete('SET NULL')
-    })
-  }
-
-  async down() {
-    this.schema.dropTable(this.tableName)
-  }
-}

database/migrations/1768620764730_create_parts_table.ts

@@ -1,24 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Segments (articles) of a binary. The primary key makes re-ingesting the
- * same range a no-op.
- */
-export default class extends BaseSchema {
-  protected tableName = 'parts'
-
-  async up() {
-    this.schema.createTable(this.tableName, (table) => {
-      table.bigInteger('binary_id').notNullable().references('binaries.id').onDelete('CASCADE')
-      table.integer('number').notNullable()
-      table.text('message_id').notNullable()
-      table.integer('bytes').notNullable()
-      table.bigInteger('article_number')
-      table.primary(['binary_id', 'number'])
-    })
-  }
-
-  async down() {
-    this.schema.dropTable(this.tableName)
-  }
-}

database/migrations/1768620764740_add_file_key_to_files_table.ts

@@ -1,17 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-export default class extends BaseSchema {
-  protected tableName = 'files'
-
-  async up() {
-    this.schema.alterTable(this.tableName, (table) => {
-      table.string('file_key', 40).unique()
-    })
-  }
-
-  async down() {
-    this.schema.alterTable(this.tableName, (table) => {
-      table.dropColumn('file_key')
-    })
-  }
-}

database/migrations/1768620764750_create_releases_table.ts

@@ -1,38 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * A release is one download: the files of one post (archive volumes, par2
- * set, nfo). See lib/release/release_key.ts and CONCEPT.md (D16).
- */
-export default class extends BaseSchema {
-  protected tableName = 'releases'
-
-  async up() {
-    this.schema.createTable(this.tableName, (table) => {
-      table.bigIncrements('id')
-      table.string('release_key', 40).notNullable().unique()
-      table.text('name').notNullable()
-      // Null for obfuscated posts, whose poster is random per article.
-      table.text('poster')
-      // Number of files announced by the [i/M] subject counter, if any.
-      table.integer('file_total')
-      table.jsonb('groups').notNullable()
-      table.timestamp('posted_at', { useTz: true })
-      table.timestamp('first_seen_at', { useTz: true }).notNullable()
-      table.timestamp('last_seen_at', { useTz: true }).notNullable()
-    })
-
-    this.schema.alterTable('binaries', (table) => {
-      table.bigInteger('release_id').references('releases.id').onDelete('SET NULL').index()
-    })
-    this.schema.alterTable('files', (table) => {
-      table.bigInteger('release_id').references('releases.id').onDelete('SET NULL').index()
-    })
-  }
-
-  async down() {
-    this.schema.alterTable('files', (table) => table.dropColumn('release_id'))
-    this.schema.alterTable('binaries', (table) => table.dropColumn('release_id'))
-    this.schema.dropTable(this.tableName)
-  }
-}

database/migrations/1768620764760_add_par2_matching.ts

@@ -1,33 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Par2-based grouping of obfuscated posts (CONCEPT.md D16): the par2 file
- * list of a release, and the exact file size of body-derived binaries to
- * match against it.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.alterTable('binaries', (table) => {
-      // Exact file size from =ybegin; only known for body-derived binaries.
-      table.bigInteger('size').index()
-    })
-    this.schema.alterTable('releases', (table) => {
-      // How the release was recognised: subject, stem, weak-stem or par2.
-      table.string('source', 16)
-    })
-    this.schema.createTable('release_files', (table) => {
-      table.bigInteger('release_id').notNullable().references('releases.id').onDelete('CASCADE')
-      table.string('par2_file_id', 32).notNullable()
-      table.text('name').notNullable()
-      table.bigInteger('size').notNullable().index()
-      table.string('md5_16k', 32).notNullable()
-      table.primary(['release_id', 'par2_file_id'])
-    })
-  }
-
-  async down() {
-    this.schema.dropTable('release_files')
-    this.schema.alterTable('releases', (table) => table.dropColumn('source'))
-    this.schema.alterTable('binaries', (table) => table.dropColumn('size'))
-  }
-}

database/migrations/1768620764770_create_fetch_ranges_table.ts

@@ -1,37 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Article ranges handed to worker:fetch, so a range only counts as indexed
- * once it was fetched successfully (CONCEPT.md D18).
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.alterTable('groups', (table) => {
-      // How far the scheduler has queued; last_indexed_id becomes the
-      // watermark below which every article was fetched.
-      table.bigInteger('last_queued_id').notNullable().defaultTo(0)
-    })
-    this.defer(async (db) => {
-      await db.rawQuery('UPDATE groups SET last_queued_id = last_indexed_id')
-    })
-
-    this.schema.createTable('fetch_ranges', (table) => {
-      table.bigIncrements('id')
-      table.integer('group_id').notNullable().references('groups.id').onDelete('CASCADE')
-      table.bigInteger('start_id').notNullable()
-      table.bigInteger('end_id').notNullable()
-      // queued | done | failed
-      table.string('status', 8).notNullable().defaultTo('queued')
-      table.integer('requeues').notNullable().defaultTo(0)
-      table.text('error')
-      table.timestamp('created_at', { useTz: true }).notNullable()
-      table.timestamp('updated_at', { useTz: true }).notNullable()
-      table.index(['group_id', 'status'])
-    })
-  }
-
-  async down() {
-    this.schema.dropTable('fetch_ranges')
-    this.schema.alterTable('groups', (table) => table.dropColumn('last_queued_id'))
-  }
-}

database/migrations/1768620764780_add_archive_contents.ts

@@ -1,25 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/** Archive listings from the first RAR volume (CONCEPT.md D19). */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.alterTable('files', (table) => {
-      // RarInfo of a first volume: format, flags and entries.
-      table.jsonb('archive')
-    })
-    this.schema.alterTable('releases', (table) => {
-      // Password protected: encrypted file data or encrypted headers.
-      table.boolean('encrypted')
-      // Entries of the release's first archive volume.
-      table.jsonb('contents')
-    })
-  }
-
-  async down() {
-    this.schema.alterTable('releases', (table) => {
-      table.dropColumn('contents')
-      table.dropColumn('encrypted')
-    })
-    this.schema.alterTable('files', (table) => table.dropColumn('archive'))
-  }
-}

database/migrations/1768620764790_create_part_conflicts_table.ts

@@ -1,32 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Splitting equal-sized obfuscated volumes (CONCEPT.md D20): the extra
- * articles of conflicted binaries are kept, and the split-off volumes point
- * to the conflicted binary they came from.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.createTable('part_conflicts', (table) => {
-      table.bigInteger('binary_id').notNullable().references('binaries.id').onDelete('CASCADE')
-      table.integer('number').notNullable()
-      table.text('message_id').notNullable()
-      table.integer('bytes').notNullable()
-      table.bigInteger('article_number')
-      table.primary(['binary_id', 'number', 'message_id'])
-    })
-    this.schema.alterTable('binaries', (table) => {
-      table.bigInteger('parent_binary_id').references('binaries.id').onDelete('SET NULL').index()
-      // Last time the splitter processed this conflicted binary.
-      table.timestamp('split_at', { useTz: true })
-    })
-  }
-
-  async down() {
-    this.schema.alterTable('binaries', (table) => {
-      table.dropColumn('split_at')
-      table.dropColumn('parent_binary_id')
-    })
-    this.schema.dropTable('part_conflicts')
-  }
-}

database/migrations/1768620764800_add_md5_to_release_files.ts

@@ -1,16 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/** Full-file MD5 from par2, to verify assembled (e.g. split) volumes. */
-export default class extends BaseSchema {
-  protected tableName = 'release_files'
-
-  async up() {
-    this.schema.alterTable(this.tableName, (table) => {
-      table.string('md5', 32)
-    })
-  }
-
-  async down() {
-    this.schema.alterTable(this.tableName, (table) => table.dropColumn('md5'))
-  }
-}

database/migrations/1768620764810_exclude_legacy_conflicts_from_split.ts

@@ -1,36 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Binaries that were flagged conflicted before part_conflicts existed lost
- * their extra articles, so splitting them only reshuffles the first article
- * per part number, a mix of different volumes. Mark them as never to be
- * split, and remove the volumes and files the first splitter run made from
- * them (see LOG.md, 2026-09-28).
- */
-export default class extends BaseSchema {
-  async up() {
-    this.defer(async (db) => {
-      await db.rawQuery(`
-        WITH boundary AS (
-          SELECT migration_time FROM adonis_schema WHERE name LIKE '%1768620764790%'
-        ), legacy AS (
-          SELECT id FROM binaries
-          WHERE conflicted_at IS NOT NULL AND parent_binary_id IS NULL
-            AND conflicted_at < (SELECT migration_time FROM boundary)
-        ), instances AS (
-          SELECT id, file_id FROM binaries WHERE parent_binary_id IN (SELECT id FROM legacy)
-        ), deleted_files AS (
-          DELETE FROM files WHERE id IN (SELECT file_id FROM instances WHERE file_id IS NOT NULL)
-        ), deleted_instances AS (
-          DELETE FROM binaries WHERE id IN (SELECT id FROM instances)
-        )
-        -- 'infinity': last_seen_at > split_at is never true, so SplitService skips them.
-        UPDATE binaries SET split_at = 'infinity' WHERE id IN (SELECT id FROM legacy)
-      `)
-    })
-  }
-
-  async down() {
-    // The removed volumes were unverifiable; nothing to restore.
-  }
-}

database/migrations/1768620764820_add_crcs_to_parts.ts

@@ -1,30 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * yEnc CRCs of body-derived articles, so split volumes can be verified
- * without downloading them (CONCEPT.md D20): the combined part CRCs of a
- * volume must equal the whole-file crc32 in its last part.
- */
-export default class extends BaseSchema {
-  async up() {
-    for (const tableName of ['parts', 'part_conflicts']) {
-      this.schema.alterTable(tableName, (table) => {
-        table.string('pcrc32', 8)
-        // Decoded length of the part (=ypart end - begin + 1).
-        table.integer('part_size')
-        // Whole-file CRC32, when the poster included it (usually in the last part).
-        table.string('file_crc32', 8)
-      })
-    }
-  }
-
-  async down() {
-    for (const tableName of ['parts', 'part_conflicts']) {
-      this.schema.alterTable(tableName, (table) => {
-        table.dropColumn('pcrc32')
-        table.dropColumn('part_size')
-        table.dropColumn('file_crc32')
-      })
-    }
-  }
-}

database/migrations/1768620764830_add_content_names.ts

@@ -1,40 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Names and groupings from file contents (CONCEPT.md D21): a fingerprint
- * (exact size + MD5 of the first 16 KiB, as par2 records it) and the real
- * filename recovered from par2.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.alterTable('binaries', (table) => {
-      table.string('md5_16k', 32).index()
-      // Filename recovered from file contents (par2); `name` stays as posted.
-      table.text('real_name')
-    })
-    this.schema.alterTable('files', (table) => {
-      table.text('real_name')
-    })
-    for (const tableName of ['parts', 'part_conflicts']) {
-      this.schema.alterTable(tableName, (table) => {
-        // Only for part 1: MD5 of the first 16 KiB of the file.
-        table.string('md5_16k', 32)
-      })
-    }
-    this.schema.alterTable('release_files', (table) => {
-      table.index(['md5_16k'])
-    })
-  }
-
-  async down() {
-    this.schema.alterTable('release_files', (table) => table.dropIndex(['md5_16k']))
-    for (const tableName of ['parts', 'part_conflicts']) {
-      this.schema.alterTable(tableName, (table) => table.dropColumn('md5_16k'))
-    }
-    this.schema.alterTable('files', (table) => table.dropColumn('real_name'))
-    this.schema.alterTable('binaries', (table) => {
-      table.dropColumn('real_name')
-      table.dropColumn('md5_16k')
-    })
-  }
-}

database/migrations/1768620764840_add_search_indexes.ts

@@ -1,43 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Web interface (CONCEPT.md D22): trigram indexes for substring search, and
- * SQL versions of the "readable name" rule used for filtering.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.defer(async (db) => {
-      await db.rawQuery('CREATE EXTENSION IF NOT EXISTS pg_trgm')
-      // Per-file suffixes stripped, as in lib/release fileStem.
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION name_stem(name text) RETURNS text IMMUTABLE LANGUAGE sql AS $$
-          SELECT regexp_replace(name,
-            '(\\.(part\\d+\\.rar|r\\d{2,3}|rar|vol\\d+[+-]\\d+\\.par2|par2|7z(\\.\\d{3})?|zip|\\d{3}|nfo|sfv|nzb|tar|zst|gz|xz|mkv|mp4|avi|m4v|ts|iso|img|flac|mp3|epub|pdf|jpg|png))+$',
-            '', 'i')
-        $$`)
-      // Not an obfuscated token (as in ArchiveService.looksRandom).
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION is_readable_name(name text) RETURNS boolean IMMUTABLE LANGUAGE sql AS $$
-          SELECT name IS NOT NULL AND NOT (
-            (name_stem(name) ~ '^[A-Za-z0-9]{16,}$' AND name_stem(name) ~ '\\d'
-              AND name_stem(name) ~ '[a-z]' AND name_stem(name) ~ '[A-Z]')
-            OR name_stem(name) ~* '^[0-9a-f]{16,}$')
-        $$`)
-      await db.rawQuery('CREATE INDEX IF NOT EXISTS releases_name_trgm ON releases USING gin (name gin_trgm_ops)')
-      await db.rawQuery('CREATE INDEX IF NOT EXISTS files_filename_trgm ON files USING gin (filename gin_trgm_ops)')
-      await db.rawQuery('CREATE INDEX IF NOT EXISTS files_real_name_trgm ON files USING gin (real_name gin_trgm_ops)')
-      await db.rawQuery('CREATE INDEX IF NOT EXISTS releases_posted_at_idx ON releases (posted_at DESC NULLS LAST)')
-    })
-  }
-
-  async down() {
-    this.defer(async (db) => {
-      await db.rawQuery('DROP INDEX IF EXISTS releases_posted_at_idx')
-      await db.rawQuery('DROP INDEX IF EXISTS files_real_name_trgm')
-      await db.rawQuery('DROP INDEX IF EXISTS files_filename_trgm')
-      await db.rawQuery('DROP INDEX IF EXISTS releases_name_trgm')
-      await db.rawQuery('DROP FUNCTION IF EXISTS is_readable_name(text)')
-      await db.rawQuery('DROP FUNCTION IF EXISTS name_stem(text)')
-    })
-  }
-}

database/migrations/1768620764850_add_release_totals.ts

@@ -1,56 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Per-release totals for browsing and sorting (CONCEPT.md D22). A trigger on
- * `files` keeps `releases.file_count` and `releases.size` current, whichever
- * code path assigns files to releases.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.alterTable('files', (table) => {
-      table.bigInteger('size')
-    })
-    this.schema.alterTable('releases', (table) => {
-      table.integer('file_count').notNullable().defaultTo(0).index()
-      table.bigInteger('size').notNullable().defaultTo(0).index()
-    })
-    this.defer(async (db) => {
-      await db.rawQuery(`
-        UPDATE files f SET size = s.size
-        FROM (SELECT id, (SELECT coalesce(sum((v->>'size')::bigint), 0) FROM jsonb_each(message_ids) e(k, v)) AS size FROM files) s
-        WHERE s.id = f.id`)
-      await db.rawQuery(`
-        UPDATE releases r SET file_count = t.n, size = t.size
-        FROM (SELECT release_id, count(*) AS n, coalesce(sum(size), 0) AS size FROM files
-              WHERE release_id IS NOT NULL GROUP BY release_id) t
-        WHERE t.release_id = r.id`)
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION files_release_totals() RETURNS trigger LANGUAGE plpgsql AS $$
-        BEGIN
-          IF TG_OP IN ('UPDATE', 'DELETE') AND OLD.release_id IS NOT NULL THEN
-            UPDATE releases SET file_count = file_count - 1, size = size - coalesce(OLD.size, 0) WHERE id = OLD.release_id;
-          END IF;
-          IF TG_OP IN ('UPDATE', 'INSERT') AND NEW.release_id IS NOT NULL THEN
-            UPDATE releases SET file_count = file_count + 1, size = size + coalesce(NEW.size, 0) WHERE id = NEW.release_id;
-          END IF;
-          RETURN NULL;
-        END $$`)
-      await db.rawQuery(`
-        CREATE TRIGGER files_release_totals
-        AFTER INSERT OR DELETE OR UPDATE OF release_id, size ON files
-        FOR EACH ROW EXECUTE FUNCTION files_release_totals()`)
-    })
-  }
-
-  async down() {
-    this.defer(async (db) => {
-      await db.rawQuery('DROP TRIGGER IF EXISTS files_release_totals ON files')
-      await db.rawQuery('DROP FUNCTION IF EXISTS files_release_totals()')
-    })
-    this.schema.alterTable('releases', (table) => {
-      table.dropColumn('size')
-      table.dropColumn('file_count')
-    })
-    this.schema.alterTable('files', (table) => table.dropColumn('size'))
-  }
-}

database/migrations/1768620764860_add_release_completeness.ts

@@ -1,107 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Stored completeness for the "Complete" filter (CONCEPT.md D22); computing
- * it per request took ~30 s on 34k releases. Triggers keep it current when
- * files, par2 lists, sizes or announced totals change.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.defer(async (db) => {
-      await db.rawQuery('CREATE INDEX IF NOT EXISTS binaries_file_id_idx ON binaries (file_id)')
-      await db.rawQuery('CREATE INDEX IF NOT EXISTS releases_contents_trgm ON releases USING gin ((contents::text) gin_trgm_ops)')
-      await db.rawQuery('ALTER TABLE releases ADD COLUMN complete boolean NOT NULL DEFAULT false')
-      await db.rawQuery('CREATE INDEX releases_complete_idx ON releases (complete)')
-
-      // All files announced by [i/M] are there, or every par2-listed file
-      // (>= 64 KiB) is present by exact size or real name.
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION release_complete(rid bigint) RETURNS boolean STABLE LANGUAGE sql AS $$
-          SELECT coalesce((
-            SELECT CASE
-              WHEN r.file_total IS NOT NULL THEN r.file_count >= r.file_total
-              WHEN EXISTS (SELECT 1 FROM release_files rf WHERE rf.release_id = r.id) THEN NOT EXISTS (
-                SELECT 1 FROM release_files rf
-                WHERE rf.release_id = r.id AND rf.size >= 65536
-                  AND NOT EXISTS (
-                    SELECT 1 FROM files f JOIN binaries b ON b.file_id = f.id
-                    WHERE f.release_id = r.id AND (b.size = rf.size OR f.real_name = rf.name)))
-              ELSE false END
-            FROM releases r WHERE r.id = rid), false)
-        $$`)
-
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION refresh_release_complete(rid bigint) RETURNS void LANGUAGE sql AS $$
-          UPDATE releases SET complete = release_complete(rid) WHERE id = rid AND complete IS DISTINCT FROM release_complete(rid)
-        $$`)
-
-      // files: after the totals trigger has updated file_count (trigger names sort alphabetically).
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION files_release_complete() RETURNS trigger LANGUAGE plpgsql AS $$
-        BEGIN
-          IF TG_OP IN ('UPDATE', 'DELETE') AND OLD.release_id IS NOT NULL THEN PERFORM refresh_release_complete(OLD.release_id); END IF;
-          IF TG_OP IN ('UPDATE', 'INSERT') AND NEW.release_id IS NOT NULL THEN PERFORM refresh_release_complete(NEW.release_id); END IF;
-          RETURN NULL;
-        END $$`)
-      await db.rawQuery(`
-        CREATE TRIGGER files_zz_release_complete
-        AFTER INSERT OR DELETE OR UPDATE OF release_id, size, real_name ON files
-        FOR EACH ROW EXECUTE FUNCTION files_release_complete()`)
-
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION release_files_release_complete() RETURNS trigger LANGUAGE plpgsql AS $$
-        BEGIN
-          PERFORM refresh_release_complete(NEW.release_id);
-          RETURN NULL;
-        END $$`)
-      await db.rawQuery(`
-        CREATE TRIGGER release_files_release_complete
-        AFTER INSERT ON release_files FOR EACH ROW EXECUTE FUNCTION release_files_release_complete()`)
-
-      // An exact size from content inspection can make a par2-listed file "present".
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION binaries_release_complete() RETURNS trigger LANGUAGE plpgsql AS $$
-        BEGIN
-          IF NEW.file_id IS NOT NULL THEN
-            PERFORM refresh_release_complete(f.release_id) FROM files f WHERE f.id = NEW.file_id AND f.release_id IS NOT NULL;
-          END IF;
-          RETURN NULL;
-        END $$`)
-      await db.rawQuery(`
-        CREATE TRIGGER binaries_release_complete
-        AFTER UPDATE OF size ON binaries FOR EACH ROW
-        WHEN (OLD.size IS DISTINCT FROM NEW.size) EXECUTE FUNCTION binaries_release_complete()`)
-
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION releases_file_total_complete() RETURNS trigger LANGUAGE plpgsql AS $$
-        BEGIN
-          PERFORM refresh_release_complete(NEW.id);
-          RETURN NULL;
-        END $$`)
-      await db.rawQuery(`
-        CREATE TRIGGER releases_file_total_complete
-        AFTER UPDATE OF file_total ON releases FOR EACH ROW
-        WHEN (OLD.file_total IS DISTINCT FROM NEW.file_total) EXECUTE FUNCTION releases_file_total_complete()`)
-
-      await db.rawQuery('UPDATE releases SET complete = release_complete(id) WHERE file_count > 0')
-    })
-  }
-
-  async down() {
-    this.defer(async (db) => {
-      await db.rawQuery('DROP TRIGGER IF EXISTS releases_file_total_complete ON releases')
-      await db.rawQuery('DROP TRIGGER IF EXISTS binaries_release_complete ON binaries')
-      await db.rawQuery('DROP TRIGGER IF EXISTS release_files_release_complete ON release_files')
-      await db.rawQuery('DROP TRIGGER IF EXISTS files_zz_release_complete ON files')
-      await db.rawQuery('DROP FUNCTION IF EXISTS releases_file_total_complete()')
-      await db.rawQuery('DROP FUNCTION IF EXISTS binaries_release_complete()')
-      await db.rawQuery('DROP FUNCTION IF EXISTS release_files_release_complete()')
-      await db.rawQuery('DROP FUNCTION IF EXISTS files_release_complete()')
-      await db.rawQuery('DROP FUNCTION IF EXISTS refresh_release_complete(bigint)')
-      await db.rawQuery('DROP FUNCTION IF EXISTS release_complete(bigint)')
-      await db.rawQuery('ALTER TABLE releases DROP COLUMN IF EXISTS complete')
-      await db.rawQuery('DROP INDEX IF EXISTS releases_contents_trgm')
-      await db.rawQuery('DROP INDEX IF EXISTS binaries_file_id_idx')
-    })
-  }
-}

database/migrations/1768620764870_strict_par2_matching.ts

@@ -1,28 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * par2 set id per listed file (traceability; see LOG.md 2026-09-28 for the
- * cross-post merges it helps diagnose), and a stricter "random name" rule:
- * long letter-only tokens with many capitals count as random too
- * (e.g. RmISakytFwXSCEzjVEpaKk).
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.alterTable('release_files', (table) => {
-      table.string('set_id', 32)
-    })
-    this.defer(async (db) => {
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION is_readable_name(name text) RETURNS boolean IMMUTABLE LANGUAGE sql AS $$
-          SELECT name IS NOT NULL AND NOT (
-            (name_stem(name) ~ '^[A-Za-z0-9]{16,}$' AND name_stem(name) ~ '[a-z]' AND name_stem(name) ~ '[A-Z]'
-              AND (name_stem(name) ~ '\\d' OR length(regexp_replace(name_stem(name), '[^A-Z]', '', 'g')) >= 6))
-            OR name_stem(name) ~* '^[0-9a-f]{16,}$')
-        $$`)
-    })
-  }
-
-  async down() {
-    this.schema.alterTable('release_files', (table) => table.dropColumn('set_id'))
-  }
-}

database/migrations/1768620764880_add_release_category.ts

@@ -1,45 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Newznab category per release (CONCEPT.md D23), derived from the name and
- * the archive contents. A stored generated column, so it follows renames
- * and newly listed contents without any application code.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.defer(async (db) => {
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION release_category(name text, contents jsonb) RETURNS integer IMMUTABLE LANGUAGE sql AS $$
-          SELECT CASE
-            -- TV: S01E02, 1x02, or a season pack (S01).
-            WHEN name ~* '(^|[._ -])(S\\d{1,2}E\\d{1,3}|\\d{1,2}x\\d{2})([._ -]|$)' OR name ~* '[._ -]S\\d{1,2}([._ -]|$)' THEN
-              CASE WHEN name ~* '2160p|(^|[._ -])(uhd|4k)([._ -]|$)' THEN 5045
-                   WHEN name ~* '720p|1080p' THEN 5040 ELSE 5030 END
-            -- Movies: a year plus a source or quality marker.
-            WHEN name ~* '(^|[._ (-])(19|20)\\d{2}([._ )-]|$)'
-                 AND name ~* '(480p|576p|720p|1080p|2160p|blu-?ray|web-?dl|webrip|bdrip|brrip|dvdrip|hdtv|x26[45]|h\\.?26[45]|hevc|remux)' THEN
-              CASE WHEN name ~* '2160p|(^|[._ -])(uhd|4k)([._ -]|$)' THEN 2045
-                   WHEN name ~* '720p|1080p|blu-?ray' THEN 2040 ELSE 2030 END
-            -- Audio: lossless or mp3 by contents, otherwise scene music names (-WEB-2026-GROUP).
-            WHEN coalesce(contents::text, '') ~* '\\.flac"' OR name ~* '(^|[._ -])flac([._ -]|$)' THEN 3040
-            WHEN coalesce(contents::text, '') ~* '\\.mp3"' OR name ~* '(^|[._ -])mp3([._ -]|$)'
-                 OR name ~* '-(WEB|CD|CDM|CDS|EP|VINYL|VLS|SAT|FM|DAB)-(19|20)\\d{2}-[A-Za-z0-9]+$' THEN 3010
-            WHEN coalesce(contents::text, '') ~* '\\.(epub|mobi|azw3)"' OR name ~* '(^|[._ -])(epub|ebook|mobi)([._ -]|$)' THEN 7020
-            WHEN coalesce(contents::text, '') ~* '\\.(exe|msi|dmg)"' THEN 4000
-            ELSE 8010
-          END
-        $$`)
-      await db.rawQuery(
-        'ALTER TABLE releases ADD COLUMN category integer GENERATED ALWAYS AS (release_category(name, contents)) STORED'
-      )
-      await db.rawQuery('CREATE INDEX releases_category_idx ON releases (category)')
-    })
-  }
-
-  async down() {
-    this.defer(async (db) => {
-      await db.rawQuery('ALTER TABLE releases DROP COLUMN IF EXISTS category')
-      await db.rawQuery('DROP FUNCTION IF EXISTS release_category(text, jsonb)')
-    })
-  }
-}

database/migrations/1768620764890_ignore_identical_reposts.ts

@@ -1,31 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * A second article with the same part CRC is a repost of the same data, not
- * a different file (CONCEPT.md D20). Drop such recorded conflicts and unflag
- * binaries whose only conflicts were reposts. Binaries flagged before CRCs
- * were stored (split_at = 'infinity') cannot be checked and stay flagged.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.defer(async (db) => {
-      await db.rawQuery(`
-        DELETE FROM part_conflicts c USING parts p
-        WHERE p.binary_id = c.binary_id AND p.number = c.number
-          AND p.pcrc32 IS NOT NULL AND c.pcrc32 IS NOT NULL AND p.pcrc32 = c.pcrc32`)
-      // Split volumes made from reposts are copies of their parent's file.
-      await db.rawQuery(`
-        DELETE FROM binaries v USING binaries b
-        WHERE v.parent_binary_id = b.id AND v.completed_at IS NULL
-          AND b.conflicted_at IS NOT NULL AND b.split_at IS DISTINCT FROM 'infinity'
-          AND NOT EXISTS (SELECT 1 FROM part_conflicts c WHERE c.binary_id = b.id)`)
-      await db.rawQuery(`
-        UPDATE binaries b SET conflicted_at = NULL
-        WHERE b.conflicted_at IS NOT NULL AND b.parent_binary_id IS NULL
-          AND b.split_at IS DISTINCT FROM 'infinity'
-          AND NOT EXISTS (SELECT 1 FROM part_conflicts c WHERE c.binary_id = b.id)`)
-    })
-  }
-
-  async down() {}
-}

database/migrations/1768620764900_add_fetch_bodies_to_groups.ts

@@ -1,24 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Body downloads per group (CONCEPT.md D24). Identifying obfuscated articles
- * costs one body download each; busy groups carry far more of them than the
- * line can fetch, so they are indexed header-only (normal subjects) unless
- * enabled.
- */
-export default class extends BaseSchema {
-  protected tableName = 'groups'
-
-  async up() {
-    this.schema.alterTable(this.tableName, (table) => {
-      table.boolean('fetch_bodies').notNullable().defaultTo(false)
-    })
-    this.defer(async (db) => {
-      await db.rawQuery(`UPDATE groups SET fetch_bodies = true WHERE name = 'alt.binaries.test'`)
-    })
-  }
-
-  async down() {
-    this.schema.alterTable(this.tableName, (table) => table.dropColumn('fetch_bodies'))
-  }
-}

database/migrations/1768620764910_readable_name_any_extension.ts

@@ -1,23 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * is_readable_name(): also judge the stem without an unknown short
- * extension, so `326678d1d07ef12a.bin` counts as random (CONCEPT.md D22).
- */
-export default class extends BaseSchema {
-  async up() {
-    this.defer(async (db) => {
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION is_readable_name(name text) RETURNS boolean IMMUTABLE LANGUAGE sql AS $$
-          WITH s AS (SELECT regexp_replace(name_stem(name), '\\.[A-Za-z0-9]{1,4}$', '') AS stem)
-          SELECT name IS NOT NULL AND NOT (
-            (stem ~ '^[A-Za-z0-9]{16,}$' AND stem ~ '[a-z]' AND stem ~ '[A-Z]'
-              AND (stem ~ '\\d' OR length(regexp_replace(stem, '[^A-Z]', '', 'g')) >= 6))
-            OR stem ~* '^[0-9a-f]{16,}$')
-          FROM s
-        $$`)
-    })
-  }
-
-  async down() {}
-}

database/migrations/1768620764920_add_content_types.ts

@@ -1,22 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * File type by magic bytes (lib/content), and releases that are only an
- * encrypted container (CONCEPT.md D25).
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.alterTable('files', (table) => {
-      table.string('content_type', 16).index()
-    })
-    this.schema.alterTable('releases', (table) => {
-      // e.g. 'nzbe': an encrypted NZB, useless without its key.
-      table.string('container', 16)
-    })
-  }
-
-  async down() {
-    this.schema.alterTable('releases', (table) => table.dropColumn('container'))
-    this.schema.alterTable('files', (table) => table.dropColumn('content_type'))
-  }
-}

database/migrations/1768620764930_add_nzb_import.ts

@@ -1,24 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * NZBs posted to Usenet are read into releases (CONCEPT.md D26): the archive
- * password they may carry, and a lookup of files by their first segment.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.alterTable('releases', (table) => {
-      // From <meta type="password"> of a posted NZB; passed on in our NZBs.
-      table.text('password')
-    })
-    this.defer(async (db) => {
-      await db.rawQuery(`CREATE INDEX IF NOT EXISTS files_first_segment_idx ON files ((message_ids->'1'->>'id'))`)
-    })
-  }
-
-  async down() {
-    this.defer(async (db) => {
-      await db.rawQuery('DROP INDEX IF EXISTS files_first_segment_idx')
-    })
-    this.schema.alterTable('releases', (table) => table.dropColumn('password'))
-  }
-}

database/migrations/1768620764940_name_stem_subtitles.ts

@@ -1,17 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/** name_stem(): also strip subtitle extensions, as lib/release fileStem does. */
-export default class extends BaseSchema {
-  async up() {
-    this.defer(async (db) => {
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION name_stem(name text) RETURNS text IMMUTABLE LANGUAGE sql AS $$
-          SELECT regexp_replace(name,
-            '(\\.(part\\d+\\.rar|r\\d{2,3}|rar|vol\\d+[+-]\\d+\\.par2|par2|7z(\\.\\d{3})?|zip|\\d{3}|nfo|sfv|nzb|tar|zst|gz|xz|mkv|mp4|avi|m4v|ts|iso|img|flac|mp3|epub|pdf|jpg|png|srt|sub|idx|ass|ssa))+$',
-            '', 'i')
-        $$`)
-    })
-  }
-
-  async down() {}
-}

database/migrations/1768620764950_name_stem_split_volumes.ts

@@ -1,21 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * name_stem(): strip a 3-digit split suffix only after a file extension
- * (x.mkv.001), not codec names like H.264 / H.265, matching lib/release.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.defer(async (db) => {
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION name_stem(name text) RETURNS text IMMUTABLE LANGUAGE sql AS $$
-          SELECT regexp_replace(
-            regexp_replace(name, '(\\.(7z|zip|rar|tar|mkv|mp4|avi|m4v|ts|m2ts|iso|img|bin|vob))\\.\\d{3}$', '\\1', 'i'),
-            '(\\.(part\\d+\\.rar|r\\d{2,3}|rar|vol\\d+[+-]\\d+\\.par2|par2|7z|zip|nfo|sfv|nzb|tar|zst|gz|xz|mkv|mp4|avi|m4v|ts|iso|img|flac|mp3|epub|pdf|jpg|png|srt|sub|idx|ass|ssa))+$',
-            '', 'i')
-        $$`)
-    })
-  }
-
-  async down() {}
-}

database/migrations/1768620764960_name_stem_more_extensions.ts

@@ -1,18 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/** name_stem(): same extension list as lib/release fileStem (adds bin, m3u, cue, log, bz2). */
-export default class extends BaseSchema {
-  async up() {
-    this.defer(async (db) => {
-      await db.rawQuery(`
-        CREATE OR REPLACE FUNCTION name_stem(name text) RETURNS text IMMUTABLE LANGUAGE sql AS $$
-          SELECT regexp_replace(
-            regexp_replace(name, '(\\.(7z|zip|rar|tar|mkv|mp4|avi|m4v|ts|m2ts|iso|img|bin|vob))\\.\\d{3}$', '\\1', 'i'),
-            '(\\.(part\\d+\\.rar|r\\d{2,3}|rar|vol\\d+[+-]\\d+\\.par2|par2|7z|zip|nfo|sfv|nzb|srr|srs|txt|md5|sha1|tar|zst|gz|xz|bz2|mkv|mp4|avi|m4v|ts|iso|img|bin|flac|mp3|epub|pdf|jpg|png|srt|sub|idx|ass|ssa|m3u|m3u8|cue|log))+$',
-            '', 'i')
-        $$`)
-    })
-  }
-
-  async down() {}
-}

database/migrations/1768620764970_create_deferred_files_table.ts

@@ -1,47 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Files of uploaders whose bodies are not fetched right away (ngPost, see
- * CONCEPT.md D28): one row per file, from the headers alone. The row keeps
- * the article range, so the file can be found again, and is the queue for
- * worker:deferred, which fetches one body per file.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.createTable('deferred_files', (table) => {
-      table.bigIncrements('id')
-      table.string('file_key', 40).notNullable().unique()
-      // File name from the subject (random for ngPost).
-      table.text('name').notNullable()
-      table.text('subject').notNullable()
-      table.text('poster').notNullable()
-      // Article numbers are counted in the group where the file was first seen.
-      table.text('group_name').notNullable()
-      table.bigInteger('first_article').notNullable()
-      table.text('first_message_id').notNullable()
-      table.bigInteger('last_article').notNullable()
-      table.integer('article_count').notNullable()
-      table.bigInteger('bytes').notNullable()
-      table.timestamp('posted_at', { useTz: true })
-      table.timestamp('first_seen_at', { useTz: true }).notNullable()
-      table.timestamp('last_seen_at', { useTz: true }).notNullable()
-      // pending → sampled (one body read) or expanded (sent to the body queue) or failed.
-      table.string('status', 16).notNullable().defaultTo('pending')
-      table.timestamp('processed_at', { useTz: true })
-      // From the sampled body.
-      table.text('yenc_name')
-      table.bigInteger('size')
-      table.integer('total_parts')
-      table.integer('sampled_part')
-      table.string('content_type', 16)
-      table.boolean('encrypted')
-      table.text('error')
-      table.index(['status', 'last_seen_at'])
-      table.index(['first_seen_at'])
-    })
-  }
-
-  async down() {
-    this.schema.dropTable('deferred_files')
-  }
-}

database/migrations/1768620764980_create_deferred_ranges_table.ts

@@ -1,28 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Deferred articles whose headers don't say which file they belong to
- * (ngPost with random subjects and a new poster every few articles, see
- * CONCEPT.md D28). Only the article range per fetch batch is kept;
- * `deferred:expand-ranges` finds the articles again and queues their bodies.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.createTable('deferred_ranges', (table) => {
-      table.bigIncrements('id')
-      table.text('group_name').notNullable()
-      table.bigInteger('first_article').notNullable()
-      table.bigInteger('last_article').notNullable()
-      table.integer('article_count').notNullable()
-      table.bigInteger('bytes').notNullable()
-      table.timestamp('created_at', { useTz: true }).notNullable()
-      table.timestamp('expanded_at', { useTz: true })
-      table.index(['expanded_at', 'id'])
-      table.index(['created_at'])
-    })
-  }
-
-  async down() {
-    this.schema.dropTable('deferred_ranges')
-  }
-}

database/migrations/1768620764990_create_spots_table.ts

@@ -1,40 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Spotnet spots (CONCEPT.md D29): signed posts in free.pt that point to a
- * compressed NZB. Filled from headers by worker:fetch, then worker:spots
- * verifies each spot, reads its NZB and imports it as a release.
- */
-export default class extends BaseSchema {
-  async up() {
-    this.schema.createTable('spots', (table) => {
-      table.text('message_id').primary()
-      table.text('group_name').notNullable()
-      table.bigInteger('article_number')
-      table.text('title').notNullable()
-      table.text('tag')
-      table.text('poster').notNullable()
-      table.integer('category')
-      table.jsonb('subcats')
-      table.bigInteger('size')
-      table.timestamp('posted_at', { useTz: true })
-      table.integer('key_id')
-      // Spotweb's spotter id (CRC of the poster's key), to check personal disposes.
-      table.text('spotter_id')
-      // pending → imported, or unverified / no-nzb / moderation / removed / failed.
-      table.string('status', 16).notNullable().defaultTo('pending')
-      table.bigInteger('release_id').references('releases.id').onDelete('SET NULL')
-      table.text('website')
-      table.text('error')
-      table.timestamp('created_at', { useTz: true }).notNullable()
-      table.timestamp('processed_at', { useTz: true })
-      table.index(['status', 'created_at'])
-      table.index(['release_id'])
-      table.index(['status', 'release_id'])
-    })
-  }
-
-  async down() {
-    this.schema.dropTable('spots')
-  }
-}

database/migrations/1768620765000_partial_indexes.ts

@@ -1,44 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Smaller indexes (LOG.md 2026-09-28 (17)). 99.5% of releases have no files
- * yet (most never will), and search only looks at releases with files; most
- * binaries have no size, fingerprint, file or parent. Indexing only the rows
- * that queries can match saves GBs and work on every insert. Built
- * concurrently, so ingest keeps running.
- */
-const INDEXES: [table: string, name: string, definition: string][] = [
-  ['releases', 'releases_name_trgm', `USING gin (name gin_trgm_ops) WHERE file_count > 0`],
-  ['releases', 'releases_contents_trgm', `USING gin ((contents::text) gin_trgm_ops) WHERE file_count > 0`],
-  ['releases', 'releases_posted_at_idx', `(posted_at DESC NULLS LAST) WHERE file_count > 0`],
-  ['releases', 'releases_size_index', `(size) WHERE file_count > 0`],
-  ['releases', 'releases_file_count_index', `(file_count) WHERE file_count > 0`],
-  ['releases', 'releases_complete_idx', `(complete) WHERE file_count > 0`],
-  ['releases', 'releases_category_idx', `(category) WHERE file_count > 0`],
-  ['binaries', 'binaries_md5_16k_index', `(md5_16k) WHERE md5_16k IS NOT NULL`],
-  ['binaries', 'binaries_size_index', `(size) WHERE size IS NOT NULL`],
-  ['binaries', 'binaries_file_id_idx', `(file_id) WHERE file_id IS NOT NULL`],
-  ['binaries', 'binaries_parent_binary_id_index', `(parent_binary_id) WHERE parent_binary_id IS NOT NULL`],
-]
-
-export default class extends BaseSchema {
-  static disableTransactions = true
-
-  async up() {
-    for (const [table, name, definition] of INDEXES) {
-      // Build the partial index under a temporary name, then swap it in.
-      await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS ${name}_partial`)
-      await this.db.rawQuery(`CREATE INDEX CONCURRENTLY ${name}_partial ON ${table} ${definition}`)
-      await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS ${name}`)
-      await this.db.rawQuery(`ALTER INDEX ${name}_partial RENAME TO ${name}`)
-    }
-  }
-
-  async down() {
-    for (const [table, name, definition] of INDEXES) {
-      const full = definition.replace(/ WHERE .*$/, '')
-      await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS ${name}`)
-      await this.db.rawQuery(`CREATE INDEX CONCURRENTLY ${name} ON ${table} ${full}`)
-    }
-  }
-}

database/migrations/1768620765010_spot_listings.ts

@@ -1,41 +0,0 @@
-import { BaseSchema } from '@adonisjs/lucid/schema'
-
-/**
- * Spot listings (CONCEPT.md D29): old spots become a release without files
- * (source `spot`), whose NZB is fetched when first downloaded. Search shows
- * releases with files or spot listings, so the partial release indexes
- * cover both. Spots are verified from their overview when fetched.
- */
-const VISIBLE = `(file_count > 0 OR source = 'spot')`
-const INDEXES: [name: string, definition: string][] = [
-  ['releases_name_trgm', `USING gin (name gin_trgm_ops)`],
-  ['releases_contents_trgm', `USING gin ((contents::text) gin_trgm_ops)`],
-  ['releases_posted_at_idx', `(posted_at DESC NULLS LAST)`],
-  ['releases_size_index', `(size)`],
-  ['releases_file_count_index', `(file_count)`],
-  ['releases_complete_idx', `(complete)`],
-  ['releases_category_idx', `(category)`],
-]
-
-export default class extends BaseSchema {
-  static disableTransactions = true
-
-  async up() {
-    await this.db.rawQuery(`ALTER TABLE spots ADD COLUMN IF NOT EXISTS verified boolean`)
-    await this.swap(VISIBLE)
-  }
-
-  async down() {
-    await this.swap('file_count > 0')
-    await this.db.rawQuery(`ALTER TABLE spots DROP COLUMN IF EXISTS verified`)
-  }
-
-  private async swap(predicate: string) {
-    for (const [name, definition] of INDEXES) {
-      await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS ${name}_next`)
-      await this.db.rawQuery(`CREATE INDEX CONCURRENTLY ${name}_next ON releases ${definition} WHERE ${predicate}`)
-      await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS ${name}`)
-      await this.db.rawQuery(`ALTER INDEX ${name}_next RENAME TO ${name}`)
-    }
-  }
-}

database/migrations/1770000000100_create_groups.ts

@@ -0,0 +1,27 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The indexed groups (REBUILD.md, group list). `priority` decides a
+ * crosspost's owner: the indexed group in its Xref with the lowest number.
+ * New groups get numbers after the existing ones, so they never take
+ * articles from groups already done.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('groups', (table) => {
+      table.increments('id')
+      table.string('name', 255).notNullable().unique()
+      table.integer('priority').notNullable().unique()
+      table.boolean('active').notNullable().defaultTo(true)
+      // From GROUP when planned: the server's oldest and newest article.
+      table.bigInteger('low').nullable()
+      table.bigInteger('high').nullable()
+      table.timestamp('planned_at', { useTz: true }).nullable()
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('groups')
+  }
+}

database/migrations/1770000000200_create_rule_sets.ts

@@ -0,0 +1,21 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * Versions of what workers drop (REBUILD.md, rules and recovery): dropped
+ * signatures and tools, the per-article limit, the sample share. Every range
+ * records the version it ran with.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('rule_sets', (table) => {
+      table.integer('version').primary()
+      table.jsonb('rules').notNullable()
+      table.text('note').nullable()
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('rule_sets')
+  }
+}

database/migrations/1770000000300_create_ranges.ts

@@ -0,0 +1,38 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * A group's history in leasable pieces (REBUILD.md, coordinator):
+ * planned → (spooled by the prefetcher) → leased → done. `next_id` is the
+ * resume point: everything before it has been stored.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('ranges', (table) => {
+      table.bigIncrements('id')
+      table.integer('group_id').notNullable().references('id').inTable('groups').onDelete('CASCADE')
+      table.bigInteger('start_id').notNullable()
+      table.bigInteger('end_id').notNullable()
+      table.bigInteger('next_id').notNullable()
+      table.string('status', 8).notNullable().defaultTo('planned')
+      // Lower first; the coordinator plans the oldest era first.
+      table.integer('priority').notNullable().defaultTo(0)
+      table.integer('rule_version').nullable()
+      table.integer('node_id').nullable()
+      table.integer('attempts').notNullable().defaultTo(0)
+      // Chunks of this range in the spool (prefetch); all of them when spooled_at is set.
+      table.integer('spooled_chunks').notNullable().defaultTo(0)
+      table.timestamp('spooled_at', { useTz: true }).nullable()
+      table.timestamp('leased_at', { useTz: true }).nullable()
+      table.timestamp('heartbeat_at', { useTz: true }).nullable()
+      table.timestamp('done_at', { useTz: true }).nullable()
+      table.bigInteger('headers').notNullable().defaultTo(0)
+      table.text('error').nullable()
+      table.unique(['group_id', 'start_id'])
+    })
+    this.schema.raw(`CREATE INDEX ranges_todo_idx ON ranges (priority, id) WHERE status IN ('planned', 'leased')`)
+  }
+
+  async down() {
+    this.schema.dropTable('ranges')
+  }
+}

database/migrations/1770000000400_create_releases.ts

@@ -0,0 +1,41 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The release catalog (REBUILD.md, part 4): one row per release, the hot
+ * part on fast disk. Updated with aggregates once per batch; the files
+ * themselves are in release_chunks.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('releases', (table) => {
+      table.bigIncrements('id')
+      table.string('release_key', 40).notNullable().unique()
+      table.text('name').notNullable()
+      table.string('source', 16).notNullable()
+      table.text('poster').nullable()
+      table.jsonb('groups').notNullable()
+      table.timestamp('posted_at', { useTz: true }).nullable()
+      // Files announced by the [i/N] counter, when present.
+      table.integer('file_total').nullable()
+      table.integer('files').notNullable().defaultTo(0)
+      table.bigInteger('segments').notNullable().defaultTo(0)
+      table.bigInteger('bytes').notNullable().defaultTo(0)
+      table.integer('chunks').notNullable().defaultTo(0)
+      table.bigInteger('first_article').nullable()
+      table.bigInteger('last_article').nullable()
+      table.boolean('readable').notNullable().defaultTo(false)
+      table.boolean('has_par2').notNullable().defaultTo(false)
+      table.boolean('complete').nullable()
+      table.timestamp('settled_at', { useTz: true }).nullable()
+      table.timestamp('first_seen_at', { useTz: true }).notNullable().defaultTo(this.now())
+      table.timestamp('last_seen_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+    this.schema.raw(`CREATE EXTENSION IF NOT EXISTS pg_trgm`)
+    this.schema.raw(`CREATE INDEX releases_name_trgm ON releases USING gin (name gin_trgm_ops)`)
+    this.schema.raw(`CREATE INDEX releases_posted_at_idx ON releases (posted_at DESC)`)
+  }
+
+  async down() {
+    this.schema.dropTable('releases')
+  }
+}

database/migrations/1770000000500_create_release_chunks.ts

@@ -0,0 +1,34 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+import env from '#start/env'
+
+/**
+ * The chunk store (REBUILD.md, part 3; decision a): one write-once row per
+ * release per batch, its files encoded by lib/chunks. Partitioned by month
+ * of writing, in CHUNK_TABLESPACE (the HDD) when set; ChunkPartitions keeps
+ * the next months' partitions ready. A re-run of a range replaces its chunks
+ * (by range_id).
+ */
+export default class extends BaseSchema {
+  async up() {
+    const tablespace = env.get('CHUNK_TABLESPACE')
+    const where = tablespace ? ` TABLESPACE ${tablespace}` : ''
+    this.schema.raw(`
+      CREATE TABLE release_chunks (
+        id bigserial,
+        release_id bigint NOT NULL,
+        range_id bigint NOT NULL,
+        batch_seq integer NOT NULL,
+        files integer NOT NULL,
+        segments integer NOT NULL,
+        payload bytea NOT NULL,
+        created_at timestamptz NOT NULL DEFAULT now(),
+        PRIMARY KEY (id, created_at)
+      ) PARTITION BY RANGE (created_at)${where}`)
+    this.schema.raw(`CREATE INDEX release_chunks_release_idx ON release_chunks (release_id)`)
+    this.schema.raw(`CREATE INDEX release_chunks_range_idx ON release_chunks (range_id)`)
+  }
+
+  async down() {
+    this.schema.raw(`DROP TABLE IF EXISTS release_chunks CASCADE`)
+  }
+}

database/migrations/1770000000600_create_fragments.ts

@@ -0,0 +1,24 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * Segments of files cut off by a range's edge (REBUILD.md, fragments): kept
+ * by range, merged when the neighbouring range is done, then written as a
+ * chunk.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('fragments', (table) => {
+      table.bigIncrements('id')
+      table.bigInteger('range_id').notNullable()
+      table.string('file_key', 40).notNullable()
+      table.bigInteger('article_number').notNullable()
+      table.jsonb('segment').notNullable()
+      table.index(['file_key'])
+      table.index(['range_id'])
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('fragments')
+  }
+}

database/migrations/1770000000700_create_probe_candidates.ts

@@ -0,0 +1,25 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The probe queue (REBUILD.md, part 6): per release the few articles phase 2
+ * should read. Phase 1 only fills it.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('probe_candidates', (table) => {
+      table.bigIncrements('id')
+      table.bigInteger('release_id').notNullable()
+      table.string('kind', 8).notNullable()
+      table.text('message_id').notNullable()
+      table.text('file').notNullable()
+      table.string('status', 8).notNullable().defaultTo('waiting')
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+      table.unique(['release_id', 'kind', 'message_id'])
+    })
+    this.schema.raw(`CREATE INDEX probe_candidates_waiting_idx ON probe_candidates (id) WHERE status = 'waiting'`)
+  }
+
+  async down() {
+    this.schema.dropTable('probe_candidates')
+  }
+}

database/migrations/1770000000800_create_rule_bookkeeping.ts

@@ -0,0 +1,35 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * What the rules dropped, per range (REBUILD.md, rules and recovery), and
+ * the sample of it kept in full.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('rule_counts', (table) => {
+      table.bigInteger('range_id').notNullable()
+      table.string('rule', 12).notNullable()
+      table.text('key').notNullable()
+      table.bigInteger('articles').notNullable().defaultTo(0)
+      table.bigInteger('files').notNullable().defaultTo(0)
+      table.primary(['range_id', 'rule', 'key'])
+    })
+    this.schema.createTable('rule_samples', (table) => {
+      table.bigIncrements('id')
+      table.bigInteger('range_id').notNullable()
+      table.string('rule', 12).notNullable()
+      table.text('rule_key').notNullable()
+      table.string('release_key', 40).notNullable()
+      table.text('name').notNullable()
+      table.integer('files').notNullable()
+      table.integer('segments').notNullable()
+      table.binary('payload').notNullable()
+      table.index(['rule', 'rule_key'])
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('rule_samples')
+    this.schema.dropTable('rule_counts')
+  }
+}