import { apiFetch } from "../../net/api";
import { getICEServers } from "./iceServers";
import { useAppStore, type CandidateBreakdown, type IceStats, type TransferPhase } from "../../store";
import { deliverIncoming, openIncomingSink, type IncomingSink } from "./incomingSink";
import type { FileSource } from "./source";

// WebRTC client wired to the brief §2 invariants:
//   - iceServers: pulled from lib/iceServers (Cloudflare Realtime TURN over
//     TLS when CDROP_CF_TURN_* is configured; STUN-only fallback otherwise).
//     Refreshed on login by RootLayout.
//   - DataChannel "cdrop-file", ordered:true
//   - chunk size 64 KiB, backpressure thresholds 16 MiB / 4 MiB
//   - 接收端流式写 OPFS（lib/incomingSink），避免 iOS Safari 内存压力
//   - control frames as JSON strings (head meta + tail done), payload bytes binary
//   - signaling via POST /api/hub/signal, peer routed by deviceName
//
// trickle ICE: each candidate ships in its own signal frame the moment the
// browser surfaces it.
//
// Same-LAN Restricted Cone NAT 排查依据：iceCandidatePoolSize=4 预热候选；
// 收集本/远端 host/mdns/srflx/prflx/relay 计数 + 选中候选对类型推到 store，
// 让 UI 暴露"实际通过哪条路径连上"（host:host = 真直连，relay:* = 走了 TURN）。
const CHANNEL_NAME = "cdrop-file";
const CHUNK_SIZE = 64 * 1024;
const HIGH_WATERMARK = 16 * 1024 * 1024;
const LOW_WATERMARK = 4 * 1024 * 1024;
// 发送端「读写解耦」：一次读 4 MiB 进内存，再从内存切 CHUNK_SIZE 子片 dc.send。原先
// 每 64 KiB 都 await file.slice().arrayBuffer()，逐片异步文件读在 iOS Safari 上极慢（实测
// 直连 P2P 仅 ~250 KB/s，远低于中继 3 MB/s）。块读把异步文件读次数降 ~64 倍，SCTP 消息
// 大小与流控（HIGH/LOW watermark）完全不变，接收端无感。
const READ_BLOCK_SIZE = 4 * 1024 * 1024;
const ICE_CANDIDATE_POOL = 4;

export interface FileMeta
{
    name: string;
    size: number;
    sha256?: string;
}

export type P2PState = "connecting" | "connected" | "completed" | "failed" | "closed";

export interface P2PProgressEvent
{
    bytes: number;
    total: number;
}

export interface P2PSession
{
    sessionId: string;
    peerName: string;
    state: P2PState;
    onState: (cb: (s: P2PState) => void) => () => void;
    onProgress: (cb: (e: P2PProgressEvent) => void) => () => void;
    cancel: () => void;
}

interface SignalPayload
{
    type: "offer" | "answer" | "ice";
    sdp?: RTCSessionDescriptionInit;
    candidate?: RTCIceCandidateInit;
}

const EMPTY_BREAKDOWN: CandidateBreakdown =
    { host: 0, mdns: 0, srflx: 0, prflx: 0, relay: 0 };

class Session
{
    public state: P2PState = "connecting";
    public stateListeners = new Set<(s: P2PState) => void>();
    public progressListeners = new Set<(p: P2PProgressEvent) => void>();
    public pc: RTCPeerConnection;
    public dc: RTCDataChannel | null = null;
    public canceled = false;

    private receivedMeta: FileMeta | null = null;
    private receivedBytes = 0;
    // 接收端字节槽：onmessage 是同步回调，但 sink.write 是 async；用一根
    // promise 链保证 chunk 顺序写入，链尾的 then 处理 close。
    private receiveQueue: Promise<IncomingSink | null> = Promise.resolve(null);
    // Track total only for sender progress reporting; receiver learns it
    // from the meta frame.
    private outgoingTotal = 0;
    private bytesSent = 0;

