usenet-indexer/app/services/IngestService.ts

import db from '@adonisjs/lucid/services/db'
import logger from '@adonisjs/core/services/logger'
import type { QueryClientContract } from '@adonisjs/lucid/types/database'
import type { Knex } from 'knex'
import QueueService from '#services/QueueService'
import NameService from '#services/NameService'
import type { IngestSegment } from '#lib/ingest/segment'

// Header parsing is pure and lives in lib, so seeding nodes need no database (CONCEPT.md D37).
export { segmentFromHeader, segmentFromYenc, type IngestSegment, type OverviewHeader } from '#lib/ingest/segment'

export interface CompletedBinary {
  id: number
  fileKey: string
}

const BINARY_CHUNK = 1000
/** last_seen_at is only rewritten when it is older than this (it drives GC at 48 h). */
const TOUCH_MS = 10 * 60 * 1000

type Merge = 'least' | 'coalesce' | 'touch'

/**
 * Inserts rows keyed by `key` and returns their ids. Existing rows are not
 * locked unless something in them changes (`least`: an earlier value,
 * `coalesce`: a value where there was none, `touch`: a timestamp more than
 * TOUCH_MS newer). Parallel fetch jobs see mostly the same rows, and
 * `INSERT … ON CONFLICT DO UPDATE` made them wait on each other for every one.
 */
async function upsertKeyed(
  knex: Knex,
  table: 'binaries' | 'releases',
  key: string,
  rows: Record<string, unknown>[],
  merge: Record<string, [Merge, string]>
): Promise<Map<string, number>> {
  const ids = new Map<string, number>()
  if (rows.length === 0) return ids
  const inserted: Record<string, any>[] = await knex(table).insert(rows).onConflict(key).ignore().returning(['id', key])
  for (const row of inserted) ids.set(row[key], Number(row.id))

  const incoming = new Map(rows.filter((r) => !ids.has(r[key] as string)).map((r) => [r[key] as string, r]))
  if (incoming.size === 0) return ids
  const columns = Object.keys(merge)
  const existing: Record<string, any>[] = await knex(table).whereIn(key, [...incoming.keys()]).select('id', key, ...columns)

  const changes: Record<string, unknown>[] = []
  for (const row of existing) {
    ids.set(row[key], Number(row.id))
    const next = incoming.get(row[key])!
    const changed = columns.some((column) => {
      const [kind] = merge[column]
      const current = row[column]
      const value = next[column]
      if (value === null || value === undefined) return false
      if (kind === 'coalesce') return current === null
      if (kind === 'least') return current === null || value < current
      return current === null || +(value as Date) - +current > TOUCH_MS
    })
    if (changed) changes.push({ id: Number(row.id), ...Object.fromEntries(columns.map((c) => [c, next[c] ?? null])) })
  }
  if (changes.length === 0) return ids

  const set = columns
    .map((c) => {
      const [kind] = merge[c]
      const fn = kind === 'least' ? 'least' : kind === 'touch' ? 'greatest' : 'coalesce'
      return `${c} = ${fn}(t.${c}, v.${c})`
    })
    .join(', ')
  const types = columns.map((c) => `${c} ${merge[c][1]}`).join(', ')
  changes.sort((a, b) => (a.id as number) - (b.id as number))
  for (let attempt = 1; ; attempt++) {
    try {
      await knex.raw(
        `UPDATE ${table} t SET ${set}
         FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, ${types})
         WHERE t.id = v.id`,
        [JSON.stringify(changes)]
      )
      return ids
    } catch (error: any) {
      // Rare now, but two jobs can still update overlapping rows in a different order.
      if (error.code !== '40P01' || attempt >= 3) throw error
    }
  }
}
/** Size-only matching below this is too ambiguous (nfo, sfv, small volumes). */
export const PAR2_SIZE_ONLY_MIN = 1024 * 1024
/** A par2 set and its files are posted together; don't match across days. */
export const PAR2_MATCH_WINDOW_SECONDS = 12 * 60 * 60
const PART_CHUNK = 2000

function chunks<T>(items: T[], size: number): T[][] {
  const result: T[][] = []
  for (let i = 0; i < items.length; i += size) result.push(items.slice(i, i + size))
  return result
}

