import assert from "node:assert/strict";
import { test } from "node:test";
import { randomUUID } from "node:crypto";
import { Worker } from "node:worker_threads";
import * as path from "node:path";
import { Logger } from "sync-client";
import { ServerControl } from "../deterministic-tests/src/server-control";
import { MemoryDisk, type DiskImage, type StoredClient } from "./storage";
import { getJson, type CanonicalSnapshot } from "./canonical";
import type { RequestRecord } from "./network";
import type { CrashWorkerInput } from "../deterministic-tests/src/crash-worker";

const root = path.resolve(__dirname, "../..");
const token = "test-token-change-me";
const headers = {
    Authorization: `Bearer ${token}`,
    "Device-Id": "protocol-test",
    "Content-Type": "application/json"
};
const makeServer = () =>
    new ServerControl(
        path.join(root, "sync-server/target/release/sync_server"),
        path.join(root, "sync-server/config-e2e.yml"),
        new Logger()
    );
async function put(url: string, body: unknown) {
    const response = await fetch(url, {
        method: "PUT",
        headers,
        body: JSON.stringify(body),
        signal: AbortSignal.timeout(5_000)
    });
    assert(
        response.ok,
        `HTTP ${response.status}: ${await response.clone().text()}`
    );
    return (await response.json()) as {
        type: string;
        vaultUpdateId: number;
        fileManifestId: number;
    };
}
const content = (text: string, parentVersionId: number | null = null) => ({
    requestId: randomUUID(),
    parentVersionId,
    content: { type: "Snapshot", value: Buffer.from(text).toString("base64") }
});

interface WorkerSnapshot {
    type: string;
    disk: DiskImage;
    stored: StoredClient;
    requests: RequestRecord[];
    editedBase?: { documentId: string; vaultUpdateId: number };
}
async function runWorker(input: CrashWorkerInput): Promise<WorkerSnapshot> {
    const worker = new Worker(
        path.join(root, "frontend/deterministic-tests/dist/crash-worker.js"),
        { workerData: input }
    );
    let timeout: ReturnType<typeof setTimeout> | undefined;
    try {
        return await new Promise<WorkerSnapshot>((resolve, reject) => {
            timeout = setTimeout(
                () =>
                    reject(new Error("Worker checkpoint/completion timed out")),
                15_000
            );
            worker.once("error", reject);
            worker.once("exit", (code) =>
                reject(
                    new Error(
                        `Worker exited (${code}) before expected checkpoint`
                    )
                )
            );
            worker.on(
                "message",
                (message: WorkerSnapshot & { error?: string }) => {
                    if (message.type === "error")
                        reject(new Error(message.error));
                    else {
                        assert.equal(
                            message.type,
                            input.stopAt ? "checkpoint" : "finished"
                        );
                        resolve(message);
                    }
                }
            );
        });
    } finally {
        clearTimeout(timeout);
        // This is the fault injection, not graceful shutdown masquerading as it.
        await worker.terminate();
    }
}

for (const [kind, format, originalPath] of [
    ["create", "Snapshot", "original.md"],
    ["content", "Snapshot", "original.bin"],
    ["content", "Diff", "original.md"],
    ["manifest", undefined, "original.md"]
] as const) {
    test(
        `abrupt client death after accepted ${kind} ${format ?? ""} reply, newer head, server SIGKILL, durable retry`,
        { timeout: 30_000 },
        async () => {
            const server = makeServer();
            try {
                await server.start();
                const vault = `crash-${randomUUID()}`;
                const url = `${server.remoteUri}/vaults/${vault}`;
                const disk = new MemoryDisk();
                await disk.userWrite(
                    originalPath,
                    new TextEncoder().encode("preserve-original\n")
                );
                let input: CrashWorkerInput = {
                    disk: disk.image(),
                    stored: {
                        settings: {
                            isSyncEnabled: true,
                            enableTelemetry: false,
                            remoteUri: server.remoteUri,
                            vaultName: vault,
                            token,
                            requestTimeoutMs: 5_000,
                            syncIntervalMs: 0
                        }
                    } as StoredClient
                };
                let originalId: string | undefined;
                let originalParent: number | undefined;
                let uploaded = "preserve-original\n";
                if (kind === "content" && format === "Diff") {
                    // Keep the established base in the worker's real content
                    // cache; a cold restart intentionally falls back to Snapshot.
                    uploaded += "local-update\n";
                    input.editAfterSync = {
                        path: originalPath,
                        content: uploaded
                    };
                } else if (kind === "content") {
                    const initialized = await runWorker(input);
                    originalId = Object.keys(
                        initialized.stored.database!.local!
                    )[0];
                    originalParent =
                        initialized.stored.database!.documents![originalId]
                            .base!.vaultUpdateId;
                    disk.restore(initialized.disk);
                    uploaded += "local-update\n";
                    await disk.userWrite(
                        originalPath,
                        new TextEncoder().encode(uploaded)
                    );
                    input = { disk: disk.image(), stored: initialized.stored };
                }
                const checkpoint = await runWorker({ ...input, stopAt: kind });
                if (format === "Diff") {
                    assert(
                        checkpoint.editedBase,
                        "Warm update baseline was not recorded"
                    );
                    originalId = checkpoint.editedBase.documentId;
                    originalParent = checkpoint.editedBase.vaultUpdateId;
                }
                const pending = checkpoint.stored.database!.pending!;
                assert(
                    pending,
                    "The uncertain request was not persisted before sending"
                );
                const interrupted = checkpoint.requests.at(-1)!;
                assert.equal(
                    interrupted.kind,
                    kind,
                    "Wrong CAS was interrupted"
                );
                if (kind !== "manifest") {
                    assert.equal(pending.type, "content");
                    assert(pending.type === "content");
                    assert.equal(pending.request.content.type, format);
                    assert.equal(
                        pending.request.parentVersionId,
                        kind === "create" ? null : originalParent
                    );
                    if (originalId)
                        assert.equal(
                            pending.documentId,
                            originalId,
                            "Update created a replacement UUID"
                        );
                }
                const accepted = await getJson<CanonicalSnapshot>(
                    `${url}/vault-snapshot`,
                    token
                );
                // Unpublished content has no manifest membership yet, so it is
                // deliberately absent from /vault-snapshot's document list.
                const doc =
                    pending.type === "content"
                        ? await getJson<{
                              documentId: string;
                              vaultUpdateId: number;
                          }>(`${url}/documents/${pending.documentId}`, token)
                        : accepted.documents[0];
                assert(doc);
                const remote = await put(
                    `${url}/documents/${doc.documentId}`,
                    content(`${uploaded}remote-update\n`, doc.vaultUpdateId)
                );
                assert.equal(remote.type, "Accepted");
                const remotePath = originalPath.replace("original", "remote");
                const renamed = await put(`${url}/file-manifest`, {
                    requestId: randomUUID(),
                    parentFileManifestId: accepted.fileManifest.fileManifestId,
                    entries: { [doc.documentId]: remotePath }
                });
                assert.equal(renamed.type, "Accepted");
                const before = await getJson<CanonicalSnapshot>(
                    `${url}/vault-snapshot`,
                    token
                );
                await server.crash();
                await server.restart();
                assert.deepEqual(
                    await getJson(`${url}/vault-snapshot`, token),
                    before,
                    "Server restart lost acknowledged state"
                );
                const restarted = await runWorker({
                    disk: checkpoint.disk,
                    stored: checkpoint.stored
                });
                const replay = restarted.requests.find(
                    (r) => r.body.requestId === pending.request.requestId
                );
                assert(replay, "Restart never retried the uncertain request");
                assert.equal(
                    replay.url,
                    interrupted.url,
                    "Restart changed the request target"
                );
                assert.deepEqual(
                    replay.body,
                    pending.request,
                    "Restart changed the persisted request"
                );
                assert.equal(restarted.stored.database!.pending, undefined);
                const recovered = new MemoryDisk();
                recovered.restore(restarted.disk);
                assert.deepEqual(
                    recovered.userFiles(),
                    new Map([
                        [
                            remotePath,
                            new TextEncoder().encode(
                                `${uploaded}remote-update\n`
                            )
                        ]
                    ])
                );
                assert.deepEqual(restarted.stored.database!.local, {
                    [doc.documentId]: remotePath
                });
                const events = await getJson<{
                    events: { requestId: string }[];
                }>(`${url}/events-since?after=0`, token);
                assert.equal(
                    events.events.filter(
                        (e) => e.requestId === pending.request.requestId
                    ).length,
                    1,
                    "Retry emitted a duplicate event"
                );
            } finally {
                await server.stop();
            }
        }
    );
}