    private phase: TransferPhase = "initializing";
    private localCandidates: CandidateBreakdown = { ...EMPTY_BREAKDOWN };
    private remoteCandidates: CandidateBreakdown = { ...EMPTY_BREAKDOWN };
    private selectedPair: { local: string; remote: string } | undefined;
    private firstByteSent = false;
    private firstByteReceived = false;
    // 节流：emitProgress 一秒可达数百次，每次 store.upsertTransfer 都触发
    // 一次 React 重渲染（含 DebugPanel 的 IceStats）。在 384 次 / 24 MB 的尺度
    // 上能把主线程拖到吃掉 dc.onmessage 的程度，使接收方 SCTP 缓冲填满 → rwnd=0
    // → sender 的 bufferedAmount 永远卡在 HIGH。把 store push 限到 10 Hz。
    private storePushScheduled: number | null = null;
    private lastMessageAt = 0;
    // 发送进度轮询：streamFile 期间每 250ms 刷一次，让"已交付字节"在 backpressure
    // 等待期间也能平滑爬升、不误报 stalled（见 streamFile / senderDeliveredBytes）。
    private sendProgressPollId: number | null = null;

    constructor(public sessionId: string, public peerName: string, public role: "sender" | "receiver")
    {
        this.pc = new RTCPeerConnection({
            iceServers: getICEServers(),
            iceCandidatePoolSize: ICE_CANDIDATE_POOL,
        });

        this.pc.onicecandidate = (ev) =>
        {
            if (!ev.candidate)
            {
                // gathering 完成；推一次包含最终计数的快照
                this.flushPushUpdate();
                return;
            }
            this.classifyAndCount(ev.candidate, this.localCandidates);
            this.flushPushUpdate();
            void this.sendSignal({ type: "ice", candidate: ev.candidate.toJSON() });
        };

        this.pc.onicegatheringstatechange = () =>
        {
            if (this.pc.iceGatheringState === "gathering" && this.phase === "initializing")
            {
                this.setPhase("ice_gathering");
            }
            else
            {
                this.flushPushUpdate();
            }
        };

        this.pc.oniceconnectionstatechange = () =>
        {
            if (this.canceled) { return; }
            const s = this.pc.iceConnectionState;
            if (s === "checking")
            {
                this.setPhase("ice_checking");
            }
            else if (s === "connected" || s === "completed")
            {
                this.setState("connected");
                if (this.phase === "ice_checking" || this.phase === "ice_gathering"
                    || this.phase === "initializing")
                {
                    this.setPhase("ice_connected");
                }
                void this.refreshSelectedPair();
                // ICE 报 connected 时 nominated pair 经常还没决定，一次 getStats
                // 拿不到就要轮询，否则 sender 调试面板永远显示"尚未选定"。
                this.startSelectedPairPolling();
            }
            else if (s === "failed")
            {
                this.setState("failed");
                void this.markServerFail("ice_failed");
            }
            this.flushPushUpdate();
        };

        if (role === "receiver")
        {
            this.pc.ondatachannel = (ev) =>
            {
                this.bindDataChannel(ev.channel);
                this.startReceiverWatchdog();
            };
        }
    }

    setState(s: P2PState): void
    {
        if (this.state === s) { return; }
        this.state = s;
        for (const cb of this.stateListeners) { cb(s); }
    }

    private setPhase(p: TransferPhase): void
    {
        if (this.phase === p) { return; }
        this.phase = p;
        // phase 变化是稀疏事件（< 10 次 / 整个传输），同步 push 不会拥塞渲染
        this.flushPushUpdate();
    }

    /** 立即 push（绕过节流），并取消任何 pending 节流 push */
    private flushPushUpdate(): void
    {
        if (this.storePushScheduled !== null)
        {
            window.clearTimeout(this.storePushScheduled);
            this.storePushScheduled = null;
        }
        this.pushUpdate();
    }

    /** 节流到 ≤10 Hz；高频 emitProgress 走这条 */
    private throttledPushUpdate(): void
    {
        if (this.storePushScheduled !== null) { return; }
        this.storePushScheduled = window.setTimeout(() =>
        {
            this.storePushScheduled = null;
            this.pushUpdate();
        }, 100);
    }