function validDate(date?: Date) {
  return date && !Number.isNaN(date.getTime()) ? date : null
}

/**
 * Stores segments in `binaries`/`parts` and queues every binary that became
 * complete exactly once. Idempotent: ingesting the same segments again is a
 * no-op. See CONCEPT.md (D12).
 */
class IngestService {
  async ingest(segments: IngestSegment[]): Promise<CompletedBinary[]> {
    if (segments.length === 0) return []

    // Short statements, no enclosing transaction: every step is idempotent, and
    // row locks held for a whole batch made parallel fetch jobs wait on each other.
    const binaryIds = await (async () => {
      const knex = db.connection().getWriteClient()
      const now = new Date()
      const releaseIds = await this.upsertReleases(knex, segments, now)

      // One row per file_key (ON CONFLICT cannot touch a row twice per statement),
      // sorted so concurrent workers lock rows in the same order and cannot deadlock.
      const binaries = new Map<string, Record<string, unknown>>()
      for (const segment of segments) {
        const postedAt = validDate(segment.postedAt)
        const existing = binaries.get(segment.fileKey)
        if (existing) {
          if (postedAt && (!existing.posted_at || postedAt < (existing.posted_at as Date))) {
            existing.posted_at = postedAt
          }
          continue
        }
        binaries.set(segment.fileKey, {
          file_key: segment.fileKey,
          poster: segment.poster,
          subject: segment.subject,
          name: segment.name,
          total_parts: segment.totalParts,
          groups: JSON.stringify(segment.groups),
          posted_at: postedAt,
          first_seen_at: now,
          last_seen_at: now,
          release_id: segment.release ? releaseIds.get(segment.release.key) : null,
          size: segment.size ?? null,
          md5_16k: segment.md5_16k ?? null,
        })
      }
      const binaryRows = [...binaries.values()].sort((a, b) =>
        (a.file_key as string).localeCompare(b.file_key as string)
      )

      const idsByKey = new Map<string, number>()
      for (const chunk of chunks(binaryRows, BINARY_CHUNK)) {
        const ids = await upsertKeyed(knex, 'binaries', 'file_key', chunk, {
          last_seen_at: ['touch', 'timestamptz'],
          posted_at: ['least', 'timestamptz'],
          release_id: ['coalesce', 'bigint'],
          md5_16k: ['coalesce', 'varchar'],
        })
        for (const [key, id] of ids) idsByKey.set(key, id)
      }

      // A binary that is a file already has its segment list in `files`; its
      // header parts were dropped then. Crossposts in other groups, reposts
      // and backfill ranges keep bringing the same articles, and inserting
      // them again grew `parts` to 56M rows nobody reads (LOG.md 2026-10-04).
      const done = new Set<number>()
      const ids = [...idsByKey.values()]
      for (const chunk of chunks(ids, BINARY_CHUNK)) {
        const { rows } = await knex.raw('SELECT id FROM binaries WHERE id = ANY(?) AND file_id IS NOT NULL', [chunk])
        for (const row of rows) done.add(Number(row.id))
      }

      const partRows = segments
        .map((segment) => ({
          binary_id: idsByKey.get(segment.fileKey)!,
          number: segment.number,
          message_id: segment.messageId,
          bytes: segment.bytes,
          article_number: segment.articleNumber?.toString() ?? null,
          pcrc32: segment.pcrc32 ?? null,
          part_size: segment.partSize ?? null,
          file_crc32: segment.fileCrc32 ?? null,
          md5_16k: segment.md5_16k ?? null,
        }))
      const weakRows = partRows.filter((row, i) => segments[i].weakIdentity && !(done.has(row.binary_id) && row.pcrc32 === null))
      // Body-derived parts (with a per-part CRC) are kept, as worker:file does.
      const newParts = partRows.filter((row) => !(done.has(row.binary_id) && row.pcrc32 === null))

      newParts.sort((a, b) => a.binary_id - b.binary_id || a.number - b.number)
      for (const chunk of chunks(newParts, PART_CHUNK)) {
        await knex('parts').insert(chunk).onConflict(['binary_id', 'number']).ignore()
      }

      if (weakRows.length > 0) await this.flagConflicts(db.connection(), weakRows)

      return [...idsByKey.values()]
    })()

    // par2 naming needs a size, which only body-derived segments have; for
    // header batches these queries were most of Postgres' work.
    const sized = [...new Set(segments.filter((s) => s.size != null).map((s) => s.fileKey))]
    if (sized.length > 0) {
      const sizedIds = (await db.from('binaries').whereIn('file_key', sized).select('id')).map((r) => Number(r.id))
      await this.matchPar2Releases(sizedIds)
      await NameService.resolve(sizedIds)
    }

    // After commit, so the last writer of a binary always sees all of its parts.
    const completed = await this.claimCompleted(binaryIds)
    await this.enqueue(completed)
    return completed
  }

