#!/usr/bin/env tsx
/**
 * `pnpm reingest:mailbox` — rebuilds a mailbox Durable Object from the raw MIME
 * that is still sitting in R2, by replaying each message through the *real*
 * inbound path instead of writing DO rows by hand.
 *
 * Why this exists: a Durable Object's id is derived from its jurisdiction, so
 * pinning mailboxes to `eu` (see src/lib/mailbox-stub.ts) does not move the old
 * object — it addresses a different, empty one. R2 still holds every inbound
 * message's raw MIME, and `message_index` in the *old* D1 still enumerates it, so
 * the mail can be put back. Everything else the old DO held (outbound bodies,
 * drafts, mailbox_meta) has no source outside it and does not come back; see
 * scripts/verify-mailbox-rebuild.ts, which names those losses instead of quietly
 * counting around them.
 *
 * ## How the replay works
 *
 * Each rebuildable row is turned back into the `InboundEmailQueueMessage` that
 * `src/cloudflare/email-handler.ts` would have produced, and pushed onto the same
 * inbound queue the Email Routing worker publishes to. From there nothing is
 * special-cased: `handleInboundQueue` heads the R2 object, calls the DO's
 * `/ingest`, and `ingestInboundEmail` re-parses the MIME, re-resolves the thread,
 * rewrites the FTS row and re-extracts attachments. No parsing, threading or
 * indexing logic is reimplemented here — the payload is validated against the
 * production `inboundEmailQueueMessageSchema` before it is sent, so a payload the
 * consumer would reject cannot leave this script.
 *
 * The queue is the only way in. The DO's `/ingest` route is not reachable over
 * HTTP (nothing proxies it), and `POST /api/debug/phase0/email` — the other
 * end-to-end path — re-derives `receivedAt` from the clock, which would restamp
 * twenty months-old messages as having arrived today and write a second copy of
 * every raw object under a new key. Replaying the queue payload keeps the
 * original `receivedAt`, `rawR2Key` and `rawSha256`.
 *
 * ## Idempotency
 *
 * Not invented here, and not a second mechanism layered on top: `ingestInboundEmail`
 * looks up `ingest_events.idempotency_key` before it does anything else and returns
 * `duplicate` without inserting. The key is rebuilt with the very same
 * `inboundIdempotencyKey()` the original delivery used, over the same mailbox id,
 * Message-ID and raw sha256, so it comes out byte-identical. Two runs therefore
 * leave one message. On top of that this script pre-checks the *target* D1's
 * `ingest_events` and skips keys already recorded as processed, so a second run is
 * also a no-op at the queue.
 *
 * ## Ordering
 *
 * Messages are pushed one at a time, oldest first, each awaited to completion.
 * Threading is order-dependent — `resolveThreadId` matches an incoming message
 * against the messages already stored — and Queues makes no ordering promise, so a
 * parallel replay could group a conversation differently than it was grouped the
 * first time. Twenty sequential round-trips is a cheap price for a deterministic
 * rebuild.
 *
 * SAFETY: dry-run by default. The dry run reads exactly what the apply run reads
 * (same remote D1, same remote R2) and prints the payload it would push, so the
 * preview is the plan. `--apply` is the only thing that writes.
 *
 * Usage:
 *   pnpm reingest:mailbox --env dev --mailbox mbx_... --source-d1 <pre-migration-index>
 *   pnpm reingest:mailbox --env dev --mailbox mbx_... --source-d1 <pre-migration-index> --apply
 *
 * Requires: `wrangler` auth (remote D1 + R2 reads) and, for `--apply`, a
 * CLOUDFLARE_API_TOKEN with Queues:Edit plus CLOUDFLARE_ACCOUNT_ID (or a
 * `wrangler whoami` that resolves one) — pushing to a queue has no wrangler command.
 */
import { execFileSync } from "node:child_process";
import { mkdtempSync, readFileSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { InboundEmailQueueMessage } from "../src/cloudflare/types";
import { inboundEmailQueueMessageSchema } from "../src/cloudflare/types";
import { randomTraceId, sha256Hex } from "../src/lib/crypto";
import { normalizeMessageId } from "../src/lib/email-metadata";
import { inboundIdempotencyKey } from "../src/lib/idempotency";
import { parseMimeBytes } from "../src/lib/mime";

/** Mirrors MAX_QUEUE_BYTES in src/cloudflare/email-handler.ts (not exported). */
const MAX_QUEUE_BYTES = 128 * 1024;

type WranglerBlock = {
	r2_buckets?: Array<{ binding: string; bucket_name: string; jurisdiction?: string }>;
	queues?: { producers?: Array<{ binding: string; queue: string }> };
	d1_databases?: Array<{ binding: string; database_name: string }>;
};

type WranglerConfig = WranglerBlock & { env?: Record<string, WranglerBlock> };

type MessageIndexRow = {
	message_local_id: string;
	rfc_message_id: string | null;
	subject: string | null;
	from_addr: string;
	received_at: string;
	labels_json: string;
	state: string;
	raw_r2_key: string;
	raw_sha256: string;
};

type IngestEventRow = {
	idempotency_key: string;
	message_local_id: string | null;
	status: string;
	error_code: string | null;
};

type Rebuildable = {
	row: MessageIndexRow;
	payload: InboundEmailQueueMessage;
	rawSize: number;
	notes: string[];
};

type Unrebuildable = {
	row: MessageIndexRow;
	reason: string;
};

type Outcome = {
	row: MessageIndexRow;
	status:
		| "inserted"
		| "duplicate"
		| "conflict"
		| "timeout"
		| "push_failed"
		| "skipped"
		| "unverified";
	detail: string;
	newLocalId: string | null;
};

function parseArgs(argv: string[]): Record<string, string> {
	const args: Record<string, string> = {};
	for (let i = 0; i < argv.length; i += 1) {
		const arg = argv[i];
		if (!arg?.startsWith("--")) continue;
		const [rawKey, inlineValue] = arg.slice(2).split("=", 2);
		if (!rawKey) continue;
		if (inlineValue !== undefined) {
			args[rawKey] = inlineValue;
			continue;
		}
		const next = argv[i + 1];
		if (!next || next.startsWith("--")) {
			args[rawKey] = "true";
			continue;
		}
		args[rawKey] = next;
		i += 1;
	}
	return args;
}

function fail(message: string): never {
	console.error(`reingest:mailbox: ${message}`);
	process.exit(1);
}

function stripJsonc(input: string): string {
	return input.replace(/\/\*[\s\S]*?\*\//g, "").replace(/^\s*\/\/.*$/gm, "");
}

/** SQLite string literal. The only untrusted values reaching SQL are labels_json
 * and state, both read back out of D1, but they still go through here. */
function sqlString(value: string): string {
	return `'${value.replace(/'/g, "''")}'`;
}

function wrangler(args: string[]): string {
	return execFileSync("pnpm", ["wrangler", ...args], {
		encoding: "utf8",
		stdio: ["ignore", "pipe", "pipe"],
		maxBuffer: 64 * 1024 * 1024,
	});
}

/**
 * `wrangler d1 execute --json` prefixes its JSON with a human banner on some
 * versions, so cut to the first `[` rather than parsing the whole stream — the
 * same shape doctor.ts relies on.
 */
// wrangler uploads a `--file` query to R2 before running it, and the JSON it
// returns for an uploaded query carries a *summary* ("Total queries executed",
// "Rows read", ...) in `results` instead of the rows themselves. Silently, with
// success: true. So SELECTs go through `--command`, whose response really does
// carry the rows. The queries here are small enough that the argv limit is not a
// concern, and a query large enough to hit it would be the wrong shape for this
// script anyway.
function d1Query<T>(database: string, sql: string, extraArgs: string[] = []): T[] {
	const raw = wrangler([
		"d1",
		"execute",
		database,
		"--remote",
		"--json",
		"--command",
		sql,
		...extraArgs,
	]);
	const start = raw.indexOf("[");
	if (start === -1) {
		throw new Error(`unexpected wrangler d1 output:\n${raw}`);
	}
	const parsed = JSON.parse(raw.slice(start)) as Array<{ results?: T[] }>;
	return parsed[0]?.results ?? [];
}

function d1Exec(database: string, sql: string, extraArgs: string[] = []): void {
	d1Query<unknown>(database, sql, extraArgs);
}

/**
 * Reads one raw object. Returns null only for a genuine miss — any other wrangler
 * failure is rethrown, because "R2 said no" and "your token expired" must not both
 * present as "this message is unrecoverable".
 */
function fetchRawObject(
	bucket: string,
	key: string,
	dir: string,
	jurisdiction: string | undefined,
): Uint8Array | null {
	const file = join(dir, "object.eml");
	// Every download reuses this path, so clear it first: a wrangler run that exits
	// zero without writing would otherwise hand back the *previous* message's bytes.
	// The sha256 check downstream would catch it, but only as a mystery.
	rmSync(file, { force: true });
	try {
		// A jurisdiction-restricted bucket is invisible without the flag: wrangler
		// looks in the default jurisdiction, finds no such bucket, and reports every
		// key as missing. That reads identically to "this mail is unrecoverable",
		// which is the most dangerous wrong answer this script can give.
		wrangler([
			"r2",
			"object",
			"get",
			`${bucket}/${key}`,
			"--remote",
			"--file",
			file,
			...(jurisdiction ? ["--jurisdiction", jurisdiction] : []),
		]);
	} catch (error) {
		const output = [
			error instanceof Error ? error.message : String(error),
			(error as { stdout?: Buffer | string }).stdout?.toString() ?? "",
			(error as { stderr?: Buffer | string }).stderr?.toString() ?? "",
		].join("\n");
		if (/not\s*found|404|NoSuchKey|does not exist/i.test(output)) {
			return null;
		}
		throw new Error(`r2 get failed for ${key}:\n${output}`);
	}
	return new Uint8Array(readFileSync(file));
}

async function cfApi<T>(token: string, path: string, init?: RequestInit): Promise<T> {
	const response = await fetch(`https://api.cloudflare.com/client/v4${path}`, {
		...init,
		headers: {
			authorization: `Bearer ${token}`,
			"content-type": "application/json",
			...(init?.headers ?? {}),
		},
	});
	const payload = (await response.json()) as {
		success?: boolean;
		result?: T;
		errors?: Array<{ message?: string; code?: number }>;
	};
	if (!response.ok || payload.success !== true) {
		const detail = (payload.errors ?? [])
			.map((item) => `${item.message ?? "unknown error"}${item.code ? ` [${item.code}]` : ""}`)
			.join("; ");
		throw new Error(detail || `Cloudflare API request failed (${response.status})`);
	}
	return payload.result as T;
}

/** Env var first, then the account `wrangler whoami` is already authenticated as. */
function resolveAccountId(): string | undefined {
	const fromEnv = process.env.CLOUDFLARE_ACCOUNT_ID?.trim();
	if (fromEnv) return fromEnv;
	try {
		const payload = JSON.parse(wrangler(["whoami", "--json"])) as {
			account?: { id?: string };
			accounts?: Array<{ id?: string }>;
		};
		return payload.account?.id ?? payload.accounts?.[0]?.id;
	} catch {
		return undefined;
	}
}

async function resolveQueueId(
	token: string,
	accountId: string,
	queueName: string,
): Promise<string> {
	// Field names have drifted across Queues API versions (queue_id/queue_name vs
	// id/name); accept either rather than pinning to whichever is current today.
	const queues = await cfApi<Array<Record<string, unknown>>>(
		token,
		`/accounts/${accountId}/queues`,
	);
	for (const queue of queues) {
		const name = (queue.queue_name ?? queue.name) as string | undefined;
		const id = (queue.queue_id ?? queue.id) as string | undefined;
		if (name === queueName && id) return id;
	}
	throw new Error(`queue ${queueName} not found in account ${accountId}`);
}

function sleep(ms: number): Promise<void> {
	return new Promise((resolve) => setTimeout(resolve, ms));
}

const args = parseArgs(process.argv.slice(2));
const envName = args.env ?? "dev";
const mailboxId = args.mailbox?.trim();
const apply = args.apply === "true";
const force = args.force === "true";
const timeoutMs = Number(args["timeout-ms"] ?? "60000");
const limit = args.limit ? Number(args.limit) : Number.POSITIVE_INFINITY;

if (!mailboxId) {
	fail("pass --mailbox <mailbox-id>.");
}
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) {
	fail("--timeout-ms must be a positive number of milliseconds.");
}

const config = JSON.parse(
	stripJsonc(readFileSync(join(process.cwd(), "wrangler.jsonc"), "utf8")),
) as WranglerConfig;
const block: WranglerBlock | undefined = envName === "production" ? config : config.env?.[envName];
if (!block) {
	fail(`wrangler.jsonc has no env "${envName}".`);
}

const targetDb = block.d1_databases?.find((db) => db.binding === "INDEX_DB")?.database_name;
const bucketBinding = block.r2_buckets?.find((r2) => r2.binding === "MAIL_OBJECTS");
const bucket = bucketBinding?.bucket_name;
const bucketJurisdiction = bucketBinding?.jurisdiction;
const queueName = block.queues?.producers?.find(
	(producer) => producer.binding === "INBOUND_EMAIL_QUEUE",
)?.queue;
if (!targetDb || !bucket || !queueName) {
	fail(`env "${envName}" is missing an INDEX_DB, MAIL_OBJECTS or INBOUND_EMAIL_QUEUE binding.`);
}

// `production` is the top-level config block, not an entry under `env`, so wrangler
// rejects `--env production`. Only named environments get the flag.
const envArgs = envName === "production" ? [] : ["--env", envName];

// The migration that motivated this script also moved D1, so what the mailbox used
// to contain and where it is being rebuilt are two different databases. Defaulting
// to the target keeps the ordinary "replay into the same env" case a one-flag run.
const sourceDb = args["source-d1"]?.trim() || targetDb;

console.log("Mailbox re-ingest");
console.log(`  Env:            ${envName}`);
console.log(`  Mailbox:        ${mailboxId}`);
console.log(`  Source D1:      ${sourceDb}${sourceDb === targetDb ? " (same as target)" : ""}`);
console.log(`  Target D1:      ${targetDb}`);
console.log(
	`  R2 bucket:      ${bucket}${bucketJurisdiction ? ` (jurisdiction ${bucketJurisdiction})` : ""}`,
);
console.log(`  Inbound queue:  ${queueName}`);
console.log(`  Mode:           ${apply ? "APPLY" : "DRY RUN (pass --apply to write)"}`);
console.log();

const mailboxRows = d1Query<{ primary_address: string }>(
	sourceDb,
	`SELECT primary_address FROM mailboxes WHERE mailbox_id = ${sqlString(mailboxId)};`,
);
const primaryAddress = mailboxRows[0]?.primary_address?.trim().toLowerCase();
if (!primaryAddress) {
	fail(`no mailboxes row for ${mailboxId} in ${sourceDb}.`);
}

// The address the mail was actually delivered to is not stored anywhere — R2's
// custom metadata does not carry it and message_index only keeps the parsed To.
// The primary address is the closest honest stand-in, and the only thing the DO
// does with it is derive the mailbox's own domain for the subject-threading
// fallback (see mailboxOwnDomains), where any address on the mailbox is equivalent.
const recipient = primaryAddress;
const domain = recipient.split("@")[1] ?? "";

const indexRows = d1Query<MessageIndexRow>(
	sourceDb,
	`SELECT message_local_id, rfc_message_id, subject, from_addr, received_at,
          labels_json, state, raw_r2_key, raw_sha256
     FROM message_index
    WHERE mailbox_id = ${sqlString(mailboxId)}
    ORDER BY received_at ASC;`,
);

if (indexRows.length === 0) {
	fail(`message_index in ${sourceDb} has no rows for ${mailboxId}; nothing to re-ingest.`);
}

console.log(`Enumerated ${indexRows.length} indexed message(s) in ${sourceDb}.`);
console.log();

const workDir = mkdtempSync(join(tmpdir(), "reccado-reingest-raw-"));
const rebuildable: Rebuildable[] = [];
const unrebuildable: Unrebuildable[] = [];

try {
	for (const row of indexRows) {
		if (rebuildable.length >= limit) break;

		const rawBytes = fetchRawObject(bucket, row.raw_r2_key, workDir, bucketJurisdiction);
		if (!rawBytes) {
			unrebuildable.push({ row, reason: `no object at R2 key ${row.raw_r2_key}` });
			continue;
		}

		const notes: string[] = [];
		const actualSha = await sha256Hex(rawBytes);
		if (actualSha !== row.raw_sha256) {
			// The raw object is the rebuild source and its digest is half the
			// idempotency key. A mismatch means the object under this key is not the
			// message the index describes, and replaying it would store something
			// nobody asked for under a key that claims otherwise.
			unrebuildable.push({
				row,
				reason: `raw_sha256 mismatch: index says ${row.raw_sha256}, R2 object hashes to ${actualSha}`,
			});
			continue;
		}

		const parsed = await parseMimeBytes(rawBytes);
		const indexedMessageId = normalizeMessageId(row.rfc_message_id);
		const parsedMessageId = normalizeMessageId(parsed.messageId);
		if (indexedMessageId && parsedMessageId && indexedMessageId !== parsedMessageId) {
			notes.push(`Message-ID differs (index ${indexedMessageId}, MIME ${parsedMessageId})`);
		}
		// The index value wins: it is what the original idempotency key was computed
		// from, so using it is what makes the replay dedupe against the first delivery.
		const messageId = indexedMessageId ?? parsedMessageId;

		// raw_r2_key embeds the original receivedAt as epoch ms (see rawEmailR2Key), so
		// key and index are two independent records of the same instant and can be
		// checked against each other for free.
		const keyEpoch = Number(row.raw_r2_key.split("/").pop()?.split("-")[0]);
		if (Number.isFinite(keyEpoch) && Math.abs(keyEpoch - Date.parse(row.received_at)) > 1000) {
			notes.push(
				`received_at ${row.received_at} disagrees with the timestamp in raw_r2_key (${new Date(keyEpoch).toISOString()})`,
			);
		}

		const idempotencyKey = inboundIdempotencyKey({
			mailboxId,
			messageId,
			rawSha256: row.raw_sha256,
		});

		const payload = inboundEmailQueueMessageSchema.parse({
			schemaVersion: 1,
			eventType: "email.received.v1",
			traceId: randomTraceId(),
			enqueuedAt: new Date().toISOString(),
			receivedAt: row.received_at,
			mailboxId,
			domain,
			recipient,
			sender: row.from_addr,
			rawR2Key: row.raw_r2_key,
			rawSha256: row.raw_sha256,
			rawSize: rawBytes.byteLength,
			messageId,
			headers: {
				subject: parsed.subject ?? row.subject,
				// The original stored the verbatim `Date:` header line; postal-mime
				// normalizes it to ISO-8601. Same instant, different spelling, and
				// `date_header` is only ever displayed or quoted — noted, not hidden.
				date: parsed.date,
				inReplyTo: normalizeMessageId(parsed.inReplyTo),
				references: parsed.references
					.map((reference) => normalizeMessageId(reference))
					.filter((reference): reference is string => Boolean(reference)),
			},
			routing: { ruleId: null, action: "store", matchedAlias: recipient },
			idempotencyKey,
		}) satisfies InboundEmailQueueMessage;

		const payloadBytes = new TextEncoder().encode(JSON.stringify(payload)).byteLength;
		if (payloadBytes > MAX_QUEUE_BYTES) {
			unrebuildable.push({
				row,
				reason: `queue payload would be ${payloadBytes} bytes, over the ${MAX_QUEUE_BYTES} limit`,
			});
			continue;
		}

		rebuildable.push({ row, payload, rawSize: rawBytes.byteLength, notes });
	}
} finally {
	rmSync(workDir, { recursive: true, force: true });
}

const duplicateKeys = new Map<string, string[]>();
for (const candidate of rebuildable) {
	const seen = duplicateKeys.get(candidate.payload.idempotencyKey) ?? [];
	seen.push(candidate.row.message_local_id);
	duplicateKeys.set(candidate.payload.idempotencyKey, seen);
}
const collisions = [...duplicateKeys.entries()].filter(([, ids]) => ids.length > 1);

// What the target already knows. `ingest_events` is written by the queue consumer
// after the DO answers, for both `inserted` and `duplicate`, so a row here means
// the message is in the rebuilt DO and there is nothing to push.
const alreadyProcessed = new Map<string, IngestEventRow>();
if (rebuildable.length > 0) {
	const keys = rebuildable.map((candidate) => sqlString(candidate.payload.idempotencyKey));
	const rows = d1Query<IngestEventRow>(
		targetDb,
		`SELECT idempotency_key, message_local_id, status, error_code
       FROM ingest_events WHERE idempotency_key IN (${keys.join(", ")});`,
		envArgs,
	);
	for (const row of rows) {
		alreadyProcessed.set(row.idempotency_key, row);
	}
}

console.log("Rebuildable from R2");
console.log("-------------------");
rebuildable.forEach((candidate, position) => {
	const { row, payload } = candidate;
	const existing = alreadyProcessed.get(payload.idempotencyKey);
	const plan = existing
		? force
			? `RE-PUSH (already ${existing.status} in ${targetDb}; --force)`
			: `SKIP (already ${existing.status} in ${targetDb} as ${existing.message_local_id ?? "?"})`
		: "PUSH";
	console.log(
		`  [${String(position + 1).padStart(2, " ")}/${rebuildable.length}] ${row.received_at}  ${plan}`,
	);
	console.log(`         subject   ${row.subject ?? "(none)"}`);
	console.log(`         from      ${row.from_addr}`);
	console.log(`         raw       ${row.raw_r2_key} (${candidate.rawSize} bytes)`);
	console.log(`         idem      ${payload.idempotencyKey}`);
	console.log(`         old id    ${row.message_local_id}  state=${row.state}`);
	for (const note of candidate.notes) {
		console.log(`         note      ${note}`);
	}
});
if (rebuildable.length === 0) {
	console.log("  (none)");
}
console.log();

console.log("Not rebuildable — no raw MIME to replay");
console.log("--------------------------------------");
for (const item of unrebuildable) {
	const outbound = item.row.state === "sent" || item.row.raw_r2_key.startsWith("sent/");
	console.log(`  ${item.row.message_local_id}  state=${item.row.state}  ${item.row.received_at}`);
	console.log(`         subject   ${item.row.subject ?? "(none)"}`);
	console.log(`         reason    ${item.reason}`);
	// Outbound messages never had a raw object: the send path computes
	// `sent/<draftId>` as a key and writes nothing there. Anything else missing its
	// raw object is a loss nobody has accounted for, and saying so here is cheaper
	// than discovering it during verification.
	console.log(
		outbound
			? "         expected  yes — outbound, its body only ever lived in the old DO"
			: "         expected  NO — inbound message with no raw object; investigate before you accept this",
	);
}
if (unrebuildable.length === 0) {
	console.log("  (none)");
}
console.log();

for (const [key, ids] of collisions) {
	console.log(`WARNING: ${ids.length} rows share idempotency key ${key}: ${ids.join(", ")}`);
	console.log("         Only one of them can exist in the DO; the rest will report as duplicates.");
}
if (collisions.length > 0) console.log();

const toPush = rebuildable.filter(
	(candidate) => force || !alreadyProcessed.has(candidate.payload.idempotencyKey),
);

console.log("Plan");
console.log("----");
console.log(`  indexed            ${indexRows.length}`);
console.log(`  rebuildable        ${rebuildable.length}`);
console.log(`  not rebuildable    ${unrebuildable.length}`);
console.log(`  already in target  ${rebuildable.length - toPush.length}`);
console.log(`  to push            ${toPush.length}`);

const stateRestores = rebuildable.filter(
	(candidate) => candidate.row.state !== "inbox" || candidate.row.labels_json !== "[]",
);
const doStateDivergence = rebuildable.filter((candidate) => candidate.row.state !== "inbox");
console.log(`  state/labels to restore in ${targetDb}: ${stateRestores.length}`);
console.log(
	`  messages whose DO state is not the ingest default 'inbox': ${doStateDivergence.length}`,
);
console.log();

if (!apply) {
	console.log("DRY RUN: nothing was pushed and no row was written.");
	console.log("Re-run with --apply to replay the messages listed above.");
	process.exit(0);
}

// Two ways to get a message onto the queue, because publishing has no wrangler
// command. The REST API is the direct one but needs an API token with Queues:Edit,
// which is a credential someone has to mint. `--push-url` is the alternative: a
// Worker holding a producer binding to this queue, which needs nothing but the
// wrangler session that deployed it. Both end at the same `send()`.
const pushUrl = args["push-url"]?.trim();
const pushToken = (args["push-token"] ?? process.env.RECCADO_PUSH_TOKEN)?.trim();
let publish: (payload: unknown) => Promise<void>;

if (pushUrl) {
	console.log(`Pushing to queue ${queueName} via ${new URL(pushUrl).origin}.`);
	publish = async (payload) => {
		const response = await fetch(pushUrl, {
			method: "POST",
			headers: {
				"content-type": "application/json",
				...(pushToken ? { authorization: `Bearer ${pushToken}` } : {}),
			},
			body: JSON.stringify(payload),
		});
		if (!response.ok) {
			throw new Error(`push endpoint returned ${response.status}: ${await response.text()}`);
		}
	};
} else {
	const apiToken = process.env.CLOUDFLARE_API_TOKEN?.trim();
	if (!apiToken) {
		fail(
			"--apply needs CLOUDFLARE_API_TOKEN (Queues:Edit), or --push-url pointing at a Worker with a producer binding for this queue.",
		);
	}
	const accountId = resolveAccountId();
	if (!accountId) {
		fail("could not resolve an account id; set CLOUDFLARE_ACCOUNT_ID.");
	}
	const queueId = await resolveQueueId(apiToken, accountId, queueName);
	console.log(`Pushing to queue ${queueName} (${queueId}) in account ${accountId}.`);
	publish = async (payload) => {
		await cfApi(apiToken, `/accounts/${accountId}/queues/${queueId}/messages`, {
			method: "POST",
			body: JSON.stringify({ body: payload, content_type: "json" }),
		});
	};
}
console.log();

const outcomes: Outcome[] = [];

for (const candidate of rebuildable) {
	const { row, payload } = candidate;
	const label = `${row.received_at} ${row.subject ?? "(no subject)"}`;

	if (!force && alreadyProcessed.has(payload.idempotencyKey)) {
		const existing = alreadyProcessed.get(payload.idempotencyKey)!;
		outcomes.push({
			row,
			status: "skipped",
			detail: `already ${existing.status} in ${targetDb}`,
			newLocalId: existing.message_local_id,
		});
		console.log(`  SKIP      ${label} — already ${existing.status} in ${targetDb}`);
		continue;
	}

	try {
		await publish(payload);
	} catch (error) {
		outcomes.push({
			row,
			status: "push_failed",
			detail: error instanceof Error ? error.message : String(error),
			newLocalId: null,
		});
		console.log(`  PUSH-FAIL ${label} — ${error instanceof Error ? error.message : error}`);
		continue;
	}

	// Wait for this message to land before pushing the next one; see the ordering
	// note at the top of this file.
	// A row carrying this idempotency key may already exist before we push -- the
	// projection is rebuildable and can be restored from a backup that predates the
	// Durable Object it describes, which is exactly the state a jurisdiction move
	// leaves behind. Waiting for "a row exists" would then be satisfied instantly by
	// that stale row and report work that never happened, reusing the old message id.
	// So completion means the id *changed*, not that a row is present.
	const priorLocalId = alreadyProcessed.get(payload.idempotencyKey)?.message_local_id ?? null;
	const deadline = Date.now() + timeoutMs;
	let settled: IngestEventRow | undefined;
	while (Date.now() < deadline) {
		await sleep(1500);
		const rows = d1Query<IngestEventRow>(
			targetDb,
			`SELECT idempotency_key, message_local_id, status, error_code
         FROM ingest_events WHERE idempotency_key = ${sqlString(payload.idempotencyKey)};`,
			envArgs,
		);
		const row0 = rows[0];
		if (!row0 || (row0.status !== "processed" && row0.status !== "failed")) continue;
		if (priorLocalId !== null && row0.message_local_id === priorLocalId) continue;
		settled = row0;
		break;
	}

	// Timing out against an unchanged pre-existing row is genuinely ambiguous: the
	// Durable Object either deduplicated this message or never processed it, and a
	// stale projection row cannot tell those apart. Say so instead of picking one.
	if (!settled && priorLocalId !== null) {
		outcomes.push({
			row,
			status: "unverified",
			detail: `a pre-existing ${targetDb} row for this key was unchanged after ${timeoutMs}ms — the DO either deduplicated it or never processed it, and this row predates the run`,
			newLocalId: priorLocalId,
		});
		console.log(
			`  UNVERIFIED ${label} — pre-existing row unchanged; cannot distinguish dedupe from no-op`,
		);
		continue;
	}

	if (!settled) {
		outcomes.push({
			row,
			status: "timeout",
			detail: `no ingest_events row in ${targetDb} after ${timeoutMs}ms`,
			newLocalId: null,
		});
		console.log(`  TIMEOUT   ${label} — no ingest_events row in ${targetDb} after ${timeoutMs}ms`);
		continue;
	}
	if (settled.status === "failed") {
		outcomes.push({
			row,
			status: "conflict",
			detail: settled.error_code ?? "failed",
			newLocalId: null,
		});
		console.log(`  CONFLICT  ${label} — ${settled.error_code ?? "failed"}`);
		continue;
	}

	// `duplicate` is only claimed when the Durable Object itself reported one, never
	// inferred from a projection row we did not watch appear.
	const reused = settled.message_local_id === priorLocalId;
	outcomes.push({
		row,
		status: reused ? "duplicate" : "inserted",
		detail: settled.message_local_id ?? "",
		newLocalId: settled.message_local_id,
	});
	console.log(
		`  ${reused ? "DUPLICATE" : "INSERTED "} ${label} → ${settled.message_local_id ?? "?"}`,
	);
}

console.log();

// Restore what the ingest path cannot know. A replayed message is inbound and lands
// in 'inbox' with no labels, because that is what an arriving email is; the folder it
// had been filed into and the labels put on it live only in message_index, so they
// are copied forward onto the row the consumer just wrote under the new local id.
const restorable = outcomes.filter(
	(outcome) =>
		outcome.newLocalId &&
		(outcome.row.state !== "inbox" || outcome.row.labels_json !== "[]") &&
		(outcome.status === "inserted" ||
			outcome.status === "duplicate" ||
			outcome.status === "skipped"),
);

if (restorable.length > 0) {
	const now = new Date().toISOString();
	const statements = restorable.map(
		(outcome) =>
			`UPDATE message_index SET state = ${sqlString(outcome.row.state)}, ` +
			`labels_json = ${sqlString(outcome.row.labels_json)}, updated_at = ${sqlString(now)} ` +
			`WHERE mailbox_id = ${sqlString(mailboxId)} AND message_local_id = ${sqlString(outcome.newLocalId!)};`,
	);
	d1Exec(targetDb, statements.join("\n"), envArgs);
	console.log(`Restored state/labels on ${restorable.length} message_index row(s) in ${targetDb}.`);
} else {
	console.log("No state/labels to restore: every replayed message was 'inbox' with no labels.");
}
console.log();

const counts = new Map<Outcome["status"], number>();
for (const outcome of outcomes) {
	counts.set(outcome.status, (counts.get(outcome.status) ?? 0) + 1);
}

console.log("Summary");
console.log("-------");
for (const status of [
	"inserted",
	"duplicate",
	"unverified",
	"skipped",
	"conflict",
	"timeout",
	"push_failed",
] as const) {
	console.log(`  ${status.padEnd(12, " ")} ${counts.get(status) ?? 0}`);
}
console.log(`  not rebuildable ${unrebuildable.length}`);
console.log();

const notClean = outcomes.filter(
	(outcome) =>
		outcome.status === "conflict" ||
		outcome.status === "timeout" ||
		outcome.status === "push_failed",
);
if (notClean.length > 0) {
	console.log("Did not re-ingest");
	console.log("-----------------");
	for (const outcome of notClean) {
		console.log(`  ${outcome.status.toUpperCase()}  ${outcome.row.message_local_id}`);
		console.log(`         subject ${outcome.row.subject ?? "(none)"}`);
		console.log(`         detail  ${outcome.detail}`);
	}
	console.log();
}

// The DO stores state too, and nothing reachable from here can set it: the only
// route that moves a message between folders is the Access-protected
// POST /api/mailboxes/:id/messages/:id/actions. Rather than pretend the restore
// above covered it, print the exact calls that finish the job.
const doFollowUps = outcomes.filter(
	(outcome) => outcome.newLocalId && outcome.row.state !== "inbox",
);
if (doFollowUps.length > 0) {
	console.log("Durable Object state still needs a follow-up");
	console.log("-------------------------------------------");
	console.log(
		"message_index now says these are filed elsewhere, but the DO — which is what the UI reads —",
	);
	console.log("has them in 'inbox'. Apply each with an authenticated request:");
	for (const outcome of doFollowUps) {
		const action =
			outcome.row.state === "archive"
				? "archive"
				: outcome.row.state === "trash"
					? "trash"
					: `(no action maps to state '${outcome.row.state}')`;
		console.log(
			`  POST /api/mailboxes/${mailboxId}/messages/${outcome.newLocalId}/actions  {"action":"${action}"}`,
		);
	}
	console.log();
}

if (sourceDb === targetDb) {
	const superseded = outcomes
		.filter((outcome) => outcome.newLocalId && outcome.newLocalId !== outcome.row.message_local_id)
		.map((outcome) => outcome.row.message_local_id);
	if (superseded.length > 0) {
		console.log("Superseded message_index rows");
		console.log("----------------------------");
		console.log(
			"Re-ingest mints a new message_local_id, so the pre-rebuild rows are still in this",
		);
		console.log("database pointing at DO messages that no longer exist. Delete them when you have");
		console.log("confirmed the rebuild (verify-mailbox-rebuild.ts), not before:");
		console.log(
			`  DELETE FROM message_index WHERE mailbox_id = ${sqlString(mailboxId)} AND message_local_id IN (${superseded
				.map((id) => sqlString(id))
				.join(", ")});`,
		);
		console.log();
	}
}

const failed =
	(counts.get("conflict") ?? 0) + (counts.get("timeout") ?? 0) + (counts.get("push_failed") ?? 0);
if (failed > 0) {
	console.error(`FAIL: ${failed} message(s) did not re-ingest cleanly. See above.`);
	process.exit(1);
}

console.log(
	`PASS: ${counts.get("inserted") ?? 0} inserted, ${counts.get("duplicate") ?? 0} already present, ` +
		`${counts.get("skipped") ?? 0} skipped. Now run pnpm verify:mailbox-rebuild.`,
);