Spots: the spotter and the spot inspector, passwords kept apart
Test and deploy / test (push) Successful in 50s
Test and deploy / edge (push) Successful in 2m1s
Test and deploy / deploy (push) Successful in 1m28s

Two processes, one NNTP connection each (the coordinator 40 -> 38):
- spots:read (service spotter): free.pt's overviews after the last article
  seen, each spot's full header verified (lib/spotnet from voyager 1, as
  Spotweb), its description, website and NZB articles; the password its
  title or text names; delete/dispose messages applied.
- spots:inspect (service spot-inspector): each spot's compressed NZB read,
  its files listed (spot_files), each matched to the header pass's release
  by the same release identity in one of its groups within 48 h
  (release_spots); unmatched spots matched again hourly, after 24 h a
  release of their own (source spot) with the spot's NZB.
- release_passwords: per release and password, its source (nzb, spot-title,
  spot-text) and spot; the catalog keeps only passworded.
XoverClient.head(); PROBE_CONNECTIONS default 10.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Daan MeijerandClaude Opus 5.5 committed 2026-10-13 12:10:10 +02:00
1 parent 3fd08b598a
commit 7df5a638ca
20 files changed
+1782 -8

No files matched your search

+7 -1
View File
@@ -1,3 +1,4 @@
import { gunzipSync } from 'node:zlib'
import db from '@adonisjs/lucid/services/db'
import { decodeChunk, type ChunkFile } from '#lib/chunks/index'
import { buildNzb } from '#lib/nzb/index'
@@ -19,8 +20,13 @@ class NzbService {
}
async build(releaseId: number): Promise<{ name: string; xml: string } | null> {
const release = await db.from('releases').where('id', releaseId).select('name').first()
const release = await db.from('releases').where('id', releaseId).select('name', 'source').first()
if (!release) return null
// A release a spot made of its own (no files matched the header pass's): the spot's NZB as read.
if (release.source === 'spot') {
const spot = await db.from('spots').where('release_id', releaseId).whereNotNull('nzb').select('nzb').first()
if (spot) return { name: release.name, xml: gunzipSync(spot.nzb).toString('utf8') }
}
const xml = buildNzb(
(await this.files(releaseId)).map((f) => ({
name: f.name,
+331
View File
@@ -0,0 +1,331 @@
import { createHash } from 'node:crypto'
import { gzipSync } from 'node:zlib'
import db from '@adonisjs/lucid/services/db'
import logger from '@adonisjs/core/services/logger'
import env from '#start/env'
import coordinatorConfig from '#config/coordinator'
import { XoverClient } from '#lib/nntp/index'
import { decodeSpotBinary } from '#lib/spotnet/index'
import { parseNzb, type ParsedNzb } from '#lib/nzb/index'
import { parseSubject } from '#lib/headers/index'
import { releaseIdentity } from '#lib/release/index'
/** Spots claimed per round, newest first. */
const BATCH = 20
/** Between rounds when there was nothing to do. */
const IDLE_MS = 60_000
/** Unmatched spots are matched again this often (the header pass may have built their releases since). */
const REMATCH_MS = 60 * 60 * 1000
/** A file's release posted further apart than this from the NZB's date is another upload of the same name. */
const MAX_SPREAD_MS = 48 * 3600 * 1000
export interface InspectRound {
inspected: number
matched: number
standalone: number
outcomes: Record<string, number>
}
interface SpotFile {
idx: number
name: string
subject: string
poster: string
groups: string[]
postedAt: Date | null
bytes: number
segments: number
}
/**
* A placeholder NZB (Dummy.rar + Dummy.nzb, shared by many spots): under 1% of the size the spot announces, or under
* 16 KB in all (text-only spots). Its files aren't listed (voyager 1's isDummyNzb).
*/
export function isDummyNzb(nzb: ParsedNzb, spotSize: number): boolean {
const bytes = nzb.files.reduce((sum, f) => sum + f.segments.reduce((s, x) => s + x.bytes, 0), 0)
return bytes < 16 * 1024 || (spotSize > 0 && bytes < spotSize / 100)
}
/** A file of an NZB as the header pass names it: the subject's file name, by the same parser. */
export function nzbFileName(subject: string): { name: string; base: string } {
const parsed = parseSubject(subject) ?? parseSubject(`${subject} (1/1)`)
return parsed ? { name: parsed.name, base: parsed.base } : { name: subject, base: subject }
}
/**
* The spot inspector (REBUILD.md, spots): looks inside each listed spot, on
* one NNTP connection of its own. It reads the spot's compressed NZB (its
* articles in alt.binaries.ftd), keeps the list of files in `spot_files`,
* and matches every file to the release the header pass built for it: the
* release identity of the file's subject and poster (lib/release, as the
* edge workers compute it), in one of the file's groups, posted within 48 h.
* Each matched release is linked in `release_spots`, and gets the spot's
* password in `release_passwords` (the NZB's own `<meta type="password">`
* first, else what the spot's title or text names) and `passworded`. A spot
* whose files match nothing is matched again every hour; after
* SPOT_STANDALONE_AFTER_HOURS it becomes a release of its own (source `spot`)
* with its NZB.
*/
export class SpotInspector {
private stopped = false
private client?: XoverClient
private groupIds = new Map<string, number | null>()
private rematchedAt = 0
stop() {
this.stopped = true
this.client?.close()
}
async run() {
while (!this.stopped) {
let round: InspectRound | null = null
try {
round = await this.round()
if (round.inspected > 0 || round.standalone > 0) {
const outcomes = Object.entries(round.outcomes).map(([k, n]) => `${n} ${k}`).join(', ')
logger.info(`Spots inspected: ${round.inspected} (${outcomes}); ${round.matched} matched releases, ${round.standalone} of their own.`)
}
} catch (error: any) {
logger.error(`Spot inspection failed: ${error.message}`)
this.client?.close()
this.client = undefined
}
if (!round || round.inspected === 0) {
for (let waited = 0; waited < IDLE_MS && !this.stopped; waited += 1000) await new Promise((r) => setTimeout(r, 1000))
}
}
}
private async connect(): Promise<XoverClient> {
this.client ??= await XoverClient.connect({
host: env.get('NNTP_HOST')!,
port: env.get('NNTP_PORT', 563),
user: env.get('NNTP_USER'),
password: env.get('NNTP_PASSWORD'),
secure: env.get('NNTP_SECURE', true),
compress: false,
timeoutMs: coordinatorConfig.nntpTimeoutMs,
})
return this.client
}
async round(): Promise<InspectRound> {
const round: InspectRound = { inspected: 0, matched: 0, standalone: 0, outcomes: {} }
const spots = await db.from('spots').where('status', 'listed').orderBy('posted_at', 'desc').limit(BATCH)
for (const spot of spots) {
if (this.stopped) break
const outcome = await this.inspect(spot)
round.inspected++
round.outcomes[outcome.status] = (round.outcomes[outcome.status] ?? 0) + 1
round.matched += outcome.matched
}
if (Date.now() - this.rematchedAt >= REMATCH_MS && !this.stopped) {
this.rematchedAt = Date.now()
const again = await this.rematch()
round.matched += again.matched
round.standalone += again.standalone
}
return round
}
/** One spot: its NZB read, its files listed and matched. */
async inspect(spot: any): Promise<{ status: string; matched: number }> {
const done = async (status: string, extra: Record<string, unknown> = {}) => {
await db.from('spots').where('message_id', spot.message_id).update({ status, inspected_at: new Date(), ...extra })
return { status, matched: 0 }
}
const segments: string[] = spot.nzb_segments ?? []
if (segments.length === 0) return done('no-nzb')
if (segments.length > coordinatorConfig.spots.maxNzbSegments) return done('no-nzb', { error: `${segments.length} NZB articles` })
const client = await this.connect()
const bodies = await client.bodies(segments)
if (bodies.some((b) => b === null)) return done('nzb-gone', { error: `${bodies.filter((b) => b === null).length} of ${segments.length} NZB articles gone` })
let xml: string
try {
xml = decodeSpotBinary(bodies as Buffer[]).toString('utf8')
} catch (error: any) {
return done('nzb-unreadable', { error: String(error.message).slice(0, 500) })
}
return this.inspectNzb(spot, xml)
}
/** The files of a spot's NZB listed and matched (the tests feed this directly). */
async inspectNzb(spot: any, xml: string): Promise<{ status: string; matched: number }> {
const nzb = parseNzb(xml)
const finish = (status: string, extra: Record<string, unknown> = {}) =>
db.from('spots').where('message_id', spot.message_id).update({ status, inspected_at: new Date(), ...extra })
if (!nzb || nzb.files.length === 0) {
await finish('no-nzb')
return { status: 'no-nzb', matched: 0 }
}
if (isDummyNzb(nzb, Number(spot.size ?? 0))) {
await finish('dummy')
return { status: 'dummy', matched: 0 }
}
const files: SpotFile[] = nzb.files.map((f, idx) => ({
idx,
name: nzbFileName(f.subject).name.slice(0, 1000),
subject: f.subject.slice(0, 2000),
poster: f.poster,
groups: f.groups,
postedAt: f.date ? new Date(f.date * 1000) : null,
bytes: f.segments.reduce((s, x) => s + x.bytes, 0),
segments: f.segments.length,
}))
// The NZB's own password over what the spot's text names.
const password = nzb.meta.password ? { password: nzb.meta.password.slice(0, 200), source: 'nzb' } : spot.password ? { password: spot.password, source: spot.password_source } : null
await db.transaction(async (trx) => {
await trx.from('spot_files').where('spot_id', spot.message_id).delete()
for (let i = 0; i < files.length; i += 1000) {
await trx.table('spot_files').multiInsert(
files.slice(i, i + 1000).map((f) => ({
spot_id: spot.message_id,
idx: f.idx,
name: f.name,
subject: f.subject,
poster: f.poster,
groups: f.groups,
posted_at: f.postedAt,
bytes: f.bytes,
segments: f.segments,
}))
)
}
await trx
.from('spots')
.where('message_id', spot.message_id)
.update({
status: 'inspected',
inspected_at: new Date(),
files: files.length,
nzb: gzipSync(xml),
...(password && password.source === 'nzb' ? { password: password.password, password_source: 'nzb' } : {}),
})
})
const matched = await this.match(spot.message_id)
return { status: 'inspected', matched }
}
private async groupId(name: string): Promise<number | null> {
if (!this.groupIds.has(name)) {
const row = await db.from('groups').where('name', name).select('id').first()
this.groupIds.set(name, row ? Number(row.id) : null)
}
return this.groupIds.get(name)!
}
/**
* Matches a spot's listed files to releases and links them; how many releases. A file's release: the same
* identity (releaseIdentity over its subject and poster) in one of its groups, the one posted nearest the file.
*/
async match(spotId: string): Promise<number> {
const spot = await db.from('spots').where('message_id', spotId).first()
if (!spot || spot.status !== 'inspected') return 0
const files = await db.from('spot_files').where('spot_id', spotId).orderBy('idx')
const wanted: { idx: number; key: string; groups: number[]; at: number | null }[] = []
for (const f of files) {
if (!f.subject) continue
const { name, base } = nzbFileName(f.subject)
const identity = releaseIdentity({ poster: f.poster ?? '', subject: base, name, weak: false })
if (!identity) continue
const groups = (await Promise.all((f.groups ?? []).map((g: string) => this.groupId(g)))).filter((g): g is number => g !== null)
if (groups.length > 0) wanted.push({ idx: f.idx, key: identity.key, groups, at: f.posted_at ? new Date(f.posted_at).getTime() : null })
}
if (wanted.length === 0) return 0
const { rows } = await db.rawQuery(
`SELECT id, identity_key, group_id, posted_at FROM releases WHERE identity_key = ANY(?) AND group_id = ANY(?)`,
[[...new Set(wanted.map((w) => w.key))], [...new Set(wanted.flatMap((w) => w.groups))]]
)
const byKey = new Map<string, { id: number; group: number; at: number | null }[]>()
for (const r of rows) {
const list = byKey.get(r.identity_key) ?? []
list.push({ id: Number(r.id), group: Number(r.group_id), at: r.posted_at ? new Date(r.posted_at).getTime() : null })
byKey.set(r.identity_key, list)
}
const fileRelease = new Map<number, number>()
for (const w of wanted) {
const near = (byKey.get(w.key) ?? [])
.filter((r) => w.groups.includes(r.group))
.map((r) => ({ ...r, spread: r.at === null || w.at === null ? 0 : Math.abs(r.at - w.at) }))
.filter((r) => r.spread <= MAX_SPREAD_MS)
.sort((a, b) => a.spread - b.spread)
if (near[0]) fileRelease.set(w.idx, near[0].id)
}
const perRelease = new Map<number, number>()
for (const id of fileRelease.values()) perRelease.set(id, (perRelease.get(id) ?? 0) + 1)
await db.transaction(async (trx) => {
for (const [idx, releaseId] of fileRelease) await trx.from('spot_files').where({ spot_id: spotId, idx }).update({ release_id: releaseId })
for (const [releaseId, n] of perRelease) {
await trx.table('release_spots').insert({ release_id: releaseId, spot_id: spotId, files: n }).onConflict(['release_id', 'spot_id']).merge({ files: n })
}
if (spot.password && perRelease.size > 0) {
await trx
.table('release_passwords')
.multiInsert([...perRelease.keys()].map((id) => ({ release_id: id, password: spot.password, source: spot.password_source ?? 'spot-text', spot_id: spotId })))
.onConflict(['release_id', 'password'])
.ignore()
await trx.from('releases').whereIn('id', [...perRelease.keys()]).update({ passworded: true })
}
const top = [...perRelease.entries()].sort((a, b) => b[1] - a[1])[0]
await trx.from('spots').where('message_id', spotId).update({ matched: fileRelease.size, release_id: top ? top[0] : null })
})
return perRelease.size
}
/** Unmatched inspected spots matched again; the ones that waited long enough become releases of their own. */
async rematch(): Promise<{ matched: number; standalone: number }> {
const out = { matched: 0, standalone: 0 }
const spots = await db.from('spots').where('status', 'inspected').where('matched', 0).whereNull('release_id').orderBy('inspected_at').limit(500)
const cutoff = Date.now() - coordinatorConfig.spots.standaloneAfterHours * 3600 * 1000
for (const spot of spots) {
if (this.stopped) break
const n = await this.match(spot.message_id)
out.matched += n
if (n === 0 && spot.posted_at && new Date(spot.posted_at).getTime() < cutoff && (await this.standalone(spot.message_id))) out.standalone++
}
return out
}
/** A release of the spot's own (source `spot`), from its listed files; false when it has none. */
async standalone(spotId: string): Promise<boolean> {
const spot = await db.from('spots').where('message_id', spotId).first()
const files = await db.from('spot_files').where('spot_id', spotId)
if (!spot || files.length === 0) return false
const groups = [...new Set(files.flatMap((f: any) => f.groups ?? []))]
const groupId = groups.length > 0 ? await this.groupId(groups[0]) : null
const key = createHash('sha1').update(`spot\n${spotId}`).digest('hex')
const [row] = await db
.table('releases')
.insert({
release_key: `spot:${key}`,
identity_key: key,
group_id: groupId,
name: spot.title.slice(0, 1000),
source: 'spot',
poster: spot.poster,
groups: JSON.stringify(groups),
posted_at: spot.posted_at,
last_posted_at: spot.posted_at,
files: files.length,
segments: files.reduce((s: number, f: any) => s + Number(f.segments), 0),
bytes: files.reduce((s: number, f: any) => s + Number(f.bytes), 0),
readable: true,
passworded: spot.password ? true : null,
settled_at: new Date(),
})
.onConflict('release_key')
.ignore()
.returning('id')
if (!row) return false
const releaseId = Number(row.id)
await db.transaction(async (trx) => {
await trx.table('release_spots').insert({ release_id: releaseId, spot_id: spotId, files: files.length }).onConflict(['release_id', 'spot_id']).ignore()
if (spot.password) {
await trx.table('release_passwords').insert({ release_id: releaseId, password: spot.password, source: spot.password_source ?? 'spot-text', spot_id: spotId }).onConflict(['release_id', 'password']).ignore()
}
await trx.from('spots').where('message_id', spotId).update({ release_id: releaseId })
})
return true
}
}
+184
View File
@@ -0,0 +1,184 @@
import db from '@adonisjs/lucid/services/db'
import logger from '@adonisjs/core/services/logger'
import env from '#start/env'
import coordinatorConfig from '#config/coordinator'
import { MissingArticle, XoverClient } from '#lib/nntp/index'
import {
decodeSpotText,
moderationTarget,
parseSpotArticleHeaders,
parseSpotHeader,
parseSpotXml,
passwordFromSpot,
spotterId,
verifySpotHeader,
} from '#lib/spotnet/index'
/** Overviews asked for at a time. */
const XOVER_STEP = 1000
/** A personal dispose counts within this long of the spot (Spotweb). */
const DISPOSE_WINDOW_SECONDS = 5 * 86400
export interface SpotterRound {
read: number
listed: number
unverified: number
moderation: number
}
/**
* The spotter (REBUILD.md, spots): Spotnet's spots, from the headers of
* free.pt into `spots`, on one NNTP connection of its own. Every few
* minutes it reads the overviews after the last article it saw (the first
* time the last SPOT_BACKFILL), and for each spot the full header: its
* signature is checked there (lib/spotnet, as Spotweb does), and the X-XML
* gives the description, the website and the NZB's articles. A password the
* title (`{{pw}}`) or the description names is kept with the spot; the spot
* inspector carries it to the releases. A delete/dispose message marks its
* target `removed` and takes back what the target gave the releases.
*/
export class Spotter {
private stopped = false
private client?: XoverClient
stop() {
this.stopped = true
this.client?.close()
}
async run() {
while (!this.stopped) {
const started = Date.now()
try {
const round = await this.round()
if (round.read > 0) {
logger.info(`Spots: ${round.read} articles read, ${round.listed} spots listed, ${round.unverified} unverified, ${round.moderation} moderation.`)
}
} catch (error: any) {
logger.error(`Spot round failed: ${error.message}`)
this.client?.close()
this.client = undefined
}
const wait = coordinatorConfig.spots.everySeconds * 1000
for (let waited = Date.now() - started; waited < wait && !this.stopped; waited += 1000) await new Promise((r) => setTimeout(r, 1000))
}
}
private async connect(): Promise<XoverClient> {
this.client ??= await XoverClient.connect({
host: env.get('NNTP_HOST')!,
port: env.get('NNTP_PORT', 563),
user: env.get('NNTP_USER'),
password: env.get('NNTP_PASSWORD'),
secure: env.get('NNTP_SECURE', true),
compress: false,
timeoutMs: coordinatorConfig.nntpTimeoutMs,
})
return this.client
}
/** The articles after the last one seen, up to the group's high. */
async round(): Promise<SpotterRound> {
const { group, backfill } = coordinatorConfig.spots
const client = await this.connect()
const { low, high } = await client.group(group)
const state = await db.from('spot_state').where('group_name', group).first()
let next = state ? Number(state.last_article) + 1 : Math.max(low, high - backfill + 1)
const round: SpotterRound = { read: 0, listed: 0, unverified: 0, moderation: 0 }
while (next <= high && !this.stopped) {
const to = Math.min(high, next + XOVER_STEP - 1)
const overviews = await client.xover(next, to)
round.read += overviews.length
for (const [number, fields] of overviews) {
if (this.stopped) return round
const messageId = (fields['message-id'] ?? '').replace(/^<|>$/g, '')
if (!messageId || !parseSpotHeader(fields.from ?? '', fields.subject ?? '', messageId, fields.date)) continue
const outcome = await this.read(client, messageId, number)
if (outcome) round[outcome]++
}
await db
.table('spot_state')
.insert({ group_name: group, last_article: to })
.onConflict('group_name')
.merge({ last_article: to })
next = to + 1
}
return round
}
/** One spot by its full header; what it was, or null when it was there already or is gone. */
async read(client: XoverClient, messageId: string, articleNumber: number | null): Promise<'listed' | 'unverified' | 'moderation' | null> {
if (await db.from('spots').where('message_id', messageId).first()) return null
let raw: Buffer
try {
raw = await client.head(messageId)
} catch (error) {
if (error instanceof MissingArticle) return null
throw error
}
return this.store(raw.toString('latin1'), messageId, articleNumber)
}
/** Verifies and stores a spot from its raw header (the tests feed this directly). */
async store(raw: string, messageId: string, articleNumber: number | null): Promise<'listed' | 'unverified' | 'moderation' | null> {
const { headers, xml: xmlText } = parseSpotArticleHeaders(raw)
const header = parseSpotHeader(headers.from ?? '', headers.subject ?? '', messageId, headers.date)
if (!header) return null
const verified = verifySpotHeader(header)
const xml = verified ? parseSpotXml(decodeSpotText(xmlText)) : null
const target = verified ? moderationTarget(header) : null
const title = xml?.title || header.title
// Kept apart by where it came from: the title's {{pw}}, else the description's "Wachtwoord: …".
const fromTitle = verified && !target ? passwordFromSpot(title, null) : null
const fromText = verified && !target && !fromTitle ? passwordFromSpot('', xml?.description) : null
const status = !verified ? 'unverified' : target ? 'moderation' : 'listed'
await db
.table('spots')
.insert({
message_id: messageId,
article_number: articleNumber,
title: title.slice(0, 2000),
tag: header.tag || null,
poster: header.poster,
category: header.category,
subcats: JSON.stringify(header.subcats),
size: header.size || null,
posted_at: header.stamp ? new Date(header.stamp * 1000) : null,
key_id: header.keyId,
spotter_id: header.selfSignedPubKey.length > 50 ? spotterId(header.selfSignedPubKey) : null,
description: xml?.description || null,
website: xml?.website || null,
nzb_segments: xml?.nzbSegments.length ? xml.nzbSegments : null,
password: fromTitle ?? fromText,
password_source: fromTitle ? 'spot-title' : fromText ? 'spot-text' : null,
status,
})
.onConflict('message_id')
.ignore()
if (target) await this.moderate(header.selfSignedPubKey, header.stamp, target)
return status
}
/**
* A delete/dispose/remove: the target is `removed`, and its links and passwords go from the releases. As in
* Spotweb, a personal dispose (the poster's own key) counts only from the same spotter within 5 days; a
* moderator's (Spotnet's key) always.
*/
private async moderate(pubKey: string, stamp: number, targetId: string) {
const target = await db.from('spots').where('message_id', targetId).first()
if (!target) return
const personal = pubKey.length > 50 ? spotterId(pubKey) : ''
if (personal) {
const age = stamp - (target.posted_at ? new Date(target.posted_at).getTime() / 1000 : 0)
if (target.spotter_id !== personal || age > DISPOSE_WINDOW_SECONDS) return
}
await db.transaction(async (trx) => {
await trx.from('spots').where('message_id', targetId).update({ status: 'removed' })
await trx.from('release_spots').where('spot_id', targetId).delete()
await trx.from('release_passwords').where('spot_id', targetId).delete()
// A release the spot made of its own goes with it.
if (target.release_id) await trx.from('releases').where('id', target.release_id).where('source', 'spot').delete()
})
logger.info(`Spot "${target.title}" removed by ${personal ? 'its poster' : 'a moderator'}.`)
}
}