    private pushUpdate(): void
    {
        const store = useAppStore.getState();
        const cur = store.activeTransfers[this.sessionId];
        if (!cur) { return; }

        const isReceiver = this.role === "receiver";
        const bytes = isReceiver ? this.receivedBytes : this.senderDeliveredBytes();
        const lastProgressAt = bytes > (cur.bytesTransferred ?? 0)
            ? Date.now()
            : cur.lastProgressAt;

        const iceStats: IceStats = {
            gathering:  this.pc.iceGatheringState,
            connection: this.pc.iceConnectionState,
            candidates: {
                local:  { ...this.localCandidates },
                remote: { ...this.remoteCandidates },
            },
            selectedPair: this.selectedPair,
            dataChannel: this.dc
                ? { readyState: this.dc.readyState, bufferedAmount: this.dc.bufferedAmount }
                : undefined,
        };

        store.upsertTransfer({
            ...cur,
            phase: this.phase,
            bytesTransferred: bytes,
            lastProgressAt,
            iceStats,
        });
    }

    private emitProgress(): void
    {
        const total = this.role === "sender" ? this.outgoingTotal : (this.receivedMeta?.size ?? 0);
        const bytes = this.role === "sender" ? this.senderDeliveredBytes() : this.receivedBytes;
        for (const cb of this.progressListeners) { cb({ bytes, total }); }
        this.throttledPushUpdate();
    }

    /**
     * 发送端"已交付字节"＝已离开本地 buffer 的量（bytesSent − bufferedAmount）。
     * bytesSent 只是"已塞进 dc 的本地 16MiB buffer"，会被 64KB 一片秒满到 16MiB 后
     * 定格；真正送达网络的是离开 buffer 的部分。用它做进度，进度随接收端实际收程
     * 平滑爬升，而非跳到 buffer 满就冻住，也避免 backpressure 等待期被误判 stalled。
     * 单调非降：send 时 bytesSent 与 bufferedAmount 同增（delivered 持平），其余时刻
     * buffer 抽干使 bufferedAmount 降（delivered 升）。
     */
    private senderDeliveredBytes(): number
    {
        const buffered = this.dc?.bufferedAmount ?? 0;
        return Math.max(0, this.bytesSent - buffered);
    }

    private startSendProgressPoll(): void
    {
        if (this.sendProgressPollId !== null) { return; }
        this.sendProgressPollId = window.setInterval(() =>
        {
            if (this.canceled) { this.stopSendProgressPoll(); return; }
            this.emitProgress();
        }, 250);
    }

    private stopSendProgressPoll(): void
    {
        if (this.sendProgressPollId !== null)
        {
            window.clearInterval(this.sendProgressPollId);
            this.sendProgressPollId = null;
        }
    }

    private classifyAndCount(c: RTCIceCandidate | RTCIceCandidateInit, into: CandidateBreakdown): void
    {
        const sdp = (c as RTCIceCandidate).candidate ?? (c as RTCIceCandidateInit).candidate ?? "";
        if (!sdp) { return; }
        const typeMatch = sdp.match(/typ (host|srflx|prflx|relay)/);
        if (!typeMatch) { return; }
        const t = typeMatch[1] as "host" | "srflx" | "prflx" | "relay";
        if (t === "host")
        {
            // candidate:<foundation> <component> <protocol> <priority> <ip> <port> typ host …
            const addr = sdp.match(/^candidate:\S+ \d+ \S+ \d+ (\S+) /);
            if (addr && addr[1].endsWith(".local")) { into.mdns += 1; return; }
            into.host += 1;
            return;
        }
        into[t] += 1;
    }

    bindDataChannel(dc: RTCDataChannel): void
    {
        this.dc = dc;
        dc.binaryType = "arraybuffer";
        dc.bufferedAmountLowThreshold = LOW_WATERMARK;

        dc.onopen = () =>
        {
            // 接收端的 dc.onopen 与发送端被分别绑定（发送端在 startOutgoing 里覆盖
            // 这个 handler 来 streamFile）。这里覆盖掉的概率仅在 receiver。
            if (this.role === "receiver" && this.phase !== "transferring")
            {
                this.setPhase("dc_open");
            }
        };
        dc.onmessage = (ev) => { this.handleMessage(ev.data); };
        dc.onerror = () => { this.setState("failed"); };
        dc.onclose = () =>
        {
            if (this.state !== "completed" && this.state !== "failed")
            {
                this.setState("closed");
            }
        };
    }