  /**
   * Creates or touches the releases referenced by `segments` and returns
   * their ids by release key. Also used by `release:rebuild`.
   */
  async upsertReleases(
    knex: Knex,
    segments: Pick<IngestSegment, 'release' | 'poster' | 'groups' | 'postedAt' | 'weakIdentity'>[],
    now = new Date()
  ): Promise<Map<string, number>> {
    const releases = new Map<string, Record<string, unknown>>()
    for (const segment of segments) {
      const release = segment.release
      if (!release) continue
      const postedAt = validDate(segment.postedAt)
      const existing = releases.get(release.key)
      if (existing) {
        if (postedAt && (!existing.posted_at || postedAt < (existing.posted_at as Date))) {
          existing.posted_at = postedAt
        }
        // Not every file carries the [i/M] counter (e.g. a par2 index): take it from any that does.
        if (existing.file_total === null && release.fileTotal) existing.file_total = release.fileTotal
        continue
      }
      releases.set(release.key, {
        release_key: release.key,
        name: release.name,
        // Obfuscating posters randomise the poster, so it says nothing about the release.
        poster: segment.weakIdentity ? null : segment.poster,
        file_total: release.fileTotal ?? null,
        source: release.source,
        groups: JSON.stringify(segment.groups),
        posted_at: postedAt,
        first_seen_at: now,
        last_seen_at: now,
      })
    }

    const ids = new Map<string, number>()
    const rows = [...releases.values()].sort((a, b) =>
      (a.release_key as string).localeCompare(b.release_key as string)
    )
    for (const chunk of chunks(rows, BINARY_CHUNK)) {
      const result = await upsertKeyed(knex, 'releases', 'release_key', chunk, {
        last_seen_at: ['touch', 'timestamptz'],
        posted_at: ['least', 'timestamptz'],
        file_total: ['coalesce', 'integer'],
        source: ['coalesce', 'varchar'],
      })
      for (const [key, id] of result) ids.set(key, id)
    }
    return ids
  }

  /**
   * Assigns body-derived binaries without a release and without a
   * fingerprint to a known par2 set by size: only when that size is listed
   * once in the set, the file is large, and exactly one set nearby lists it.
   * Binaries with a fingerprint are matched exactly by NameService instead.
   */
  async matchPar2Releases(binaryIds: number[]) {
    if (binaryIds.length === 0) return
    // Listed-once first (index-only on release_files (release_id, size): most candidate
    // sizes are a set's equal-sized volumes), the release's date only for what's left;
    // OFFSET 0 keeps that order. Was: every size match's date and count, 18 s per batch.
    await db.rawQuery(
      `UPDATE binaries b SET release_id = m.release_id
       FROM (
         SELECT c.id, min(c.release_id) AS release_id
         FROM (
           SELECT b2.id, b2.posted_at, rf.release_id
           FROM binaries b2
           JOIN release_files rf ON rf.size = b2.size
           WHERE b2.id = ANY(?) AND b2.release_id IS NULL AND b2.md5_16k IS NULL AND b2.size >= ?
             AND NOT EXISTS (
               SELECT 1 FROM release_files x
               WHERE x.release_id = rf.release_id AND x.size = rf.size AND x.par2_file_id <> rf.par2_file_id)
           OFFSET 0
         ) c
         JOIN releases r ON r.id = c.release_id
         WHERE abs(extract(epoch FROM (r.posted_at - c.posted_at))) < ?
         GROUP BY c.id
         HAVING count(DISTINCT c.release_id) = 1
       ) m
       WHERE b.id = m.id`,
      [binaryIds, PAR2_SIZE_ONLY_MIN, PAR2_MATCH_WINDOW_SECONDS]
    )
  }

  /**
   * Flags binaries where a stored part has a different message-id and a
   * different part CRC than the incoming one: two files were merged under
   * one identity. Identical reposts (same CRC) are ignored.
   */
  private async flagConflicts(
    trx: QueryClientContract,
    rows: {
      binary_id: number
      number: number
      message_id: string
      bytes: number
      article_number: string | null
      pcrc32: string | null
      part_size: number | null
      file_crc32: string | null
      md5_16k: string | null
    }[]
  ) {
    const bindings = [
      rows.map((r) => r.binary_id),
      rows.map((r) => r.number),
      rows.map((r) => r.message_id),
      rows.map((r) => r.bytes),
      rows.map((r) => r.article_number),
      rows.map((r) => r.pcrc32),
      rows.map((r) => r.part_size),
      rows.map((r) => r.file_crc32),
      rows.map((r) => r.md5_16k),
    ]
    // Keep the extra articles: SplitService separates the volumes from them.
    const conflicts = await trx.rawQuery(
      `INSERT INTO part_conflicts (binary_id, number, message_id, bytes, article_number, pcrc32, part_size, file_crc32, md5_16k)
       SELECT v.binary_id, v.number, v.message_id, v.bytes, v.article_number, v.pcrc32, v.part_size, v.file_crc32, v.md5_16k
       FROM unnest(?::bigint[], ?::int[], ?::text[], ?::int[], ?::bigint[], ?::text[], ?::int[], ?::text[], ?::text[])
         AS v(binary_id, number, message_id, bytes, article_number, pcrc32, part_size, file_crc32, md5_16k)
       JOIN parts p ON p.binary_id = v.binary_id AND p.number = v.number
       WHERE p.message_id <> v.message_id
         -- Same part CRC: the same data posted again (a repost), not another file.
         AND (p.pcrc32 IS NULL OR v.pcrc32 IS NULL OR p.pcrc32 <> v.pcrc32)
       ON CONFLICT DO NOTHING
       RETURNING binary_id`,
      bindings as any
    )
    const binaryIds: number[] = [
      ...new Set<number>(conflicts.rows.map((r: { binary_id: string }) => Number(r.binary_id))),
    ]
    if (binaryIds.length === 0) return
    const result = await trx.rawQuery(
      `UPDATE binaries SET conflicted_at = now()
       WHERE conflicted_at IS NULL AND id = ANY(?) RETURNING name`,
      [binaryIds]
    )
    for (const row of result.rows) {
      logger.warn(`Binary "${row.name}" has conflicting parts; it will be split into volumes.`)
    }
  }

  /**
   * Marks complete binaries among `binaryIds` as claimed. The single UPDATE
   * locks each row, so concurrent callers never claim the same binary twice.
   */
  async claimCompleted(binaryIds: number[]): Promise<CompletedBinary[]> {
    if (binaryIds.length === 0) return []
    const result = await db.rawQuery(
      `UPDATE binaries b SET completed_at = now()
       WHERE b.id = ANY(?) AND b.completed_at IS NULL AND b.conflicted_at IS NULL
         AND (SELECT count(*) FROM parts p WHERE p.binary_id = b.id) >= b.total_parts
       RETURNING b.id, b.file_key`,
      [binaryIds]
    )
    return result.rows.map((row: { id: string; file_key: string }) => ({
      id: Number(row.id),
      fileKey: row.file_key,
    }))
  }

  async enqueue(completed: CompletedBinary[]) {
    if (completed.length === 0) return
    // jobId = file_key: BullMQ ignores an add() for an id that already exists.
    await QueueService.fileQueue.addBulk(
      completed.map(({ id, fileKey }) => ({
        name: 'process-file',
        data: { binaryId: id },
        opts: { jobId: fileKey },
      }))
    )
  }
}

export default new IngestService()