usenet-indexer/app/services/spot_inspector.ts

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()
      // Its poster for poster search, as the batch store adds its releases' (migration 4000).
      if (spot.poster) await trx.rawQuery(`INSERT INTO posters (poster) VALUES (left(?, 200)) ON CONFLICT DO NOTHING`, [spot.poster])
      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
  }
}