Daan MeijerandClaude Opus 5.5
3f6bb8b162
Oldest inspection first went through the 3-month backfill before today's spots, whose releases the tail has: 3 h after reset 24, 2 of 16.7k matched. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
360 lines
16 KiB
TypeScript
360 lines
16 KiB
TypeScript
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
|
|
/**
|
|
* Unmatched spots matched again per hour: after a reset every spot is (16.5k at 500 an hour took 33 h). Never tried
|
|
* first, then the newest: the oldest inspection first tried a 3-month backfill before today's spots, the ones whose
|
|
* releases the tail has (2 of 16.7k matched 3 h after reset 24).
|
|
*/
|
|
const REMATCH_BATCH = 5000
|
|
/** Unmatched spots are matched again for this long after their inspection. */
|
|
const REMATCH_DAYS = 30
|
|
/** 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, for 30 days; once
|
|
* SPOT_STANDALONE_AFTER_HOURS have passed since its inspection it becomes a
|
|
* release of its own (source `spot`) with its NZB, until the header pass
|
|
* builds the releases after all (a spot older than what the catalog has
|
|
* indexed): then it moves to those, and its own release goes.
|
|
*/
|
|
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,
|
|
matched: 0,
|
|
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)
|
|
if (perRelease.size === 0) return 0
|
|
await db.transaction(async (trx) => {
|
|
// A release of the spot's own goes now that the header pass built its files' releases.
|
|
if (spot.release_id) {
|
|
const own = await trx.from('releases').where('id', spot.release_id).where('source', 'spot').select('id').first()
|
|
if (own) {
|
|
await trx.from('release_spots').where('release_id', own.id).delete()
|
|
await trx.from('release_passwords').where('release_id', own.id).delete()
|
|
await trx.from('releases').where('id', own.id).delete()
|
|
}
|
|
}
|
|
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]
|
|
// The NZB is only kept for a release of the spot's own: the header pass's releases have their chunks.
|
|
await trx.from('spots').where('message_id', spotId).update({ matched: fileRelease.size, release_id: top ? top[0] : null, nzb: 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 { rows: spots } = await db.rawQuery(
|
|
`SELECT message_id, inspected_at, release_id FROM spots
|
|
WHERE status = 'inspected' AND matched = 0 AND inspected_at > now() - make_interval(days => ?)
|
|
ORDER BY rematched_at NULLS FIRST, posted_at DESC NULLS LAST LIMIT ?`,
|
|
[REMATCH_DAYS, REMATCH_BATCH]
|
|
)
|
|
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)
|
|
await db.from('spots').where('message_id', spot.message_id).update({ rematched_at: new Date() })
|
|
out.matched += n
|
|
if (n === 0 && !spot.release_id && new Date(spot.inspected_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
|
|
}
|
|
}
|