usenet-indexer/app/services/GarbageFilter.ts

import db from '@adonisjs/lucid/services/db'
import logger from '@adonisjs/core/services/logger'
import indexerConfig from '#config/indexer'
import { looksRandom } from '#services/ArchiveService'
import { fileStem } from '#lib/release/index'
import { explored, signatureGroup, signatureKey, uploadTool } from '#lib/garbage/index'

/** What the filter needs to know about an upload before it is stored. */
export interface Upload {
  /** The posted file name. */
  name: string
  /** Any of its articles' Message-IDs (one tool posts all of a file). */
  messageId: string | null | undefined
  groups: unknown
  /** Its release's key (the file key when it has none): kept or dropped as a whole. */
  releaseKey: string
}

/** Contents that can't name a random-named release without a key (LOG.md 2026-10-08 (5)). */
const CONTAINERS = ['rclone', 'nzbe', 'yencbin']

/** How long the dropped signatures are cached per process. */
const CACHE_MS = 5 * 60 * 1000

export function isRandomName(name: string) {
  return looksRandom(fileStem(name))
}

/**
 * Garbage: random-named uploads that never get a real name and have no
 * known password (CONCEPT.md D42). Two filters:
 * - before storing (seeding, ingest): random-named uploads whose signature
 *   (posting tool, first group) is dropped, or whose release was dropped
 *   after its probe, aren't stored;
 * - after a probe: a random-named release whose probes are all done, that
 *   is still without a real name, password, names from its files or spot,
 *   and is encrypted, a container or missing its first article, is deleted.
 * The dropped signatures are learned from a sample of probed releases
 * (learn(), gc:scheduler).
 */
class GarbageFilter {
  private dropped: Set<string> | null = null
  private loadedAt = 0

  /** The dropped signatures, cached for CACHE_MS. */
  async droppedSignatures(): Promise<Set<string>> {
    if (this.dropped && Date.now() - this.loadedAt < CACHE_MS) return this.dropped
    const rows = await db.from('upload_signatures').where('dropped', true).select('tool', 'group_name')
    this.dropped = new Set(rows.map((row) => signatureKey(row.tool, row.group_name)))
    this.loadedAt = Date.now()
    return this.dropped
  }

  /** Forgets the cached signatures (after learn(), and in tests). */
  reset() {
    this.dropped = null
  }

  /** Whether a random-named upload's signature is dropped and it isn't in the explored share. */
  private droppedBySignature(upload: Upload, signatures: Set<string>) {
    if (signatures.size === 0) return false
    if (!signatures.has(signatureKey(uploadTool(upload.messageId), signatureGroup(upload.groups)))) return false
    return !explored(upload.releaseKey, indexerConfig.garbage.explorePercent)
  }

  /**
   * The items worth storing: drops random-named uploads of dropped signatures
   * (but explorePercent of their releases) and of dropped releases.
   */
  async keep<T>(items: T[], upload: (item: T) => Upload): Promise<{ kept: T[]; dropped: number }> {
    if (!indexerConfig.garbage.enabled || items.length === 0) return { kept: items, dropped: 0 }
    const signatures = await this.droppedSignatures()
    // Per release, kept as a whole when any of its files has a readable name. A file is
    // looked at once: an ingest batch has many articles of each.
    const verdicts = new Map<string, 'keep' | 'drop' | 'check'>()
    const seen = new Set<string>()
    for (const item of items) {
      const u = upload(item)
      if (verdicts.get(u.releaseKey) === 'keep' || seen.has(`${u.releaseKey}\n${u.name}`)) continue
      seen.add(`${u.releaseKey}\n${u.name}`)
      if (!isRandomName(u.name)) verdicts.set(u.releaseKey, 'keep')
      else if (!verdicts.has(u.releaseKey)) verdicts.set(u.releaseKey, this.droppedBySignature(u, signatures) ? 'drop' : 'check')
    }
    const check = [...verdicts].filter(([, v]) => v === 'check').map(([key]) => key)
    for (let i = 0; i < check.length; i += 5000) {
      const rows = await db.from('dropped_releases').whereIn('release_key', check.slice(i, i + 5000)).select('release_key')
      for (const row of rows) verdicts.set(row.release_key, 'drop')
    }
    const kept = items.filter((item) => verdicts.get(upload(item).releaseKey) !== 'drop')
    return { kept, dropped: items.length - kept.length }
  }