    private handleMessage(data: string | ArrayBuffer | Blob): void
    {
        this.lastMessageAt = Date.now();
        if (typeof data === "string")
        {
            try
            {
                const msg = JSON.parse(data);
                if (msg.type === "meta")
                {
                    this.receivedMeta = { name: msg.name, size: msg.size, sha256: msg.sha256 };
                    // 提前打开 sink；后续 chunk 入队等 sink 就绪。带上文件名，桌面大文件
                    // spill 落盘后据此命名。
                    this.receiveQueue = openIncomingSink(this.sessionId, msg.name);
                    this.emitProgress();
                }
                else if (msg.type === "done")
                {
                    this.setPhase("completing");
                    void this.finalizeIncoming();
                }
            }
            catch
            {
                // ignore malformed control frame
            }
            return;
        }
        if (data instanceof ArrayBuffer)
        {
            if (!this.firstByteReceived) { this.firstByteReceived = true; this.setPhase("transferring"); }
            this.receivedBytes += data.byteLength;
            // 串到 promise 链尾；sink.write 完成才会处理下一个 chunk，与 SCTP
            // rwnd 共同提供 backpressure。链上失败会沿着 finalize 抛出来。
            this.receiveQueue = this.receiveQueue.then(async (sink) =>
            {
                if (sink) { await sink.write(data); }
                return sink;
            });
            this.emitProgress();
        }
    }

    /** 收方看门狗：超过 5s 没收到任何消息（onmessage 没 fire）即 console.warn，
     *  方便诊断主线程被微任务堵住的情形。仅 receiver 角色起作用。 */
    private startReceiverWatchdog(): void
    {
        if (this.role !== "receiver") { return; }
        this.lastMessageAt = Date.now();
        const id = window.setInterval(() =>
        {
            if (this.canceled || this.state === "completed" || this.state === "failed"
                || this.state === "closed")
            {
                window.clearInterval(id);
                return;
            }
            if (!this.firstByteReceived) { return; }
            const since = Date.now() - this.lastMessageAt;
            if (since > 5_000)
            {
                // eslint-disable-next-line no-console
                console.warn("p2p receiver: no onmessage for >5s", {
                    sinceMs: since,
                    bytesReceived: this.receivedBytes,
                    bufferedAmount: this.dc?.bufferedAmount,
                    readyState: this.dc?.readyState,
                });
            }
        }, 5_000);
    }

    private async finalizeIncoming(): Promise<void>
    {
        if (!this.receivedMeta) { return; }
        try
        {
            const sink = await this.receiveQueue;
            if (!sink) { throw new Error("receive sink missing"); }
            const result = await sink.close();
            deliverIncoming(result, this.receivedMeta.name, this.sessionId);
            this.setState("completed");
            void this.markServerDone(this.receivedBytes);
        }
        catch (e)
        {
            // eslint-disable-next-line no-console
            console.error("finalize incoming failed", e);
            this.setState("failed");
            void this.markServerFail((e as Error).message ?? "finalize_failed");
        }
    }

    async sendSignal(payload: SignalPayload): Promise<void>
    {
        const r = await apiFetch("/api/hub/signal", {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({ to: this.peerName, payload }),
        });
        if (!r.ok && r.status !== 410)
        {
            // 410 means peer offline; signaling fails silently (caller will time out)
            // eslint-disable-next-line no-console
            console.warn("p2p signal HTTP", r.status, await r.text().catch(() => ""));
        }
    }

