refactor(worker): promote ranged-listing dispatcher to a shared module

This commit is contained in:
2026-07-31 04:46:39 +02:00
parent eda882dc90
commit 26be615918
3 changed files with 78 additions and 46 deletions
@@ -0,0 +1,15 @@
import { describe, it, expect } from "vitest";
import { readScannedListingRanged } from "./dispatch.js";
describe("readScannedListingRanged", () => {
it("returns null for an unknown archive type without calling the reader", async () => {
const read = async () => Buffer.alloc(0);
const result = await readScannedListingRanged(
"DOCUMENT",
{ invoke: async () => ({}) } as never,
[{ fileId: "1", fileSize: 100n, fileName: "a.pdf" }],
);
expect(result).toBeNull();
void read; // unused placeholder kept out of the dispatch call — DOCUMENT never reaches a reader
});
});
+62
View File
@@ -0,0 +1,62 @@
import type { Client } from "tdl";
import { downloadFileRange } from "../../tdlib/range-download.js";
import { parseZipCentralDirectoryFromTail, MIN_ZIP_TAIL_BYTES } from "../central-directory.js";
import { childLogger } from "../../util/logger.js";
import type { FileEntry } from "../zip-reader.js";
import { readSevenZListingRanged, type RangedPart } from "./sevenz-ranged.js";
import { readRarListingRanged } from "./rar-ranged.js";
import { tdlibRangeReader } from "./range-reader.js";
const log = childLogger("ranged-dispatch");
/**
* Read a ZIP central directory from the tail of a (possibly multipart)
* archive. `parts` is ordered; only the LAST part carries the EOCD record.
* `fileSize` on each part is that part's own size (NOT the whole-archive
* total) so the download offset stays within that part's bounds, while
* `tailStart` passed to the parser is the logical whole-archive offset
* (preceding parts' sizes + the offset within the last part).
*/
export async function readScannedZipListing(
client: Client,
parts: { fileId: string; fileSize: bigint }[],
): Promise<FileEntry[] | null> {
if (parts.length === 0) return null;
const lastPart = parts[parts.length - 1];
const precedingSize = parts.slice(0, -1).reduce((sum, p) => sum + Number(p.fileSize), 0);
const lastSize = Number(lastPart.fileSize);
for (const tailBytes of [MIN_ZIP_TAIL_BYTES, MIN_ZIP_TAIL_BYTES * 4]) {
const partOffset = Math.max(0, lastSize - tailBytes);
const downloadLen = Math.min(tailBytes, lastSize);
try {
const buf = await downloadFileRange(client, lastPart.fileId, partOffset, downloadLen, lastPart.fileSize);
const tailStart = precedingSize + partOffset;
return parseZipCentralDirectoryFromTail(buf, tailStart);
} catch (err) {
if (err instanceof RangeError) continue; // try a larger tail
log.warn({ err, fileId: lastPart.fileId }, "ranged ZIP listing failed");
return null;
}
}
return null;
}
/**
* Dispatch a (no-download) inner-file listing read by archive type. Used both
* by the provenance-backfill path (reading an already-uploaded copy) and the
* forward-priority ingestion path (reading the source channel's copy before
* any download/forward decision is made) — the read itself only needs
* {fileId, fileSize, fileName}, so it doesn't matter which channel the file
* currently lives in.
*/
export async function readScannedListingRanged(
archiveType: string,
client: Client,
parts: RangedPart[],
): Promise<FileEntry[] | null> {
const read = tdlibRangeReader(client);
if (archiveType === "ZIP") return readScannedZipListing(client, parts);
if (archiveType === "SEVEN_Z") return readSevenZListingRanged(parts, read);
if (archiveType === "RAR") return readRarListingRanged(parts, read);
return null;
}
+1 -46
View File
@@ -1,8 +1,6 @@
import { db } from "./db/client.js"; import { db } from "./db/client.js";
import { childLogger } from "./util/logger.js"; import { childLogger } from "./util/logger.js";
import { downloadFileRange } from "./tdlib/range-download.js";
import { invokeWithTimeout } from "./tdlib/download.js"; import { invokeWithTimeout } from "./tdlib/download.js";
import { parseZipCentralDirectoryFromTail, MIN_ZIP_TAIL_BYTES } from "./archive/central-directory.js";
import { fingerprintsMatch, crcFingerprint } from "./archive/fingerprint.js"; import { fingerprintsMatch, crcFingerprint } from "./archive/fingerprint.js";
import { import {
findPlaceholderCandidates, findPlaceholderCandidates,
@@ -15,6 +13,7 @@ import { readSevenZListingRanged, type RangedPart } from "./archive/ranged/seven
import { readRarListingRanged } from "./archive/ranged/rar-ranged.js"; import { readRarListingRanged } from "./archive/ranged/rar-ranged.js";
import { tdlibRangeReader } from "./archive/ranged/range-reader.js"; import { tdlibRangeReader } from "./archive/ranged/range-reader.js";
import { fullDownloadListing } from "./archive/ranged/fallback.js"; import { fullDownloadListing } from "./archive/ranged/fallback.js";
import { readScannedZipListing, readScannedListingRanged } from "./archive/ranged/dispatch.js";
import type { Client } from "tdl"; import type { Client } from "tdl";
const log = childLogger("provenance-backfill"); const log = childLogger("provenance-backfill");
@@ -36,38 +35,6 @@ export interface BackfillArgs {
previewMsgId?: bigint | null; previewMsgId?: bigint | null;
} }
/**
* Read a ZIP central directory from the tail of a (possibly multipart)
* archive. `parts` is ordered; only the LAST part carries the EOCD record.
* `fileSize` on each part is that part's own size (NOT the whole-archive
* total) so the download offset stays within that part's bounds, while
* `tailStart` passed to the parser is the logical whole-archive offset
* (preceding parts' sizes + the offset within the last part).
*/
async function readScannedZipListing(
client: Client,
parts: { fileId: string; fileSize: bigint }[],
): Promise<FileEntry[] | null> {
if (parts.length === 0) return null;
const lastPart = parts[parts.length - 1];
const precedingSize = parts.slice(0, -1).reduce((sum, p) => sum + Number(p.fileSize), 0);
const lastSize = Number(lastPart.fileSize);
for (const tailBytes of [MIN_ZIP_TAIL_BYTES, MIN_ZIP_TAIL_BYTES * 4]) {
const partOffset = Math.max(0, lastSize - tailBytes);
const downloadLen = Math.min(tailBytes, lastSize);
try {
const buf = await downloadFileRange(client, lastPart.fileId, partOffset, downloadLen, lastPart.fileSize);
const tailStart = precedingSize + partOffset;
return parseZipCentralDirectoryFromTail(buf, tailStart);
} catch (err) {
if (err instanceof RangeError) continue; // try a larger tail
log.warn({ err, fileId: lastPart.fileId }, "ranged ZIP listing failed");
return null;
}
}
return null;
}
/** /**
* Resolve the destination copy's message(s) into ranged parts (file id + * Resolve the destination copy's message(s) into ranged parts (file id +
* size + name), in order, so a multipart destination copy is reconstructed * size + name), in order, so a multipart destination copy is reconstructed
@@ -106,18 +73,6 @@ async function resolveDestParts(
} }
} }
async function readScannedListingRanged(
archiveType: string,
client: Client,
parts: RangedPart[],
): Promise<FileEntry[] | null> {
const read = tdlibRangeReader(client);
if (archiveType === "ZIP") return readScannedZipListing(client, parts);
if (archiveType === "SEVEN_Z") return readSevenZListingRanged(parts, read);
if (archiveType === "RAR") return readRarListingRanged(parts, read);
return null;
}
/** /**
* Build the CRC fingerprint entries for a placeholder candidate: start from * Build the CRC fingerprint entries for a placeholder candidate: start from
* its stored PackageFile CRCs, and if those are incomplete (e.g. a rebuild * its stored PackageFile CRCs, and if those are incomplete (e.g. a rebuild