usenet-indexer/tests/functional/ingest.spec.ts

import { test } from '@japa/runner'
import db from '@adonisjs/lucid/services/db'
import testUtils from '@adonisjs/core/services/test_utils'
import IngestService, { type IngestSegment } from '#services/IngestService'
import QueueService from '#services/QueueService'
import { fileKey } from '#lib/subject/index'
import FileStore from '#services/FileStore'

function segment(overrides: Partial<IngestSegment>): IngestSegment {
  const poster = 'ingest-test@example.invalid'
  const subject = '"test-file.bin" yEnc'
  return {
    fileKey: fileKey(poster, subject, 3),
    poster,
    subject,
    name: 'test-file.bin',
    totalParts: 3,
    number: 1,
    messageId: 'part1@test',
    bytes: 100,
    groups: ['alt.binaries.test'],
    ...overrides,
  }
}

async function binaryFor(key: string) {
  return db.from('binaries').where('file_key', key).first()
}

// None of these cases complete a binary, so nothing is queued in Redis.
test.group('IngestService', (group) => {
  group.each.setup(() => testUtils.db().withGlobalTransaction())
  // IngestService opens the BullMQ queues; close them so the test process can exit.
  group.teardown(() => QueueService.closeAll())

  test('existing binaries are only rewritten when something changes', async ({ assert }) => {
    const first = segment({ number: 1, postedAt: new Date('2026-09-28T12:00:00Z') })
    await IngestService.ingest([first])
    const before = await binaryFor(first.fileKey)

    // A later part within TOUCH_MS: nothing to change, so the row is left alone.
    await IngestService.ingest([segment({ number: 2, messageId: 'part2@test', postedAt: new Date('2026-09-28T12:05:00Z') })])
    const same = await binaryFor(first.fileKey)
    assert.equal(+same.last_seen_at, +before.last_seen_at)
    assert.equal(await db.from('parts').where('binary_id', before.id).count('* as n').then((r) => Number(r[0].n)), 2)

    // An earlier post date and a stale last_seen_at are merged.
    await db.from('binaries').where('id', before.id).update({ last_seen_at: new Date(Date.now() - 60 * 60 * 1000) })
    await IngestService.ingest([segment({ number: 1, postedAt: new Date('2026-09-28T11:00:00Z') })])
    const after = await binaryFor(first.fileKey)
    assert.equal(after.posted_at.toISOString(), '2026-09-28T11:00:00.000Z')
    assert.isAbove(+after.last_seen_at, Date.now() - 60_000)
  })

  test('a binary that is a file already takes no new header parts', async ({ assert }) => {
    await IngestService.ingest([segment({ number: 1 })])
    const binary = await binaryFor(segment({}).fileKey)
    const now = new Date()
    const [file] = await FileStore.insert([
      { filename: 'test-file.bin', poster: 'x', date: now.getTime(), parts: 3, size: 300, groups: '[]', segments: [], created_at: now, updated_at: now },
    ])
    await db.from('binaries').where('id', binary.id).update({ file_id: file.id })
    await db.from('parts').where('binary_id', binary.id).delete()

    // The same articles again (a crosspost, a backfill range): nothing to store.
    await IngestService.ingest([segment({ number: 2, messageId: 'part2@test' }), segment({ number: 3, messageId: 'part3@test' })])
    assert.equal(await db.from('parts').where('binary_id', binary.id).count('* as n').then((r) => Number(r[0].n)), 0)

    // Body-derived parts (with a per-part CRC) are still kept.
    await IngestService.ingest([segment({ number: 2, messageId: 'part2@test', pcrc32: 'abcd1234' })])
    assert.equal(await db.from('parts').where('binary_id', binary.id).count('* as n').then((r) => Number(r[0].n)), 1)
  })

  test('ingesting the same segments twice is a no-op', async ({ assert }) => {
    const segments = [segment({ number: 1 }), segment({ number: 2, messageId: 'part2@test' })]
    await IngestService.ingest(segments)
    await IngestService.ingest(segments)

    const binary = await binaryFor(segments[0].fileKey)
    const [{ count }] = await db.from('parts').where('binary_id', binary.id).count('* as count')
    assert.equal(Number(count), 2)
    assert.isNull(binary.completed_at)
    assert.isNull(binary.conflicted_at)
  })

  test('a weak identity with two articles for one part number is flagged and never completes', async ({
    assert,
  }) => {
    const weak = { weakIdentity: true }
    await IngestService.ingest([segment({ ...weak, number: 1, messageId: 'volume-a-1@test' })])
    // Same name, poster and size, but another file: the second volume of a set.
    await IngestService.ingest([
      segment({ ...weak, number: 1, messageId: 'volume-b-1@test' }),
      segment({ ...weak, number: 2, messageId: 'volume-b-2@test' }),
      segment({ ...weak, number: 3, messageId: 'volume-b-3@test' }),
    ])

    const binary = await binaryFor(segment({}).fileKey)
    assert.isNotNull(binary.conflicted_at)
    assert.isNull(binary.completed_at, 'all part numbers are present, but it must not complete')
  })

  test('a conflict inside a single batch is detected too', async ({ assert }) => {
    await IngestService.ingest([
      segment({ weakIdentity: true, number: 2, messageId: 'a@test' }),
      segment({ weakIdentity: true, number: 2, messageId: 'b@test' }),
    ])
    assert.isNotNull((await binaryFor(segment({}).fileKey)).conflicted_at)
  })

  test('on the subject path a second article for a part is a repost, not a conflict', async ({
    assert,
  }) => {
    await IngestService.ingest([segment({ number: 1, messageId: 'original@test' })])
    await IngestService.ingest([segment({ number: 1, messageId: 'repost@test' })])

    const binary = await binaryFor(segment({}).fileKey)
    assert.isNull(binary.conflicted_at)
    const part = await db.from('parts').where({ binary_id: binary.id, number: 1 }).first()
    assert.equal(part.message_id, 'original@test')
  })

  test('an obfuscated file joins a known par2 set of exactly its size', async ({ assert }) => {
    const now = new Date()
    const [release] = await db
      .table('releases')
      .insert({ release_key: 'k'.repeat(40), name: 'Some.Show.S01E01', source: 'par2', groups: '[]', posted_at: now, first_seen_at: now, last_seen_at: now })
      .returning('id')
    await db.table('release_files').insert({ release_id: release.id, par2_file_id: 'f'.repeat(32), name: 'Some.Show.S01E01.mkv', size: 5467240532, md5_16k: '0'.repeat(32) })

    const weak = { weakIdentity: true, postedAt: now, release: null }
    await IngestService.ingest([
      segment({ ...weak, fileKey: 'a'.repeat(40), size: 5467240532, totalParts: 7628, messageId: 'match@test' }),
      segment({ ...weak, fileKey: 'b'.repeat(40), size: 5467240533, totalParts: 7628, messageId: 'other@test' }),
    ])

    const matched = await binaryFor('a'.repeat(40))
    const other = await binaryFor('b'.repeat(40))
    assert.equal(Number(matched.release_id), Number(release.id))
    assert.isNull(other.release_id)
  })

  test('an identical repost (same part CRC) is not a conflict', async ({ assert }) => {
    const weak = { weakIdentity: true, pcrc32: 'abcd1234', partSize: 700 }
    await IngestService.ingest([segment({ ...weak, number: 1, messageId: 'first@test' })])
    await IngestService.ingest([segment({ ...weak, number: 1, messageId: 'repost@test' })])
    const binary = await binaryFor(segment({}).fileKey)
    assert.isNull(binary.conflicted_at)

    await IngestService.ingest([segment({ ...weak, pcrc32: 'ffff0000', number: 1, messageId: 'other@test' })])
    assert.isNotNull((await binaryFor(segment({}).fileKey)).conflicted_at)
  })

  test('a release takes its file total from whichever file carries the counter', async ({ assert }) => {
    const { releaseIdentity } = await import('#lib/release/index')
    const poster = 'archiver@example.invalid'
    const file = (subject: string, name: string, key: string) =>
      segment({ fileKey: key.padEnd(40, 'k'), poster, subject, name, totalParts: 1, messageId: `${key}@test`,
                release: releaseIdentity({ poster, subject, name, weak: false }) })
    // The counter-less par2 index comes first in the batch.
    await IngestService.ingest([
      file('"x.tar.zst.par2" yEnc 3100', 'x.tar.zst.par2', 'index'),
      file('[1/2] - "x.tar.zst" yEnc 1000', 'x.tar.zst', 'archive'),
    ])
    const release = await db.from('releases').where('name', 'x').where('poster', poster).first()
    assert.equal(release.file_total, 2)
  })
})