    async handleSignal(payload: SignalPayload): Promise<void>
    {
        if (payload.type === "offer" && payload.sdp)
        {
            await this.pc.setRemoteDescription(payload.sdp);
            const answer = await this.pc.createAnswer();
            await this.pc.setLocalDescription(answer);
            await this.sendSignal({ type: "answer", sdp: answer });
        }
        else if (payload.type === "answer" && payload.sdp)
        {
            await this.pc.setRemoteDescription(payload.sdp);
        }
        else if (payload.type === "ice" && payload.candidate)
        {
            this.classifyAndCount(payload.candidate, this.remoteCandidates);
            this.pushUpdate();
            try { await this.pc.addIceCandidate(payload.candidate); }
            catch (e)
            {
                // eslint-disable-next-line no-console
                console.warn("p2p addIceCandidate", e);
            }
        }
    }

    async startOutgoing(src: FileSource): Promise<void>
    {
        this.outgoingTotal = src.size;
        const dc = this.pc.createDataChannel(CHANNEL_NAME, { ordered: true });
        this.bindDataChannel(dc);
        // bindDataChannel 给 receiver 绑了 onopen；sender 这里覆盖一下，等
        // ICE 通道一就绪立刻开始 streamFile（receiver 不会走这条分支）。
        dc.onopen = () =>
        {
            this.setPhase("dc_open");
            void this.streamFile(src).catch((e) =>
            {
                // eslint-disable-next-line no-console
                console.error("p2p send failed", e);
                this.stopSendProgressPoll();
                this.setState("failed");
            });
        };

        const offer = await this.pc.createOffer();
        await this.pc.setLocalDescription(offer);
        await this.sendSignal({ type: "offer", sdp: offer });
    }

    private async streamFile(src: FileSource): Promise<void>
    {
        if (!this.dc) { throw new Error("no data channel"); }
        const dc = this.dc;

        // Tell the backend that the channel is live so /done is later legal:
        // the state machine demands ACCEPTED → P2P_ACTIVE → DONE.
        await this.markP2PActive();

        // Header.
        const meta: FileMeta = { name: src.name, size: src.size };
        dc.send(JSON.stringify({ type: "meta", ...meta }));

        // buffer 会被 64KB 一片秒满到 16MiB，之后发送循环长时间阻塞在 waitForBufferLow
        // 等抽干，期间没有 send、也没有 emitProgress。这个轮询在等待期持续刷新进度，让
        // "已交付字节"随接收端实际收程平滑爬升、lastProgressAt 保持新鲜，不误报"疑似卡住"。
        this.startSendProgressPoll();

        // 发送块大小固定 64 KiB：实测放大到 256 KiB 在 iOS Safari 上反而把吞吐打到
        // <100 KB/s（SCTP 大消息分片/重组代价），故维持 CHUNK_SIZE。读写解耦（块读）保留。
        const sendChunk = CHUNK_SIZE;

        let offset = 0;
        while (offset < src.size)
        {
            // 一次读一大块进内存（异步文件读从每 64 KiB 一次降到每 4 MiB 一次）。iOS 下
            // src.slice 经 Range 向原生拉取该块，整文件不进 WebView 内存（R-iOS-4）。
            const blockEnd = Math.min(offset + READ_BLOCK_SIZE, src.size);
            const block = await src.slice(offset, blockEnd);

            // 从内存块按 sendChunk 子片发送；流控（bufferedAmount/水位）逻辑不变。
            let inBlock = 0;
            while (inBlock < block.byteLength)
            {
                // 预测式检查：在 send 之前看"加了这一片之后会不会撞上 HIGH"。
                // Chrome WebRTC 的 RTCDataChannel.send 在 bufferedAmount + size 超过
                // kMaxQueuedSendDataBytes（16 MiB）时直接抛 OperationError；我们 HIGH
                // 也设 16 MiB，原本的 `bufferedAmount > HIGH` 检查在 buffer = HIGH - 1
                // 时不触发，下一片就会跨过硬上限挂掉。
                if (dc.bufferedAmount + sendChunk > HIGH_WATERMARK)
                {
                    await waitForBufferLow(dc, LOW_WATERMARK);
                }
                const end = Math.min(inBlock + sendChunk, block.byteLength);
                // subarray 是零拷贝视图；dc.send 接受 ArrayBufferView，只发该视图字节范围。
                const chunk = block.subarray(inBlock, end);
                try
                {
                    dc.send(chunk);
                }
                catch (e)
                {
                    // 兜底：极端情况下 bufferedAmount 在 await 与 send 之间漂移，仍可能
                    // 撞到 OperationError。退一步等 buffer 抽干到 LOW 再重试一次；仍失
                    // 败说明通道本身有问题，让外层 .catch 走 markServerFail。
                    if (e instanceof DOMException && e.name === "OperationError")
                    {
                        // eslint-disable-next-line no-console
                        console.warn("p2p send queue full, backing off then retrying", {
                            bufferedAmount: dc.bufferedAmount,
                            readyState: dc.readyState,
                            bytesSent: this.bytesSent,
                        });
                        await waitForBufferLow(dc, LOW_WATERMARK);
                        dc.send(chunk);
                    }
                    else
                    {
                        throw e;
                    }
                }
                const sent = end - inBlock;
                inBlock += sent;
                offset += sent;
                this.bytesSent = offset;
                if (!this.firstByteSent) { this.firstByteSent = true; this.setPhase("transferring"); }
                this.emitProgress();
            }
        }

        dc.send(JSON.stringify({ type: "done" }));
        this.setPhase("completing");
        // 抽干整个 buffer（数据 + done 帧）再宣告完成：dc.send 只是入本地 buffer，真正
        // 送达要等 SCTP 发出。早先送完入 buffer 即 setState("completed")，会在还有多达
        // 16MiB 在途时就标记完成，与接收端"仍在下载"不一致。等 bufferedAmount 落到 0＝
        // 字节已全部交付网络，接收端随后 finalize 并 POST /done（那条权威 /done 仍由接收
        // 端发，这里只在抽干后驱动发送端自身 UI 收尾）。
        await waitForBufferLow(dc, 0);
        this.stopSendProgressPoll();
        this.bytesSent = src.size;
        this.emitProgress();
        this.setState("completed");
        // /done is intentionally posted by the *receiver* only — they are the
        // authoritative "got it" signal. Doubling it up here just produces
        // 409 console noise once the receiver's call lands first.
    }