  /**
   * Whether a stored file's probe can be skipped: a random-named file of a
   * dropped signature, outside the explored share (files stored before its
   * signature was dropped). Its binary stops waiting for inspection.
   */
  async skipProbe(file: { id: number | string; filename: string; real_name?: string | null; first_message_id?: string | null; groups: unknown; release_id?: number | string | null }) {
    if (!indexerConfig.garbage.enabled || file.real_name || !isRandomName(file.filename)) return false
    const signatures = await this.droppedSignatures()
    if (!signatures.has(signatureKey(uploadTool(file.first_message_id), signatureGroup(file.groups)))) return false
    const release = file.release_id ? await db.from('releases').where('id', file.release_id).select('release_key').first() : null
    if (explored(release?.release_key ?? String(file.id), indexerConfig.garbage.explorePercent)) return false
    await db.from('binaries').where('file_id', file.id).where('awaiting_inspection', true).update({ awaiting_inspection: null })
    return true
  }

  /**
   * After a file's probe has been applied: notes its release for learning
   * (a sample) and deletes the release when it is garbage. Never throws:
   * the probe's result is stored already. Returns why the release was
   * deleted, or null.
   */
  async afterProbe(fileId: number, contentType: string | null): Promise<string | null> {
    try {
      const file = await db.from('files').where('id', fileId).select('id', 'filename', 'real_name', 'first_message_id', 'groups', 'release_id').first()
      if (!file?.release_id || !isRandomName(String(file.filename))) return null
      const releaseId = Number(file.release_id)
      if (releaseId % indexerConfig.garbage.probeSample === 0) {
        await db.rawQuery(`INSERT INTO signature_probes (release_id, tool, group_name) VALUES (?, ?, ?) ON CONFLICT DO NOTHING`, [
          releaseId,
          uploadTool(file.first_message_id),
          signatureGroup(file.groups),
        ])
      }
      if (!indexerConfig.garbage.enabled) return null
      return await this.dropIfGarbage(releaseId, contentType)
    } catch (error: any) {
      logger.warn(`Garbage check after probing file ${fileId}: ${error.message}`)
      return null
    }
  }

  private async dropIfGarbage(releaseId: number, contentType: string | null): Promise<string | null> {
    const { rows } = await db.rawQuery(
      `SELECT r.release_key, r.container, r.encrypted,
         is_real_release_name(r.name) OR r.password IS NOT NULL OR coalesce(r.search_files, '') <> ''
           OR EXISTS (SELECT 1 FROM spots s WHERE s.release_id = r.id) AS named,
         EXISTS (SELECT 1 FROM binaries b WHERE b.release_id = r.id AND b.awaiting_inspection) AS waiting
       FROM releases r WHERE r.id = ?`,
      [releaseId]
    )
    const release = rows[0]
    if (!release || release.named || release.waiting) return null
    const reason = CONTAINERS.includes(release.container) ? release.container : release.encrypted === true ? 'encrypted' : contentType === 'missing' ? 'missing' : null
    if (!reason) return null
    await db.transaction(async (trx) => {
      await trx.rawQuery(`INSERT INTO dropped_releases (release_key, reason) VALUES (?, ?) ON CONFLICT DO NOTHING`, [release.release_key, reason])
      await trx.rawQuery(`DELETE FROM binaries WHERE release_id = ?`, [releaseId])
      await trx.rawQuery(`DELETE FROM files WHERE release_id = ?`, [releaseId])
      await trx.rawQuery(`DELETE FROM releases WHERE id = ?`, [releaseId])
    })
    return reason
  }

  /**
   * Learns the signatures from the probed releases of the last windowDays:
   * dropped when at least minProbes are known and at most maxNamedRate got a
   * real name, password, names from their files or a spot. A signature with
   * fewer probes keeps its verdict (a dropped one only gets explored probes).
   * Releases deleted as garbage count as unnamed. Returns the signatures
   * whose verdict changed.
   */
  async learn(): Promise<{ signatures: number; dropped: number; changed: string[] }> {
    const { minProbes, maxNamedRate, windowDays } = indexerConfig.garbage
    await db.rawQuery(`DELETE FROM signature_probes WHERE probed_at < now() - make_interval(days => ?)`, [windowDays])
    const before = new Map((await db.from('upload_signatures').select('tool', 'group_name', 'dropped')).map((row) => [signatureKey(row.tool, row.group_name), row.dropped]))
    await db.rawQuery(
      `INSERT INTO upload_signatures (tool, group_name, probed, named, dropped, updated_at)
       SELECT tool, group_name, probed, named, probed >= ?::int AND named <= probed * ?::float8, now()
       FROM (
         SELECT p.tool, p.group_name, count(*)::int AS probed,
           (count(*) FILTER (WHERE is_real_release_name(r.name) OR r.password IS NOT NULL OR coalesce(r.search_files, '') <> ''
              OR EXISTS (SELECT 1 FROM spots s WHERE s.release_id = r.id)))::int AS named
         FROM signature_probes p LEFT JOIN releases r ON r.id = p.release_id
         GROUP BY p.tool, p.group_name
       ) s
       ON CONFLICT (tool, group_name) DO UPDATE SET
         probed = excluded.probed, named = excluded.named, updated_at = now(),
         dropped = CASE WHEN excluded.probed >= ?::int THEN excluded.dropped ELSE upload_signatures.dropped END`,
      [minProbes, maxNamedRate, minProbes]
    )
    const after = await db.from('upload_signatures').select('tool', 'group_name', 'dropped', 'probed', 'named')
    const changed = after
      .filter((row) => (before.get(signatureKey(row.tool, row.group_name)) ?? false) !== row.dropped)
      .map((row) => `${row.tool} ${row.group_name || '(no group)'}: ${row.dropped ? 'dropped' : 'kept'} (${row.named}/${row.probed} named)`)
    this.reset()
    return { signatures: after.length, dropped: after.filter((row) => row.dropped).length, changed }
  }

  /**
   * Fills signature_probes from a sample of files probed before it existed
   * (garbage:learn --seed): random-named files with a content type, from
   * TABLESAMPLE SYSTEM (percent) of files. Returns the releases added.
   */
  async seed(percent: number): Promise<number> {
    const { rows } = await db.rawQuery(
      `SELECT DISTINCT ON (f.release_id) f.release_id, f.first_message_id, f.groups
       FROM files f TABLESAMPLE SYSTEM (?)
       WHERE f.content_type IS NOT NULL AND f.release_id IS NOT NULL AND NOT is_readable_name(f.filename)`,
      [percent]
    )
    let added = 0
    for (let i = 0; i < rows.length; i += 2000) {
      const chunk = rows.slice(i, i + 2000)
      const result = await db.rawQuery(
        `INSERT INTO signature_probes (release_id, tool, group_name)
         SELECT * FROM unnest(?::bigint[], ?::text[], ?::text[]) ON CONFLICT DO NOTHING`,
        [chunk.map((row: any) => row.release_id), chunk.map((row: any) => uploadTool(row.first_message_id)), chunk.map((row: any) => signatureGroup(row.groups))]
      )
      added += result.rowCount ?? 0
    }
    return added
  }
}

export default new GarbageFilter()