test(
    "WebSocket hints trigger HTTP catchup in commit order, including after reconnect",
    { timeout: 20_000 },
    async () => {
        const server = makeServer();
        let ws: WebSocket | undefined;
        try {
            await server.start();
            const url = `${server.remoteUri}/vaults/order-${randomUUID()}`;
            const received: number[] = [];
            let catchup = Promise.resolve();
            let catchupError: unknown;
            const connect = async () => {
                ws = new WebSocket(`${url.replace(/^http/, "ws")}/ws`);
                ws.addEventListener("message", (message) => {
                    const hint = JSON.parse(String(message.data));
                    if (hint.type !== "vaultChanged") return;
                    catchup = catchup
                        .then(async () => {
                            assert.deepEqual(hint, { type: "vaultChanged" });
                            const batch = await getJson<{
                                events: { eventId: number }[];
                            }>(
                                `${url}/events-since?after=${received.at(-1) ?? 0}`,
                                token
                            );
                            received.push(
                                ...batch.events.map((event) => event.eventId)
                            );
                        })
                        .catch((error) => {
                            catchupError = error;
                        });
                });
                await new Promise<void>((resolve, reject) => {
                    ws!.onopen = () => resolve();
                    ws!.onerror = () => reject(new Error("WebSocket failed"));
                });
                ws.send(
                    JSON.stringify({
                        type: "handshake",
                        token,
                        deviceId: "order-observer"
                    })
                );
            };
            await connect();
            const ids = Array.from({ length: 12 }, () => randomUUID());
            const requests = ids.map((id, i) => ({
                id,
                body: content(`bytes-${i}`)
            }));
            const replies = await Promise.all(
                requests.map(({ id, body }) =>
                    put(`${url}/documents/${id}`, body)
                )
            );
            assert.deepEqual(
                replies.map((r) => r.vaultUpdateId).sort((a, b) => a - b),
                ids.map((_, i) => i + 1)
            );
            const emptyManifest = await getJson<{
                entries: Record<string, string>;
            }>(`${url}/file-manifest`, token);
            assert.deepEqual(
                emptyManifest.entries,
                {},
                "Content-only CAS changed membership"
            );
            const entries = Object.fromEntries(
                ids.map((id, i) => [id, `${i}.md`])
            );
            await put(`${url}/file-manifest`, {
                requestId: randomUUID(),
                parentFileManifestId: 0,
                entries
            });
            // All retries must return original receipts without emitting events.
            assert.deepEqual(
                await Promise.all(
                    requests.map(({ id, body }) =>
                        put(`${url}/documents/${id}`, body)
                    )
                ),
                replies
            );
            const waitForEvents = async (head: number) => {
                const deadline = Date.now() + 5_000;
                while (received.length < head && Date.now() < deadline)
                    await new Promise((resolve) => setTimeout(resolve, 10));
                if (catchupError) throw catchupError;
                assert.deepEqual(
                    received,
                    Array.from({ length: head }, (_, i) => i + 1),
                    "HTTP catchup after WebSocket hints lost or duplicated an event"
                );
            };
            await waitForEvents(13);
            await new Promise<void>((resolve) => {
                ws!.onclose = () => resolve();
                ws!.close();
            });
            await put(
                `${url}/documents/${ids[0]}`,
                content("next", replies[0].vaultUpdateId)
            );
            await connect();
            await waitForEvents(14);
            const log = await getJson<{ events: { eventId: number }[] }>(
                `${url}/events-since?after=0`,
                token
            );
            assert.deepEqual(
                received,
                log.events.map((event) => event.eventId)
            );
            const stale = await put(`${url}/file-manifest`, {
                requestId: randomUUID(),
                parentFileManifestId: 0,
                entries: {}
            });
            assert.equal(stale.type, "StaleBase");
            const invalid = await fetch(`${url}/file-manifest`, {
                method: "PUT",
                headers,
                body: JSON.stringify({
                    requestId: randomUUID(),
                    parentFileManifestId: 13,
                    entries: { [ids[0]]: "same", [ids[1]]: "same" }
                })
            });
            assert.equal(invalid.status, 422);
            assert.equal(
                (
                    await getJson<CanonicalSnapshot>(
                        `${url}/vault-snapshot`,
                        token
                    )
                ).headEventId,
                14,
                "Rejected CAS emitted an event"
            );
        } finally {
            ws?.close();
            await server.stop();
        }
    }
);