    private selectedPairPollId: number | null = null;

    private async refreshSelectedPair(): Promise<boolean>
    {
        try
        {
            const stats = await this.pc.getStats();
            const cands = new Map<string, RTCStats & { candidateType?: string; protocol?: string }>();
            let pairLocalId: string | undefined;
            let pairRemoteId: string | undefined;

            stats.forEach((report) =>
            {
                const r = report as RTCStats & {
                    candidateType?: string; protocol?: string;
                    state?: string; nominated?: boolean;
                    localCandidateId?: string; remoteCandidateId?: string;
                };
                if (r.type === "local-candidate" || r.type === "remote-candidate")
                {
                    cands.set(r.id, r);
                }
                if (r.type === "candidate-pair" && r.state === "succeeded" && r.nominated)
                {
                    pairLocalId = r.localCandidateId;
                    pairRemoteId = r.remoteCandidateId;
                }
            });
            if (pairLocalId && pairRemoteId)
            {
                this.selectedPair = {
                    local:  formatCandidateLabel(cands.get(pairLocalId)),
                    remote: formatCandidateLabel(cands.get(pairRemoteId)),
                };
                this.flushPushUpdate();
                return true;
            }
            return false;
        }
        catch
        {
            // 老浏览器或权限问题，吞掉，调试面板就显示不出 selectedPair
            return true; // 不再重试
        }
    }

    /** 一次 refreshSelectedPair 经常拿不到（连接刚 connected，nominated pair
     *  还没确定）。每 2s 轮询一次直到拿到，或 30s 超时，或会话进入终态。 */
    private startSelectedPairPolling(): void
    {
        if (this.selectedPairPollId !== null) { return; }
        const start = Date.now();
        this.selectedPairPollId = window.setInterval(async () =>
        {
            if (this.canceled
                || this.selectedPair
                || this.state === "completed"
                || this.state === "failed"
                || this.state === "closed"
                || Date.now() - start > 30_000)
            {
                if (this.selectedPairPollId !== null)
                {
                    window.clearInterval(this.selectedPairPollId);
                    this.selectedPairPollId = null;
                }
                return;
            }
            await this.refreshSelectedPair();
        }, 2_000);
    }

