diff --git a/worker/src/db/queries.ts b/worker/src/db/queries.ts index a899336..3676648 100644 --- a/worker/src/db/queries.ts +++ b/worker/src/db/queries.ts @@ -1022,15 +1022,18 @@ export interface PlaceholderCandidate { destChannel: { telegramId: bigint } | null; } -export async function findPlaceholderCandidate( +/** + * Find every placeholder Package matching name+size (oldest first). Package + * has no direct `destChannel` relation (only the scalar `destChannelId`), so + * each row's destination TelegramChannel telegramId is resolved with a + * follow-up lookup rather than a Prisma include. + */ +export async function findPlaceholderCandidates( destChannelId: string, fileName: string, fileSize: bigint, -): Promise { - // Package has no direct `destChannel` relation (only the scalar - // `destChannelId`), so resolve the destination TelegramChannel's - // telegramId with a follow-up lookup rather than a Prisma include. - const row = await db.package.findFirst({ +): Promise { + const rows = await db.package.findMany({ where: { fileName, fileSize, @@ -1048,22 +1051,36 @@ export async function findPlaceholderCandidate( }, orderBy: { indexedAt: "asc" }, }); - if (!row) return null; - const destChannel = row.destChannelId - ? await db.telegramChannel.findUnique({ - where: { id: row.destChannelId }, - select: { telegramId: true }, + if (rows.length === 0) return []; + + const destChannelIds = [...new Set(rows.map((r) => r.destChannelId).filter((id): id is string => !!id))]; + const channels = destChannelIds.length + ? await db.telegramChannel.findMany({ + where: { id: { in: destChannelIds } }, + select: { id: true, telegramId: true }, }) - : null; - return { + : []; + const telegramIdById = new Map(channels.map((c) => [c.id, c.telegramId])); + + return rows.map((row) => ({ id: row.id, archiveType: row.archiveType, fileCount: row.fileCount, fileSize: row.fileSize, destMessageId: row.destMessageId, destMessageIds: row.destMessageIds, - destChannel, - }; + destChannel: row.destChannelId && telegramIdById.has(row.destChannelId) + ? { telegramId: telegramIdById.get(row.destChannelId)! } + : null, + })); +} + +export async function findPlaceholderCandidate( + destChannelId: string, + fileName: string, + fileSize: bigint, +): Promise { + return (await findPlaceholderCandidates(destChannelId, fileName, fileSize))[0] ?? null; } export async function getPackageFileCrcs(packageId: string): Promise<(string | null)[]> { diff --git a/worker/src/provenance-backfill.ts b/worker/src/provenance-backfill.ts index b656a8e..bc4dba3 100644 --- a/worker/src/provenance-backfill.ts +++ b/worker/src/provenance-backfill.ts @@ -1,12 +1,14 @@ +import { db } from "./db/client.js"; import { childLogger } from "./util/logger.js"; import { downloadFileRange } from "./tdlib/range-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 { - findPlaceholderCandidate, + findPlaceholderCandidates, getPackageFileCrcs, backfillProvenance, + type PlaceholderCandidate, } from "./db/queries.js"; import type { FileEntry } from "./archive/zip-reader.js"; import type { Client } from "tdl"; @@ -72,49 +74,103 @@ async function readZipListingFromDestination( } } +/** + * Build the CRC fingerprint entries for a placeholder candidate: start from + * its stored PackageFile CRCs, and if those are incomplete (e.g. a rebuild + * candidate with fileCount === 0), fall back to a fresh ranged read of the + * candidate's own copy in the destination channel (Task 9). + */ +async function resolveCandidateFingerprintEntries( + client: Client, + candidate: PlaceholderCandidate, +): Promise { + const candidateCrcs = await getPackageFileCrcs(candidate.id); + let candidateEntries: FileEntry[] = candidateCrcs.map((crc) => ({ + path: "", fileName: "", extension: null, compressedSize: 0n, uncompressedSize: 0n, crc32: crc, + })); + const destMessageId = + candidate.destMessageIds.length > 0 + ? candidate.destMessageIds[candidate.destMessageIds.length - 1] + : candidate.destMessageId; + if (!crcFingerprint(candidateEntries).complete && destMessageId && candidate.destChannel) { + const destEntries = await readZipListingFromDestination( + client, + candidate.destChannel.telegramId, + destMessageId, + candidate.fileSize, + ); + if (destEntries) { + candidateEntries = destEntries; + } + } + return candidateEntries; +} + export async function tryProvenanceBackfill( args: BackfillArgs, ): Promise<{ backfilled: boolean; confidence?: "fingerprint" | "name-size" }> { - const candidate = await findPlaceholderCandidate(args.destChannelId, args.fileName, args.fileSize); - if (!candidate) return { backfilled: false }; + const candidates = await findPlaceholderCandidates(args.destChannelId, args.fileName, args.fileSize); + if (candidates.length === 0) return { backfilled: false }; - let entries: FileEntry[] | null = null; + let scannedEntries: FileEntry[] | null = null; + if (args.archiveType === "ZIP") { + scannedEntries = await readScannedZipListing(args.client, args.scannedFileId, args.fileSize); + } + + let chosen = candidates[0]; let confidence: "fingerprint" | "name-size" = "name-size"; - if (args.archiveType === "ZIP") { - entries = await readScannedZipListing(args.client, args.scannedFileId, args.fileSize); - if (entries) { - const candidateCrcs = await getPackageFileCrcs(candidate.id); - let candidateEntries: FileEntry[] = candidateCrcs.map((crc) => ({ - path: "", fileName: "", extension: null, compressedSize: 0n, uncompressedSize: 0n, crc32: crc, - })); - const destMessageId = - candidate.destMessageIds.length > 0 - ? candidate.destMessageIds[candidate.destMessageIds.length - 1] - : candidate.destMessageId; - if (!crcFingerprint(candidateEntries).complete && destMessageId && candidate.destChannel) { - const destEntries = await readZipListingFromDestination( - args.client, - candidate.destChannel.telegramId, - destMessageId, - candidate.fileSize, - ); - if (destEntries) { - candidateEntries = destEntries; - } + if (candidates.length > 1) { + // Multiple placeholder packages share this name+size. Try to + // disambiguate by fingerprint (ZIP only); if we can't uniquely resolve + // it, notify instead of guessing which one is the real match. + if (args.archiveType === "ZIP" && scannedEntries) { + const matches: PlaceholderCandidate[] = []; + for (const c of candidates) { + const candidateEntries = await resolveCandidateFingerprintEntries(args.client, c); + if (fingerprintsMatch(scannedEntries, candidateEntries)) matches.push(c); } - if (fingerprintsMatch(entries, candidateEntries)) { + if (matches.length === 1) { + chosen = matches[0]; confidence = "fingerprint"; } else { - // Fingerprint mismatch: NOT the same content despite name+size. Do not backfill. - log.info({ candidateId: candidate.id, fileName: args.fileName }, "fingerprint mismatch — not backfilling"); + await db.systemNotification.create({ + data: { + type: "INTEGRITY_AUDIT", + severity: "WARNING", + title: `Ambiguous provenance match: ${args.fileName}`, + message: `${candidates.length} placeholder packages share this name+size and the fingerprint did not uniquely disambiguate. No provenance was backfilled.`, + context: { fileName: args.fileName, candidateIds: candidates.map((c) => c.id) }, + }, + }); return { backfilled: false }; } + } else { + // Can't disambiguate without a fingerprint — notify, don't guess. + await db.systemNotification.create({ + data: { + type: "INTEGRITY_AUDIT", + severity: "WARNING", + title: `Ambiguous provenance match: ${args.fileName}`, + message: `${candidates.length} placeholder packages share this name+size (archive type ${args.archiveType} — no cheap fingerprint). No provenance was backfilled.`, + context: { fileName: args.fileName, candidateIds: candidates.map((c) => c.id) }, + }, + }); + return { backfilled: false }; + } + } else if (scannedEntries) { + const candidateEntries = await resolveCandidateFingerprintEntries(args.client, chosen); + if (fingerprintsMatch(scannedEntries, candidateEntries)) { + confidence = "fingerprint"; + } else { + // Fingerprint mismatch: NOT the same content despite name+size. Do not backfill. + log.info({ candidateId: chosen.id, fileName: args.fileName }, "fingerprint mismatch — not backfilling"); + return { backfilled: false }; } } const ok = await backfillProvenance({ - packageId: candidate.id, + packageId: chosen.id, destChannelId: args.destChannelId, sourceChannelId: args.scannedSourceChannelId, sourceMessageId: args.sourceMessageId, @@ -122,14 +178,14 @@ export async function tryProvenanceBackfill( sourceCaption: args.sourceCaption, remoteUniqueId: args.remoteUniqueId, creator: args.creator, - entries: candidate.fileCount === 0 && entries ? entries : undefined, + entries: chosen.fileCount === 0 && scannedEntries ? scannedEntries : undefined, previewData: args.previewData ?? undefined, previewMsgId: args.previewMsgId ?? undefined, }); if (!ok) return { backfilled: false }; log.info( - { candidateId: candidate.id, fileName: args.fileName, confidence, source: args.scannedSourceChannelId }, + { candidateId: chosen.id, fileName: args.fileName, confidence, source: args.scannedSourceChannelId }, "provenance backfilled", ); return { backfilled: true, confidence };