usenet-indexer/app/services/SeedService.ts

import db from '@adonisjs/lucid/services/db'
import IngestService from '#services/IngestService'
import FileStore, { insertLimits } from '#services/FileStore'
import SfvService from '#services/SfvService'
import ContentService from '#services/ContentService'
import { fromFragment, type SeedFile, type SeedFragment } from '#lib/seed/wire'
import InspectionQueue from '#services/InspectionQueue'
import GarbageFilter from '#services/GarbageFilter'

/** Articles from a range's edge within which a file may continue in the neighbouring range. */
const EDGE_ARTICLES = 50_000

export interface SeedStoreResult {
  files: number
  /** Files that existed already (same file_key, e.g. from live indexing or a crosspost). */
  existing: number
  releases: number
  inspections: number
  /** Random-named files of garbage signatures or dropped releases, not stored (D42). */
  dropped: number
}

/**
 * Stores what seeding nodes found (CONCEPT.md D37): per complete file its
 * release, the file with its segment list, and one completed binary, as
 * worker:file would have left it, so par2 naming, inspection and GC treat
 * it like any other file; fragments take the regular ingest path. A batch
 * is a few statements, not one per article.
 */
class SeedService {
  async store(batch: SeedFile[]): Promise<SeedStoreResult> {
    const result: SeedStoreResult = { files: 0, existing: 0, releases: 0, inspections: 0, dropped: 0 }
    // A repost can complete the same file twice in one batch; ON CONFLICT DO UPDATE takes each row once.
    const unique = [...new Map(batch.map((file) => [file.fileKey, file])).values()]
    // Garbage isn't stored at all: no release, file or binary (CONCEPT.md D42).
    const { kept: files, dropped } = await GarbageFilter.keep(unique, (file) => ({
      name: file.name,
      messageId: file.segments[0]?.[1],
      groups: file.groups,
      releaseKey: file.release?.key ?? file.fileKey,
    }))
    result.dropped = dropped
    if (files.length === 0) return result
    const knex = db.connection().getWriteClient()
    const now = new Date()
    const postedOf = (file: SeedFile) => (file.postedAt ? new Date(file.postedAt) : undefined)

    const releaseIds = await IngestService.upsertReleases(
      knex,
      files.map((file) => ({ release: file.release, poster: file.poster, groups: file.groups, postedAt: postedOf(file) })),
      now
    )
    result.releases = releaseIds.size

    const rows = files.map((file) => ({
      file_key: file.fileKey,
      filename: file.name,
      poster: file.poster,
      date: (postedOf(file) ?? now).getTime(),
      parts: file.segments.length,
      segments: file.segments.map(([number, id, size]) => ({ number, id, size })),
      groups: JSON.stringify(file.groups),
      release_id: file.release ? (releaseIds.get(file.release.key) ?? null) : null,
      size: file.segments.reduce((sum, [, , size]) => sum + size, 0),
      created_at: now,
      updated_at: now,
    }))
    const inserted = await FileStore.insert(rows)
    result.files = inserted.length
    result.existing = rows.length - inserted.length

    // File ids for the binaries below (files queued for inspection only): new ones from the insert,
    // existing ones looked up. Looking up every key of the batch probed all files partitions again,
    // 15% of Postgres' active time (LOG.md 2026-10-06 (12)).
    const fileIds = new Map<string, number>()
    for (const row of inserted) if (row.file_key) fileIds.set(row.file_key, row.id)
    const missing = files.filter((file) => !fileIds.has(file.fileKey) && ContentService.isCandidate(file.name)).map((file) => file.fileKey)
    for (let i = 0; i < missing.length; i += 5000) {
      for (const row of await db.from('files').whereIn('file_key', missing.slice(i, i + 5000)).select('id', 'file_key')) {
        fileIds.set(row.file_key, Number(row.id))
      }
    }
    // A completed binary, as worker:file would have left behind, only for files queued for content
    // inspection: par2 naming (D16) needs the fingerprint inspection stores on it. For the others
    // (~85% in a.b.ath) it was only GC churn: 1.6M rows an hour inserted and deleted (LOG.md 2026-10-06 (11)).
    const insertedKeys = new Set(inserted.map((row) => row.file_key))
    const binaries = files.flatMap((file, i) => {
      if (!ContentService.isCandidate(file.name)) return []
      return [{
        file_key: file.fileKey,
        poster: file.poster,
        subject: file.subject,
        name: file.name,
        total_parts: file.totalParts,
        groups: rows[i].groups,
        posted_at: postedOf(file) ?? null,
        first_seen_at: now,
        last_seen_at: now,
        completed_at: now,
        release_id: rows[i].release_id,
        file_id: fileIds.get(file.fileKey) ?? null,
        // New files are queued for inspection below; an existing file was queued when it was added.
        awaiting_inspection: insertedKeys.has(file.fileKey) ? true : null,
      }]
    })
    binaries.sort((a, b) => a.file_key.localeCompare(b.file_key))
    for (let i = 0; i < binaries.length; i += insertLimits.rows) {
      await knex('binaries')
        .insert(binaries.slice(i, i + insertLimits.rows))
        .onConflict('file_key')
        .merge({
          file_id: knex.raw('coalesce(binaries.file_id, excluded.file_id)') as any,
          awaiting_inspection: knex.raw('coalesce(binaries.awaiting_inspection, excluded.awaiting_inspection)') as any,
        })
    }

    // As worker:file does for a new file: an SFV posted earlier may list it (D34), then inspection.
    const rowByKey = new Map(rows.map((row) => [row.file_key, row]))
    const added = inserted.flatMap((file) => {
      const row = file.file_key ? rowByKey.get(file.file_key) : undefined
      return row ? [{ id: file.id, name: row.filename, poster: row.poster, date: row.date, release_id: row.release_id }] : []
    })
    await SfvService.claimAll(added)
    const candidates = added
      .filter((file) => ContentService.isCandidate(file.name))
      .map((file) => file.id)
    if (candidates.length > 0) await InspectionQueue.add(candidates)
    result.inspections = candidates.length
    return result
  }

  /**
   * Segments of files that didn't complete on a node. Only files near the
   * range's edges can continue in the neighbouring range (worked on by some
   * node, maybe hours later): those go straight into binaries and parts
   * (staging would drop them after 30 minutes). The rest are files with
   * articles missing for good, and are dropped: staging them filled 22 GB
   * of churn on the NVMe (LOG.md 2026-10-05 (10)). Returns the segments kept.
   */
  async storeFragments(fragments: SeedFragment[], range: { start: number; end: number }): Promise<number> {
    const nearEdge = new Set<string>()
    for (const fragment of fragments) {
      const n = Number(fragment.articleNumber)
      if (Number.isFinite(n) && (n < range.start + EDGE_ARTICLES || n > range.end - EDGE_ARTICLES)) nearEdge.add(fragment.fileKey)
    }
    const kept = fragments.filter((fragment) => nearEdge.has(fragment.fileKey))
    if (kept.length > 0) await IngestService.ingest(kept.map(fromFragment))
    return kept.length
  }
}

export default new SeedService()