    private async markP2PActive(): Promise<void>
    {
        try
        {
            // 409 is fine — receiver may have already moved the state, or the
            // session might be in a state ineligible for P2P_ACTIVE.
            await apiFetch(`/api/transfer/${this.sessionId}/p2p`, { method: "POST" });
        }
        catch (e)
        {
            // eslint-disable-next-line no-console
            console.warn("p2p mark p2p_active", e);
        }
    }

    private async markServerDone(bytes: number): Promise<void>
    {
        // 重试到服务端真的接到为止：发完文件之后这条 POST 不到，整个状态机
        // 会停在 P2P_ACTIVE，UI 永远不进入 history。但 409 表示状态已经不允许
        // 转 DONE（多半是已 DONE / FAILED / CANCELLED 在终态），直接放弃。
        for (let attempt = 0; attempt < 3; attempt += 1)
        {
            try
            {
                const r = await apiFetch(`/api/transfer/${this.sessionId}/done`, {
                    method: "POST",
                    headers: { "Content-Type": "application/json" },
                    body: JSON.stringify({ bytes_transferred: bytes }),
                });
                if (r.ok) { return; }
                if (r.status === 409) { return; }
                const text = await r.text().catch(() => r.statusText);
                // eslint-disable-next-line no-console
                console.warn(`p2p mark done attempt ${attempt + 1} → ${r.status}: ${text}`);
            }
            catch (e)
            {
                // eslint-disable-next-line no-console
                console.warn(`p2p mark done attempt ${attempt + 1} error`, e);
            }
            if (attempt < 2)
            {
                await new Promise((r) => setTimeout(r, 500 * (attempt + 1)));
            }
        }
    }

    async markServerFail(reason: string): Promise<void>
    {
        try
        {
            await apiFetch(`/api/transfer/${this.sessionId}/fail`, {
                method: "POST",
                headers: { "Content-Type": "application/json" },
                body: JSON.stringify({ reason }),
            });
        }
        catch (e)
        {
            // eslint-disable-next-line no-console
            console.warn("p2p mark fail", e);
        }
    }

    cancel(): void
    {
        this.canceled = true;
        if (this.storePushScheduled !== null)
        {
            window.clearTimeout(this.storePushScheduled);
            this.storePushScheduled = null;
        }
        if (this.selectedPairPollId !== null)
        {
            window.clearInterval(this.selectedPairPollId);
            this.selectedPairPollId = null;
        }
        this.stopSendProgressPoll();
        this.dc?.close();
        this.pc.close();
        this.setState("closed");
        // 把可能开着的 OPFS 临时文件删干净
        this.receiveQueue = this.receiveQueue.then(async (sink) =>
        {
            try { await sink?.cancel(); } catch { /* swallow */ }
            return null;
        });
    }
}

function formatCandidateLabel(c: undefined | (RTCStats & { candidateType?: string; protocol?: string })): string
{
    if (!c) { return "?"; }
    return `${c.candidateType ?? "?"}/${c.protocol ?? "?"}`;
}

// waitForBufferLow 三重保险等 dc.bufferedAmount 落到 threshold 以下：
//   1. bufferedamountlow 事件（标准路径）
//   2. addEventListener 之后再 re-check（防 attach-before-fire 竞态）
//   3. 100ms 轮询（兜底某些 Chromium 版本事件不触发的历史 bug）
// 30s 没下来则 console.warn 但不中断（同时由 polling 接力，最终一定会
// 等到）；这条日志能定位"sender 静默卡住"是不是这条 await。
function waitForBufferLow(dc: RTCDataChannel, threshold: number): Promise<void>
{
    if (dc.bufferedAmount <= threshold) { return Promise.resolve(); }
    return new Promise<void>((resolve) =>
    {
        let resolved = false;
        const start = Date.now();
        const cleanup = () =>
        {
            window.clearInterval(pollId);
            window.clearTimeout(watchdog);
            dc.removeEventListener("bufferedamountlow", onLow);
        };
        const finish = () =>
        {
            if (resolved) { return; }
            resolved = true;
            cleanup();
            resolve();
        };
        const onLow = () => { finish(); };
        const pollId = window.setInterval(() =>
        {
            if (dc.bufferedAmount <= threshold) { finish(); }
        }, 100);
        const watchdog = window.setTimeout(() =>
        {
            // eslint-disable-next-line no-console
            console.warn("p2p waitForBufferLow stuck", {
                bufferedAmount: dc.bufferedAmount,
                threshold,
                readyState: dc.readyState,
                elapsedMs: Date.now() - start,
            });
        }, 30_000);
        dc.addEventListener("bufferedamountlow", onLow);
        // re-check: addEventListener 之后 bufferedAmount 可能已经在 attach
        // 前的最近 task 里跨过 threshold（事件已 fire 但我们没有监听），
        // 此时 polling 会兜底但有 100ms 延迟；这里同步再看一眼短路掉。
        if (dc.bufferedAmount <= threshold) { finish(); }
    });
}

