diff --git a/REBUILD.md b/REBUILD.md index bb825d5..c2f869d 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -531,12 +531,51 @@ releases; the inspector matches them again), `nntp_shares`, `nodes`, rule read all of `release_inspections` every round, 17 min on the HDD; the judge must read it by release or keep what it needs on the NVMe. -Next: the batch store (write-once releases, posters and group sets by id, -bytea keys, blocks, the name into `release_texts`), then the window and -albums in the Rust edge, block -leases and the block merger, the corpus replay harness. Until the batch -store is done the 38 functional tests that write the old columns fail on -`voyager-3`. +### Batch store (built 2026-10-25) + +`app/services/batch_store.ts`: every release chunk of a batch is one +release, inserted once. Its keys (`lib/release/keys.ts`): `identity_key` +the identity's SHA-1 as 20 bytes; `release_key` the SHA-1 of the identity, +the owner group and the release's first article, so it is deterministic +(no numbering under a group lock, no catalog lookups before writing) and a +repost, starting at another article, is a release of its own. Its poster +and group set by id (`app/services/catalog_refs.ts`: insert what's new, +read the ids, cache them per process; the tests forget the cache per +test), its source as a code, its name into `release_texts` (kind 1). The +chunk's own counts go in as the release's; `complete` is decided then from +the counter. A finished release's probe candidates are claimable at once +(`ready`, `unreadable`); a provisional tail release (`finished: false` on +the chunk, `settled_at` null) is finished by a later chunk of the same +release key, which adds the files not sent before (sums, one more chunk), +sets `settled_at` and makes its candidates claimable; a finished release +sent again is left as it is (`ON CONFLICT … WHERE settled_at IS NULL`). +Gone: the join of a chunk into an earlier release of its key (the window +keeps an upload together; the block merger joins what a block's end cut), +the join across groups (FTD's spread uploads: a gap, to be handled by the +block merger or the par2 merger across groups), the numbering of release +keys, the group lock, `fragments` (ignored when sent; the Rust edge still +sends them). The chunk type carries `finished` and `edge` (both optional; +the Rust edge doesn't send them yet, so its releases are finished, not +edges). Interim until blocks are leased: chunks, rule counts and samples +are keyed by `ranges.block_id`, and a range without a block gets a block +of its own in the store (`blockOf`). + +Readers moved with it: search joins `posters` and `group_sets`, matches +words in `release_texts` (`r.id IN (SELECT release_id … ILIKE)`), searches +posters by id, filters `source` by code and returns the source name; the +par2 merger, the sidecar folder and the judge's scraped rule use +`poster_id` (`release_merges.from_key` holds the key as hex); the spot +inspector matches by bytea identity keys and inserts its spot releases by +id references with their title in `release_texts`; the spotter and the NZB +service compare the source code. Tests insert fixtures through +`tests/functional/helpers/releases.ts` (`insertRelease`: text keys, poster +and groups by name, source by name). The unit and functional suites pass. + +Next: the window and albums in the Rust edge (`finished`, `edge`, the +provisional tail and its finish), block leases and the planner (then the +store keys by the leased block and the lease columns leave `ranges`), the +block merger, the prober writing `release_texts` kind 2, the corpus replay +harness. ## Decisions diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index a6c9c8d..a916cb5 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -1,14 +1,18 @@ import db from '@adonisjs/lucid/services/db' import { pruneSevenZipTails } from '#services/seven_zip_tails' import type { TransactionClientContract } from '@adonisjs/lucid/types/database' -import { UPLOAD_GAP_MS, type EdgeBatch, type ProbeCandidate, type ReleaseChunk } from '#lib/edge/index' +import type { EdgeBatch, ProbeCandidate, ReleaseChunk } from '#lib/edge/index' import ChunkPartitions from '#services/chunk_partitions' +import CatalogRefs, { groupSetKey, posterText } from '#services/catalog_refs' import { decodeChunk } from '#lib/chunks/index' import { SIDECAR } from '#services/sidecar_folder' +import { identityKey, releaseKey } from '#lib/release/keys' +import { sourceCode } from '#lib/release/source' +import { TEXT_KINDS, TEXT_MAX } from '#lib/release/text_kind' /** * A small new release (at most 3 files, under 1 MB) whose files are all sidecars (.nfo, .sfv, .nzb, …): the sidecar - * fold's candidates (releases.sidecar_only; the settler queues only these). Its file list decoded here, small as it is. + * fold's candidates (releases.sidecar_only). Its file list decoded here, small as it is. */ function sidecarOnly(r: ReleaseChunk): boolean | null { if (r.files > 3 || r.bytes >= 1_000_000) return null @@ -19,18 +23,32 @@ function sidecarOnly(r: ReleaseChunk): boolean | null { export class StaleBatch extends Error {} export interface StoredBatch { + /** Release chunks in the batch. */ releases: number + /** Releases the batch started. */ newReleases: number + /** Provisional (tail) releases the batch finished. */ + finalized: number candidates: number - fragments: number } +/** A chunk's slot in its batch: the key and the upload the edge numbered it with (ProbeCandidate points at it). */ +const slot = (key: string, upload: number) => `${key}\n${upload}` + /** - * Stores one batch of a leased range (REBUILD.md, parts 3–7), in one - * transaction: release aggregates in the catalog, one chunk per release, - * probe candidates, rule counts and samples, edge fragments, and the - * range's resume point. A batch that doesn't start at the resume point was + * Stores one batch of a leased range (REBUILD.md, "Voyager 3: window indexing", 2 and 4) in one transaction: each + * release chunk is a whole release the edge's window finished, written once (its poster and group set by id, its + * keys as bytea, its name into release_texts), with one chunk of its files, its probe candidates, and the batch's + * rule counts and samples; then the range's resume point moves. A batch that doesn't start at the resume point was * sent twice or too late, and is refused (StaleBatch), so the sums stay right. + * + * A tail release comes twice: provisionally (finished: false, settled_at null, its candidates waiting) and then + * finished, under the same release key (the identity, the group and its first article: lib/release/keys.ts), with + * the files not sent before (a chunk's counts are its own; the finish adds them up); the finish is the one update a + * release gets from the header pass. A finished release sent again is left as it is. + * + * Interim, until blocks are leased (the next step): chunks, counts and samples are keyed by the range's block + * (ranges.block_id), and a range without one gets a block of its own here. */ class BatchStore { async store(rangeId: number, nodeId: number, batch: EdgeBatch): Promise { @@ -39,7 +57,9 @@ class BatchStore { // One batch of a range at a time, without holding the range's row for the whole transaction: the // prefetcher's and heartbeats' updates of it waited on it (0.5 s on average, NNTP connections idle; // LOG.md 2026-10-09 (13)). The resume point is checked again when it moves, at the end. - await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended('range:' || ?::text, 0))`, [rangeId]) + await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended('range:' || ?::text, 0))`, [ + rangeId, + ]) const range = await trx.from('ranges').where('id', rangeId).first() if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new StaleBatch(`Range ${rangeId} is not leased to this node`) @@ -47,35 +67,37 @@ class BatchStore { throw new StaleBatch( `Range ${rangeId} is at ${range.next_id}, the batch starts at ${batch.from}` ) + const blockId = await this.blockOf(trx, range) const result: StoredBatch = { releases: batch.releases.length, newReleases: 0, + finalized: 0, candidates: 0, - fragments: batch.fragments.length, } const written = await this.writeReleases( trx, - rangeId, + { groupId: Number(range.group_id), blockId }, batch.from, batch.releases, batch.candidates ) result.newReleases = written.newReleases + result.finalized = written.finalized result.candidates = written.candidates if (batch.counts.length > 0) { await trx.rawQuery( - `INSERT INTO rule_counts (range_id, rule, key, articles, files) + `INSERT INTO rule_counts (block_id, rule, key, articles, files) SELECT ?, v.rule, v.key, v.articles, v.files FROM jsonb_to_recordset(?::jsonb) AS v(rule text, key text, articles bigint, files bigint) - ON CONFLICT (range_id, rule, key) DO UPDATE SET articles = rule_counts.articles + excluded.articles, files = rule_counts.files + excluded.files`, - [rangeId, JSON.stringify(batch.counts)] + ON CONFLICT (block_id, rule, key) DO UPDATE SET articles = rule_counts.articles + excluded.articles, files = rule_counts.files + excluded.files`, + [blockId, JSON.stringify(batch.counts)] ) } for (const s of batch.samples) { await trx.insertQuery().table('rule_samples').insert({ - range_id: rangeId, + block_id: blockId, rule: s.rule, rule_key: s.ruleKey, release_key: s.key, @@ -95,106 +117,75 @@ class BatchStore { .update({ next_id: batch.next, headers: db.raw('headers + ?', [batch.headers]), - // The range's posting time span (release_final): the batches' trimmed spans, min and max (least and greatest skip nulls). - posted_low: db.raw('least(posted_low, to_timestamp(?::double precision / 1000))', [batch.postedLow ?? null]), - posted_high: db.raw('greatest(posted_high, to_timestamp(?::double precision / 1000))', [batch.postedHigh ?? null]), + // The range's posting time span: the batches' trimmed spans, min and max (least and greatest skip nulls). + posted_low: db.raw('least(posted_low, to_timestamp(?::double precision / 1000))', [ + batch.postedLow ?? null, + ]), + posted_high: db.raw('greatest(posted_high, to_timestamp(?::double precision / 1000))', [ + batch.postedHigh ?? null, + ]), heartbeat_at: new Date(), }) - if (Number(moved) !== 1 && !(Array.isArray(moved) && moved.length === 1)) throw new StaleBatch(`Range ${rangeId} moved on meanwhile`) + if (Number(moved) !== 1 && !(Array.isArray(moved) && moved.length === 1)) + throw new StaleBatch(`Range ${rangeId} moved on meanwhile`) return result }) } - /** - * An upload spread over groups (FTD: par2 and part1 to a.b.music.misc, - * part2 to a.b.sounds.music, each article in one group only; LOG.md - * 2026-10-11 (9)): a chunk without a release in its own group joins one of - * its key in another group whose window is within UPLOAD_GAP_MS, when none - * of its file names are in that release yet. A post of the same files to - * another group (other message-ids, the same names) stays a release of its - * own, or the NZB would hold every file twice. Reads the candidate's file - * names from its chunks; candidates are few. - */ - private async joinAcrossGroups( - trx: TransactionClientContract, - part: ReleaseChunk[], - windows: { idx: number; key: string; first: number | null | undefined; last: number | null | undefined }[], - groupId: number, - joined: Map - ) { - const open = windows.filter((w) => !joined.has(w.idx) && w.first) - if (open.length === 0) return - const { rows: candidates } = await trx.rawQuery( - `SELECT v.idx, r.id - FROM jsonb_to_recordset(?::jsonb) AS v(idx int, key text, first bigint, last bigint) - JOIN releases r ON r.identity_key = v.key AND r.group_id <> ? - WHERE r.posted_at <= to_timestamp((v.last + ?) / 1000.0) AND coalesce(r.last_posted_at, r.posted_at) >= to_timestamp((v.first - ?) / 1000.0) - ORDER BY v.idx, r.last_posted_at DESC NULLS LAST, r.id DESC`, - [JSON.stringify(open), groupId, UPLOAD_GAP_MS, UPLOAD_GAP_MS] - ) - if (candidates.length === 0) return - const ids: number[] = [...new Set(candidates.map((c: { id: string }) => Number(c.id)))] - const names = new Map>() - const { rows: chunks } = await trx.rawQuery(`SELECT release_id, payload FROM release_chunks WHERE release_id = ANY(?)`, [ids]) - for (const c of chunks) { - const set = names.get(Number(c.release_id)) ?? new Set() - for (const f of decodeChunk(c.payload)) set.add(f.name) - names.set(Number(c.release_id), set) - } - for (const c of candidates as { idx: number; id: string }[]) { - if (joined.has(c.idx)) continue - const have = names.get(Number(c.id)) ?? new Set() - const incoming = decodeChunk(part[c.idx].payload).map((f) => f.name) - if (incoming.every((name) => !have.has(name))) { - joined.set(c.idx, Number(c.id)) - for (const name of incoming) have.add(name) - names.set(Number(c.id), have) - } - } + /** The range's block; a range without one (no planner yet) is a block of its own, leased as the range is. */ + private async blockOf(trx: TransactionClientContract, range: any): Promise { + if (range.block_id) return Number(range.block_id) + const [block] = await trx + .table('blocks') + .insert({ + group_id: range.group_id, + start_id: range.start_id, + end_id: range.end_id, + next_id: range.next_id, + status: 'leased', + priority: range.priority ?? 0, + node_id: range.node_id, + leased_at: range.leased_at ?? new Date(), + heartbeat_at: new Date(), + }) + .returning('id') + await trx.from('ranges').where('id', range.id).update({ block_id: block.id }) + return Number(block.id) } /** - * Release aggregates in the catalog, one write-once chunk per release, and - * probe candidates (a batch's, or files the fragment merger completed). - * A chunk joins the release of its key whose articles the range's group - * owns and whose posting window is within UPLOAD_GAP_MS of its own, else - * starts one: re-uploads are releases of their own (decision d). The - * group is locked, so concurrent writers don't both start one. + * The releases of a batch into the catalog, one chunk each, and their probe candidates. Every chunk is one + * release: inserted, or (a provisional tail release sent again) finished. */ async writeReleases( trx: TransactionClientContract, - rangeId: number, + where: { groupId: number; blockId: number }, batchFrom: number, releases: ReleaseChunk[], candidates: ProbeCandidate[] ) { - const out = { newReleases: 0, candidates: 0 } + const out = { newReleases: 0, finalized: 0, candidates: 0 } + if (releases.length === 0) return out + /** Release ids by slot, for the chunks and candidates; a slot not here was a finished release sent again. */ const ids = new Map() - const slot = (key: string, upload: number) => `${key}\n${upload}` - const { group_id: groupId } = await trx.from('ranges').where('id', rangeId).select('group_id').firstOrFail() - // One writer per group at a time: a chunk only matches releases of its group, and a lock per key - // ran out of shared memory (a picture group's batch has 50k keys; LOG.md 2026-10-09 (12)). - await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended('releases:' || ?::text, 0))`, [groupId]) + const finalized = new Set() + + const posters = await CatalogRefs.posterIds(trx, releases.map((r) => r.poster).filter(Boolean)) + const groupSets = await CatalogRefs.groupSetIds( + trx, + releases.map((r) => r.groups) + ) const sorted = [...releases].sort((a, b) => a.key.localeCompare(b.key) || a.upload - b.upload) for (let i = 0; i < sorted.length; i += 1000) { const part = sorted.slice(i, i + 1000) - const windows = part.map((r, idx) => ({ idx, key: r.key, first: r.postedAt, last: r.postedLast ?? r.postedAt })) - const { rows: matches } = await trx.rawQuery( - `SELECT DISTINCT ON (v.idx) v.idx, r.id - FROM jsonb_to_recordset(?::jsonb) AS v(idx int, key text, first bigint, last bigint) - JOIN releases r ON r.identity_key = v.key AND r.group_id = ? - WHERE v.first IS NULL OR r.posted_at IS NULL - OR (r.posted_at <= to_timestamp((v.last + ?) / 1000.0) AND coalesce(r.last_posted_at, r.posted_at) >= to_timestamp((v.first - ?) / 1000.0)) - ORDER BY v.idx, r.last_posted_at DESC NULLS LAST, r.id DESC`, - [JSON.stringify(windows), groupId, UPLOAD_GAP_MS, UPLOAD_GAP_MS] - ) - const joined = new Map(matches.map((m: { idx: number; id: string }) => [m.idx, Number(m.id)])) - await this.joinAcrossGroups(trx, part, windows, groupId, joined) - const row = (r: ReleaseChunk) => ({ + const rows = part.map((r, idx) => ({ + idx, + release_key: releaseKey(r.key, where.groupId, r.firstArticle).toString('hex'), + identity_key: identityKey(r.key).toString('hex'), name: r.name, - source: r.source, - poster: r.poster, - groups: r.groups, + source: sourceCode(r.source), + poster_id: r.poster ? (posters.get(posterText(r.poster)) ?? null) : null, + group_set_id: groupSets.get(groupSetKey(r.groups))!, first: r.postedAt, last: r.postedLast ?? r.postedAt, file_total: r.fileTotal, @@ -205,107 +196,105 @@ class BatchStore { last_article: r.lastArticle, readable: r.readable, has_par2: r.hasPar2, - }) - const updates = part.flatMap((r, idx) => (joined.has(idx) ? [{ id: joined.get(idx)!, ...row(r) }] : [])) - if (updates.length > 0) { - await trx.rawQuery( - `WITH v AS (SELECT * FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, first bigint, last bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean)), - a AS (SELECT id, sum(files) AS files, sum(segments) AS segments, sum(bytes) AS bytes, count(*) AS chunks, min(first) AS first, max(last) AS last, - max(file_total) AS file_total, min(first_article) AS first_article, max(last_article) AS last_article, bool_or(readable) AS readable, bool_or(has_par2) AS has_par2 - FROM v GROUP BY id) - UPDATE releases r SET - files = r.files + a.files, - segments = r.segments + a.segments, - bytes = r.bytes + a.bytes, - chunks = r.chunks + a.chunks, - posted_at = least(r.posted_at, to_timestamp(a.first / 1000.0)), - last_posted_at = greatest(r.last_posted_at, to_timestamp(a.last / 1000.0)), - file_total = coalesce(r.file_total, a.file_total), - first_article = least(r.first_article, a.first_article), - last_article = greatest(r.last_article, a.last_article), - readable = r.readable OR a.readable, - has_par2 = r.has_par2 OR a.has_par2, - -- As it stands (the counter's files all found); settling makes it final. Without a counter, what its par2 - -- set decided stays (app/services/par2_completeness.ts). - complete = CASE WHEN coalesce(r.file_total, a.file_total) IS NULL THEN r.complete ELSE r.files + a.files >= coalesce(r.file_total, a.file_total) END, - last_seen_at = now() - FROM a WHERE r.id = a.id`, - [JSON.stringify(updates)] + finished: r.finished !== false, + edge: r.edge === true, + sidecar_only: sidecarOnly(r), + })) + // Inserted, or a provisional release of the key finished (the chunk's files added up); a finished release + // sent again (no row returned) is left as it is. + const { rows: written } = await trx.rawQuery( + `INSERT INTO releases (release_key, identity_key, group_id, group_set_id, poster_id, name, source, posted_at, last_posted_at, file_total, files, segments, bytes, chunks, + first_article, last_article, readable, has_par2, complete, edge, sidecar_only, settled_at) + SELECT decode(v.release_key, 'hex'), decode(v.identity_key, 'hex'), ?, v.group_set_id, v.poster_id, v.name, v.source, + to_timestamp(v.first / 1000.0), to_timestamp(v.last / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, + v.first_article, v.last_article, v.readable, v.has_par2, CASE WHEN v.file_total IS NULL THEN NULL ELSE v.files >= v.file_total END, + v.edge, v.sidecar_only, CASE WHEN v.finished THEN now() END + FROM jsonb_to_recordset(?::jsonb) AS v(idx int, release_key text, identity_key text, name text, source int, poster_id int, group_set_id int, first bigint, last bigint, + file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean, + finished boolean, edge boolean, sidecar_only boolean) + ON CONFLICT (release_key) DO UPDATE SET + files = releases.files + excluded.files, segments = releases.segments + excluded.segments, bytes = releases.bytes + excluded.bytes, + chunks = releases.chunks + CASE WHEN excluded.files > 0 THEN 1 ELSE 0 END, + posted_at = least(releases.posted_at, excluded.posted_at), last_posted_at = greatest(releases.last_posted_at, excluded.last_posted_at), + file_total = coalesce(excluded.file_total, releases.file_total), + first_article = least(releases.first_article, excluded.first_article), last_article = greatest(releases.last_article, excluded.last_article), + readable = releases.readable OR excluded.readable, has_par2 = releases.has_par2 OR excluded.has_par2, + complete = CASE WHEN coalesce(excluded.file_total, releases.file_total) IS NULL THEN releases.complete + ELSE releases.files + excluded.files >= coalesce(excluded.file_total, releases.file_total) END, + edge = excluded.edge, sidecar_only = coalesce(releases.sidecar_only, excluded.sidecar_only), settled_at = excluded.settled_at + WHERE releases.settled_at IS NULL + RETURNING id, encode(release_key, 'hex') AS release_key, (xmax = 0) AS inserted, settled_at`, + [where.groupId, JSON.stringify(rows)] + ) + const byKey = new Map( + written.map( + (w: { id: string; release_key: string; inserted: boolean; settled_at: Date | null }) => [ + w.release_key, + { id: Number(w.id), inserted: w.inserted, settled: w.settled_at !== null }, + ] ) - part.forEach((r, idx) => joined.has(idx) && ids.set(slot(r.key, r.upload), joined.get(idx)!)) + ) + const texts: { release_id: number; text: string }[] = [] + for (const row of rows) { + const w = byKey.get(row.release_key) + if (!w) continue + const r = part[row.idx] + ids.set(slot(r.key, r.upload), w.id) + if (w.inserted) out.newReleases++ + else if (w.settled) { + out.finalized++ + finalized.add(w.id) + } + texts.push({ release_id: w.id, text: row.name.slice(0, TEXT_MAX) }) } - const fresh = part.filter((_, idx) => !joined.has(idx)) - if (fresh.length > 0) { - // The first upload of a key keeps the key; later ones are key:2, key:3, … after the highest number: merges - // delete releases, and a count fell behind it (83 releases of a key up to key:129: 20 numbers tried, all - // taken, the range's batch failed for good; LOG.md 2026-10-20 (89)). Spot releases (spot:key) aren't numbered. - const { rows: counts } = await trx.rawQuery( - `SELECT identity_key, max(CASE WHEN release_key = identity_key THEN 1 - WHEN left(release_key, length(identity_key) + 1) = identity_key || ':' AND substr(release_key, length(identity_key) + 2) ~ '^[0-9]{1,9}$' - THEN substr(release_key, length(identity_key) + 2)::int END) AS n - FROM releases WHERE identity_key = ANY(?) GROUP BY 1`, - [[...new Set(fresh.map((r) => r.key))]] + // The name, searchable (release_texts; a name sent again is a no-op). + if (texts.length > 0) { + await trx.rawQuery( + `INSERT INTO release_texts (release_id, kind, text) + SELECT v.release_id, ?, v.text FROM jsonb_to_recordset(?::jsonb) AS v(release_id bigint, text text) WHERE v.text <> '' + ON CONFLICT DO NOTHING`, + [TEXT_KINDS.name, JSON.stringify(texts)] ) - const taken = new Map(counts.map((c: { identity_key: string; n: string }) => [c.identity_key, Number(c.n)])) - const numbered = (r: ReleaseChunk) => { - const n = (taken.get(r.key) ?? 0) + 1 - taken.set(r.key, n) - return { release_key: n === 1 ? r.key : `${r.key}:${n}`, key: r.key, upload: r.upload, ...row(r), sidecar_only: sidecarOnly(r) } - } - // Their posters, for poster search (posters: one row per poster, searched by trigram; migration 4000). - const posters = [...new Set(fresh.map((r) => r.poster).filter((p): p is string => !!p))] - if (posters.length > 0) await trx.rawQuery(`INSERT INTO posters (poster) SELECT DISTINCT left(p, 200) FROM unnest(?::text[]) AS p ON CONFLICT DO NOTHING`, [posters]) - let inserts = fresh.map(numbered) - // Writers of other groups number the same key at the same time (a spam post crossposted to 67 - // groups; only the group is locked): a number taken meanwhile is skipped, and the next one tried. - for (let attempt = 0; inserts.length > 0; attempt++) { - if (attempt === 20) throw new Error(`No free release key for ${inserts[0].key}`) - const { rows } = await trx.rawQuery( - `INSERT INTO releases (release_key, identity_key, group_id, name, source, poster, groups, posted_at, last_posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2, complete, sidecar_only) - SELECT v.release_key, v.key, ?, v.name, v.source, v.poster, v.groups, to_timestamp(v.first / 1000.0), to_timestamp(v.last / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, - v.first_article, v.last_article, v.readable, v.has_par2, CASE WHEN v.file_total IS NULL THEN NULL ELSE v.files >= v.file_total END, v.sidecar_only - FROM jsonb_to_recordset(?::jsonb) AS v(release_key text, key text, name text, source text, poster text, groups jsonb, first bigint, last bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean, sidecar_only boolean) - ON CONFLICT (release_key) DO NOTHING - RETURNING id, release_key`, - [groupId, JSON.stringify(inserts)] - ) - const byReleaseKey = new Map(rows.map((x: { id: string; release_key: string }) => [x.release_key, Number(x.id)])) - const missed: typeof inserts = [] - for (const r of inserts) { - const id = byReleaseKey.get(r.release_key) - if (id === undefined) missed.push(r) - else ids.set(slot(r.key, r.upload), id) - } - out.newReleases += rows.length - const chunks = new Map(fresh.map((r) => [slot(r.key, r.upload), r])) - inserts = missed.map((r) => numbered(chunks.get(slot(r.key, r.upload))!)) - } } } - // One write-once chunk per release. + // One write-once chunk per release (a finish adds the files not sent before). for (let i = 0; i < sorted.length; i += 500) { - const rows = sorted.slice(i, i + 500).map((r) => ({ - release_id: ids.get(slot(r.key, r.upload))!, - range_id: rangeId, - batch_from: batchFrom, - files: r.files, - segments: r.segments, - payload: r.payload, - })) + const rows = sorted + .slice(i, i + 500) + .filter((r) => ids.has(slot(r.key, r.upload)) && r.files > 0) + .map((r) => ({ + release_id: ids.get(slot(r.key, r.upload))!, + block_id: where.blockId, + batch_from: batchFrom, + files: r.files, + segments: r.segments, + payload: r.payload, + })) if (rows.length > 0) await trx.insertQuery().table('release_chunks').multiInsert(rows) } + // A finished release's candidates are claimable at once; a provisional one's wait for its finish. + const settled = new Set() + for (const r of sorted) { + const id = ids.get(slot(r.key, r.upload)) + if (id !== undefined && r.finished !== false) settled.add(id) + } + const chunks = new Map(sorted.map((r) => [slot(r.key, r.upload), r])) const kept = candidates.filter((c) => ids.has(slot(c.releaseKey, c.upload ?? 0))) for (let i = 0; i < kept.length; i += 2000) { - const rows = kept - .slice(i, i + 2000) - .map((c) => ({ - release_id: ids.get(slot(c.releaseKey, c.upload ?? 0))!, + const rows = kept.slice(i, i + 2000).map((c) => { + const releaseId = ids.get(slot(c.releaseKey, c.upload ?? 0))! + const chunk = chunks.get(slot(c.releaseKey, c.upload ?? 0))! + return { + release_id: releaseId, kind: c.kind, message_id: c.messageId, file: c.file, - })) + ready: settled.has(releaseId), + unreadable: !chunk.readable, + } + }) const added = await trx .knexQuery() .table('probe_candidates') @@ -317,15 +306,14 @@ class BatchStore { // A later piece of a split 7z: the tails of lower volumes go (app/services/seven_zip_tails.ts). const tails = new Set(rows.filter((r) => r.kind === '7z-tail').map((r) => r.release_id)) if (added.length > 0 && tails.size > 0) await pruneSevenZipTails(trx, [...tails]) - // Files added to a settled release (a later range of its upload): the settler made its candidates claimable - // already and won't again (93,642 waited since 2026-10-13; LOG.md 2026-10-20 (87)). - if (added.length > 0) { - await trx.rawQuery( - `UPDATE probe_candidates pc SET ready = true, unreadable = NOT r.readable FROM releases r - WHERE pc.id = ANY(?) AND r.id = pc.release_id AND r.settled_at IS NOT NULL`, - [added.map((a: { id: string | number }) => String(a.id))] - ) - } + } + // The finish of a provisional release: its waiting candidates become claimable. + if (finalized.size > 0) { + await trx.rawQuery( + `UPDATE probe_candidates pc SET ready = true, unreadable = NOT r.readable FROM releases r + WHERE pc.release_id = ANY(?) AND r.id = pc.release_id AND pc.status = 'waiting' AND NOT pc.ready`, + [[...finalized]] + ) } return out diff --git a/app/services/catalog_refs.ts b/app/services/catalog_refs.ts new file mode 100644 index 0000000..c4cb27d --- /dev/null +++ b/app/services/catalog_refs.ts @@ -0,0 +1,93 @@ +import type db from '@adonisjs/lucid/services/db' + +type Client = Pick + +/** Posters are kept to this length (as the trigram index and poster search always did; a btree entry holds 2,704 bytes). */ +export const POSTER_MAX = 200 +/** Cached ids per process: group sets are ~2,600 in all; posters many more, so the cache is emptied when it grows past this. */ +const CACHE_MAX = 200_000 + +const posterCache = new Map() +const groupSetCache = new Map() + +/** A group list as group_sets stores it: sorted, without repeats, as JSON. */ +export function groupSetKey(groups: string[]): string { + return JSON.stringify([...new Set(groups)].sort()) +} + +/** A poster as posters stores it. */ +export function posterText(poster: string): string { + return poster.slice(0, POSTER_MAX) +} + +async function ids( + client: Client, + cache: Map, + table: 'posters' | 'group_sets', + column: 'poster' | 'groups', + cast: 'text' | 'jsonb', + values: string[] +): Promise> { + const out = new Map() + const missing: string[] = [] + for (const v of values) { + const id = cache.get(v) + if (id === undefined) missing.push(v) + else out.set(v, id) + } + if (missing.length === 0) return out + // Insert what's new, then read every id (ON CONFLICT DO NOTHING returns nothing for the rows that existed). + await client.rawQuery( + `INSERT INTO ${table} (${column}) SELECT DISTINCT v::${cast} FROM unnest(?::text[]) AS v ON CONFLICT DO NOTHING`, + [missing] + ) + const { rows } = await client.rawQuery( + `SELECT id, ${column}::text AS value FROM ${table} WHERE ${column} = ANY(?::${cast}[])`, + [missing] + ) + if (cache.size + rows.length > CACHE_MAX) cache.clear() + for (const r of rows) { + // jsonb comes back in Postgres' spelling ("["a", "b"]"); the caller's key is the JSON we sent. + const key = cast === 'jsonb' ? JSON.stringify(JSON.parse(r.value)) : r.value + cache.set(key, Number(r.id)) + out.set(key, Number(r.id)) + } + return out +} + +/** + * Ids in the catalog's lookup tables (voyager 3, migration 6000): a release's poster_id and group_set_id. New + * values are inserted; known ones come from a cache per process (the batch store sees the same posters and + * group sets batch after batch). + */ +export default { + /** Ids by poster (posterText applied), for a batch of posters. */ + async posterIds(client: Client, posters: string[]): Promise> { + return ids(client, posterCache, 'posters', 'poster', 'text', [ + ...new Set(posters.map(posterText)), + ]) + }, + + async posterId(client: Client, poster: string): Promise { + const found = await this.posterIds(client, [poster]) + return found.get(posterText(poster))! + }, + + /** Ids by groupSetKey, for a batch of group lists. */ + async groupSetIds(client: Client, sets: string[][]): Promise> { + return ids(client, groupSetCache, 'group_sets', 'groups', 'jsonb', [ + ...new Set(sets.map(groupSetKey)), + ]) + }, + + async groupSetId(client: Client, groups: string[]): Promise { + const found = await this.groupSetIds(client, [groups]) + return found.get(groupSetKey(groups))! + }, + + /** Forgets the cached ids (tests roll their transactions back, so the ids they saw are gone). */ + forget() { + posterCache.clear() + groupSetCache.clear() + }, +} diff --git a/app/services/nzb_service.ts b/app/services/nzb_service.ts index 1d547a5..f1df2d8 100644 --- a/app/services/nzb_service.ts +++ b/app/services/nzb_service.ts @@ -2,6 +2,7 @@ import { gunzipSync } from 'node:zlib' import db from '@adonisjs/lucid/services/db' import { decodeChunk, type ChunkFile } from '#lib/chunks/index' import { buildNzb } from '#lib/nzb/index' +import { SOURCES } from '#lib/release/source' /** Where a known password came from, the most trusted first: the spot's NZB meta, its title, its text. */ const PASSWORD_SOURCES = ['nzb', 'spot-title', 'spot-text'] @@ -35,7 +36,7 @@ class NzbService { if (!release) return null const password = (await this.passwords(releaseId))[0]?.password // A release a spot made of its own (no files matched the header pass's): the spot's NZB as read. - if (release.source === 'spot') { + if (Number(release.source) === SOURCES.spot) { const spot = await db.from('spots').where('release_id', releaseId).whereNotNull('nzb').select('nzb').first() if (spot) return { name: release.name, xml: withPassword(gunzipSync(spot.nzb).toString('utf8'), password) } } diff --git a/app/services/par2_merger.ts b/app/services/par2_merger.ts index 76e5927..f1e1c48 100644 --- a/app/services/par2_merger.ts +++ b/app/services/par2_merger.ts @@ -93,11 +93,12 @@ const HOLDERS = `SELECT c.release_id FROM ( interface Member { id: number - release_key: string group_id: number files: number posted_at: Date | null readable: boolean + poster_id?: number | null + /** The poster's text (joined from posters) where a query needs it. */ poster?: string | null file_total?: number | null } @@ -295,7 +296,8 @@ export class Par2Merger { */ async nameMatches(trx: any, setId: string): Promise | null> { const { rows: all } = await trx.rawQuery( - `SELECT r.id, r.release_key, r.group_id, r.files, r.posted_at, r.readable, r.poster FROM par2_sets s JOIN releases r ON r.id = s.release_id + `SELECT r.id, r.group_id, r.files, r.posted_at, r.readable, r.poster_id, p.poster + FROM par2_sets s JOIN releases r ON r.id = s.release_id LEFT JOIN posters p ON p.id = r.poster_id WHERE s.set_id = ? ORDER BY r.posted_at DESC NULLS LAST, r.id`, [setId] ) @@ -310,12 +312,12 @@ export class Par2Merger { const names = new Set(described.map((d: { name: string }) => fileKey(d.name))) const at = new Date(anchor.posted_at).getTime() const { rows: candidates } = await trx.rawQuery( - `SELECT r.id, r.release_key, r.group_id, r.files, r.posted_at, r.readable, r.poster FROM releases r - WHERE left(r.poster, 200) = left(?, 200) AND r.poster = ? AND r.group_id = ? AND r.posted_at BETWEEN ? AND ? AND r.id <> ALL(?) + `SELECT r.id, r.group_id, r.files, r.posted_at, r.readable, r.poster_id FROM releases r + WHERE r.poster_id = ? AND r.group_id = ? AND r.posted_at BETWEEN ? AND ? AND r.id <> ALL(?) AND r.files <= ? - ORDER BY left(r.poster, 200), r.posted_at DESC NULLS LAST, r.id DESC LIMIT ?`, + ORDER BY r.posted_at DESC NULLS LAST, r.id DESC LIMIT ?`, // More files than the set describes, with its sidecars, can't all be named in it. - [anchor.poster, anchor.poster, anchor.group_id, new Date(at - NAME_SPREAD_MS), new Date(at + NAME_SPREAD_MS), holders.map((h: Member) => h.id), described.length + SLACK_FILES, NAME_CANDIDATES] + [anchor.poster_id, anchor.group_id, new Date(at - NAME_SPREAD_MS), new Date(at + NAME_SPREAD_MS), holders.map((h: Member) => h.id), described.length + SLACK_FILES, NAME_CANDIDATES] ) const found = new Map(holders.map((h: Member) => [String(h.id), h])) let files = holders.reduce((n: number, h: Member) => n + Number(h.files), 0) @@ -344,7 +346,7 @@ export class Par2Merger { async mergeStem(stem: string): Promise { return db.transaction(async (trx) => { const { rows } = await trx.rawQuery( - `SELECT DISTINCT r.id, r.release_key, r.group_id, r.files, r.posted_at, r.readable, r.poster, r.file_total + `SELECT DISTINCT r.id, r.group_id, r.files, r.posted_at, r.readable, r.poster_id, r.file_total FROM release_inspections i JOIN releases r ON r.id = i.release_id WHERE i.kind = 'lone' AND ${INNER_STEM} = ?`, [stem] @@ -356,7 +358,7 @@ export class Par2Merger { const total = (r: Member) => (r.file_total ? Number(r.file_total) : null) const near = (r: Member) => r.group_id === anchor.group_id && - (r.poster ?? '') === (anchor.poster ?? '') && + (r.poster_id ?? null) === (anchor.poster_id ?? null) && (total(r) === null || total(anchor) === null || total(r) === total(anchor)) && (at(r) === null || at(anchor) === null || Math.abs(at(r)! - at(anchor)!) <= MAX_SPREAD_MS) const found = new Map() @@ -374,7 +376,7 @@ export class Par2Merger { async mergeSet(setId: string): Promise { return db.transaction(async (trx) => { const { rows: holders } = await trx.rawQuery( - `SELECT r.id, r.release_key, r.group_id, r.files, r.posted_at, r.readable FROM par2_sets s JOIN releases r ON r.id = s.release_id + `SELECT r.id, r.group_id, r.files, r.posted_at, r.readable FROM par2_sets s JOIN releases r ON r.id = s.release_id WHERE s.set_id = ? ORDER BY r.id`, [setId] ) @@ -383,7 +385,7 @@ export class Par2Merger { const at = (r: Member) => (r.posted_at ? new Date(r.posted_at).getTime() : null) const near = (r: Member) => r.group_id === anchor.group_id && (at(r) === null || at(anchor) === null || Math.abs(at(r)! - at(anchor)!) <= MAX_SPREAD_MS) const { rows: matches } = await trx.rawQuery( - `SELECT f.file_id, r.id, r.release_key, r.group_id, r.files, r.posted_at, r.readable + `SELECT f.file_id, r.id, r.group_id, r.files, r.posted_at, r.readable FROM par2_files f CROSS JOIN LATERAL (${HOLDERS}) h JOIN releases r ON r.id = h.release_id WHERE f.set_id = ?`, [setId] @@ -426,7 +428,7 @@ export class Par2Merger { await trx.rawQuery( `INSERT INTO release_merges (from_id, from_key, into_id, set_id, files, via) - SELECT id, release_key, ?, ?, files, ? FROM releases WHERE id = ANY(?)`, + SELECT id, encode(release_key, 'hex'), ?, ?, files, ? FROM releases WHERE id = ANY(?)`, [target.id, key, via, others] ) // A candidate of an article the target, or another member of lower id, has already (the same file in two releases) @@ -462,13 +464,13 @@ export class Par2Merger { files = t.files + s.files, segments = t.segments + s.segments, bytes = t.bytes + s.bytes, chunks = t.chunks + s.chunks, first_article = least(t.first_article, s.first_article), last_article = greatest(t.last_article, s.last_article), posted_at = least(t.posted_at, s.posted_at), last_posted_at = greatest(t.last_posted_at, s.last_posted_at), - first_seen_at = least(t.first_seen_at, s.first_seen_at), last_seen_at = greatest(t.last_seen_at, s.last_seen_at), + first_seen_at = least(t.first_seen_at, s.first_seen_at), readable = t.readable OR s.readable, has_par2 = t.has_par2 OR s.has_par2, passworded = t.passworded OR s.passworded, file_total = greatest(t.file_total, s.file_total), inner_name = coalesce(t.inner_name, s.inner_name), name = coalesce(nullif(?, ''), t.name) FROM (SELECT sum(files) AS files, sum(segments) AS segments, sum(bytes) AS bytes, sum(chunks) AS chunks, min(first_article) AS first_article, max(last_article) AS last_article, min(posted_at) AS posted_at, - max(last_posted_at) AS last_posted_at, min(first_seen_at) AS first_seen_at, max(last_seen_at) AS last_seen_at, + max(last_posted_at) AS last_posted_at, min(first_seen_at) AS first_seen_at, bool_or(readable) AS readable, bool_or(has_par2) AS has_par2, bool_or(passworded) AS passworded, max(file_total) AS file_total, max(inner_name) AS inner_name FROM releases WHERE id = ANY(?)) s diff --git a/app/services/release_judge.ts b/app/services/release_judge.ts index f0b23e3..79e7901 100644 --- a/app/services/release_judge.ts +++ b/app/services/release_judge.ts @@ -1,5 +1,6 @@ import db from '@adonisjs/lucid/services/db' import logger from '@adonisjs/core/services/logger' +import { SOURCES } from '#lib/release/source' /** Raised when a rule changes: verdicts of an older version are judged again (a full round, right away). */ export const JUDGE_VERSION = 1 @@ -88,10 +89,10 @@ export const RULES: Rule[] = [ // regex on all of releases was 45 s, every round (LOG.md 2026-10-19 (80)). sql: (scoped) => scoped - ? `SELECT r.id FROM releases r ${only(scoped, 'r.id')} - WHERE r.poster ~* '${SCRAPED_POSTERS}' AND r.source = 'subject' AND r.file_total IS NULL` - : `SELECT r.id FROM posters p JOIN releases r ON left(r.poster, 200) = p.poster - WHERE p.poster ~* '${SCRAPED_POSTERS}' AND r.poster ~* '${SCRAPED_POSTERS}' AND r.source = 'subject' AND r.file_total IS NULL`, + ? `SELECT r.id FROM releases r ${only(scoped, 'r.id')} JOIN posters p ON p.id = r.poster_id + WHERE p.poster ~* '${SCRAPED_POSTERS}' AND r.source = ${SOURCES.subject} AND r.file_total IS NULL` + : `SELECT r.id FROM posters p JOIN releases r ON r.poster_id = p.id + WHERE p.poster ~* '${SCRAPED_POSTERS}' AND r.source = ${SOURCES.subject} AND r.file_total IS NULL`, }, { name: 'locked', diff --git a/app/services/search_service.ts b/app/services/search_service.ts index 9deb74d..9ebace1 100644 --- a/app/services/search_service.ts +++ b/app/services/search_service.ts @@ -1,6 +1,7 @@ import db from '@adonisjs/lucid/services/db' import logger from '@adonisjs/core/services/logger' import NzbService from '#services/nzb_service' +import { SOURCES, sourceName } from '#lib/release/source' export type SortOrder = 'newest' | 'oldest' | 'largest' | 'files' | 'name' @@ -91,10 +92,11 @@ class SearchService { // Judged junk (app/services/release_judge.ts) is left out unless asked for. if (!params.judged) where.push('NOT EXISTS (SELECT 1 FROM release_verdicts v WHERE v.release_id = r.id)') const bindings: unknown[] = [] - // Every word in the name or the inner name the prober found, through their trigram indexes; "The.Block.AU" and "the block au" alike. + // Every word in one of the release's texts (its name, the names the prober found inside; release_texts, by its + // trigram index, which yields ids: never an OR across the two tables); "The.Block.AU" and "the block au" alike. for (const word of (params.q ?? '').split(/[\s._-]+/).filter(Boolean).slice(0, 8)) { - where.push('(r.name ILIKE ? OR r.inner_name ILIKE ?)') - bindings.push(contains(word), contains(word)) + where.push('r.id IN (SELECT t.release_id FROM release_texts t WHERE t.text ILIKE ?)') + bindings.push(contains(word)) } if (params.readable) where.push('(r.readable OR r.inner_name IS NOT NULL)') if (params.complete) where.push('r.complete') @@ -102,16 +104,20 @@ class SearchService { if (params.notIncomplete) where.push('(r.complete IS NOT FALSE OR r.settled_at IS NULL)') if (params.unlocked) where.push('r.passworded IS NOT TRUE') if (params.source) { - where.push('r.source = ?') - bindings.push(params.source) + const code = (SOURCES as Record)[params.source] + if (code === undefined) where.push('false') + else { + where.push('r.source = ?') + bindings.push(code) + } } // A poster search: the matching posters first (posters, by trigram), each one's releases then by // releases_poster_newest_idx (below). "poster ILIKE" on releases walked all 50M rows newest first: 30+ min for a // poster with few releases; as a filter on the posters found the planner still walked them (LOG.md 2026-10-16 (70)). - let posters: string[] | null = null + let posters: number[] | null = null if (params.poster) { - const { rows: matched } = await db.rawQuery(`SELECT poster FROM posters WHERE poster ILIKE ? LIMIT ${MAX_POSTERS}`, [contains(params.poster)]) - posters = matched.map((m: { poster: string }) => m.poster) + const { rows: matched } = await db.rawQuery(`SELECT id FROM posters WHERE poster ILIKE ? LIMIT ${MAX_POSTERS}`, [contains(params.poster)]) + posters = matched.map((m: { id: string }) => Number(m.id)) } if (params.categories?.length) { where.push('r.category = ANY(?)') @@ -143,14 +149,14 @@ class SearchService { const whereSql = where.join(' AND ') const countLimit = offset + (params.countLimit ?? COUNT_LIMIT) if (posters !== null && posters.length === 0) return { page, perPage, offset, total: 0, totalCapped: false, releases: [] as ReleaseSummary[] } - // Per poster found (by first 200 characters, as posters and the index keep them; a btree entry holds at most 2,704 - // bytes): its releases matching everything else, the first offset + perPage in the order asked; then merged. + // Per poster found: its releases matching everything else (releases_poster_newest_idx), the first offset + perPage + // in the order asked; then merged. const source = posters === null ? { from: 'releases r', where: whereSql, bindings: [] as unknown[] } : { - from: `unnest(?::text[]) AS p(poster) CROSS JOIN LATERAL ( - SELECT r.* FROM releases r WHERE left(r.poster, 200) = p.poster AND ${whereSql} ORDER BY ${order} LIMIT ${offset + perPage}) r`, + from: `unnest(?::int[]) AS p(id) CROSS JOIN LATERAL ( + SELECT r.* FROM releases r WHERE r.poster_id = p.id AND ${whereSql} ORDER BY ${order} LIMIT ${offset + perPage}) r`, where: 'true', bindings: [posters] as unknown[], } @@ -158,16 +164,17 @@ class SearchService { posters === null ? db.rawQuery(`SELECT count(*) AS total FROM (SELECT 1 FROM releases r WHERE ${whereSql} LIMIT ${countLimit + 1}) x`, bindings) : db.rawQuery( - `SELECT count(*) AS total FROM (SELECT 1 FROM unnest(?::text[]) AS p(poster) CROSS JOIN LATERAL ( - SELECT 1 FROM releases r WHERE left(r.poster, 200) = p.poster AND ${whereSql} LIMIT ${countLimit + 1}) x LIMIT ${countLimit + 1}) y`, + `SELECT count(*) AS total FROM (SELECT 1 FROM unnest(?::int[]) AS p(id) CROSS JOIN LATERAL ( + SELECT 1 FROM releases r WHERE r.poster_id = p.id AND ${whereSql} LIMIT ${countLimit + 1}) x LIMIT ${countLimit + 1}) y`, [posters, ...bindings] ) const [{ rows }, { rows: counted }] = await Promise.all([ db.rawQuery( - `SELECT r.id, r.name, r.source, r.poster, r.groups, r.posted_at, r.file_total, r.files, r.bytes, r.readable, r.complete, r.category, r.settled_at, r.passworded, r.inner_name, + `SELECT r.id, r.name, r.source, po.poster, gs.groups, r.posted_at, r.file_total, r.files, r.bytes, r.readable, r.complete, r.category, r.settled_at, r.passworded, r.inner_name, EXISTS (SELECT 1 FROM release_passwords p WHERE p.release_id = r.id) AS password_known, r.settled_at IS NOT NULL AS final - FROM ${source.from} WHERE ${source.where} ORDER BY ${order} LIMIT ? OFFSET ?`, + FROM ${source.from} LEFT JOIN posters po ON po.id = r.poster_id LEFT JOIN group_sets gs ON gs.id = r.group_set_id + WHERE ${source.where} ORDER BY ${order} LIMIT ? OFFSET ?`, [...source.bindings, ...bindings, perPage, offset] ), counting, @@ -182,9 +189,9 @@ class SearchService { releases: (rows as any[]).map((r): ReleaseSummary => ({ id: Number(r.id), name: r.name, - source: r.source, - poster: r.poster, - groups: r.groups, + source: sourceName(r.source), + poster: r.poster ?? null, + groups: r.groups ?? [], postedAt: r.posted_at, fileTotal: r.file_total, fileCount: Number(r.files), @@ -203,7 +210,12 @@ class SearchService { /** A release with its files, read from its chunks. */ async release(id: number) { - const r = await db.from('releases').where('id', id).first() + const { + rows: [r], + } = await db.rawQuery( + `SELECT r.*, po.poster, gs.groups FROM releases r LEFT JOIN posters po ON po.id = r.poster_id LEFT JOIN group_sets gs ON gs.id = r.group_set_id WHERE r.id = ?`, + [id] + ) if (!r) return null const [files, { rows: inspections }, passwords] = await Promise.all([ NzbService.files(id), @@ -213,9 +225,9 @@ class SearchService { return { id: Number(r.id), name: r.name, - source: r.source, - poster: r.poster, - groups: r.groups, + source: sourceName(r.source), + poster: r.poster ?? null, + groups: r.groups ?? [], postedAt: r.posted_at, fileTotal: r.file_total, fileCount: Number(r.files), @@ -227,7 +239,6 @@ class SearchService { hasPar2: r.has_par2, category: r.category, firstSeenAt: r.first_seen_at, - lastSeenAt: r.last_seen_at, settledAt: r.settled_at, passworded: r.passworded, /** Known passwords (spots, their NZBs), the most trusted first; the first is in the NZB. */ diff --git a/app/services/sidecar_folder.ts b/app/services/sidecar_folder.ts index 965b10e..55b5c87 100644 --- a/app/services/sidecar_folder.ts +++ b/app/services/sidecar_folder.ts @@ -56,7 +56,7 @@ export class SidecarFolder { // A candidate whose release went (merged, deleted) leaves with its batch: a separate scan for those over the whole // queue took 5+ min a round at 13M (LOG.md 2026-10-17 (72)). const { rows: queued } = await db.rawQuery( - `SELECT f.release_id AS queued_id, r.id, r.release_key, r.group_id, r.poster, r.posted_at, r.files, r.file_total, r.readable + `SELECT f.release_id AS queued_id, r.id, r.group_id, r.poster_id, r.posted_at, r.files, r.file_total, r.readable FROM fold_candidates f LEFT JOIN releases r ON r.id = f.release_id ORDER BY f.created_at LIMIT ?`, [BATCH] ) @@ -70,10 +70,10 @@ export class SidecarFolder { round.sidecars++ if (!c.posted_at) continue const { rows: siblings } = await db.rawQuery( - `SELECT r.id, r.release_key, r.group_id, r.files, r.file_total, r.posted_at, r.readable, r.name FROM releases r + `SELECT r.id, r.group_id, r.files, r.file_total, r.posted_at, r.readable, r.name FROM releases r WHERE r.posted_at BETWEEN ?::timestamptz - make_interval(mins => ?) AND ?::timestamptz + make_interval(mins => ?) - AND r.group_id = ? AND r.poster = ? AND r.id <> ? AND r.files > ?`, - [c.posted_at, WINDOW_MINUTES, c.posted_at, WINDOW_MINUTES, c.group_id, c.poster, c.id, c.files] + AND r.group_id = ? AND r.poster_id = ? AND r.id <> ? AND r.files > ?`, + [c.posted_at, WINDOW_MINUTES, c.posted_at, WINDOW_MINUTES, c.group_id, c.poster_id, c.id, c.files] ) const sibling = pickSibling(files, c.file_total ? Number(c.file_total) : null, siblings.map((s: any) => ({ ...s, id: Number(s.id), files: Number(s.files), file_total: s.file_total === null ? null : Number(s.file_total) }))) if (!sibling) continue diff --git a/app/services/spot_inspector.ts b/app/services/spot_inspector.ts index 1da2fa1..e48577c 100644 --- a/app/services/spot_inspector.ts +++ b/app/services/spot_inspector.ts @@ -9,6 +9,9 @@ import { decodeSpotBinary } from '#lib/spotnet/index' import { parseNzb, type ParsedNzb } from '#lib/nzb/index' import { parseSubject } from '#lib/headers/index' import { releaseIdentity } from '#lib/release/index' +import { identityKey, releaseKey } from '#lib/release/keys' +import { SOURCES } from '#lib/release/source' +import CatalogRefs from '#services/catalog_refs' /** Spots claimed per round, newest first. */ const BATCH = 20 @@ -251,8 +254,8 @@ export class SpotInspector { } if (wanted.length === 0) return 0 const { rows } = await db.rawQuery( - `SELECT id, identity_key, group_id, posted_at FROM releases WHERE identity_key = ANY(?) AND group_id = ANY(?)`, - [[...new Set(wanted.map((w) => w.key))], [...new Set(wanted.flatMap((w) => w.groups))]] + `SELECT id, encode(identity_key, 'hex') AS identity_key, group_id, posted_at FROM releases WHERE identity_key = ANY(?::bytea[]) AND group_id = ANY(?)`, + [[...new Set(wanted.map((w) => w.key))].map((k) => identityKey(k)), [...new Set(wanted.flatMap((w) => w.groups))]] ) const byKey = new Map() for (const r of rows) { @@ -275,7 +278,7 @@ export class SpotInspector { await db.transaction(async (trx) => { // A release of the spot's own goes now that the header pass built its files' releases. if (spot.release_id) { - const own = await trx.from('releases').where('id', spot.release_id).where('source', 'spot').select('id').first() + const own = await trx.from('releases').where('id', spot.release_id).where('source', SOURCES.spot).select('id').first() if (own) { await trx.from('release_spots').where('release_id', own.id).delete() await trx.from('release_passwords').where('release_id', own.id).delete() @@ -335,16 +338,19 @@ export class SpotInspector { const groups = [...new Set(files.flatMap((f: any) => f.groups ?? []))] const groupId = groups.length > 0 ? await this.groupId(groups[0]) : null const key = createHash('sha1').update(`spot\n${spotId}`).digest('hex') + // Its poster and group set by id (voyager 3, migration 6000), as the batch store stores its releases'. + const posterId = spot.poster ? await CatalogRefs.posterId(db, spot.poster) : null + const groupSetId = await CatalogRefs.groupSetId(db, groups) const [row] = await db .table('releases') .insert({ - release_key: `spot:${key}`, - identity_key: key, + release_key: releaseKey(key, groupId, null), + identity_key: identityKey(key), group_id: groupId, + group_set_id: groupSetId, + poster_id: posterId, name: spot.title.slice(0, 1000), - source: 'spot', - poster: spot.poster, - groups: JSON.stringify(groups), + source: SOURCES.spot, posted_at: spot.posted_at, last_posted_at: spot.posted_at, files: files.length, @@ -363,8 +369,8 @@ export class SpotInspector { const releaseId = Number(row.id) await db.transaction(async (trx) => { await trx.table('release_spots').insert({ release_id: releaseId, spot_id: spotId, files: files.length }).onConflict(['release_id', 'spot_id']).ignore() - // Its poster for poster search, as the batch store adds its releases' (migration 4000). - if (spot.poster) await trx.rawQuery(`INSERT INTO posters (poster) VALUES (left(?, 200)) ON CONFLICT DO NOTHING`, [spot.poster]) + // Its title, searchable (release_texts). + await trx.rawQuery(`INSERT INTO release_texts (release_id, kind, text) VALUES (?, 1, left(?, 255)) ON CONFLICT DO NOTHING`, [releaseId, spot.title]) if (spot.password) { await trx.table('release_passwords').insert({ release_id: releaseId, password: spot.password, source: spot.password_source ?? 'spot-text', spot_id: spotId }).onConflict(['release_id', 'password']).ignore() } diff --git a/app/services/spotter.ts b/app/services/spotter.ts index 823d67c..c42a9f2 100644 --- a/app/services/spotter.ts +++ b/app/services/spotter.ts @@ -1,5 +1,6 @@ import db from '@adonisjs/lucid/services/db' import logger from '@adonisjs/core/services/logger' +import { SOURCES } from '#lib/release/source' import env from '#start/env' import coordinatorConfig from '#config/coordinator' import { MissingArticle, XoverClient } from '#lib/nntp/index' @@ -183,7 +184,7 @@ export class Spotter { await trx.from('release_spots').where('spot_id', targetId).delete() await trx.from('release_passwords').where('spot_id', targetId).delete() // A release the spot made of its own goes with it. - if (target.release_id) await trx.from('releases').where('id', target.release_id).where('source', 'spot').delete() + if (target.release_id) await trx.from('releases').where('id', target.release_id).where('source', SOURCES.spot).delete() }) logger.info(`Spot "${target.title}" removed by ${personal ? 'its poster' : 'a moderator'}.`) } diff --git a/commands/edge_bench.ts b/commands/edge_bench.ts index cdb07e9..2284d25 100644 --- a/commands/edge_bench.ts +++ b/commands/edge_bench.ts @@ -108,7 +108,7 @@ export default class EdgeBench extends BaseCommand { let text = 0 let peakRss = 0 let peakDetail = '' - const out = { headers: 0, releases: 0, files: 0, segments: 0, payload: 0, candidates: 0, fragments: 0, samples: 0 } + const out = { headers: 0, releases: 0, files: 0, segments: 0, payload: 0, candidates: 0, samples: 0 } const counts = new Map() const keys = new Set() let batches = 0 @@ -164,7 +164,6 @@ export default class EdgeBench extends BaseCommand { out.payload += r.payload.length } out.candidates += batch.candidates.length - out.fragments += batch.fragments.length out.samples += batch.samples.length for (const c of batch.counts) counts.set(`${c.rule} ${c.key}`.trim(), (counts.get(`${c.rule} ${c.key}`.trim()) ?? 0) + c.articles) const m = process.memoryUsage() @@ -182,7 +181,7 @@ export default class EdgeBench extends BaseCommand { this.logger.info(`${out.headers.toLocaleString('en')} headers in ${(total / 1000).toFixed(1)} s (${Math.round(out.headers / (total / 1000)).toLocaleString('en')}/s on one core; CPU ${cpuS.toFixed(1)} s, wall ${((Date.now() - wall) / 1000).toFixed(1)} s)`) this.logger.info(` inflate ${(t.inflate / 1000).toFixed(1)} s, parse ${(t.parse / 1000).toFixed(1)} s, index ${(t.index / 1000).toFixed(1)} s; peak RSS ${(peakRss / 1e6).toFixed(0)} MB (${peakDetail})`) this.logger.info(` ${batches} batches of ${this.batchChunks} chunks: ${out.releases} release chunks for ${keys.size} releases (${(out.releases / Math.max(1, keys.size)).toFixed(2)} chunks per release)`) - this.logger.info(` kept: ${out.releases} release chunks, ${out.files} files, ${out.segments} segments, payload ${(out.payload / 1e6).toFixed(1)} MB (${(out.payload / Math.max(1, out.segments)).toFixed(1)} B/segment); ${out.candidates} candidates, ${out.fragments} fragments, ${out.samples} samples`) + this.logger.info(` kept: ${out.releases} release chunks, ${out.files} files, ${out.segments} segments, payload ${(out.payload / 1e6).toFixed(1)} MB (${(out.payload / Math.max(1, out.segments)).toFixed(1)} B/segment); ${out.candidates} candidates, ${out.samples} samples`) for (const [k, v] of [...counts].sort((a, b) => b[1] - a[1]).slice(0, 8)) this.logger.info(` dropped ${k}: ${v.toLocaleString('en')} articles`) if (this.store) this.logger.info(` stored ${batches} batches in ${(storeMs / 1000).toFixed(1)} s (${(storeMs / Math.max(1, out.releases)).toFixed(2)} ms per release chunk)`) } diff --git a/commands/releases_regroup.ts b/commands/releases_regroup.ts index f66b2dc..72ce628 100644 --- a/commands/releases_regroup.ts +++ b/commands/releases_regroup.ts @@ -3,6 +3,7 @@ import type { CommandOptions } from '@adonisjs/core/types/ace' import db from '@adonisjs/lucid/services/db' import { decodeChunk } from '#lib/chunks/index' import { releaseIdentity } from '#lib/release/index' +import { sourceCode } from '#lib/release/source' /** * What lib/release's identity makes of releases already in the catalog: a @@ -27,7 +28,7 @@ export default class ReleasesRegroup extends BaseCommand { async run() { for (const name of this.group ?? []) { const { rows } = await db.rawQuery( - `SELECT DISTINCT c.release_id FROM release_chunks c JOIN ranges r ON r.id = c.range_id JOIN groups g ON g.id = r.group_id WHERE g.name = ?`, + `SELECT DISTINCT c.release_id FROM release_chunks c JOIN blocks b ON b.id = c.block_id JOIN groups g ON g.id = b.group_id WHERE g.name = ?`, [name] ) const ids = rows.map((r: { release_id: string }) => r.release_id) @@ -54,7 +55,7 @@ export default class ReleasesRegroup extends BaseCommand { } if (this.show?.length) return for (const source of ['subject', 'post', 'stem']) { - const rows = await db.from('releases').where('source', source).where('files', '>', 1).orderByRaw('random()').limit(this.sample).select('id', 'name', 'files') + const rows = await db.from('releases').where('source', sourceCode(source)).where('files', '>', 1).orderByRaw('random()').limit(this.sample).select('id', 'name', 'files') let split = 0 let pieces = 0 const examples: string[] = [] diff --git a/lib/edge/index.ts b/lib/edge/index.ts index 9adc6af..6a38955 100644 --- a/lib/edge/index.ts +++ b/lib/edge/index.ts @@ -2,7 +2,6 @@ export { RangeIndexer, postedSpan, chunkOf, candidatesOf, releaseKeyOf, uploadsO export { encodeFrame, decodeFrames, isKeepalive, KEEPALIVE, type Frame } from './stream.js' export { toWire, fromWire, type WireBatch } from './wire.js' export type { Lease } from './lease.js' -export { mergeFragments } from './merge.js' export type { EdgeSegment, FileMeta } from './scan.js' export { throttle } from './throttle.js' export { filterOverviews, undelta, type ChunkDrops } from './corpus_filter.js' diff --git a/lib/edge/merge.ts b/lib/edge/merge.ts deleted file mode 100644 index 87c64bf..0000000 --- a/lib/edge/merge.ts +++ /dev/null @@ -1,35 +0,0 @@ -import type { AggregatedFile } from '../files/index.js' -import type { Fragment } from './range_indexer.js' -import type { EdgeSegment, FileMeta } from './scan.js' - -/** - * The fragments of one file from neighbouring ranges, put together - * (REBUILD.md, fragments): the complete file when every part from 1 to its - * total is there (the first article seen of a part wins), else null. - */ -export function mergeFragments(fragments: Fragment[]): AggregatedFile | null { - if (fragments.length === 0) return null - const first = fragments[0] - const meta: FileMeta = { - fileKey: first.fileKey, - name: first.name, - poster: first.poster, - subject: first.subject, - totalParts: first.totalParts, - groups: first.groups, - release: first.release ? { ...first.release, source: first.release.source as any } : null, - } - const parts = new Map() - for (const fragment of fragments) { - const postedAt = fragment.postedAt ?? Number.NaN - for (const [number, messageId, bytes, articleNumber] of fragment.segments) { - const known = parts.get(number) - if (!known || articleNumber < known.articleNumber) { - parts.set(number, { fileKey: meta.fileKey, number, totalParts: meta.totalParts, messageId, bytes, articleNumber, postedAt, file: meta }) - } - } - } - if (parts.size < meta.totalParts) return null - for (let n = 1; n <= meta.totalParts; n++) if (!parts.has(n)) return null - return { fileKey: meta.fileKey, segments: [...parts.values()].sort((a, b) => a.number - b.number) } -} diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts index 1b6426b..68f1938 100644 --- a/lib/edge/range_indexer.ts +++ b/lib/edge/range_indexer.ts @@ -26,6 +26,13 @@ export interface ReleaseChunk { /** A file with a readable name; a par2 file. */ readable: boolean hasPar2: boolean + /** + * Voyager 3: the window closed on the release (default), or it is a tail release sent provisionally, finished by a + * later chunk of the same release key (the totals then are the release's; the payload only the files not sent yet). + */ + finished?: boolean + /** Still open at its block's end: the block merger joins it with the neighbour block's part. */ + edge?: boolean payload: Buffer } diff --git a/lib/release/keys.ts b/lib/release/keys.ts new file mode 100644 index 0000000..755bd9a --- /dev/null +++ b/lib/release/keys.ts @@ -0,0 +1,34 @@ +import { createHash } from 'node:crypto' + +/** + * The catalog's keys as voyager 3 stores them (migration 6000): 20-byte SHA-1s in bytea columns. + * - identity_key: the identity's hash (lib/release/identity.ts gives it as 40 hex characters). + * - release_key: one upload of an identity, unique: the identity, the owner group and the first article the window + * saw of it. The same release sent twice (a tail release provisionally, then finished) hits the same row; a repost + * starts at another article and is a release of its own (decision d). Deterministic, so no numbering under a lock. + */ +const HEX40 = /^[0-9a-f]{40}$/i + +function sha1(text: string): Buffer { + return createHash('sha1').update(text).digest() +} + +/** The identity key's bytes: 40 hex characters decoded, any other text (a spot's, a test's) hashed. */ +export function identityKey(key: string): Buffer { + return HEX40.test(key) ? Buffer.from(key, 'hex') : sha1(key) +} + +/** The release key's bytes for an identity in a group, first seen at an article (null for releases without one: a spot's). */ +export function releaseKey( + identity: string, + groupId: number | null, + firstArticle: number | null +): Buffer { + return sha1(`release\n${identity}\n${groupId ?? ''}\n${firstArticle ?? ''}`) +} + +/** A stored key as hex, for logs, the API and release_merges.from_key. */ +export function keyHex(key: Buffer | Uint8Array | string | null | undefined): string | null { + if (key === null || key === undefined) return null + return typeof key === 'string' ? key : Buffer.from(key).toString('hex') +} diff --git a/lib/release/source.ts b/lib/release/source.ts index f8d11fb..f82e409 100644 --- a/lib/release/source.ts +++ b/lib/release/source.ts @@ -15,6 +15,8 @@ export const SOURCES = { nzb: 5, /** One poster's files that differ only by a number sequence, or share a prefix (voyager 3). */ album: 6, + /** An obfuscated post's stem, read from the yEnc body (identity.ts: weak). */ + 'weak-stem': 7, } as const export type Source = keyof typeof SOURCES diff --git a/tests/bootstrap.ts b/tests/bootstrap.ts index bee79df..571323b 100644 --- a/tests/bootstrap.ts +++ b/tests/bootstrap.ts @@ -8,6 +8,7 @@ import testUtils from '@adonisjs/core/services/test_utils' import { authApiClient } from '@adonisjs/auth/plugins/api_client' import { sessionApiClient } from '@adonisjs/session/plugins/api_client' import type { Registry } from '../.adonisjs/client/registry/schema.d.ts' +import CatalogRefs from '#services/catalog_refs' /** * This file is imported by the "bin/test.ts" entrypoint file @@ -51,6 +52,8 @@ export const runnerHooks: Required> = { */ export const configureSuite: Config['configureSuite'] = (suite) => { if (['browser', 'functional', 'e2e'].includes(suite.name)) { + // Each test's transaction is rolled back: the poster and group-set ids cached meanwhile are gone with it. + suite.onGroup((group) => group.each.setup(() => CatalogRefs.forget())) return suite.setup(() => testUtils.httpServer().start()) } } diff --git a/tests/functional/batch_store.spec.ts b/tests/functional/batch_store.spec.ts index 10fe71c..1ead8ed 100644 --- a/tests/functional/batch_store.spec.ts +++ b/tests/functional/batch_store.spec.ts @@ -2,19 +2,35 @@ import { test } from '@japa/runner' import db from '@adonisjs/lucid/services/db' import testUtils from '@adonisjs/core/services/test_utils' import BatchStore, { StaleBatch } from '#services/batch_store' -import { RangeIndexer } from '#lib/edge/index' -import { decodeChunk } from '#lib/chunks/index' +import { RangeIndexer, type EdgeBatch, type ReleaseChunk } from '#lib/edge/index' +import { decodeChunk, encodeChunk } from '#lib/chunks/index' import { DEFAULT_RULES } from '#lib/rules/index' import type { Overview } from '#lib/nntp/index' const GROUP = 'alt.binaries.test' let id = 0 -function post(at: number, name: string, parts: number, counter: string, tool: 'ngPost' | 'plain' = 'plain'): Overview[] { +function post( + at: number, + name: string, + parts: number, + counter: string, + tool: 'ngPost' | 'plain' = 'plain' +): Overview[] { return Array.from({ length: parts }, (_, i) => { id++ const mid = tool === 'ngPost' ? `${id.toString(16).padStart(32, '0')}@ngPost` : `art${id}@test` - return [at + i, { 'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`, 'from': 'p ', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', 'message-id': `<${mid}>`, ':bytes': '700000', 'xref': `news.example ${GROUP}:${at + i}` }] as Overview + return [ + at + i, + { + 'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`, + 'from': 'p ', + 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', + 'message-id': `<${mid}>`, + ':bytes': '700000', + 'xref': `news.example ${GROUP}:${at + i}`, + }, + ] as Overview }) } @@ -23,73 +39,218 @@ async function leasedRange() { const [node] = await db.table('nodes').insert({ name: 'test-node' }).returning('id') const [range] = await db .table('ranges') - .insert({ group_id: group.id, start_id: 1, end_id: 1_000_000, next_id: 1, status: 'leased', node_id: node.id, rule_version: 1 }) + .insert({ + group_id: group.id, + start_id: 1, + end_id: 1_000_000, + next_id: 1, + status: 'leased', + node_id: node.id, + rule_version: 1, + }) .returning('id') return { rangeId: Number(range.id), nodeId: Number(node.id) } } function indexer() { - return new RangeIndexer({ group: GROUP, start: 1, end: 1_000_000, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } }, edgeArticles: 1000 }) + return new RangeIndexer({ + group: GROUP, + start: 1, + end: 1_000_000, + rules: { + ...DEFAULT_RULES, + samplePercent: 0, + droppedSignatures: [], + priorities: { [GROUP]: 1 }, + }, + edgeArticles: 1000, + }) } /** One file's articles, posted `minutes` apart from `start` (an ISO time), in `group`. */ -function upload(at: number, name: string, parts: number, counter: string, start: string, group = GROUP, minutes = 1): Overview[] { +function upload( + at: number, + name: string, + parts: number, + counter: string, + start: string, + group = GROUP, + minutes = 1 +): Overview[] { return Array.from({ length: parts }, (_, i) => { id++ - const date = new Date(Date.parse(start) + i * minutes * 60_000).toUTCString().replace('GMT', '+0000') - return [at + i, { 'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`, 'from': 'p ', 'date': date, 'message-id': ``, ':bytes': '700000', 'xref': `news.example ${group}:${at + i}` }] as Overview + const date = new Date(Date.parse(start) + i * minutes * 60_000) + .toUTCString() + .replace('GMT', '+0000') + return [ + at + i, + { + 'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`, + 'from': 'p ', + 'date': date, + 'message-id': ``, + ':bytes': '700000', + 'xref': `news.example ${group}:${at + i}`, + }, + ] as Overview }) } async function rangeIn(groupName: string, start: number, nodeId?: number) { - const g = (await db.from('groups').where('name', groupName).first()) ?? (await db.table('groups').insert({ name: groupName, priority: groupName === GROUP ? 1 : 2 }).returning('*'))[0] - const node = nodeId ?? Number((await db.table('nodes').insert({ name: `n${groupName}${start}` }).returning('id'))[0].id) - const [range] = await db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: start + 99_999, next_id: start, status: 'leased', node_id: node, rule_version: 1 }).returning('id') + const g = + (await db.from('groups').where('name', groupName).first()) ?? + ( + await db + .table('groups') + .insert({ name: groupName, priority: groupName === GROUP ? 1 : 2 }) + .returning('*') + )[0] + const node = + nodeId ?? + Number( + ( + await db + .table('nodes') + .insert({ name: `n${groupName}${start}` }) + .returning('id') + )[0].id + ) + const [range] = await db + .table('ranges') + .insert({ + group_id: g.id, + start_id: start, + end_id: start + 99_999, + next_id: start, + status: 'leased', + node_id: node, + rule_version: 1, + }) + .returning('id') return { rangeId: Number(range.id), nodeId: node } } +function batchOf(groupName: string, start: number, overviews: Overview[]): EdgeBatch { + const ix = new RangeIndexer({ + group: groupName, + start, + end: start + 99_999, + rules: { + ...DEFAULT_RULES, + samplePercent: 0, + droppedSignatures: [], + priorities: { [GROUP]: 1, 'alt.binaries.other': 2 }, + }, + edgeArticles: 1, + }) + ix.add(overviews, start + 100_000) + ix.finish() + return ix.take() +} + async function storeIn(groupName: string, start: number, overviews: Overview[]) { const { rangeId, nodeId } = await rangeIn(groupName, start) - const ix = new RangeIndexer({ group: groupName, start, end: start + 99_999, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1, 'alt.binaries.other': 2 } }, edgeArticles: 1 }) - ix.add(overviews, start + 100_000) - ix.finish() - await BatchStore.store(rangeId, nodeId, ix.take()) + return BatchStore.store(rangeId, nodeId, batchOf(groupName, start, overviews)) } -const releasesOf = () => db.from('releases').orderBy('id').select('release_key', 'identity_key', 'files', 'posted_at', 'last_posted_at') +const releasesOf = () => + db + .from('releases') + .orderBy('id') + .select( + db.raw( + `encode(release_key, 'hex') AS release_key, encode(identity_key, 'hex') AS identity_key, files, chunks, posted_at, last_posted_at, settled_at` + ) + ) test.group('BatchStore', (group) => { group.each.setup(() => testUtils.db().withGlobalTransaction()) - test('stores releases, their chunks, probe candidates, rule counts and fragments, and moves the resume point', async ({ assert }) => { + test('stores releases by id references, their texts, chunks, probe candidates and rule counts, and moves the resume point', async ({ + assert, + }) => { const { rangeId, nodeId } = await leasedRange() const ix = indexer() - ix.add([ - ...post(100_000, 'Some.Show.S01E01.1080p.WEB.part1.rar', 2, 'Some.Show.S01E01 [1/2]'), - ...post(100_010, 'Some.Show.S01E01.1080p.WEB.par2', 1, 'Some.Show.S01E01 [2/2]'), - ...post(100_020, 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 3, '[1/1]', 'ngPost'), - ...post(10, 'Edge.Cut.2020.part1.rar', 3, 'Edge.Cut [1/1]').slice(1), - ], 500_000) + ix.add( + [ + ...post(100_000, 'Some.Show.S01E01.1080p.WEB.part1.rar', 2, 'Some.Show.S01E01 [1/2]'), + ...post(100_010, 'Some.Show.S01E01.1080p.WEB.par2', 1, 'Some.Show.S01E01 [2/2]'), + ...post(100_020, 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 3, '[1/1]', 'ngPost'), + ], + 500_000 + ) const first = ix.take() const stored = await BatchStore.store(rangeId, nodeId, first) - assert.include(stored, { releases: 1, newReleases: 1, candidates: 2, fragments: 0 }) + assert.deepEqual(stored, { releases: 1, newReleases: 1, finalized: 0, candidates: 2 }) const release = await db.from('releases').first() - assert.include(release, { files: 2, segments: '3', chunks: 1, readable: true, has_par2: true, file_total: 2 }) + assert.include(release, { + files: 2, + segments: '3', + chunks: 1, + readable: true, + has_par2: true, + file_total: 2, + complete: true, + edge: false, + source: 1, + }) + assert.isNotNull(release.settled_at) + // The poster and the groups by id, the keys 20 bytes, the name searchable. + assert.equal( + (await db.from('posters').where('id', release.poster_id).firstOrFail()).poster, + 'p ' + ) + assert.deepEqual( + (await db.from('group_sets').where('id', release.group_set_id).firstOrFail()).groups, + [GROUP] + ) + assert.lengthOf(release.release_key, 20) + assert.lengthOf(release.identity_key, 20) + assert.deepEqual(await db.from('release_texts').select('release_id', 'kind', 'text'), [ + { release_id: release.id, kind: 1, text: 'Some Show S01E01' }, + ]) const chunk = await db.from('release_chunks').where('release_id', release.id).first() - assert.deepEqual(decodeChunk(chunk.payload).map((f) => f.name).sort(), ['Some.Show.S01E01.1080p.WEB.par2', 'Some.Show.S01E01.1080p.WEB.part1.rar']) - assert.deepEqual((await db.from('probe_candidates').orderBy('kind')).map((c) => c.kind), ['archive', 'par2']) - assert.deepEqual((await db.from('rule_counts')).map((c) => [c.rule, c.key, Number(c.articles)]), [['garbage', 'ngPost alt.binaries.test', 3]]) - assert.equal(Number((await db.from('ranges').where('id', rangeId).first()).next_id), 500_000) + assert.deepEqual( + decodeChunk(chunk.payload) + .map((f) => f.name) + .sort(), + ['Some.Show.S01E01.1080p.WEB.par2', 'Some.Show.S01E01.1080p.WEB.part1.rar'] + ) + // Claimable at once: the release is finished. + assert.deepEqual( + (await db.from('probe_candidates').orderBy('kind')).map((c) => [ + c.kind, + c.ready, + c.unreadable, + ]), + [ + ['archive', true, false], + ['par2', true, false], + ] + ) + // Keyed by the range's block (a range without one is a block of its own, until blocks are leased). + const range = await db.from('ranges').where('id', rangeId).first() + assert.isNotNull(range.block_id) + assert.equal(Number(chunk.block_id), Number(range.block_id)) + assert.deepEqual( + (await db.from('rule_counts')).map((c) => [ + Number(c.block_id), + c.rule, + c.key, + Number(c.articles), + ]), + [[Number(range.block_id), 'garbage', 'ngPost alt.binaries.test', 3]] + ) + assert.equal(Number(range.next_id), 500_000) // The same batch again (a retry after a lost reply) is refused, and changes nothing. await assert.rejects(() => BatchStore.store(rangeId, nodeId, first), StaleBatch) assert.equal((await db.from('releases').first()).files, 2) - // The end of the range: the cut-off file near the start edge is kept as a fragment. + // The end of the range. ix.finish() - const last = ix.take() - assert.include(await BatchStore.store(rangeId, nodeId, last), { fragments: 1 }) + await BatchStore.store(rangeId, nodeId, ix.take()) assert.equal(Number((await db.from('ranges').where('id', rangeId).first()).next_id), 1_000_001) }) @@ -97,52 +258,192 @@ test.group('BatchStore', (group) => { const { rangeId, nodeId } = await leasedRange() const ix = indexer() ix.add(post(100_000, 'A.Release.2020.mkv', 1, 'A [1/1]'), 200_000) - await assert.rejects(() => BatchStore.store(rangeId, nodeId + 1, ix.take()), /not leased to this node/) + await assert.rejects( + () => BatchStore.store(rangeId, nodeId + 1, ix.take()), + /not leased to this node/ + ) }) - test('a re-upload is a release of its own: in one range, or a later range; a continuation joins', async ({ assert }) => { - // One range: the same file posted Monday and again Thursday. + + test('every chunk is a release of its own: a re-upload, and a part read by another range (the block merger joins those)', async ({ + assert, + }) => { + // One range: the same file posted Monday and again Thursday are two uploads. await storeIn(GROUP, 1, [ - ...upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/2]', '2026-10-05T10:00:00Z'), - ...upload(20, 'Some.Show.S01E01.1080p.par2', 1, 'Some.Show.S01E01.1080p [2/2]', '2026-10-05T10:05:00Z'), - ...upload(500, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/2]', '2026-10-08T10:00:00Z'), + ...upload( + 10, + 'Some.Show.S01E01.1080p.mkv', + 2, + 'Some.Show.S01E01.1080p [1/2]', + '2026-10-05T10:00:00Z' + ), + ...upload( + 20, + 'Some.Show.S01E01.1080p.par2', + 1, + 'Some.Show.S01E01.1080p [2/2]', + '2026-10-05T10:05:00Z' + ), + ...upload( + 500, + 'Some.Show.S01E01.1080p.mkv', + 2, + 'Some.Show.S01E01.1080p [1/2]', + '2026-10-08T10:00:00Z' + ), ]) let rows = await releasesOf() - assert.lengthOf(rows, 2) - assert.equal(rows[1].release_key, `${rows[0].release_key}:2`) - assert.deepEqual(rows.map((r) => [r.identity_key, r.files]), [[rows[0].release_key, 2], [rows[0].release_key, 1]]) + assert.deepEqual( + rows.map((r) => r.files), + [2, 1] + ) + assert.equal(rows[0].identity_key, rows[1].identity_key) + assert.notEqual(rows[0].release_key, rows[1].release_key) - // The next range: Thursday's par2 an hour later joins Thursday's release; a Saturday copy is a third. - await storeIn(GROUP, 100_001, upload(100_010, 'Some.Show.S01E01.1080p.par2', 1, 'Some.Show.S01E01.1080p [2/2]', '2026-10-08T11:00:00Z')) - await storeIn(GROUP, 200_001, upload(200_010, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/2]', '2026-10-10T10:00:00Z')) + // The next range: Thursday's par2 an hour later is written as a release of its own (its key: the identity, the + // group and its first article); the window keeps such parts together, and the block merger joins the edges. + await storeIn( + GROUP, + 100_001, + upload( + 100_010, + 'Some.Show.S01E01.1080p.par2', + 1, + 'Some.Show.S01E01.1080p [2/2]', + '2026-10-08T11:00:00Z' + ) + ) rows = await releasesOf() - assert.deepEqual(rows.map((r) => [r.release_key.split(':')[1] ?? '1', r.files]), [['1', 2], ['2', 2], ['3', 1]]) - const chunks = await db.from('release_chunks').count('* as n').first() - assert.equal(Number(chunks!.n), 4) + assert.deepEqual( + rows.map((r) => r.files), + [2, 1, 1] + ) + assert.lengthOf(new Set(rows.map((r) => r.release_key)), 3) + assert.lengthOf(new Set(rows.map((r) => r.identity_key)), 1) + assert.equal(Number((await db.from('release_chunks').count('* as n').first())!.n), 3) }) - test('candidates added to a settled release (a later range of its upload) are claimable at once', async ({ assert }) => { - await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E02.1080p.mkv', 2, 'Some.Show.S01E02.1080p [1/2]', '2026-10-05T10:00:00Z')) - // Settled before the next range came (the settler marks only what it settles). - await db.from('releases').update({ settled_at: new Date() }) - await db.from('probe_candidates').update({ ready: true }) - await storeIn(GROUP, 100_001, upload(100_010, 'Some.Show.S01E02.1080p.par2', 1, 'Some.Show.S01E02.1080p [2/2]', '2026-10-05T10:20:00Z')) - const candidates = await db.from('probe_candidates').orderBy('id').select('kind', 'ready') - assert.deepEqual(candidates.map((c) => c.kind), ['par2']) - assert.isTrue(candidates.every((c) => c.ready)) - assert.lengthOf(await releasesOf(), 1) + test('a provisional tail release is finished once by a later chunk of its key, which adds the files not sent before', async ({ + assert, + }) => { + const { rangeId, nodeId } = await rangeIn(GROUP, 1) + const batch = batchOf( + GROUP, + 1, + upload( + 10, + 'Some.Show.S01E02.1080p.mkv', + 2, + 'Some.Show.S01E02.1080p [1/2]', + '2026-10-05T10:00:00Z' + ) + ) + assert.lengthOf(batch.releases, 1) + const provisional: ReleaseChunk = { ...batch.releases[0], finished: false } + const stored = await BatchStore.store(rangeId, nodeId, { + ...batch, + next: 50_000, + releases: [provisional], + }) + assert.include(stored, { newReleases: 1, finalized: 0 }) + let [row] = await releasesOf() + assert.isNull(row.settled_at) + assert.include(row, { files: 1, chunks: 1 }) + + // The finish: the par2 that came within the window, under the same key (the identity, the group, the first + // article), with its probe candidate: claimable at once, the release being finished. + const second = batchOf( + GROUP, + 1, + upload( + 30, + 'Some.Show.S01E02.1080p.par2', + 1, + 'Some.Show.S01E02.1080p [2/2]', + '2026-10-05T10:20:00Z' + ) + ) + const [par2] = second.releases + const finish: ReleaseChunk = { + ...par2, + key: provisional.key, + upload: provisional.upload, + firstArticle: provisional.firstArticle, + finished: true, + } + const candidates = second.candidates.map((c) => ({ + ...c, + releaseKey: provisional.key, + upload: provisional.upload, + })) + assert.deepEqual( + candidates.map((c) => c.kind), + ['par2'] + ) + const finished = await BatchStore.store(rangeId, nodeId, { + ...batch, + from: 50_000, + next: 60_000, + releases: [finish], + candidates, + counts: [], + samples: [], + fragments: [], + }) + assert.include(finished, { releases: 1, newReleases: 0, finalized: 1, candidates: 1 }) + ;[row] = await releasesOf() + assert.isNotNull(row.settled_at) + assert.include(row, { files: 2, chunks: 2 }) + const release = await db.from('releases').firstOrFail() + assert.include(release, { complete: true, has_par2: true, last_article: '30' }) + assert.deepEqual(await db.from('probe_candidates').select('kind', 'ready'), [ + { kind: 'par2', ready: true }, + ]) + assert.deepEqual( + (await db.from('release_chunks').orderBy('id')).flatMap((c) => + decodeChunk(c.payload).map((f) => f.name) + ), + ['Some.Show.S01E02.1080p.mkv', 'Some.Show.S01E02.1080p.par2'] + ) + + // Sent once more, finished already: left as it is. + const again = await BatchStore.store(rangeId, nodeId, { + ...batch, + from: 60_000, + next: 70_000, + releases: [{ ...finish, payload: encodeChunk([]), files: 0, segments: 0, bytes: 0 }], + candidates: [], + counts: [], + samples: [], + fragments: [], + }) + assert.include(again, { releases: 1, newReleases: 0, finalized: 0 }) + assert.include((await releasesOf())[0], { files: 2, chunks: 2 }) + assert.equal(Number((await db.from('release_chunks').count('* as n').first())!.n), 2) }) - test('a split 7z read in pieces keeps the tail of its highest volume only; RAR gets none', async ({ assert }) => { - // As the live tail reads it while posted: each piece queues a tail for the last volume it saw so far. + test('a split 7z keeps the tail of its highest volume only; RAR gets none', async ({ + assert, + }) => { await storeIn(GROUP, 1, [ ...upload(10, 'TsIgNBewuFbuwWy.7z.001', 2, '[1/4]', '2026-10-23T08:24:00Z'), ...upload(20, 'TsIgNBewuFbuwWy.7z.002', 2, '[2/4]', '2026-10-23T08:26:00Z'), + ...upload(30, 'TsIgNBewuFbuwWy.7z.003', 2, '[3/4]', '2026-10-23T08:28:00Z'), + ...upload(40, 'TsIgNBewuFbuwWy.7z.004', 2, '[4/4]', '2026-10-23T08:30:00Z'), + ...upload( + 50, + 'Some.Show.S01E03.part01.rar', + 2, + 'Some.Show.S01E03 [1/2]', + '2026-10-23T09:00:00Z' + ), + ...upload( + 60, + 'Some.Show.S01E03.part02.rar', + 2, + 'Some.Show.S01E03 [2/2]', + '2026-10-23T09:02:00Z' + ), ]) - await storeIn(GROUP, 100_001, upload(100_010, 'TsIgNBewuFbuwWy.7z.004', 2, '[4/4]', '2026-10-23T08:30:00Z')) - // A lower volume posted late: its tail goes at once. - await storeIn(GROUP, 200_001, upload(200_010, 'TsIgNBewuFbuwWy.7z.003', 2, '[3/4]', '2026-10-23T08:28:00Z')) - await storeIn(GROUP, 300_001, upload(300_010, 'Some.Show.S01E03.part01.rar', 2, 'Some.Show.S01E03 [1/2]', '2026-10-23T09:00:00Z')) - await storeIn(GROUP, 400_001, upload(400_010, 'Some.Show.S01E03.part02.rar', 2, 'Some.Show.S01E03 [2/2]', '2026-10-23T09:02:00Z')) const candidates = await db.from('probe_candidates').orderBy('file').select('kind', 'file') assert.deepEqual( candidates.map((c) => [c.kind, c.file]), @@ -154,43 +455,36 @@ test.group('BatchStore', (group) => { ) }) - test('the same upload in another group (multi-posted) is a release of its own', async ({ assert }) => { - await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:00:00Z')) - await storeIn('alt.binaries.other', 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:01:00Z', 'alt.binaries.other')) + test('the same upload in another group (multi-posted) is a release of its own', async ({ + assert, + }) => { + await storeIn( + GROUP, + 1, + upload( + 10, + 'Some.Show.S01E01.1080p.mkv', + 2, + 'Some.Show.S01E01.1080p [1/1]', + '2026-10-05T10:00:00Z' + ) + ) + await storeIn( + 'alt.binaries.other', + 1, + upload( + 10, + 'Some.Show.S01E01.1080p.mkv', + 2, + 'Some.Show.S01E01.1080p [1/1]', + '2026-10-05T10:01:00Z', + 'alt.binaries.other' + ) + ) const rows = await releasesOf() assert.lengthOf(rows, 2) assert.equal(rows[0].identity_key, rows[1].identity_key) - }) - - test('an upload spread over groups (other files in another group) is one release', async ({ assert }) => { - await storeIn(GROUP, 1, upload(10, 'Some.Album-WEB-2026-FTD.part1.rar', 2, 'Some.Album-WEB-2026-FTD [2/3]', '2026-10-05T10:00:00Z')) - await storeIn('alt.binaries.other', 1, upload(10, 'Some.Album-WEB-2026-FTD.part2.rar', 2, 'Some.Album-WEB-2026-FTD [3/3]', '2026-10-05T10:00:30Z', 'alt.binaries.other')) - const rows = await releasesOf() - assert.lengthOf(rows, 1) - assert.equal(rows[0].files, 2) - }) - - test('a release key taken meanwhile by another group\'s writer is skipped for the next number', async ({ assert }) => { - await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:00:00Z')) - const [first] = await releasesOf() - // As if another group's batch numbered the key at the same moment: key:2 exists, not counted for this key. - const [g] = await db.table('groups').insert({ name: 'alt.binaries.third', priority: 3 }).returning('id') - await db.table('releases').insert({ release_key: `${first.identity_key}:2`, identity_key: 'someone-else', group_id: g.id, name: 'x', source: 'subject', poster: 'p', groups: JSON.stringify([]), files: 0, segments: 0, bytes: 0, chunks: 0 }) - await storeIn('alt.binaries.other', 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:01:00Z', 'alt.binaries.other')) - const keys = (await releasesOf()).filter((r) => r.identity_key === first.identity_key).map((r) => r.release_key) - assert.deepEqual(keys, [first.identity_key, `${first.identity_key}:3`]) - }) - - test('a re-upload is numbered after the highest number, not into the holes merges left', async ({ assert }) => { - for (let day = 0; day < 4; day++) { - await storeIn(GROUP, 1 + day * 100_000, upload(10 + day * 100_000, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', `2026-10-0${day * 2 + 1}T10:00:00Z`)) - } - const [first] = await releasesOf() - // Merged away: key:2 and key:3 gone, key and key:4 left. - await db.from('releases').whereIn('release_key', [`${first.identity_key}:2`, `${first.identity_key}:3`]).delete() - await storeIn(GROUP, 400_001, upload(400_010, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-09T10:00:00Z')) - const keys = (await releasesOf()).map((r) => r.release_key.split(':')[1] ?? '1') - assert.deepEqual(keys, ['1', '4', '5']) + assert.notEqual(rows[0].release_key, rows[1].release_key) }) test('a small release of sidecars only is marked for the sidecar fold', async ({ assert }) => { @@ -199,7 +493,12 @@ test.group('BatchStore', (group) => { ...upload(1_000_010, 'Some.Tool.v1.zip', 1, '[01/01]', '2026-10-05T10:00:00Z'), ...upload(1_000_020, 'Some.Show.S01E01.part01.rar', 2, '[02/30]', '2026-10-05T10:00:00Z'), ]) - const flags = new Map((await db.from('releases').select('name', 'sidecar_only')).map((r) => [r.name, r.sidecar_only])) + const flags = new Map( + (await db.from('releases').select('name', 'sidecar_only')).map((r) => [ + r.name, + r.sidecar_only, + ]) + ) assert.isTrue(flags.get('Murdoch Mysteries S8D1')) assert.isFalse(flags.get('Some.Tool.v1')) // Over 1 MB: not looked at. diff --git a/tests/functional/helpers/releases.ts b/tests/functional/helpers/releases.ts new file mode 100644 index 0000000..09ab211 --- /dev/null +++ b/tests/functional/helpers/releases.ts @@ -0,0 +1,57 @@ +import db from '@adonisjs/lucid/services/db' +import CatalogRefs from '#services/catalog_refs' +import { identityKey, releaseKey } from '#lib/release/keys' +import { sourceCode } from '#lib/release/source' + +/** A release as the tests describe one: text keys, a poster and groups by name, the source by name. */ +export interface ReleaseFixture { + /** Any text; the stored key is its hash (keyOf), or 40 hex characters decoded. Default: from the identity, group and first article. */ + release_key?: string + /** Any text (default: release_key, else the name). */ + identity_key?: string + name: string + source?: string + poster?: string | null + groups?: string[] | string + group_id?: number | null + first_article?: number | null + [column: string]: unknown +} + +/** The bytes a text key is stored as (identityKey: 40 hex characters decoded, anything else hashed). */ +export const keyOf = identityKey + +/** Inserts releases with their poster, group set, keys and source as the catalog stores them; their ids in order. */ +export async function insertReleases(fixtures: ReleaseFixture[]): Promise { + const ids: number[] = [] + for (const f of fixtures) { + const { release_key: textKey, identity_key: textIdentity, source, poster, groups, ...rest } = f + const identity = textIdentity ?? textKey ?? f.name + const groupList: string[] = typeof groups === 'string' ? JSON.parse(groups) : (groups ?? []) + const [row] = await db + .table('releases') + .insert({ + ...rest, + release_key: textKey + ? identityKey(textKey) + : releaseKey(identity, f.group_id ?? null, f.first_article ?? null), + identity_key: identityKey(identity), + source: sourceCode(source ?? 'stem'), + poster_id: poster ? await CatalogRefs.posterId(db, poster) : null, + group_set_id: await CatalogRefs.groupSetId(db, groupList), + }) + .returning('id') + const id = Number(row.id) + await db.rawQuery( + `INSERT INTO release_texts (release_id, kind, text) VALUES (?, 1, left(?, 255)) ON CONFLICT DO NOTHING`, + [id, f.name] + ) + ids.push(id) + } + return ids +} + +export async function insertRelease(fixture: ReleaseFixture): Promise { + const [id] = await insertReleases([fixture]) + return id +} diff --git a/tests/functional/node_api.spec.ts b/tests/functional/node_api.spec.ts index e811892..f5dd984 100644 --- a/tests/functional/node_api.spec.ts +++ b/tests/functional/node_api.spec.ts @@ -77,7 +77,9 @@ test.group('Node API', (group) => { // The same batch again: refused. ;(await client.post(`/api/v1/node/ranges/${rangeId}/batches`).bearerToken(token).json(batch)).assertStatus(409) - assert.deepEqual((await db.from('rule_counts').where('range_id', rangeId)).map((c) => [c.rule, c.key, Number(c.articles)]), [['obfuscated', '', 3]]) + // Counts by the range's block (a range without one is a block of its own, until blocks are leased). + const { block_id: blockId } = await db.from('ranges').where('id', rangeId).firstOrFail() + assert.deepEqual((await db.from('rule_counts').where('block_id', blockId)).map((c) => [c.rule, c.key, Number(c.articles)]), [['obfuscated', '', 3]]) ;(await client.post(`/api/v1/node/ranges/${rangeId}/finish`).bearerToken(token)).assertStatus(200) const range = await db.from('ranges').where('id', rangeId).first() diff --git a/tests/functional/par2_completeness.spec.ts b/tests/functional/par2_completeness.spec.ts index 96957aa..f3f5fc8 100644 --- a/tests/functional/par2_completeness.spec.ts +++ b/tests/functional/par2_completeness.spec.ts @@ -45,23 +45,23 @@ test.group('par2 completeness', (group) => { assert.isFalse(par2Complete(set, [{ name: 'Qx81.bin', size: 500 }, { name: 'Zr02.bin', size: 12 }], [])) }) - test('decided once settled with a par2 set, and kept when more files come', async ({ assert }) => { + test('decided once finished, by the par2 set the release holds', async ({ assert }) => { await store(1, [...file('Some.Upload.mkv', 3, 10), ...file('Some.Upload.par2', 1, 20)]) const release = await db.from('releases').firstOrFail() assert.isNull(release.file_total) await db.table('par2_files').multiInsert([setFile('Some.Upload.mkv', 2_000_000), setFile('Some.Upload.nfo', 100)].map((f, i) => ({ set_id: 's'.repeat(32), file_id: String(i).repeat(32), ...f }))) await db.table('par2_sets').insert({ set_id: 's'.repeat(32), release_id: release.id, files: 2 }) - // Not settled yet: left alone. + // A provisional tail release (not finished yet): left alone. + await db.from('releases').where('id', release.id).update({ settled_at: null }) assert.equal(await decidePar2Completeness(db, [release.id]), 0) await db.from('releases').where('id', release.id).update({ settled_at: new Date() }) assert.equal(await decidePar2Completeness(db, [release.id]), 1) assert.isFalse((await db.from('releases').where('id', release.id).firstOrFail()).complete) - // The nfo comes in a later range: complete, and the batch store's own update keeps it. - await store(1001, file('Some.Upload.nfo', 1, 1010)) - await decidePar2Completeness(db, [release.id]) - await store(2001, file('Some.Upload.sfv', 1, 2010)) - const after = await db.from('releases').where('id', release.id).firstOrFail() - assert.equal(after.files, 4) - assert.isTrue(after.complete) + // Another upload with the nfo: complete. + await store(1001, [...file('Some.Upload.mkv', 3, 1010), ...file('Some.Upload.par2', 1, 1020), ...file('Some.Upload.nfo', 1, 1030)]) + const whole = await db.from('releases').whereNot('id', release.id).firstOrFail() + await db.table('par2_sets').insert({ set_id: 's'.repeat(32), release_id: whole.id, files: 2 }) + assert.equal(await decidePar2Completeness(db, [whole.id]), 1) + assert.isTrue((await db.from('releases').where('id', whole.id).firstOrFail()).complete) }) }) diff --git a/tests/functional/par2_merger.spec.ts b/tests/functional/par2_merger.spec.ts index ccce686..9a134d9 100644 --- a/tests/functional/par2_merger.spec.ts +++ b/tests/functional/par2_merger.spec.ts @@ -4,6 +4,7 @@ import testUtils from '@adonisjs/core/services/test_utils' import { Par2Merger } from '#services/par2_merger' import { encodeChunk } from '#lib/chunks/index' import ChunkPartitions from '#services/chunk_partitions' +import { insertRelease } from '#tests/functional/helpers/releases' const hash = (c: string) => c.repeat(32) @@ -13,14 +14,11 @@ test.group('Par2Merger', (group) => { test('merges the one-file releases of an obfuscated upload by its par2 set, named by the par2', async ({ assert }) => { const at = new Date('2022-07-08T03:46:00Z') const release = async (key: string, posted: Date, groupId: number, files = 1) => { - const [row] = await db - .table('releases') - .insert({ + const row = { id: await insertRelease({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: groupId, files, segments: files * 10, bytes: files * 1000, first_article: 100, last_article: 200, posted_at: posted, readable: false, file_total: 20, settled_at: posted, - }) - .returning('id') + }) } return Number(row.id) } // The par2 index, two data files of the upload, the same file reposted a week later, and one in another group. @@ -95,10 +93,7 @@ test.group('Par2Merger', (group) => { const at = new Date('2022-07-08T03:46:00Z') const ids: number[] = [] for (const key of ['dd000001', 'dd000002']) { - const [row] = await db - .table('releases') - .insert({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: 1, files: 1, posted_at: at, readable: false }) - .returning('id') + const row = { id: await insertRelease({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: 1, files: 1, posted_at: at, readable: false }) } ids.push(Number(row.id)) } // The par2 is held by the first; the second holds the file, and so do 20 releases elsewhere and years apart. @@ -120,10 +115,7 @@ test.group('Par2Merger', (group) => { const at = new Date('2022-07-08T03:46:00Z') const ids: number[] = [] for (const [key, files] of [['bb000001', 1], ['bb000002', 1], ['bb000003', 3]] as const) { - const [row] = await db - .table('releases') - .insert({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: 1, files, posted_at: at, readable: false }) - .returning('id') + const row = { id: await insertRelease({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: 1, files, posted_at: at, readable: false }) } ids.push(Number(row.id)) } const setId = hash('7') @@ -145,10 +137,7 @@ test.group('Par2Merger', (group) => { const ids: number[] = [] for (let i = 1; i <= 6; i++) { const key = `cc00000${i}` - const [row] = await db - .table('releases') - .insert({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: `kantodo${i}`, source: 'stem', groups: '[]', group_id: 1, files: 11, posted_at: at, readable: true }) - .returning('id') + const row = { id: await insertRelease({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: `kantodo${i}`, source: 'stem', groups: '[]', group_id: 1, files: 11, posted_at: at, readable: true }) } ids.push(Number(row.id)) await db.table('par2_sets').insert({ set_id: setId, release_id: row.id, files: 1 }) } @@ -163,10 +152,7 @@ test.group('Par2Merger', (group) => { const release = async (poster: string, posted: Date, total: number, inner: string | null, groupId = 1) => { n++ const key = `dd${String(n).padStart(6, '0')}` - const [row] = await db - .table('releases') - .insert({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: `hex${n}`, source: 'stem', groups: '[]', group_id: groupId, files: 1, posted_at: posted, readable: false, file_total: total, poster }) - .returning('id') + const row = { id: await insertRelease({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: `hex${n}`, source: 'stem', groups: '[]', group_id: groupId, files: 1, posted_at: posted, readable: false, file_total: total, poster }) } if (inner) { await db.table('release_inspections').insert({ release_id: row.id, candidate_id: 5000 + n, kind: 'lone', result: JSON.stringify({ yencName: inner, type: '', size: 1 }) }) } @@ -202,10 +188,7 @@ test.group('Par2Merger', (group) => { const ids: number[] = [] for (let i = 1; i <= 3; i++) { const key = `ee00000${i}` - const [row] = await db - .table('releases') - .insert({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: 1, files: 10, posted_at: at, readable: false, file_total: 5, poster: 'p' }) - .returning('id') + const row = { id: await insertRelease({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: 1, files: 10, posted_at: at, readable: false, file_total: 5, poster: 'p' }) } ids.push(Number(row.id)) await db.table('release_inspections').insert({ release_id: row.id, candidate_id: 6000 + i, kind: 'lone', result: JSON.stringify({ yencName: `${stem}.7z.00${i}` }) }) } @@ -217,10 +200,7 @@ test.group('Par2Merger', (group) => { const at = new Date('2022-07-08T03:46:00Z') const ids: number[] = [] for (const key of ['aa000001', 'aa000002', 'aa000003']) { - const [row] = await db - .table('releases') - .insert({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: 1, files: 1, posted_at: at, readable: false }) - .returning('id') + const row = { id: await insertRelease({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: 1, files: 1, posted_at: at, readable: false }) } ids.push(Number(row.id)) } // Two test uploads whose 1 GiB files start alike, and a set describing one such file, held by the third. @@ -242,13 +222,10 @@ test.group('Par2Merger', (group) => { const release = async (name: string, files: number, total: number | null, posted: Date, chunkFiles: string[]) => { n++ const key = `ff${String(n).padStart(6, '0')}`.padEnd(40, '0') - const [row] = await db - .table('releases') - .insert({ release_key: key, identity_key: key, name, source: 'stem', groups: '[]', group_id: 1, files, file_total: total, poster, posted_at: posted, readable: true, settled_at: posted }) - .returning('id') + const row = { id: await insertRelease({ release_key: key, identity_key: key, name, source: 'stem', groups: '[]', group_id: 1, files, file_total: total, poster, posted_at: posted, readable: true, settled_at: posted }) } const id = Number(row.id) const payload = encodeChunk(chunkFiles.map((f, i) => ({ name: f, poster, subject: `${name} [${String(i).padStart(2, '0')}/${total ?? 0}] - "${f}" yEnc`, postedAt: posted.getTime(), groups: ['a.b.test'], segments: [[1, `${id}-${i}@x`, 1000]] }))) - await db.table('release_chunks').insert({ release_id: id, range_id: 1, batch_from: 1, files: chunkFiles.length, segments: chunkFiles.length, payload }) + await db.table('release_chunks').insert({ release_id: id, block_id: 1, batch_from: 1, files: chunkFiles.length, segments: chunkFiles.length, payload }) await db.table('fold_candidates').insert({ release_id: id }).onConflict('release_id').ignore() return id } diff --git a/tests/functional/par2_name_merge.spec.ts b/tests/functional/par2_name_merge.spec.ts index d1506d9..87ffad8 100644 --- a/tests/functional/par2_name_merge.spec.ts +++ b/tests/functional/par2_name_merge.spec.ts @@ -48,7 +48,7 @@ test.group('Par2Merger / by file names', (group) => { ...file('Track One.flac', 100_060, 5, 'other '), ]) await store(file('Track Two.flac', 200_070, 24 * 60 + 10)) - const ids = Object.fromEntries((await db.from('releases').select('id', 'name', 'poster', 'posted_at')).map((r) => [`${r.name}${r.poster.startsWith('other') ? ' (other)' : ''}${new Date(r.posted_at).getUTCDate() === 21 ? ' (later)' : ''}`, Number(r.id)])) + const ids = Object.fromEntries((await db.from('releases as r').leftJoin('posters as p', 'p.id', 'r.poster_id').select('r.id', 'r.name', 'p.poster', 'r.posted_at')).map((r) => [`${r.name}${r.poster.startsWith('other') ? ' (other)' : ''}${new Date(r.posted_at).getUTCDate() === 21 ? ' (later)' : ''}`, Number(r.id)])) const index = ids['Some.Album-WEB-2026'] const setId = 'a'.repeat(32) await db.table('par2_files').multiInsert( @@ -57,7 +57,7 @@ test.group('Par2Merger / by file names', (group) => { await db.table('par2_sets').insert({ set_id: setId, release_id: index, files: 3 }) assert.equal(await new Par2Merger().mergeByName(setId), 3) - const left = (await db.from('releases').orderBy('name').select('name', 'files', 'poster')).map((r) => `${r.name} ${r.files}${r.poster.startsWith('other') ? ' (other)' : ''}`) + const left = (await db.from('releases as r').leftJoin('posters as p', 'p.id', 'r.poster_id').orderBy('r.name').select('r.name', 'r.files', 'p.poster')).map((r) => `${r.name} ${r.files}${r.poster.startsWith('other') ? ' (other)' : ''}`) assert.includeMembers(left, ['Another Song 1', 'Track One 1 (other)', 'Track Two 1']) assert.lengthOf(left, 4) const merged = await db.from('release_merges').where('via', 'name').count('* as n') diff --git a/tests/functional/par2_pruner.spec.ts b/tests/functional/par2_pruner.spec.ts index ff80d0b..d1de368 100644 --- a/tests/functional/par2_pruner.spec.ts +++ b/tests/functional/par2_pruner.spec.ts @@ -2,6 +2,7 @@ import { test } from '@japa/runner' import db from '@adonisjs/lucid/services/db' import testUtils from '@adonisjs/core/services/test_utils' import { Par2Pruner } from '#services/par2_pruner' +import { insertRelease } from '#tests/functional/helpers/releases' test.group('Par2Pruner', (group) => { group.each.setup(() => testUtils.db().withGlobalTransaction()) @@ -12,10 +13,7 @@ test.group('Par2Pruner', (group) => { const release = async () => { n++ const key = `pp${String(n).padStart(6, '0')}`.padEnd(40, '0') - const [row] = await db - .table('releases') - .insert({ release_key: key, identity_key: key, name: key, source: 'stem', groups: '[]', group_id: 1, files: 2, readable: false }) - .returning('id') + const row = { id: await insertRelease({ release_key: key, identity_key: key, name: key, source: 'stem', groups: '[]', group_id: 1, files: 2, readable: false }) } return Number(row.id) } let candidate = 5000 @@ -59,12 +57,12 @@ test.group('Par2Pruner', (group) => { const posted = new Date('2024-03-01T12:00:00Z') const [group] = await db.table('groups').insert({ name: 'alt.binaries.pruned', priority: 1 }).returning('id') const groupId = Number(group.id) - await db.from('releases').where('id', f).update({ group_id: groupId, posted_at: posted, last_posted_at: posted, first_article: 5000, last_article: 5100 }) + // Finished (the window closed on it) long ago; its range done since. + await db.from('releases').where('id', f).update({ group_id: groupId, posted_at: posted, last_posted_at: posted, first_article: 5000, last_article: 5100, settled_at: old }) const [range] = await db .table('ranges') .insert({ group_id: groupId, start_id: 1, end_id: 10000, next_id: 10001, status: 'done', done_at: old, posted_low: new Date('2024-02-28T00:00:00Z'), posted_high: new Date('2024-03-03T00:00:00Z') }) .returning('id') - await db.table('range_blocks').insert({ group_id: groupId, start_id: 1, end_id: 10000, posted_low: new Date('2024-02-28T00:00:00Z'), posted_high: new Date('2024-03-03T00:00:00Z'), at_group_start: false }) // A lone candidate unprobed since before the range was done: its hash may still match. Kept. await db.table('probe_candidates').insert({ id: 9001, release_id: b, kind: 'lone', message_id: '<9001@x>', file: 'x', status: 'waiting', created_at: new Date(old.getTime() - 3600_000) }) @@ -87,7 +85,7 @@ test.group('Par2Pruner', (group) => { // A hash of its files since the merger's last window: the merger hasn't matched it yet. Kept. const g = await release() const unseen = await set('g', g, 1, null) - await db.from('releases').where('id', g).update({ group_id: groupId, posted_at: posted, last_posted_at: posted, first_article: 6000, last_article: 6100 }) + await db.from('releases').where('id', g).update({ group_id: groupId, posted_at: posted, last_posted_at: posted, first_article: 6000, last_article: 6100, settled_at: old }) await db.table('worker_cursors').insert({ name: 'merger', at: new Date(Date.now() - 3600_000) }) await db.table('probe_candidates').insert({ id: 9002, release_id: await release(), kind: 'lone', message_id: '<9002@x>', file: 'x', status: 'done' }) await db.table('file_hashes').insert({ candidate_id: 9002, release_id: 1, hash16k: 'g0'.padEnd(32, '0'), size: 1000 }) @@ -105,7 +103,7 @@ test.group('Par2Pruner', (group) => { let candidate = 7000 for (const letter of ['p', 'q', 'r']) { const key = `pr${letter}`.padEnd(40, '0') - const [row] = await db.table('releases').insert({ release_key: key, identity_key: key, name: key, source: 'stem', groups: '[]', group_id: 1, files: 1, readable: false }).returning('id') + const row = { id: await insertRelease({ release_key: key, identity_key: key, name: key, source: 'stem', groups: '[]', group_id: 1, files: 1, readable: false }) } const id = Number(row.id) const setId = letter.repeat(32) const h = `${letter}0`.padEnd(32, '0') diff --git a/tests/functional/prober.spec.ts b/tests/functional/prober.spec.ts index 3291cbe..d2a6b3d 100644 --- a/tests/functional/prober.spec.ts +++ b/tests/functional/prober.spec.ts @@ -2,15 +2,13 @@ import { test } from '@japa/runner' import db from '@adonisjs/lucid/services/db' import testUtils from '@adonisjs/core/services/test_utils' import { Prober } from '#services/prober' +import { insertRelease } from '#tests/functional/helpers/releases' test.group('Prober', (group) => { group.each.setup(() => testUtils.db().withGlobalTransaction()) test('a claim takes each tier: the readable tiers also take unreadable candidates, a kind with only one of them too', async ({ assert }) => { - const [row] = await db - .table('releases') - .insert({ release_key: 'pb'.padEnd(40, '0'), identity_key: 'pb'.padEnd(40, '0'), name: 'pb', source: 'subject', groups: '[]', files: 3 }) - .returning('id') + const row = { id: await insertRelease({ release_key: 'pb'.padEnd(40, '0'), identity_key: 'pb'.padEnd(40, '0'), name: 'pb', source: 'subject', groups: '[]', files: 3 }) } const candidate = (kind: string, unreadable: boolean, ready = true) => ({ release_id: row.id, kind, message_id: `<${kind}-${unreadable}-${ready}@x>`, file: kind, ready, unreadable, }) @@ -26,7 +24,7 @@ test.group('Prober', (group) => { test('a flush stores a name with a NUL in it, and one article that two releases have as a candidate', async ({ assert }) => { const release = async (key: string) => - Number((await db.table('releases').insert({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'subject', groups: '[]', files: 1 }).returning('id'))[0].id) + insertRelease({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'subject', groups: '[]', files: 1 }) const a = await release('fa') const b = await release('fb') const candidate = async (releaseId: number) => diff --git a/tests/functional/release_category.spec.ts b/tests/functional/release_category.spec.ts index ea4e582..4d65e03 100644 --- a/tests/functional/release_category.spec.ts +++ b/tests/functional/release_category.spec.ts @@ -1,6 +1,7 @@ import { test } from '@japa/runner' import db from '@adonisjs/lucid/services/db' import testUtils from '@adonisjs/core/services/test_utils' +import { insertReleases } from '#tests/functional/helpers/releases' const categoryOf = async (name: string) => Number((await db.rawQuery(`SELECT release_category(?) AS c`, [name])).rows[0].c) @@ -37,20 +38,21 @@ test.group('Release categories', (group) => { }) test("the name's, else the inner name's, else a given one (a spot's), else the groups'", async ({ assert }) => { - const row = (key: string, extra: Record) => ({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), source: 'stem', files: 1, groups: '["alt.binaries.misc"]', ...extra }) - await db.table('releases').multiInsert([ + const row = (key: string, extra: { name: string } & Record) => ({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), source: 'stem', files: 1, groups: '["alt.binaries.misc"]', ...extra }) + const [a, b, c, d] = await insertReleases([ row('a', { name: 'Show.S01E02.1080p.WEB' }), row('b', { name: 'TsIgNBewuFbuwWy' }), row('c', { name: 'TsIgNBewuFbuwWy', groups: '["alt.binaries.pictures.erotica.puffies"]' }), // A spot's release of its own, inserted with the spot's category. row('d', { name: 'Some spot title', category: 2040 }), ]) - const byName = async () => Object.fromEntries((await db.from('releases').select('release_key', 'category')).map((r) => [r.release_key[0], r.category])) + const letters = new Map([[a, 'a'], [b, 'b'], [c, 'c'], [d, 'd']]) + const byName = async () => Object.fromEntries((await db.from('releases').select('id', 'category')).map((r) => [letters.get(Number(r.id)), r.category])) assert.deepEqual(await byName(), { a: 5040, b: 8010, c: 6060, d: 2040 }) // The prober finds an inner name; a spot links a release (the spot inspector sets a Misc one's category). - await db.from('releases').where('release_key', 'b'.padEnd(40, '0')).update({ inner_name: 'Some.Movie.2020.1080p.BluRay.x264-GRP.mkv' }) - await db.from('releases').where('release_key', 'c'.padEnd(40, '0')).update({ category: 5030 }) - await db.from('releases').where('release_key', 'c'.padEnd(40, '0')).update({ inner_name: 'nothing.bin' }) + await db.from('releases').where('id', b).update({ inner_name: 'Some.Movie.2020.1080p.BluRay.x264-GRP.mkv' }) + await db.from('releases').where('id', c).update({ category: 5030 }) + await db.from('releases').where('id', c).update({ inner_name: 'nothing.bin' }) assert.deepEqual(await byName(), { a: 5040, b: 2040, c: 5030, d: 2040 }) }) diff --git a/tests/functional/release_judge.spec.ts b/tests/functional/release_judge.spec.ts index ba22011..9aa5da5 100644 --- a/tests/functional/release_judge.spec.ts +++ b/tests/functional/release_judge.spec.ts @@ -3,6 +3,7 @@ import db from '@adonisjs/lucid/services/db' import testUtils from '@adonisjs/core/services/test_utils' import { ReleaseJudge } from '#services/release_judge' import SearchService from '#services/search_service' +import { insertRelease } from '#tests/functional/helpers/releases' test.group('ReleaseJudge', (group) => { group.each.setup(() => testUtils.db().withGlobalTransaction()) @@ -15,9 +16,7 @@ test.group('ReleaseJudge', (group) => { const release = async (name: string, extra: Record = {}) => { n++ const key = `jj${String(n).padStart(6, '0')}`.padEnd(40, '0') - const [row] = await db - .table('releases') - .insert({ + const row = { id: await insertRelease({ release_key: key, identity_key: key, name, @@ -29,8 +28,7 @@ test.group('ReleaseJudge', (group) => { readable: true, settled_at: settled, ...extra, - }) - .returning('id') + }) } return Number(row.id) } let candidate = 9000 @@ -93,10 +91,6 @@ test.group('ReleaseJudge', (group) => { poster: 'celeborn@not.here (Celeborn)', source: 'subject', }) - // Scraped posters are found through posters (the batch store adds each release's). - await db - .table('posters') - .multiInsert([{ poster: '@_CelebMovie_@' }, { poster: 'celeborn@not.here (Celeborn)' }]) const locked = await release('Locked.Thing', { passworded: true }) const unlocked = await release('Known.Password', { passworded: true }) @@ -104,30 +98,19 @@ test.group('ReleaseJudge', (group) => { .table('release_passwords') .insert({ release_id: unlocked, password: 'pw', source: 'spot-text' }) - // A block of done ranges around the unusable ones (release_final); one beside an unread range isn't judged yet. + // Unusable once finished (the window closed on it); a provisional tail release isn't judged yet. await db .table('groups') .insert({ id: 1, name: 'alt.binaries.judged', priority: 1 }) .onConflict('id') .ignore() - await db.table('ranges').multiInsert([ - { - group_id: 1, - start_id: 1, - end_id: 1000, - next_id: 1001, - status: 'done', - posted_low: new Date(settled.getTime() - 86400_000), - posted_high: new Date(settled.getTime() + 86400_000), - }, - { group_id: 1, start_id: 1001, end_id: 2000, next_id: 1001, status: 'planned' }, - ]) const placed = { first_article: 100, last_article: 110 } const unusable = await release('ab12CD34ef56GH78ij90', { readable: false, ...placed }) const pending = await release('pp12CD34ef56GH78ij90', { readable: false, first_article: 990, last_article: 1000, + settled_at: null, posted_at: new Date(settled.getTime() + 86400_000 - 3600_000), }) const fresh = await release('zz12CD34ef56GH78ij90', { @@ -188,9 +171,7 @@ test.group('ReleaseJudge', (group) => { const release = async (name: string, extra: Record = {}) => { n++ const key = `jc${String(n).padStart(6, '0')}`.padEnd(40, '0') - const [row] = await db - .table('releases') - .insert({ + const row = { id: await insertRelease({ release_key: key, identity_key: key, name, @@ -201,8 +182,7 @@ test.group('ReleaseJudge', (group) => { readable: true, settled_at: settled, ...extra, - }) - .returning('id') + }) } return Number(row.id) } const judge = new ReleaseJudge() diff --git a/tests/functional/releases_api.spec.ts b/tests/functional/releases_api.spec.ts index e8f5c63..da88559 100644 --- a/tests/functional/releases_api.spec.ts +++ b/tests/functional/releases_api.spec.ts @@ -94,6 +94,8 @@ test.group('Releases API', (group) => { // As the prober leaves them: the movie's archive is encrypted, the show holds a differently named file. await db.from('releases').where('id', movie.id).update({ passworded: true, inspected_at: new Date() }) await db.from('releases').where('id', show.id).update({ passworded: false, inner_name: 'Hidden.Gem.2023.mkv', readable: false, inspected_at: new Date() }) + // The name inside, searchable (release_texts, kind file: lib/release/text_kind.ts). + await db.table('release_texts').insert({ release_id: show.id, kind: 2, text: 'Hidden.Gem.2023.mkv' }) await db.table('release_inspections').insert({ release_id: show.id, candidate_id: 1, kind: 'archive', result: JSON.stringify({ yencName: 'x.rar', size: 1, type: 'rar5', names: ['Hidden.Gem.2023.mkv'], entries: 1, passworded: false }) }) const byInner = (await client.get('/api/v1/releases?q=hidden gem')).body() as { releases: { id: number; innerName: string }[] } diff --git a/tests/functional/seven_zip_tails.spec.ts b/tests/functional/seven_zip_tails.spec.ts index f0944a9..39023f5 100644 --- a/tests/functional/seven_zip_tails.spec.ts +++ b/tests/functional/seven_zip_tails.spec.ts @@ -2,13 +2,14 @@ import { test } from '@japa/runner' import db from '@adonisjs/lucid/services/db' import testUtils from '@adonisjs/core/services/test_utils' import { pruneSevenZipTails } from '#services/seven_zip_tails' +import { insertReleases } from '#tests/functional/helpers/releases' test.group('7z tails', (group) => { group.each.setup(() => testUtils.db().withGlobalTransaction()) test("of a release's waiting tails, those below its highest volume go; the highest volume's all stay", async ({ assert }) => { const row = (key: string) => ({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', files: 4 }) - const [merged, other] = (await db.table('releases').multiInsert([row('a'), row('b')]).returning('id')).map((r) => r.id) + const [merged, other] = await insertReleases([row('a'), row('b')]) const tail = (releaseId: number, volume: string, mid: string, status = 'waiting') => ({ release_id: releaseId, kind: '7z-tail', message_id: mid, file: `x.7z.${volume}`, status }) await db.table('probe_candidates').multiInsert([ // Two merged members' tails, one of the highest volume probed already. diff --git a/tests/functional/spots.spec.ts b/tests/functional/spots.spec.ts index dc717da..eac9867 100644 --- a/tests/functional/spots.spec.ts +++ b/tests/functional/spots.spec.ts @@ -5,6 +5,8 @@ import { Spotter } from '#services/spotter' import { SpotInspector } from '#services/spot_inspector' import { releaseIdentity } from '#lib/release/index' import { parseSubject } from '#lib/headers/index' +import { SOURCES } from '#lib/release/source' +import { insertRelease } from '#tests/functional/helpers/releases' // Message ids whose sha1('') starts with 0000: self-signed spots (key 7) pass on that hashcash alone. const SPOT = 'spottest731@spot.net' @@ -68,10 +70,7 @@ test.group('Spots', (group) => { // The release the header pass built for these files: the same identity, in the group. const parsed = parseSubject(subjects[0])! const identity = releaseIdentity({ poster, subject: parsed.base, name: parsed.name, weak: false })! - const [r] = await db - .table('releases') - .insert({ release_key: identity.key, identity_key: identity.key, name: identity.name, source: identity.source, poster, groups: '[]', group_id: groupId, files: 3, posted_at: posted, readable: true }) - .returning('id') + const r = { id: await insertRelease({ release_key: identity.key, identity_key: identity.key, name: identity.name, source: identity.source, poster, groups: '[]', group_id: groupId, files: 3, posted_at: posted, readable: true }) } const releaseId = Number(r.id) const nzb = nzbXml(subjects, poster, posted.getTime() / 1000, 'nzbpw') await new Spotter().store(rawSpot(SPOT, 'Some.Movie.2024.1080p', 'Wachtwoord: geheim123'), SPOT, 10) @@ -100,13 +99,14 @@ test.group('Spots', (group) => { const inspector = new SpotInspector() assert.deepEqual(await inspector.inspectNzb(await db.from('spots').where('message_id', SPOT).firstOrFail(), nzb), { status: 'inspected', matched: 0 }) assert.isTrue(await inspector.standalone(SPOT)) - const release = await db.from('releases').where('source', 'spot').firstOrFail() + const release = await db.from('releases').where('source', SOURCES.spot).firstOrFail() assert.equal(release.name, 'Lonely File') // Its title says nothing: the spot's category (01, a09: a movie, x264). assert.equal(release.category, 2040) assert.equal(release.files, 1) - assert.isNotNull(release.poster) - assert.isNotNull(await db.from('posters').where('poster', release.poster).first()) + // Its poster (the spot's) by id, in posters for poster search. + assert.isNotNull(release.poster_id) + assert.isNotEmpty((await db.from('posters').where('id', release.poster_id).firstOrFail()).poster) const { default: NzbService } = await import('#services/nzb_service') assert.include((await NzbService.build(Number(release.id)))!.xml, 'lonely.file.mkv') // Not again while it has one; once the header pass builds the file's release, the spot moves there. @@ -114,10 +114,7 @@ test.group('Spots', (group) => { const group = await db.from('groups').where('name', 'alt.binaries.test.spots').firstOrFail() const parsed = parseSubject('[1/1] - "lonely.file.mkv" yEnc (1/9)')! const identity = releaseIdentity({ poster: 'p', subject: parsed.base, name: parsed.name, weak: false })! - const [h] = await db - .table('releases') - .insert({ release_key: identity.key, identity_key: identity.key, name: identity.name, source: identity.source, poster: 'p', groups: '[]', group_id: group.id, files: 1, posted_at: new Date(STAMP * 1000), readable: true }) - .returning('id') + const h = { id: await insertRelease({ release_key: identity.key, identity_key: identity.key, name: identity.name, source: identity.source, poster: 'p', groups: '[]', group_id: group.id, files: 1, posted_at: new Date(STAMP * 1000), readable: true }) } assert.equal(await inspector.match(SPOT), 1) assert.isNull(await db.from('releases').where('id', release.id).first()) assert.equal(Number((await db.from('spots').where('message_id', SPOT).firstOrFail()).release_id), Number(h.id)) diff --git a/tests/unit/release/source.spec.ts b/tests/unit/release/source.spec.ts index d9c7644..4992ca0 100644 --- a/tests/unit/release/source.spec.ts +++ b/tests/unit/release/source.spec.ts @@ -3,7 +3,7 @@ import { SOURCES, sourceCode, sourceName } from '#lib/release/source' test.group('release source codes', () => { test('names and codes map both ways; the codes are fixed', ({ assert }) => { - assert.deepEqual(SOURCES, { subject: 1, post: 2, stem: 3, spot: 4, nzb: 5, album: 6 }) + assert.deepEqual(SOURCES, { 'subject': 1, 'post': 2, 'stem': 3, 'spot': 4, 'nzb': 5, 'album': 6, 'weak-stem': 7 }) for (const [name, code] of Object.entries(SOURCES)) { assert.equal(sourceCode(name), code) assert.equal(sourceName(code), name)