usenet-indexer/app/services/SeedService.ts

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

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))
    await knex('binaries')
      .insert(binaries)
      .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: staging (D35) for lone ones, binaries and parts for the rest. */
  async storeFragments(group: string, fragments: SeedFragment[]): Promise<void> {
    if (fragments.length === 0) return
    const segments = fragments.map(fromFragment)
    const { ingest } = StagingService.enabled(group)
      ? await StagingService.filter(segments, { counters: StagingService.countersEnabled(group) })
      : { ingest: segments }
    await IngestService.ingest(ingest)
  }
}

export default new SeedService()