diff --git a/worker/src/archive/listing-plan.test.ts b/worker/src/archive/listing-plan.test.ts new file mode 100644 index 0000000..e7d0315 --- /dev/null +++ b/worker/src/archive/listing-plan.test.ts @@ -0,0 +1,130 @@ +import { describe, it, expect } from "vitest"; +import { + planListingRead, + planRangedFallback, + classifySourceShape, + isConcatRepackName, + concatRepackBase, + concatChunkIndex, + isVolumeSet, +} from "./listing-plan.js"; + +const GB = 1024n * 1024n * 1024n; + +/** Defaults for a plan input; individual tests override what they care about. */ +function input(over: Partial[0]>) { + return { + archiveType: "ZIP", + sourceFileName: "Pack.z01", + destFileNames: ["Pack.z01", "Pack.zip"], + totalSize: 10n * GB, + maxDownloadBytes: 200n * GB, + rangedOnly: false, + ...over, + }; +} + +describe("classifySourceShape", () => { + it("separates spanned ZIP volumes from a raw byte split", () => { + expect(classifySourceShape("Pack.z01")).toBe("spanned-zip"); + expect(classifySourceShape("Pack.z12")).toBe("spanned-zip"); + expect(classifySourceShape("Pack.zip.001")).toBe("byte-split"); + expect(classifySourceShape("Pack.7z.001")).toBe("byte-split"); + }); + + it("recognizes RAR volume sets and lone archives", () => { + expect(classifySourceShape("Pack.part1.rar")).toBe("rar-volume-set"); + expect(classifySourceShape("Pack.r00")).toBe("rar-volume-set"); + expect(classifySourceShape("Pack.zip")).toBe("single"); + expect(classifySourceShape("notes.txt")).toBe("unknown"); + }); + + it("marks exactly the layouts whose volumes are independent containers", () => { + expect(isVolumeSet("spanned-zip")).toBe(true); + expect(isVolumeSet("rar-volume-set")).toBe(true); + expect(isVolumeSet("byte-split")).toBe(false); + expect(isVolumeSet("single")).toBe(false); + }); +}); + +describe("concat repack naming", () => { + it("recognizes the repack chunk names the uploader produces", () => { + expect(isConcatRepackName("Pack.concat.001")).toBe(true); + expect(isConcatRepackName("Pack.concat.017")).toBe(true); + expect(isConcatRepackName("Pack.concat")).toBe(true); + expect(isConcatRepackName("Pack.z01")).toBe(false); + expect(isConcatRepackName("Pack.zip.001")).toBe(false); + // "concat" appearing mid-name must not count + expect(isConcatRepackName("concat-models.zip")).toBe(false); + }); + + it("groups and orders chunks of one repack", () => { + expect(concatRepackBase("Pack.concat.002")).toBe("pack.concat"); + expect(concatRepackBase("Pack.concat")).toBe("pack.concat"); + expect(concatChunkIndex("Pack.concat.017")).toBe(17); + expect(concatChunkIndex("Pack.concat")).toBe(0); + }); +}); + +describe("planListingRead", () => { + it("routes a spanned ZIP set to the ranged read", () => { + const plan = planListingRead(input({})); + expect(plan.route).toBe("ranged"); + expect(plan.reason).toContain("spanned-zip"); + }); + + it("skips a concatenated spanned ZIP set — no reader can ever list it", () => { + const plan = planListingRead( + input({ sourceFileName: "Pack.z01", destFileNames: ["Pack.concat.001", "Pack.concat.002"] }) + ); + expect(plan.route).toBe("skip"); + expect(plan.reason).toContain("not a valid archive"); + }); + + it("skips a concatenated RAR volume set for the same reason", () => { + const plan = planListingRead( + input({ + archiveType: "RAR", + sourceFileName: "Pack.part1.rar", + destFileNames: ["Pack.concat.001", "Pack.concat.002"], + }) + ); + expect(plan.route).toBe("skip"); + expect(plan.reason).toContain("rar-volume-set"); + }); + + it("still reads a concatenated BYTE SPLIT — re-cutting one stream is lossless", () => { + const plan = planListingRead( + input({ sourceFileName: "Pack.zip.001", destFileNames: ["Pack.concat.001", "Pack.concat.002"] }) + ); + expect(plan.route).toBe("ranged"); + expect(plan.reason).toContain("byte split"); + }); + + it("skips archive types with no file-list reader without touching the API", () => { + expect(planListingRead(input({ archiveType: "DOCUMENT" }))).toMatchObject({ route: "skip" }); + }); + + it("skips when no destination part could be resolved", () => { + expect(planListingRead(input({ destFileNames: [] }))).toMatchObject({ route: "skip" }); + }); +}); + +describe("planRangedFallback", () => { + it("refuses to download when rangedOnly is set, however large the archive", () => { + const plan = planRangedFallback({ totalSize: 35n * GB, maxDownloadBytes: 200n * GB, rangedOnly: true }); + expect(plan.route).toBe("skip"); + expect(plan.reason).toContain("rangedOnly"); + }); + + it("refuses to download past the size cap", () => { + const plan = planRangedFallback({ totalSize: 300n * GB, maxDownloadBytes: 200n * GB, rangedOnly: false }); + expect(plan.route).toBe("skip"); + expect(plan.reason).toContain("exceeds the download cap"); + }); + + it("falls back to a download when it is allowed and affordable", () => { + const plan = planRangedFallback({ totalSize: 2n * GB, maxDownloadBytes: 200n * GB, rangedOnly: false }); + expect(plan.route).toBe("download"); + }); +}); diff --git a/worker/src/archive/listing-plan.ts b/worker/src/archive/listing-plan.ts new file mode 100644 index 0000000..d8618bc --- /dev/null +++ b/worker/src/archive/listing-plan.ts @@ -0,0 +1,164 @@ +import { detectArchive } from "./detect.js"; + +/** + * Decide *how* to read an already-uploaded archive's inner-file listing before + * spending a single byte of Telegram traffic on it. + * + * Three outcomes matter: + * + * - `ranged` — read only the header/tail bytes that carry the listing. + * Tens of kilobytes regardless of archive size. + * - `download` — the ranged read can't work (or already failed); pull the whole + * archive down and let the on-disk reader handle it. + * - `skip` — no reader can ever list this destination copy. Saying so + * up-front is the whole point: the alternative is burning API + * calls and bandwidth on something structurally unreadable. + * + * The `skip` case that motivated this module: when any source volume exceeded + * the upload cap, the ingestion worker concatenated every volume into one file + * and re-split it into uniform `.concat.NNN` chunks. For a *byte split* + * (`pack.zip.001`, …) that round-trips fine — the bytes are the same stream. + * For a ZIP-spec **spanned** set (`pack.z01`, …, `pack.zip`) or a RAR volume + * set it does not: those volumes are separate containers, and their + * concatenation is not a valid archive in any format. The destination copy of + * such a package is permanently unlistable, and no amount of downloading will + * change that. + */ + +/** `.concat`, `.concat.001`, … — the re-split repack naming. */ +const CONCAT_REPACK_RE = /\.concat(?:\.\d{2,})?$/i; + +export function isConcatRepackName(fileName: string): boolean { + return CONCAT_REPACK_RE.test(fileName.trim().replace(/[.\s]+$/, "")); +} + +/** Strip the `.NNN` chunk suffix, so every chunk of one repack shares a key. */ +export function concatRepackBase(fileName: string): string { + return fileName + .trim() + .replace(/[.\s]+$/, "") + .replace(/\.(\d{2,})$/, "") + .toLowerCase(); +} + +/** Numeric chunk index of a `.concat.NNN` name; 0 for a bare `.concat`. */ +export function concatChunkIndex(fileName: string): number { + const m = fileName.trim().replace(/[.\s]+$/, "").match(/\.(\d{2,})$/); + return m ? parseInt(m[1], 10) : 0; +} + +/** + * How the *source* archive was laid out, derived from the Package's own + * fileName. This is what decides whether a `.concat.NNN` repack is survivable: + * only a stream that was contiguous to begin with can be re-cut. + */ +export type SourceShape = + /** `.z01`, `.z02`, … + `.zip` — ZIP-spec spanned, one container per volume. */ + | "spanned-zip" + /** `.partN.rar` or `.rNN` — RAR volume set, one container per volume. */ + | "rar-volume-set" + /** `.zip.001`, `.7z.001`, … — one file cut into chunks. */ + | "byte-split" + /** A lone `.zip` / `.rar` / `.7z` / document. */ + | "single" + | "unknown"; + +export function classifySourceShape(fileName: string): SourceShape { + const info = detectArchive(fileName); + if (!info) return "unknown"; + switch (info.pattern) { + case "ZIP_LEGACY": + return "spanned-zip"; + case "RAR_PART": + case "RAR_LEGACY": + return "rar-volume-set"; + case "ARCHIVE_NUMBERED": + return "byte-split"; + case "SINGLE": + return "single"; + } +} + +/** True for layouts whose volumes are independent containers, not one stream. */ +export function isVolumeSet(shape: SourceShape): boolean { + return shape === "spanned-zip" || shape === "rar-volume-set"; +} + +export type ListingRoute = + | { route: "ranged"; reason: string } + | { route: "download"; reason: string } + | { route: "skip"; reason: string }; + +export interface ListingPlanInput { + /** Package.archiveType. */ + archiveType: string; + /** Package.fileName — the *source* name, which encodes the original layout. */ + sourceFileName: string; + /** File names of the resolved destination parts, in upload order. */ + destFileNames: string[]; + /** Total size of the destination parts. */ + totalSize: bigint; + /** Cap on a full download (WORKER_MAX_ZIP_SIZE_MB, in bytes). */ + maxDownloadBytes: bigint; + /** When true, the full-download fallback is off the table entirely. */ + rangedOnly: boolean; +} + +const RANGED_TYPES = new Set(["ZIP", "RAR", "SEVEN_Z"]); + +/** + * Pick the first route to try for a package. `download` is only returned when + * ranged reading is structurally impossible for the type; a ranged read that + * fails at runtime is handled by {@link planRangedFallback}. + */ +export function planListingRead(input: ListingPlanInput): ListingRoute { + if (!RANGED_TYPES.has(input.archiveType)) { + return { route: "skip", reason: `archiveType ${input.archiveType} has no file-list reader` }; + } + if (input.destFileNames.length === 0) { + return { route: "skip", reason: "no destination parts could be resolved" }; + } + + const shape = classifySourceShape(input.sourceFileName); + const repacked = input.destFileNames.some(isConcatRepackName); + + if (repacked && isVolumeSet(shape)) { + return { + route: "skip", + reason: + `destination copy is a .concat.NNN repack of a ${shape} — concatenated volumes ` + + "are not a valid archive, so no reader can ever list it", + }; + } + + if (repacked) { + // A re-cut byte split is still the original contiguous stream, so the + // ranged reader's whole-archive offset arithmetic applies unchanged. + return { route: "ranged", reason: `.concat.NNN repack of a ${shape} — readable as a byte split` }; + } + + return { route: "ranged", reason: `${shape} destination copy, ${input.destFileNames.length} part(s)` }; +} + +/** + * What to do once a ranged read has come back empty. Kept separate from + * {@link planListingRead} so the "we tried cheap and it didn't work" decision + * is explicit and testable rather than buried in a conditional. + */ +export function planRangedFallback( + input: Pick +): { route: "download"; reason: string } | { route: "skip"; reason: string } { + if (input.rangedOnly) { + return { + route: "skip", + reason: `ranged read failed and rangedOnly is set — not downloading ${input.totalSize} bytes`, + }; + } + if (input.totalSize > input.maxDownloadBytes) { + return { + route: "skip", + reason: `ranged read failed and ${input.totalSize} bytes exceeds the download cap of ${input.maxDownloadBytes}`, + }; + } + return { route: "download", reason: `ranged read failed — falling back to a ${input.totalSize} byte download` }; +} diff --git a/worker/src/backfill-scope.test.ts b/worker/src/backfill-scope.test.ts new file mode 100644 index 0000000..192d050 --- /dev/null +++ b/worker/src/backfill-scope.test.ts @@ -0,0 +1,121 @@ +import { describe, it, expect } from "vitest"; +import { + parseBackfillPayload, + parseFileNameLike, + MAX_BACKFILL_LIMIT, + DEFAULT_BACKFILL_LIMIT, +} from "./backfill-scope.js"; + +/** The plan of a payload that must parse; fails the test if it doesn't. */ +function planOf(payload: string) { + const parsed = parseBackfillPayload(payload); + if (!parsed.ok) throw new Error(`expected payload to parse, got: ${parsed.error}`); + return parsed.plan; +} + +function errorOf(payload: string): string { + const parsed = parseBackfillPayload(payload); + if (parsed.ok) throw new Error(`expected payload to be rejected, got plan: ${parsed.plan.describe}`); + return parsed.error; +} + +describe("parseBackfillPayload — refusing the unbounded sweep", () => { + it("rejects an empty payload rather than selecting every empty package", () => { + expect(errorOf("{}")).toMatch(/refusing an unbounded backfill/); + expect(errorOf("")).toMatch(/refusing an unbounded backfill/); + }); + + it("rejects a payload that only names an archiveType", () => { + // This is the shape that used to select 4,330 packages / 5.4TB. + expect(errorOf('{"archiveType":"ZIP"}')).toMatch(/refusing an unbounded backfill/); + }); + + it("rejects invalid JSON and non-object payloads", () => { + expect(errorOf("not json")).toMatch(/not valid JSON/); + expect(errorOf("[]")).toMatch(/must be a JSON object/); + expect(errorOf("null")).toMatch(/must be a JSON object/); + }); + + it("rejects an unknown field instead of ignoring it", () => { + // A typo'd selector must not silently degrade into "no selector". + expect(errorOf('{"packageIDs":["abc"]}')).toMatch(/unknown field\(s\): packageIDs/); + }); + + it("allows the broad sweep only when it asks for itself", () => { + const plan = planOf('{"archiveType":"RAR","limit":50,"allowBroadSweep":true}'); + expect(plan.selector).toEqual({ archiveType: "RAR" }); + expect(plan.limit).toBe(50); + }); + + it("rejects allowBroadSweep combined with a selector as a contradiction", () => { + expect(errorOf('{"fileNameLike":"%.z01","allowBroadSweep":true}')).toMatch(/unscoped sweeps only/); + }); +}); + +describe("parseBackfillPayload — packageIds", () => { + it("accepts, trims and de-duplicates an explicit id list", () => { + const plan = planOf('{"packageIds":[" abc123 ","abc123","def456"]}'); + expect(plan.selector.packageIds).toEqual(["abc123", "def456"]); + }); + + it("rejects an empty list, non-strings and non-identifier ids", () => { + expect(errorOf('{"packageIds":[]}')).toMatch(/must not be empty/); + expect(errorOf('{"packageIds":[1,2]}')).toMatch(/only strings/); + expect(errorOf('{"packageIds":["a\'; DROP TABLE packages--"]}')).toMatch(/not a valid identifier/); + expect(errorOf('{"packageIds":"abc"}')).toMatch(/must be an array/); + }); +}); + +describe("parseFileNameLike", () => { + it("maps the four accepted wildcard positions to Prisma filters", () => { + expect(parseFileNameLike("%.z01")).toEqual({ endsWith: ".z01" }); + expect(parseFileNameLike("Dragon%")).toEqual({ startsWith: "Dragon" }); + expect(parseFileNameLike("%dragon%")).toEqual({ contains: "dragon" }); + expect(parseFileNameLike("Pack.z01")).toEqual({ equals: "Pack.z01" }); + }); + + it("refuses patterns that would match far more than intended", () => { + expect(parseFileNameLike("%")).toHaveProperty("error"); + expect(parseFileNameLike("%%")).toHaveProperty("error"); + expect(parseFileNameLike(" ")).toHaveProperty("error"); + // Too little literal text to be a meaningful scope + expect(parseFileNameLike("%a%")).toHaveProperty("error"); + }); + + it("refuses interior and underscore wildcards rather than mis-translating them", () => { + expect(parseFileNameLike("%.z%1")).toMatchObject({ error: expect.stringMatching(/interior wildcard/) }); + expect(parseFileNameLike("%.z0_")).toMatchObject({ error: expect.stringMatching(/_ wildcard/) }); + }); +}); + +describe("parseBackfillPayload — limits and flags", () => { + it("defaults the limit and caps it", () => { + expect(planOf('{"fileNameLike":"%.z01"}').limit).toBe(DEFAULT_BACKFILL_LIMIT); + expect(errorOf(`{"fileNameLike":"%.z01","limit":${MAX_BACKFILL_LIMIT + 1}}`)).toMatch(/exceeds the maximum/); + expect(errorOf('{"fileNameLike":"%.z01","limit":0}')).toMatch(/positive integer/); + expect(errorOf('{"fileNameLike":"%.z01","limit":1.5}')).toMatch(/positive integer/); + }); + + it("defaults rangedOnly and recoverDestIds off, and requires booleans", () => { + const plan = planOf('{"fileNameLike":"%.z01"}'); + expect(plan.rangedOnly).toBe(false); + expect(plan.recoverDestIds).toBe(false); + expect(errorOf('{"fileNameLike":"%.z01","rangedOnly":"yes"}')).toMatch(/rangedOnly must be a boolean/); + }); + + it("parses the full spanned-ZIP repair payload", () => { + const plan = planOf( + '{"fileNameLike":"%.z01","archiveType":"ZIP","limit":250,"rangedOnly":true,"recoverDestIds":true}' + ); + expect(plan.selector).toEqual({ fileName: { endsWith: ".z01" }, archiveType: "ZIP" }); + expect(plan.limit).toBe(250); + expect(plan.rangedOnly).toBe(true); + expect(plan.recoverDestIds).toBe(true); + expect(plan.describe).toContain('fileName.endsWith=".z01"'); + expect(plan.describe).toContain("rangedOnly"); + }); + + it("rejects an unsupported archiveType", () => { + expect(errorOf('{"fileNameLike":"%.z01","archiveType":"TAR"}')).toMatch(/archiveType must be one of/); + }); +}); diff --git a/worker/src/backfill-scope.ts b/worker/src/backfill-scope.ts new file mode 100644 index 0000000..5b3ea6c --- /dev/null +++ b/worker/src/backfill-scope.ts @@ -0,0 +1,244 @@ +/** + * Payload parsing for the `backfill_filelists` pg_notify request. + * + * The original payload shape was `{limit, archiveType}` — both optional, both + * defaulted. That made the *unbounded* sweep the easiest thing to trigger: + * `SELECT pg_notify('backfill_filelists', '{}')` selected every Package with + * `fileCount = 0` of every archive type, oldest first, and started downloading. + * On a real catalogue that is multiple terabytes of traffic aimed at packages + * nobody asked to repair. + * + * So the rule here is: **a request must name what it wants**. A payload with no + * selector is rejected outright, and the broad "every empty package of type X" + * sweep has to opt in explicitly via `allowBroadSweep`. Omitting a field can + * only ever narrow the job or fail it — never widen it. + */ + +export type BackfillArchiveType = "ZIP" | "RAR" | "SEVEN_Z"; + +const ARCHIVE_TYPES: BackfillArchiveType[] = ["ZIP", "RAR", "SEVEN_Z"]; + +/** Hard ceiling on `limit`, so even a deliberate broad sweep stays bounded. */ +export const MAX_BACKFILL_LIMIT = 2000; +/** Hard ceiling on an explicit id list — keeps the SQL `IN (...)` sane. */ +export const MAX_BACKFILL_PACKAGE_IDS = 2000; +export const DEFAULT_BACKFILL_LIMIT = 200; + +/** Minimum literal characters in a `fileNameLike` pattern, so `%` can't stand alone. */ +const MIN_FILENAME_LITERAL = 3; + +/** Prisma-shaped filename filter — a closed set of forms, never raw SQL. */ +export type FileNameFilter = + | { equals: string } + | { startsWith: string } + | { endsWith: string } + | { contains: string }; + +export interface BackfillSelector { + /** Explicit package ids: the most tightly bounded selector there is. */ + packageIds?: string[]; + fileName?: FileNameFilter; + archiveType?: BackfillArchiveType; +} + +export interface BackfillPlan { + selector: BackfillSelector; + limit: number; + /** + * Never fall back to a full `downloadFile` of the archive, even when the + * cheap ranged read fails. Set this for repairs where the full-download cost + * would be absurd (a spanned set's listing lives in ~64KB of its final + * volume; downloading the set to reach it can be hundreds of gigabytes). + */ + rangedOnly: boolean; + /** + * Allow one scan of the destination channel to recover `destMessageIds` for + * candidates that have none. Opt-in because the scan itself costs a few + * hundred paginated `searchChatMessages` calls. + */ + recoverDestIds: boolean; + /** Compact description of the scope, for the batch log line. */ + describe: string; +} + +export type ParsedBackfillPayload = + | { ok: true; plan: BackfillPlan } + | { ok: false; error: string }; + +/** + * Translate a restricted LIKE pattern into a Prisma filter. + * + * Only leading and/or trailing `%` are accepted — an interior `%`, a `_` + * wildcard, or a pattern with too little literal text is rejected rather than + * quietly matching far more than the caller meant. + */ +export function parseFileNameLike(pattern: string): FileNameFilter | { error: string } { + if (typeof pattern !== "string") return { error: "fileNameLike must be a string" }; + const raw = pattern.trim(); + if (raw.length === 0) return { error: "fileNameLike must not be empty" }; + + const leading = raw.startsWith("%"); + const trailing = raw.endsWith("%"); + const literal = raw.slice(leading ? 1 : 0, trailing && raw.length > 1 ? -1 : undefined); + + if (literal.includes("%")) { + return { error: "fileNameLike supports a leading and/or trailing % only (no interior wildcard)" }; + } + if (literal.includes("_")) { + return { error: "fileNameLike does not support the _ wildcard — use % or a literal name" }; + } + if (literal.length < MIN_FILENAME_LITERAL) { + return { + error: `fileNameLike needs at least ${MIN_FILENAME_LITERAL} literal characters (got "${literal}")`, + }; + } + + if (leading && trailing) return { contains: literal }; + if (leading) return { endsWith: literal }; + if (trailing) return { startsWith: literal }; + return { equals: literal }; +} + +function parsePackageIds(value: unknown): string[] | { error: string } { + if (!Array.isArray(value)) return { error: "packageIds must be an array of package ids" }; + if (value.length === 0) return { error: "packageIds must not be empty" }; + if (value.length > MAX_BACKFILL_PACKAGE_IDS) { + return { error: `packageIds holds ${value.length} ids — the maximum is ${MAX_BACKFILL_PACKAGE_IDS}` }; + } + const ids: string[] = []; + const seen = new Set(); + for (const entry of value) { + if (typeof entry !== "string") return { error: "packageIds must contain only strings" }; + const id = entry.trim(); + // cuid()s are alphanumeric; the character class also rules out anything + // that could confuse a log line or a hand-written SQL check. + if (!/^[A-Za-z0-9_-]{1,64}$/.test(id)) { + return { error: `packageIds contains an id that is not a valid identifier: "${entry}"` }; + } + if (seen.has(id)) continue; + seen.add(id); + ids.push(id); + } + return ids; +} + +function describeSelector(selector: BackfillSelector, plan: Pick): string { + const bits: string[] = []; + if (selector.packageIds) bits.push(`packageIds=${selector.packageIds.length}`); + if (selector.fileName) { + const [key, value] = Object.entries(selector.fileName)[0]; + bits.push(`fileName.${key}=${JSON.stringify(value)}`); + } + bits.push(`archiveType=${selector.archiveType ?? "ZIP|RAR|SEVEN_Z"}`); + bits.push(`limit=${plan.limit}`); + if (plan.rangedOnly) bits.push("rangedOnly"); + if (plan.recoverDestIds) bits.push("recoverDestIds"); + return bits.join(" "); +} + +/** + * Parse and validate a `backfill_filelists` payload. + * + * Accepted fields (all optional except that *some* selector is required): + * packageIds string[] — repair exactly these packages + * fileNameLike string — restricted LIKE: "%.z01", "Pack%", "%dragon%" + * archiveType ZIP|RAR|SEVEN_Z + * limit number — 1..MAX_BACKFILL_LIMIT, default DEFAULT_BACKFILL_LIMIT + * rangedOnly boolean — refuse the full-download fallback + * recoverDestIds boolean — allow one destination-channel scan to recover ids + * allowBroadSweep boolean — required to run with no narrowing selector + */ +export function parseBackfillPayload(payloadJson: string): ParsedBackfillPayload { + let parsed: unknown; + try { + parsed = JSON.parse(payloadJson === "" ? "{}" : payloadJson); + } catch { + return { ok: false, error: "payload is not valid JSON" }; + } + if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) { + return { ok: false, error: "payload must be a JSON object" }; + } + const raw = parsed as Record; + + const known = new Set([ + "packageIds", + "fileNameLike", + "archiveType", + "limit", + "rangedOnly", + "recoverDestIds", + "allowBroadSweep", + ]); + const unknownKeys = Object.keys(raw).filter((k) => !known.has(k)); + if (unknownKeys.length > 0) { + // Fail rather than ignore: a typo'd `packageIDs` would otherwise silently + // become "no selector" and — worse, if allowBroadSweep were also set — + // a full sweep. + return { ok: false, error: `unknown field(s): ${unknownKeys.join(", ")}` }; + } + + let limit = DEFAULT_BACKFILL_LIMIT; + if (raw.limit !== undefined) { + if (typeof raw.limit !== "number" || !Number.isInteger(raw.limit) || raw.limit < 1) { + return { ok: false, error: "limit must be a positive integer" }; + } + if (raw.limit > MAX_BACKFILL_LIMIT) { + return { ok: false, error: `limit ${raw.limit} exceeds the maximum of ${MAX_BACKFILL_LIMIT}` }; + } + limit = raw.limit; + } + + for (const flag of ["rangedOnly", "recoverDestIds", "allowBroadSweep"] as const) { + if (raw[flag] !== undefined && typeof raw[flag] !== "boolean") { + return { ok: false, error: `${flag} must be a boolean` }; + } + } + const rangedOnly = raw.rangedOnly === true; + const recoverDestIds = raw.recoverDestIds === true; + const allowBroadSweep = raw.allowBroadSweep === true; + + const selector: BackfillSelector = {}; + + if (raw.packageIds !== undefined) { + const ids = parsePackageIds(raw.packageIds); + if (!Array.isArray(ids)) return { ok: false, error: ids.error }; + selector.packageIds = ids; + } + + if (raw.fileNameLike !== undefined) { + const filter = parseFileNameLike(raw.fileNameLike as string); + if ("error" in filter) return { ok: false, error: filter.error }; + selector.fileName = filter; + } + + if (raw.archiveType !== undefined) { + if (typeof raw.archiveType !== "string" || !ARCHIVE_TYPES.includes(raw.archiveType as BackfillArchiveType)) { + return { ok: false, error: `archiveType must be one of ${ARCHIVE_TYPES.join(", ")}` }; + } + selector.archiveType = raw.archiveType as BackfillArchiveType; + } + + const isNarrowed = selector.packageIds !== undefined || selector.fileName !== undefined; + if (!isNarrowed && !allowBroadSweep) { + return { + ok: false, + error: + "refusing an unbounded backfill: pass packageIds or fileNameLike to scope it, " + + 'or set {"allowBroadSweep":true} to deliberately sweep every empty package', + }; + } + if (allowBroadSweep && isNarrowed) { + return { ok: false, error: "allowBroadSweep is for unscoped sweeps only — drop it or drop the selector" }; + } + + return { + ok: true, + plan: { + selector, + limit, + rangedOnly, + recoverDestIds, + describe: describeSelector(selector, { limit, rangedOnly, recoverDestIds }), + }, + }; +} diff --git a/worker/src/backfill.ts b/worker/src/backfill.ts index fd14595..dba0b9e 100644 --- a/worker/src/backfill.ts +++ b/worker/src/backfill.ts @@ -1,79 +1,71 @@ import path from "path"; +import type { Client } from "tdl"; import { mkdir, rm } from "fs/promises"; +import type { Prisma } from "@prisma/client"; import { db } from "./db/client.js"; import { config } from "./util/config.js"; import { childLogger } from "./util/logger.js"; import { withTdlibMutex } from "./util/mutex.js"; import { createTdlibClient, closeTdlibClient } from "./tdlib/client.js"; -import { downloadFile } from "./tdlib/download.js"; +import { downloadFile, invokeWithTimeout } from "./tdlib/download.js"; +import { scanChatDocuments, type ChatDocument } from "./tdlib/chat-documents.js"; import { getActiveAccounts } from "./db/queries.js"; import { readZipCentralDirectory } from "./archive/zip-reader.js"; import { readRarContents } from "./archive/rar-reader.js"; import { read7zContents } from "./archive/sevenz-reader.js"; import { extractSlicerTags } from "./archive/slicer-tags.js"; +import { readScannedListingRanged } from "./archive/ranged/dispatch.js"; +import type { RangedPart } from "./archive/ranged/sevenz-ranged.js"; +import { planListingRead, planRangedFallback } from "./archive/listing-plan.js"; +import { buildDestIndex, resolveDestPartSet, type DestIndex } from "./dest-index.js"; +import { parseBackfillPayload, type BackfillPlan } from "./backfill-scope.js"; import type { FileEntry } from "./archive/zip-reader.js"; const log = childLogger("backfill"); /** - * Re-extract file listings for Packages whose fileCount is 0 — usually - * caused by historical bugs in the archive readers (e.g. the RAR parser - * that silently returned [] for every archive before 0bdd4ba). + * Re-extract file listings for Packages whose fileCount is 0 — historically a + * reader bug (the RAR parser that silently returned [] before 0bdd4ba; the + * spanned-ZIP offset bug before 402c317) rather than a genuinely empty archive. * - * For each candidate Package: - * 1. Download all destMessageIds from the destination channel - * 2. Run the appropriate reader (ZIP / RAR / 7Z) on the assembled files - * 3. Insert PackageFile rows + update Package.fileCount - * 4. Clean up the temp files + * Two things dominate the cost of doing this, and both are handled here: * - * Triggered via pg_notify "backfill_filelists" with optional payload - * `{"limit": N, "archiveType": "RAR"}` — both fields optional, defaults - * are limit=100, archiveType=any. + * 1. **Reading the listing.** A file list lives in a few tens of kilobytes of + * an archive's header or tail. `readScannedListingRanged` fetches exactly + * those bytes, so a 35GB spanned set costs ~64KB instead of 35GB. The + * full-download path remains as a fallback for archives ranged reading + * genuinely cannot handle — and `rangedOnly` turns it off when the + * difference would be terabytes. + * + * 2. **Knowing which messages the archive's parts are.** A Package whose + * `destMessageIds` array is empty falls back to `[destMessageId]`, which is + * the *first* uploaded part. A lone `.z01` has no central directory at all, + * so that package can never be listed. With `recoverDestIds` the batch pays + * for one destination-channel scan and recovers the complete, ordered part + * set for every candidate at once — then persists it, so the package stays + * repairable. + * + * Triggered via pg_notify "backfill_filelists"; see `backfill-scope.ts` for the + * payload contract. A payload with no narrowing selector is rejected — the + * unscoped sweep has to ask for itself. */ export async function processBackfillRequest(payloadJson: string): Promise { - let limit = 100; - let archiveTypeFilter: "ZIP" | "RAR" | "SEVEN_Z" | undefined; - try { - const parsed = JSON.parse(payloadJson) as { limit?: number; archiveType?: string }; - if (typeof parsed.limit === "number" && parsed.limit > 0) limit = parsed.limit; - if (parsed.archiveType === "ZIP" || parsed.archiveType === "RAR" || parsed.archiveType === "SEVEN_Z") { - archiveTypeFilter = parsed.archiveType; - } - } catch { - // Empty / invalid payload — use defaults + const parsed = parseBackfillPayload(payloadJson); + if (!parsed.ok) { + log.warn({ payload: payloadJson, error: parsed.error }, "Backfill request rejected — nothing was read or written"); + return; } + const plan = parsed.plan; - const candidates = await db.package.findMany({ - where: { - fileCount: 0, - destChannelId: { not: null }, - destMessageId: { not: null }, - archiveType: archiveTypeFilter - ? archiveTypeFilter - : { in: ["ZIP", "RAR", "SEVEN_Z"] }, - }, - select: { - id: true, - fileName: true, - fileSize: true, - archiveType: true, - destChannelId: true, - destMessageId: true, - destMessageIds: true, - isMultipart: true, - partCount: true, - }, - orderBy: { createdAt: "asc" }, - take: limit, - }); - + const candidates = await findBackfillCandidates(plan); if (candidates.length === 0) { - log.info({ archiveTypeFilter }, "Backfill: no candidates with fileCount=0"); + log.info({ scope: plan.describe }, "Backfill: no candidates with fileCount=0"); return; } + const totalBytes = candidates.reduce((sum, c) => sum + c.fileSize, 0n); log.info( - { count: candidates.length, archiveTypeFilter }, + { count: candidates.length, scope: plan.describe, totalBytes: totalBytes.toString() }, "Backfill: starting batch" ); @@ -101,33 +93,47 @@ export async function processBackfillRequest(payloadJson: string): Promise // May already be loaded } - let processed = 0; - let succeeded = 0; - let failed = 0; + const destIndexes = await buildDestIndexes(client, plan, candidates); + + const counters: BackfillCounters = { + processed: 0, + listedRanged: 0, + listedDownload: 0, + skipped: 0, + failed: 0, + idsRecovered: 0, + concatUnlistable: 0, + }; for (const pkg of candidates) { - processed++; + counters.processed++; const ctx = { packageId: pkg.id, fileName: pkg.fileName }; try { - await processOnePackage(client, pkg, ctx); - succeeded++; + await repairOnePackage(client, pkg, ctx, plan, destIndexes, counters); } catch (err) { - failed++; + counters.failed++; log.warn({ err, ...ctx }, "Backfill failed for package"); } } - log.info( - { processed, succeeded, failed, archiveTypeFilter }, - "Backfill batch complete" - ); + log.info({ ...counters, scope: plan.describe }, "Backfill batch complete"); } finally { await closeTdlibClient(client).catch(() => {}); } }); } +interface BackfillCounters { + processed: number; + listedRanged: number; + listedDownload: number; + skipped: number; + failed: number; + idsRecovered: number; + concatUnlistable: number; +} + interface BackfillPackage { id: string; fileName: string; @@ -140,143 +146,368 @@ interface BackfillPackage { partCount: number; } -async function processOnePackage( - // eslint-disable-next-line @typescript-eslint/no-explicit-any - client: any, +/** Build the `where` clause from a validated plan. */ +export function backfillCandidateWhere(plan: BackfillPlan): Prisma.PackageWhereInput { + const { selector } = plan; + return { + fileCount: 0, + destChannelId: { not: null }, + destMessageId: { not: null }, + archiveType: selector.archiveType ?? { in: ["ZIP", "RAR", "SEVEN_Z"] }, + ...(selector.packageIds ? { id: { in: selector.packageIds } } : {}), + ...(selector.fileName ? { fileName: selector.fileName } : {}), + }; +} + +async function findBackfillCandidates(plan: BackfillPlan): Promise { + return db.package.findMany({ + where: backfillCandidateWhere(plan), + select: { + id: true, + fileName: true, + fileSize: true, + archiveType: true, + destChannelId: true, + destMessageId: true, + destMessageIds: true, + isMultipart: true, + partCount: true, + }, + orderBy: { createdAt: "asc" }, + take: plan.limit, + }); +} + +/** + * Scan each destination channel that has candidates needing id recovery, once. + * + * The scan is the expensive part of a repair run (a few hundred paginated + * `searchChatMessages` calls on a large channel), so it is opt-in via + * `recoverDestIds` and it is amortised across the whole batch. Its by-product — + * a fileId and size for every destination document — also removes the need for + * a per-part `getMessage` on every candidate, recovered or not. + */ +async function buildDestIndexes( + client: Client, + plan: BackfillPlan, + candidates: BackfillPackage[] +): Promise> { + const indexes = new Map(); + if (!plan.recoverDestIds) return indexes; + + const channelIds = new Set(); + for (const pkg of candidates) { + if (pkg.destChannelId && pkg.destMessageIds.length === 0) channelIds.add(pkg.destChannelId); + } + if (channelIds.size === 0) { + log.info("Backfill: every candidate already has destMessageIds — skipping the destination scan"); + return indexes; + } + + for (const channelId of channelIds) { + const channel = await db.telegramChannel.findUnique({ + where: { id: channelId }, + select: { telegramId: true, title: true }, + }); + if (!channel) { + log.warn({ channelId }, "Backfill: destination channel not found in DB — cannot recover ids for it"); + continue; + } + log.info({ channelId, title: channel.title }, "Backfill: scanning destination channel to recover destMessageIds"); + const scan = await scanChatDocuments(client, channel.telegramId); + if (scan.truncated) { + log.warn( + { channelId, pages: scan.pages }, + "Backfill: destination scan hit the page limit — recovery may be incomplete for older packages" + ); + } + indexes.set(channelId, buildDestIndex(scan.documents)); + } + return indexes; +} + +/** Resolved destination parts plus how they were obtained. */ +interface ResolvedDestParts { + parts: RangedPart[]; + /** Message ids in upload order, when a complete set was recovered from a scan. */ + recoveredIds: bigint[] | null; +} + +/** + * Turn a Package's destination messages into ranged parts (fileId + size + name), + * in upload order. Prefers the scan index (free) over `getMessage` (one API call + * per part), and recovers the full set from the anchor message when + * `destMessageIds` is empty. + */ +async function resolveDestParts( + client: Client, pkg: BackfillPackage, - ctx: { packageId: string; fileName: string } + chatTelegramId: bigint, + index: DestIndex | undefined +): Promise { + const toRangedPart = (doc: ChatDocument): RangedPart => ({ + fileId: doc.fileId, + fileSize: doc.fileSize, + fileName: doc.fileName, + }); + + if (pkg.destMessageIds.length > 0) { + const parts: RangedPart[] = []; + for (const msgId of pkg.destMessageIds) { + const cached = index?.byMessageId.get(msgId.toString()); + if (cached) { + parts.push(toRangedPart(cached)); + continue; + } + const resolved = await fetchDocumentPart(client, chatTelegramId, msgId, pkg); + if ("error" in resolved) return resolved; + parts.push(resolved.part); + } + return { parts, recoveredIds: null }; + } + + if (!pkg.destMessageId) return { error: "package has no destination message id" }; + + // A single-part package needs no recovery: its one message id is complete. + if (pkg.partCount <= 1 && !pkg.isMultipart) { + const cached = index?.byMessageId.get(pkg.destMessageId.toString()); + if (cached) return { parts: [toRangedPart(cached)], recoveredIds: null }; + const resolved = await fetchDocumentPart(client, chatTelegramId, pkg.destMessageId, pkg); + if ("error" in resolved) return resolved; + return { parts: [resolved.part], recoveredIds: null }; + } + + if (!index) { + return { + error: + `destMessageIds is empty and the package has ${pkg.partCount} parts; ` + + "destMessageId alone is only the first part. Re-run with recoverDestIds to scan the destination channel", + }; + } + + const resolution = resolveDestPartSet(index, pkg.destMessageId, pkg.partCount); + if (!resolution.ok) return { error: resolution.reason }; + + return { + parts: resolution.parts.map(toRangedPart), + recoveredIds: resolution.parts.map((p) => p.id), + }; +} + +async function fetchDocumentPart( + client: Client, + chatTelegramId: bigint, + messageId: bigint, + pkg: BackfillPackage +): Promise<{ part: RangedPart } | { error: string }> { + const message = await invokeWithTimeout<{ + content?: { document?: { file_name?: string; document?: { id: number; size: number } } }; + }>(client, { + _: "getMessage", + chat_id: Number(chatTelegramId), + message_id: Number(messageId), + }); + const doc = message?.content?.document; + if (!doc?.document?.id) { + return { error: `destination message ${messageId} has no document` }; + } + return { + part: { + fileId: String(doc.document.id), + fileSize: BigInt(doc.document.size), + fileName: doc.file_name ?? pkg.fileName, + }, + }; +} + +async function repairOnePackage( + client: Client, + pkg: BackfillPackage, + ctx: { packageId: string; fileName: string }, + plan: BackfillPlan, + destIndexes: Map, + counters: BackfillCounters ): Promise { if (!pkg.destChannelId || !pkg.destMessageId) { - log.debug(ctx, "Skipping: no destination channel/message"); + counters.skipped++; + log.info({ ...ctx, reason: "no destination channel/message" }, "Backfill skipped"); return; } - // Look up the destination channel's Telegram ID const destChannel = await db.telegramChannel.findUnique({ where: { id: pkg.destChannelId }, select: { telegramId: true }, }); - if (!destChannel) { - throw new Error("Destination channel not found in DB"); + if (!destChannel) throw new Error("Destination channel not found in DB"); + + const index = destIndexes.get(pkg.destChannelId); + const resolved = await resolveDestParts(client, pkg, destChannel.telegramId, index); + if ("error" in resolved) { + counters.skipped++; + log.info({ ...ctx, reason: resolved.error }, "Backfill skipped"); + return; } - const chatId = Number(destChannel.telegramId); + const { parts, recoveredIds } = resolved; - // Resolve which message IDs to download. The Package may carry a - // single destMessageId or multiple destMessageIds (for multipart). - const messageIds: bigint[] = - pkg.destMessageIds.length > 0 - ? pkg.destMessageIds - : pkg.destMessageId - ? [pkg.destMessageId] - : []; - - if (messageIds.length === 0) { - throw new Error("Package has no destination message IDs"); + // Persist a recovered part set before attempting the read: the ids are correct + // regardless of whether the listing turns out to be readable, and recording + // them is what makes the package repairable on any later attempt (and lets the + // bot deliver every part rather than just the first). + if (recoveredIds) { + const written = await persistRecoveredDestIds(pkg.id, recoveredIds); + if (written) { + counters.idsRecovered++; + log.info({ ...ctx, destMessageIds: recoveredIds.map(Number) }, "Recovered destination message ids"); + } } + const route = planListingRead({ + archiveType: pkg.archiveType, + sourceFileName: pkg.fileName, + destFileNames: parts.map((p) => p.fileName), + totalSize: parts.reduce((sum, p) => sum + p.fileSize, 0n), + maxDownloadBytes: BigInt(config.maxZipSizeMB) * 1024n * 1024n, + rangedOnly: plan.rangedOnly, + }); + + if (route.route === "skip") { + counters.skipped++; + if (/concat/.test(route.reason)) counters.concatUnlistable++; + log.info( + { ...ctx, reason: route.reason, destFileNames: parts.map((p) => p.fileName) }, + "Backfill skipped — destination copy cannot be listed" + ); + return; + } + + // ── Cheap path: read only the bytes that hold the listing ── + log.info( + { ...ctx, parts: parts.length, reason: route.reason }, + "Backfill reading listing via RANGED read (no full download)" + ); + let entries = await readScannedListingRanged(pkg.archiveType, client, parts); + let pathTaken: "ranged" | "download" = "ranged"; + + if (!entries || entries.length === 0) { + const totalSize = parts.reduce((sum, p) => sum + p.fileSize, 0n); + const fallback = planRangedFallback({ + totalSize, + maxDownloadBytes: BigInt(config.maxZipSizeMB) * 1024n * 1024n, + rangedOnly: plan.rangedOnly, + }); + if (fallback.route === "skip") { + counters.skipped++; + log.info({ ...ctx, reason: fallback.reason }, "Backfill skipped after ranged read returned nothing"); + return; + } + log.warn( + { ...ctx, reason: fallback.reason, bytes: totalSize.toString() }, + "Backfill falling back to FULL DOWNLOAD" + ); + entries = await downloadAndRead(client, pkg, parts); + pathTaken = "download"; + } + + if (!entries || entries.length === 0) { + counters.skipped++; + log.warn({ ...ctx, pathTaken }, "Reader returned 0 entries — archive may be encrypted or corrupt"); + return; + } + + await writeListing(pkg, entries, ctx); + if (pathTaken === "ranged") counters.listedRanged++; + else counters.listedDownload++; + log.info({ ...ctx, fileCount: entries.length, pathTaken }, "Backfilled file list"); +} + +/** + * Write `destMessageIds` for a package that had none. Guarded on the array still + * being empty so a value written concurrently — by another worker, or by hand — + * is never clobbered. Returns whether a row was actually updated. + */ +async function persistRecoveredDestIds(packageId: string, ids: bigint[]): Promise { + const result = await db.package.updateMany({ + where: { id: packageId, destMessageIds: { isEmpty: true } }, + data: { destMessageIds: ids }, + }); + return result.count > 0; +} + +/** The original full-download path, kept as the fallback for un-ranged archives. */ +async function downloadAndRead( + client: Client, + pkg: BackfillPackage, + parts: RangedPart[] +): Promise { const tempDir = path.join(config.tempDir, `backfill_${pkg.id}`); await mkdir(tempDir, { recursive: true }); - try { const partPaths: string[] = []; - - for (let i = 0; i < messageIds.length; i++) { - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const message = (await client.invoke({ - _: "getMessage", - chat_id: chatId, - message_id: Number(messageIds[i]), - })) as unknown as { - content?: { document?: { file_name?: string; document?: { id: number; size: number } } }; - }; - - const doc = message?.content?.document; - if (!doc?.document?.id) { - throw new Error(`Destination message ${messageIds[i]} has no document`); - } - - const fileId = String(doc.document.id); - const fileName = doc.file_name ?? `${pkg.id}.part${i + 1}`; - const localPath = path.join(tempDir, fileName); - - await downloadFile( - client, - fileId, - localPath, - BigInt(doc.document.size), - fileName - ); - + for (let i = 0; i < parts.length; i++) { + const part = parts[i]; + const localPath = path.join(tempDir, part.fileName || `${pkg.id}.part${i + 1}`); + await downloadFile(client, part.fileId, localPath, part.fileSize, part.fileName); partPaths.push(localPath); } - // Run the appropriate reader on the assembled file(s) - let entries: FileEntry[] = []; - if (pkg.archiveType === "ZIP") { - entries = await readZipCentralDirectory(partPaths); - } else if (pkg.archiveType === "RAR") { - // unrar auto-discovers sibling parts when in the same directory - entries = await readRarContents(partPaths[0]); - } else if (pkg.archiveType === "SEVEN_Z") { - entries = await read7zContents(partPaths[0]); - } else { - log.debug({ ...ctx, archiveType: pkg.archiveType }, "Skipping unsupported archive type"); - return; - } - - if (entries.length === 0) { - log.warn(ctx, "Reader returned 0 entries — archive may be encrypted or corrupt"); - return; - } - - // Also derive slicer tags from the file list so the backfilled packages - // gain the same search/filter context as newly-ingested ones. - const slicerTags = extractSlicerTags(entries); - - // Write everything in a single transaction so a partial backfill never - // leaves the Package half-indexed. - await db.$transaction(async (tx) => { - // Re-check fileCount inside the transaction: another worker might - // have backfilled this package between our read and write. - const current = await tx.package.findUnique({ - where: { id: pkg.id }, - select: { fileCount: true, tags: true }, - }); - if (current && current.fileCount > 0) { - log.debug({ ...ctx, existingFileCount: current.fileCount }, "Already backfilled by another worker — skipping"); - return; - } - - await tx.packageFile.deleteMany({ where: { packageId: pkg.id } }); - await tx.packageFile.createMany({ - data: entries.map((e) => ({ - packageId: pkg.id, - path: e.path, - fileName: e.fileName, - extension: e.extension, - compressedSize: e.compressedSize, - uncompressedSize: e.uncompressedSize, - crc32: e.crc32, - })), - }); - - // Merge slicer tags with whatever's already on the Package (preserve - // channel category, manual tags, etc.). - const existingTags = current?.tags ?? []; - const mergedTags = [...new Set([...existingTags, ...slicerTags])]; - - await tx.package.update({ - where: { id: pkg.id }, - data: { fileCount: entries.length, tags: mergedTags }, - }); - }); - - log.info({ ...ctx, fileCount: entries.length }, "Backfilled file list"); + if (pkg.archiveType === "ZIP") return readZipCentralDirectory(partPaths); + // unrar / 7z auto-discover sibling parts when in the same directory + if (pkg.archiveType === "RAR") return readRarContents(partPaths[0]); + if (pkg.archiveType === "SEVEN_Z") return read7zContents(partPaths[0]); + return []; } finally { await rm(tempDir, { recursive: true, force: true }).catch(() => {}); } } +async function writeListing( + pkg: BackfillPackage, + entries: FileEntry[], + ctx: { packageId: string; fileName: string } +): Promise { + // Also derive slicer tags from the file list so the backfilled packages + // gain the same search/filter context as newly-ingested ones. + const slicerTags = extractSlicerTags(entries); + + // Write everything in a single transaction so a partial backfill never + // leaves the Package half-indexed. + await db.$transaction(async (tx) => { + // Re-check fileCount inside the transaction: another worker might + // have backfilled this package between our read and write. + const current = await tx.package.findUnique({ + where: { id: pkg.id }, + select: { fileCount: true, tags: true }, + }); + if (current && current.fileCount > 0) { + log.debug({ ...ctx, existingFileCount: current.fileCount }, "Already backfilled by another worker — skipping"); + return; + } + + await tx.packageFile.deleteMany({ where: { packageId: pkg.id } }); + await tx.packageFile.createMany({ + data: entries.map((e) => ({ + packageId: pkg.id, + path: e.path, + fileName: e.fileName, + extension: e.extension, + compressedSize: e.compressedSize, + uncompressedSize: e.uncompressedSize, + crc32: e.crc32, + })), + }); + + // Merge slicer tags with whatever's already on the Package (preserve + // channel category, manual tags, etc.). + const existingTags = current?.tags ?? []; + const mergedTags = [...new Set([...existingTags, ...slicerTags])]; + + await tx.package.update({ + where: { id: pkg.id }, + data: { fileCount: entries.length, tags: mergedTags }, + }); + }); +} + /** * Cheap pure-DB backfill: walk Packages that already have PackageFile rows * but no slicer tags, recompute the tags from their extensions, and merge diff --git a/worker/src/dest-index.test.ts b/worker/src/dest-index.test.ts new file mode 100644 index 0000000..2324660 --- /dev/null +++ b/worker/src/dest-index.test.ts @@ -0,0 +1,111 @@ +import { describe, it, expect } from "vitest"; +import { buildDestIndex, resolveDestPartSet } from "./dest-index.js"; +import type { ChatDocument } from "./tdlib/chat-documents.js"; + +let nextId = 100; +function doc(fileName: string, opts: { id?: number; size?: number } = {}): ChatDocument { + const id = opts.id ?? nextId++; + return { + id: BigInt(id), + fileName, + fileId: `f${id}`, + fileSize: BigInt(opts.size ?? 1024), + date: new Date("2026-01-01T00:00:00Z"), + }; +} + +describe("buildDestIndex + resolveDestPartSet — spanned ZIP recovery", () => { + it("recovers the complete ordered volume set from the .z01 message alone", () => { + const z01 = doc("Pack.z01", { id: 10 }); + const z02 = doc("Pack.z02", { id: 11 }); + const zip = doc("Pack.zip", { id: 12 }); + const index = buildDestIndex([z01, z02, zip, doc("Unrelated.zip", { id: 13 })]); + + const resolved = resolveDestPartSet(index, 10n, 3); + expect(resolved.ok).toBe(true); + if (!resolved.ok) return; + // .z01, .z02, then the bare .zip as the FINAL volume — the order the + // EOCD-bearing tail read depends on. + expect(resolved.parts.map((p) => p.fileName)).toEqual(["Pack.z01", "Pack.z02", "Pack.zip"]); + expect(resolved.parts.map((p) => Number(p.id))).toEqual([10, 11, 12]); + expect(resolved.kind).toBe("archive-set"); + }); + + it("carries fileId and size through, so no getMessage call is needed per part", () => { + const index = buildDestIndex([ + doc("Pack.z01", { id: 10, size: 500 }), + doc("Pack.zip", { id: 11, size: 700 }), + ]); + const resolved = resolveDestPartSet(index, 10n, 2); + expect(resolved.ok).toBe(true); + if (!resolved.ok) return; + expect(resolved.parts.map((p) => p.fileId)).toEqual(["f10", "f11"]); + expect(resolved.parts.map((p) => Number(p.fileSize))).toEqual([500, 700]); + }); +}); + +describe("resolveDestPartSet — refusing to guess", () => { + it("reports a missing anchor message instead of inventing a set", () => { + const index = buildDestIndex([doc("Pack.z01", { id: 10 }), doc("Pack.zip", { id: 11 })]); + const resolved = resolveDestPartSet(index, 999n, 2); + expect(resolved).toMatchObject({ ok: false }); + if (resolved.ok) return; + expect(resolved.reason).toMatch(/was not found in the channel scan/); + }); + + it("refuses a set whose part count disagrees with the package", () => { + // Two uploads sharing a base name get merged by groupArchiveSets into one + // oversized set — writing that back would mix two archives together. + const index = buildDestIndex([ + doc("Pack.z01", { id: 10 }), + doc("Pack.z02", { id: 11 }), + doc("Pack.zip", { id: 12 }), + ]); + const resolved = resolveDestPartSet(index, 10n, 2); + expect(resolved).toMatchObject({ ok: false }); + if (resolved.ok) return; + expect(resolved.reason).toMatch(/refusing to write an incomplete or merged set/); + }); + + it("resolves a genuine single-part package from its one message", () => { + const index = buildDestIndex([doc("Solo.zip", { id: 20 })]); + const resolved = resolveDestPartSet(index, 20n, 1); + expect(resolved).toMatchObject({ ok: true }); + if (!resolved.ok) return; + expect(resolved.parts.map((p) => p.fileName)).toEqual(["Solo.zip"]); + }); +}); + +describe("buildDestIndex — .concat.NNN repacks", () => { + it("groups repack chunks that no archive pattern matches", () => { + // These names are invisible to archive/detect.ts, so without their own + // grouping a repacked package looks identical to one whose messages are gone. + const index = buildDestIndex([ + doc("Pack.concat.003", { id: 32 }), + doc("Pack.concat.001", { id: 30 }), + doc("Pack.concat.002", { id: 31 }), + ]); + const resolved = resolveDestPartSet(index, 30n, 3); + expect(resolved.ok).toBe(true); + if (!resolved.ok) return; + expect(resolved.kind).toBe("concat-repack"); + expect(resolved.parts.map((p) => p.fileName)).toEqual([ + "Pack.concat.001", + "Pack.concat.002", + "Pack.concat.003", + ]); + }); + + it("keeps two different repacks apart", () => { + const index = buildDestIndex([ + doc("A.concat.001", { id: 40 }), + doc("A.concat.002", { id: 41 }), + doc("B.concat.001", { id: 50 }), + doc("B.concat.002", { id: 51 }), + ]); + const a = resolveDestPartSet(index, 40n, 2); + expect(a).toMatchObject({ ok: true }); + if (!a.ok) return; + expect(a.parts.map((p) => Number(p.id))).toEqual([40, 41]); + }); +}); diff --git a/worker/src/dest-index.ts b/worker/src/dest-index.ts new file mode 100644 index 0000000..8f5314f --- /dev/null +++ b/worker/src/dest-index.ts @@ -0,0 +1,116 @@ +import { isArchiveAttachment } from "./archive/detect.js"; +import { groupArchiveSets } from "./archive/multipart.js"; +import { isConcatRepackName, concatRepackBase, concatChunkIndex } from "./archive/listing-plan.js"; +import type { ChatDocument } from "./tdlib/chat-documents.js"; + +/** + * An index over one destination-channel scan, used to recover the destination + * message ids of packages whose `destMessageIds` array was never populated. + * + * Two kinds of part set live in here: + * + * - `archive-set` — grouped by `groupArchiveSets`, i.e. the same grouping the + * ingestion path uses, so a `.z01 … .zip` spanned set or a + * `.zip.001 …` byte split comes back in upload order. + * - `concat-repack` — `.concat.NNN` chunks, which match no archive + * pattern at all and so are invisible to `groupArchiveSets`. + * They still need grouping: a package repacked this way has + * real destination messages worth recording even though its + * listing can never be read back. + */ +export type DestPartSetKind = "archive-set" | "concat-repack"; + +export interface DestPartSet { + kind: DestPartSetKind; + parts: ChatDocument[]; +} + +export interface DestIndex { + byMessageId: Map; + setByMessageId: Map; + documentCount: number; +} + +export function buildDestIndex(documents: ChatDocument[]): DestIndex { + const byMessageId = new Map(); + for (const doc of documents) byMessageId.set(doc.id.toString(), doc); + + const setByMessageId = new Map(); + + // Recognized archive names: reuse the ingestion grouping verbatim so the part + // order here matches the order the parts were uploaded in. + const archives = documents.filter((d) => isArchiveAttachment(d.fileName)); + for (const set of groupArchiveSets(archives)) { + if (set.parts.length === 0) continue; + const entry: DestPartSet = { kind: "archive-set", parts: set.parts }; + for (const part of set.parts) setByMessageId.set(part.id.toString(), entry); + } + + // `.concat.NNN` repack chunks, grouped by base and ordered by chunk number. + const concatGroups = new Map(); + for (const doc of documents) { + if (!isConcatRepackName(doc.fileName)) continue; + const key = concatRepackBase(doc.fileName); + const group = concatGroups.get(key) ?? []; + group.push(doc); + concatGroups.set(key, group); + } + for (const group of concatGroups.values()) { + group.sort((a, b) => concatChunkIndex(a.fileName) - concatChunkIndex(b.fileName)); + const entry: DestPartSet = { kind: "concat-repack", parts: group }; + for (const part of group) setByMessageId.set(part.id.toString(), entry); + } + + return { byMessageId, setByMessageId, documentCount: documents.length }; +} + +export type DestResolution = + | { ok: true; kind: DestPartSetKind; parts: ChatDocument[] } + | { ok: false; reason: string }; + +/** + * Resolve a package's full destination part set from one known message id. + * + * Refuses anything it cannot corroborate. In particular the recovered set must + * hold exactly `expectedPartCount` parts: the destination channel can legitimately + * contain two uploads sharing a base name (a re-post, a duplicate ingestion), and + * `groupArchiveSets` merges those into one oversized set. Writing that merged set + * back to `destMessageIds` would hand the bot a mix of two archives, so a count + * mismatch is reported and the row is left alone. + */ +export function resolveDestPartSet( + index: DestIndex, + anchorMessageId: bigint, + expectedPartCount: number +): DestResolution { + const key = anchorMessageId.toString(); + const anchor = index.byMessageId.get(key); + if (!anchor) { + return { + ok: false, + reason: `destination message ${key} was not found in the channel scan (deleted, or outside the scanned range)`, + }; + } + + const set = index.setByMessageId.get(key); + if (!set) { + if (expectedPartCount === 1) { + return { ok: true, kind: "archive-set", parts: [anchor] }; + } + return { + ok: false, + reason: `destination message ${key} ("${anchor.fileName}") matched no part set, but the package expects ${expectedPartCount} parts`, + }; + } + + if (set.parts.length !== expectedPartCount) { + return { + ok: false, + reason: + `resolved ${set.parts.length} destination part(s) for "${anchor.fileName}" but the package records ` + + `${expectedPartCount} — refusing to write an incomplete or merged set`, + }; + } + + return { ok: true, kind: set.kind, parts: set.parts }; +} diff --git a/worker/src/fetch-listener.ts b/worker/src/fetch-listener.ts index 2a9f9ff..bafecbb 100644 --- a/worker/src/fetch-listener.ts +++ b/worker/src/fetch-listener.ts @@ -538,12 +538,30 @@ function handleManualUpload(uploadId: string): void { // ── Backfill file-list handler ── // -// Trigger via: -// SELECT pg_notify('backfill_filelists', '{"limit":50,"archiveType":"RAR"}'); +// A request must say what it wants: a payload with no narrowing selector is +// rejected outright rather than sweeping every empty package in the catalogue. +// See `backfill-scope.ts` for the full contract. // -// Both fields are optional. archiveType filters to one of ZIP/RAR/SEVEN_Z. -// Default limit is 100. The handler queues so multiple notifications run -// sequentially (no concurrent TDLib downloads competing for the mutex). +// -- repair the ZIP-spec spanned sets (.z01 … .zip) that the pre-402c317 +// -- reader bug left with no file list. Ranged reads only (~64KB each, vs +// -- ~944GB of full downloads), and one destination scan to recover the +// -- destMessageIds that were never recorded. +// SELECT pg_notify('backfill_filelists', '{ +// "fileNameLike": "%.z01", +// "archiveType": "ZIP", +// "limit": 250, +// "rangedOnly": true, +// "recoverDestIds": true +// }'); +// +// -- repair a specific handful +// SELECT pg_notify('backfill_filelists', '{"packageIds":["ckxyz…","ckabc…"],"rangedOnly":true}'); +// +// -- the old broad sweep, now explicit about being one +// SELECT pg_notify('backfill_filelists', '{"archiveType":"RAR","limit":50,"allowBroadSweep":true}'); +// +// The handler queues so multiple notifications run sequentially (no concurrent +// TDLib downloads competing for the mutex). function handleBackfillFilelists(payload: string): void { fetchQueue = fetchQueue .then(() => processBackfillRequest(payload)) diff --git a/worker/src/rebuild.ts b/worker/src/rebuild.ts index f84748e..a35746f 100644 --- a/worker/src/rebuild.ts +++ b/worker/src/rebuild.ts @@ -1,8 +1,7 @@ import type { Client } from "tdl"; -import { config } from "./util/config.js"; import { childLogger } from "./util/logger.js"; import { createTdlibClient, closeTdlibClient } from "./tdlib/client.js"; -import { invokeWithTimeout, MAX_SCAN_PAGES } from "./tdlib/download.js"; +import { scanChatDocuments } from "./tdlib/chat-documents.js"; import { isArchiveAttachment } from "./archive/detect.js"; import { extractCreatorFromFileName } from "./archive/creator.js"; import { groupArchiveSets } from "./archive/multipart.js"; @@ -263,127 +262,38 @@ export async function rebuildPackageDatabase( } /** - * Scan the destination channel for document messages using searchChatMessages. - * Returns archive messages in chronological order (oldest first). + * Scan the destination channel and keep only the documents whose names + * `archive/detect.ts` recognizes. The paging itself lives in + * `tdlib/chat-documents.ts` and is shared with the file-list repair path. */ async function scanDestinationChannel( client: Client, chatId: bigint, onProgress?: (messagesScanned: number) => Promise ): Promise { + const scan = await scanChatDocuments(client, chatId, onProgress); + const archives: TelegramMessage[] = []; - let currentFromId = 0; - let totalScanned = 0; - let pageCount = 0; - let lastProgressUpdate = 0; - - // eslint-disable-next-line no-constant-condition - while (true) { - if (pageCount >= MAX_SCAN_PAGES) { - log.warn( - { chatId: chatId.toString(), pageCount, totalScanned }, - "Hit max page limit for destination scan, stopping" + for (const doc of scan.documents) { + if (isArchiveAttachment(doc.fileName)) { + archives.push(doc); + } else { + // Not matched by any pattern in archive/detect.ts, so it is dropped without a + // packages/skipped_packages row. Grep "unrecognized attachment" to find naming + // schemes we do not handle yet. + log.debug( + { chatId: chatId.toString(), messageId: Number(doc.id), fileName: doc.fileName }, + "Skipping unrecognized attachment (no archive/document pattern matched)" ); - break; } - pageCount++; - - const previousFromId = currentFromId; - - const result = await invokeWithTimeout<{ - messages?: { - id: number; - date: number; - content: { - _: string; - document?: { - file_name?: string; - document?: { - id: number; - size: number; - }; - }; - }; - }[]; - }>(client, { - _: "searchChatMessages", - chat_id: Number(chatId), - // No topic context for a flat destination scan. TDLib 1.8.64+ replaced - // `message_thread_id` / `saved_messages_topic_id` with a single - // optional `topic_id`; for a flat scan we just omit it. - query: "", - from_message_id: currentFromId, - offset: 0, - limit: 100, - filter: { _: "searchMessagesFilterDocument" }, - sender_id: null, - }); - - if (!result.messages || result.messages.length === 0) break; - - totalScanned += result.messages.length; - - for (const msg of result.messages) { - const doc = msg.content?.document; - if (doc?.file_name && doc.document && isArchiveAttachment(doc.file_name)) { - archives.push({ - id: BigInt(msg.id), - fileName: doc.file_name, - fileId: String(doc.document.id), - fileSize: BigInt(doc.document.size), - date: new Date(msg.date * 1000), - }); - } else if (doc?.file_name) { - // Not matched by any pattern in archive/detect.ts, so it is dropped without a - // packages/skipped_packages row. Grep "unrecognized attachment" to find naming - // schemes we do not handle yet. - log.debug( - { chatId: chatId.toString(), messageId: msg.id, fileName: doc.file_name }, - "Skipping unrecognized attachment (no archive/document pattern matched)" - ); - } - } - - // Throttle progress updates to every 2 seconds - const now = Date.now(); - if (onProgress && now - lastProgressUpdate >= 2000) { - lastProgressUpdate = now; - await onProgress(totalScanned); - } - - currentFromId = result.messages[result.messages.length - 1].id; - - // Stuck detection - if (currentFromId === previousFromId) { - log.warn( - { chatId: chatId.toString(), currentFromId, totalScanned }, - "Pagination stuck, breaking" - ); - break; - } - - if (result.messages.length < 100) break; - - await sleep(config.apiDelayMs); - } - - // Final progress update - if (onProgress) { - await onProgress(totalScanned); } log.info( - { - chatId: chatId.toString(), - archives: archives.length, - totalScanned, - pages: pageCount, - }, + { chatId: chatId.toString(), archives: archives.length, totalScanned: scan.totalScanned, pages: scan.pages }, "Destination channel scan complete" ); - // Reverse to chronological order (oldest first) - return archives.reverse(); + return archives; } /** @@ -414,7 +324,3 @@ async function updateRebuildProgress( } } } - -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} diff --git a/worker/src/tdlib/chat-documents.ts b/worker/src/tdlib/chat-documents.ts new file mode 100644 index 0000000..b0e573e --- /dev/null +++ b/worker/src/tdlib/chat-documents.ts @@ -0,0 +1,146 @@ +import type { Client } from "tdl"; +import { config } from "../util/config.js"; +import { childLogger } from "../util/logger.js"; +import { invokeWithTimeout, MAX_SCAN_PAGES } from "./download.js"; + +const log = childLogger("chat-documents"); + +/** One document message in a chat, with everything a ranged read needs. */ +export interface ChatDocument { + id: bigint; + fileName: string; + fileId: string; + fileSize: bigint; + date: Date; +} + +export interface ChatDocumentScan { + /** Every document message with a file name, oldest first. */ + documents: ChatDocument[]; + /** Total messages returned by the search, including ones without a document. */ + totalScanned: number; + pages: number; + /** True when the scan stopped on MAX_SCAN_PAGES rather than running out. */ + truncated: boolean; +} + +/** + * Page through every document message in a chat. + * + * `searchChatMessages` rather than `getChatHistory` because the destination + * channel may be a hidden-history supergroup, where history reads come back + * empty. + * + * Deliberately returns **all** documents and leaves filtering to the caller. + * The rebuild path wants only names `archive/detect.ts` recognizes; the repair + * path specifically needs the ones it does *not* — a `.concat.NNN` chunk + * matches no archive pattern, and if the scan dropped those the repair could + * not tell "this package was repacked into an unlistable concatenation" apart + * from "its destination messages are gone". + */ +export async function scanChatDocuments( + client: Client, + chatId: bigint, + onProgress?: (messagesScanned: number) => Promise | void +): Promise { + const documents: ChatDocument[] = []; + let currentFromId = 0; + let totalScanned = 0; + let pageCount = 0; + let lastProgressUpdate = 0; + let truncated = false; + + for (;;) { + if (pageCount >= MAX_SCAN_PAGES) { + log.warn( + { chatId: chatId.toString(), pageCount, totalScanned }, + "Hit max page limit for chat document scan, stopping" + ); + truncated = true; + break; + } + pageCount++; + + const previousFromId = currentFromId; + + const result = await invokeWithTimeout<{ + messages?: { + id: number; + date: number; + content: { + _: string; + document?: { + file_name?: string; + document?: { id: number; size: number }; + }; + }; + }[]; + }>(client, { + _: "searchChatMessages", + chat_id: Number(chatId), + // No topic context for a flat scan. TDLib 1.8.64+ replaced + // `message_thread_id` / `saved_messages_topic_id` with a single optional + // `topic_id`; for a flat scan we just omit it. + query: "", + from_message_id: currentFromId, + offset: 0, + limit: 100, + filter: { _: "searchMessagesFilterDocument" }, + sender_id: null, + }); + + if (!result.messages || result.messages.length === 0) break; + + totalScanned += result.messages.length; + + for (const msg of result.messages) { + const doc = msg.content?.document; + if (doc?.file_name && doc.document) { + documents.push({ + id: BigInt(msg.id), + fileName: doc.file_name, + fileId: String(doc.document.id), + fileSize: BigInt(doc.document.size), + date: new Date(msg.date * 1000), + }); + } + } + + // Throttle progress updates to every 2 seconds + const now = Date.now(); + if (onProgress && now - lastProgressUpdate >= 2000) { + lastProgressUpdate = now; + await onProgress(totalScanned); + } + + currentFromId = result.messages[result.messages.length - 1].id; + + // Stuck detection + if (currentFromId === previousFromId) { + log.warn( + { chatId: chatId.toString(), currentFromId, totalScanned }, + "Pagination stuck, breaking" + ); + break; + } + + if (result.messages.length < 100) break; + + await sleep(config.apiDelayMs); + } + + if (onProgress) await onProgress(totalScanned); + + log.info( + { chatId: chatId.toString(), documents: documents.length, totalScanned, pages: pageCount, truncated }, + "Chat document scan complete" + ); + + // Reverse to chronological order (oldest first) + documents.reverse(); + return { documents, totalScanned, pages: pageCount, truncated }; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +}