usenet-indexer/app/services/SeedService.ts

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

/** 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
}

/**
 * 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 }
    // A repost can complete the same file twice in one batch; ON CONFLICT DO UPDATE takes each row once.
    const files = [...new Map(batch.map((file) => [file.fileKey, file])).values()]
    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

    const fileIds = new Map<string, number>()
    for (const row of await db.from('files').whereIn('file_key', rows.map((r) => r.file_key)).select('id', 'file_key')) {
      fileIds.set(row.file_key, Number(row.id))
    }
    // The completed binary worker:file would have left behind; an existing one just learns its file.
    const binaries = files.map((file, i) => ({
      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,
    }))
    binaries.sort((a, b) => a.file_key.localeCompare(b.file_key))
    for (let i = 0; i < binaries.length; i += INSERT_ROWS) {
      await knex('binaries')
        .insert(binaries.slice(i, i + INSERT_ROWS))
        .onConflict('file_key')
        .merge({ file_id: knex.raw('coalesce(binaries.file_id, excluded.file_id)') as any })
    }

    // As worker:file does for a new file: an SFV posted earlier may list it (D34), then inspection.
    const rowById = new Map(rows.map((row) => [fileIds.get(row.file_key), row]))
    const added = inserted.flatMap((file) => {
      const row = rowById.get(file.id)
      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) => ({ name: 'process-collection', data: { fileId: file.id }, opts: { jobId: `file-${file.id}` } }))
    if (candidates.length > 0) await QueueService.collectionQueue.addBulk(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()