voyager 3: the batch store writes releases once, by id references and bytea keys
Each release chunk is a release inserted once: release_key the SHA-1 of the identity, the owner group and the first article (deterministic; no numbering under a group lock), identity_key the identity's SHA-1 bytes, poster_id and group_set_id through catalog_refs (cached per process), source as a code (weak-stem added), the name into release_texts. A finished release's candidates are claimable at once; a provisional tail release is finished by a later chunk of its key, a finished one sent again is left alone. Gone: joins into earlier releases and across groups, key numbering, the group lock, the fragment merger. Interim: chunks, counts and samples keyed by ranges.block_id, a range without a block gets one. Readers follow: search (posters and group_sets joined, words through release_texts, posters by id, source by code), the par2 merger, the sidecar folder, the judge's scraped rule, the spot inspector, the spotter, the NZB service, releases:regroup, edge:bench. Tests insert fixtures through tests/functional/helpers/releases.ts; the suites pass again. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Daan MeijerandClaude Fable 5.1 committed
1 parent
bc4e17e86f
commit
2580f20a43
32 files changed
+962
-497
No files matched your search
+176
-188
@@ -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<StoredBatch> {
|
||||
@@ -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<number, number>
|
||||
) {
|
||||
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<number>(candidates.map((c: { id: string }) => Number(c.id)))]
|
||||
const names = new Map<number, Set<string>>()
|
||||
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<string>()
|
||||
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<string>()
|
||||
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<number> {
|
||||
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<string, number>()
|
||||
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<number>()
|
||||
|
||||
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<number, number>(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<string, { id: number; inserted: boolean; settled: boolean }>(
|
||||
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<string, number>(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<string, number>(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<number>()
|
||||
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
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
import type db from '@adonisjs/lucid/services/db'
|
||||
|
||||
type Client = Pick<typeof db, 'rawQuery'>
|
||||
|
||||
/** 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<string, number>()
|
||||
const groupSetCache = new Map<string, number>()
|
||||
|
||||
/** 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<string, number>,
|
||||
table: 'posters' | 'group_sets',
|
||||
column: 'poster' | 'groups',
|
||||
cast: 'text' | 'jsonb',
|
||||
values: string[]
|
||||
): Promise<Map<string, number>> {
|
||||
const out = new Map<string, number>()
|
||||
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<Map<string, number>> {
|
||||
return ids(client, posterCache, 'posters', 'poster', 'text', [
|
||||
...new Set(posters.map(posterText)),
|
||||
])
|
||||
},
|
||||
|
||||
async posterId(client: Client, poster: string): Promise<number> {
|
||||
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<Map<string, number>> {
|
||||
return ids(client, groupSetCache, 'group_sets', 'groups', 'jsonb', [
|
||||
...new Set(sets.map(groupSetKey)),
|
||||
])
|
||||
},
|
||||
|
||||
async groupSetId(client: Client, groups: string[]): Promise<number> {
|
||||
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()
|
||||
},
|
||||
}
|
||||
@@ -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) }
|
||||
}
|
||||
|
||||
+15
-13
@@ -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<Map<string, Member> | 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<string, Member>(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<number> {
|
||||
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<string, Member>()
|
||||
@@ -374,7 +376,7 @@ export class Par2Merger {
|
||||
async mergeSet(setId: string): Promise<number> {
|
||||
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
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -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<string, number>)[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. */
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<string, { id: number; group: number; at: number | null }[]>()
|
||||
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()
|
||||
}
|
||||
|
||||
@@ -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'}.`)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user