const p2pSessions = new Map<string, Session>();

export function p2pHandleSignal(from: string, payload: SignalPayload): void
{
    const sess = p2pSessions.get(from);
    if (!sess)
    {
        // eslint-disable-next-line no-console
        console.warn("p2p: signal from unknown peer", from, payload?.type);
        return;
    }
    void sess.handleSignal(payload);
}

export function p2pCleanup(peerName: string): void
{
    const sess = p2pSessions.get(peerName);
    if (sess) { sess.cancel(); }
    p2pSessions.delete(peerName);
}

function buildPublicSession(sess: Session): P2PSession
{
    return {
        sessionId: sess.sessionId,
        peerName: sess.peerName,
        get state() { return sess.state; },
        onState: (cb) =>
        {
            sess.stateListeners.add(cb);
            return () => sess.stateListeners.delete(cb);
        },
        onProgress: (cb) =>
        {
            sess.progressListeners.add(cb);
            return () => sess.progressListeners.delete(cb);
        },
        cancel: () => { p2pCleanup(sess.peerName); },
    };
}

// p2pStartOutgoing returns *synchronously* — startOutgoing's createOffer /
// setLocalDescription / signal POST are all kicked off as a background promise.
// This is critical for the 30s ICE-fallback watchdog (transfer.ts) to actually
// arm: previously an `await p2pStartOutgoing` could hang forever if e.g.
// createOffer was monkey-patched to never resolve (M11 fallback test scenario).
export function p2pStartOutgoing(
    sessionId: string,
    receiverName: string,
    src: FileSource,
): P2PSession
{
    const sess = new Session(sessionId, receiverName, "sender");
    p2pSessions.set(receiverName, sess);
    sess.stateListeners.add((s) => updateStoreState(sessionId, s));
    void sess.startOutgoing(src).catch((e) =>
    {
        // Cancellation (fallback path closes pc → pending createOffer rejects)
        // is expected, not a transport failure — let the relay flow drive state.
        if (sess.canceled) { return; }
        // eslint-disable-next-line no-console
        console.error("p2p start outgoing failed", e);
        sess.setState("failed");
        void sess.markServerFail((e as Error).message ?? "p2p_start_failed");
    });
    return buildPublicSession(sess);
}

export function p2pStartIncoming(sessionId: string, senderName: string): P2PSession
{
    const sess = new Session(sessionId, senderName, "receiver");
    p2pSessions.set(senderName, sess);
    sess.stateListeners.add((s) => updateStoreState(sessionId, s));
    return buildPublicSession(sess);
}

function updateStoreState(sessionId: string, p2pState: P2PState): void
{
    const store = useAppStore.getState();
    const cur = store.activeTransfers[sessionId];
    if (!cur) { return; }
    if (p2pState === "connected") { store.upsertTransfer({ ...cur, state: "P2P_ACTIVE" }); }
    else if (p2pState === "completed") { store.completeTransfer(sessionId, "DONE"); }
    else if (p2pState === "failed") { store.completeTransfer(sessionId, "FAILED"); }
    // closed 不主动移到 history：可能是 fallback 路径主动 cancel，让上层
    // /fallback / /fail 流程决定终态。
}