import Foundation
import NucleicProtocol

/// Keeps this Mac's transcript mirrors caught up with the mesh (full-transcript mesh sync). On
/// each reconcile pass it compares what it holds (`TranscriptMirrorStore.heldSeq`) against every
/// active session's tip (`SessionSummary.lastSeq`) as reported by the peers, and backfills the
/// gap from a holder — so opening a peer's session renders full history instantly instead of a
/// 200-event tail, and this Mac can in turn serve that history to a third device.
///
/// Holder selection probes rather than gossips: the session's **owner** is tried first (freshest,
/// a live controller), then any other reachable peer that might already mirror it — a peer without
/// it replies `notHeld` and the coordinator falls through. That makes the "owner offline, pull
/// from another synced Mac" case fall out for free.
///
/// This runs only on Macs — it lives in `AppStore` (the Mac authority); phones fetch history on
/// demand instead of proactively mirroring. The `fetch` dependency is injected so the coordinator
/// is testable against a fake holder without a live mesh.
public actor TranscriptSyncCoordinator {
    /// One session the mesh currently reports, for a reconcile pass.
    public struct SessionRef: Sendable {
        public let sessionID: SessionID
        /// The owner's current transcript tip — the target the mirror catches up to.
        public let lastSeq: UInt64
        /// The owner's advertised revert epoch (`SessionSummary.revertEpoch`) — how many times
        /// the transcript has been reverted/rewritten. A mirror synced at a different epoch holds
        /// (potentially) stale content at reused seqs even when its cursor is at/below `lastSeq`,
        /// so it must be healed, not topped up. `nil` = an owner that predates the field (no
        /// epoch comparison possible; only the `held > lastSeq` heal applies).
        public let revertEpoch: UInt64?
        /// Archived sessions are out of proactive scope — their mirror is dropped, not fetched.
        public let archived: Bool
        /// The device that owns/runs the session — the preferred (freshest) holder to fetch from.
        public let ownerDeviceID: String

        public init(
            sessionID: SessionID, lastSeq: UInt64, revertEpoch: UInt64? = nil,
            archived: Bool, ownerDeviceID: String
        ) {
            self.sessionID = sessionID
            self.lastSeq = lastSeq
            self.revertEpoch = revertEpoch
            self.archived = archived
            self.ownerDeviceID = ownerDeviceID
        }
    }

    private let mirrorStore: TranscriptMirrorStore
    private let fetch: @Sendable (SessionID, UInt64, String) async
        -> AsyncStream<PeerClient.TranscriptFetchUpdate>
    private let maxConcurrent: Int
    /// Sessions with a backfill in flight — the single-fetch-per-session guard. A `sessionUpdated`
    /// that advances the tip mid-fetch is harmless: the next reconcile tick sees the residual gap
    /// once the fetch clears the guard, and tops it up.
    private var inFlight: Set<SessionID> = []
    private var activeCount = 0
    /// Foreground opens join an existing mirror fetch instead of issuing a second full-history
    /// subscription. Waiters are resumed after success or failure; the caller then reads whatever
    /// mirror is available and falls back to the ordinary live subscription when necessary.
    private var completionWaiters: [SessionID: [CheckedContinuation<Void, Never>]] = [:]
    /// The owner epoch each session was last *epoch-healed* at — the loop guard for a heal that
    /// can't be confirmed. A heal refetches from any holder; an old-build holder re-serves a
    /// header whose epoch was stripped at its own mirror write, so the refetched copy still can't
    /// prove it matches the owner's epoch and would re-trigger the mismatch every tick. While the
    /// owner itself is unreachable we heal once per advertised epoch and then hold what we got
    /// (best effort); the owner appearing among the candidates bypasses this guard, and an
    /// owner-served refetch stamps a definitive epoch that ends the cycle. In-memory only — a
    /// relaunch retries at most one extra heal per session.
    private var attemptedHealEpochs: [SessionID: UInt64] = [:]

    public init(
        mirrorStore: TranscriptMirrorStore,
        maxConcurrent: Int = 3,
        fetch: @escaping @Sendable (SessionID, UInt64, String) async
            -> AsyncStream<PeerClient.TranscriptFetchUpdate>
    ) {
        self.mirrorStore = mirrorStore
        self.maxConcurrent = maxConcurrent
        self.fetch = fetch
    }

    /// Reconcile the mesh's active sessions against local mirrors: drop mirrors for archived
    /// sessions, self-heal mirrors that missed a revert (tip regression, or a revert-epoch
    /// mismatch when the owner regrew past our cursor), and start backfills (up to
    /// `maxConcurrent`) for active ones that are behind. `candidates` are the
    /// reachable holder deviceIDs to probe — the caller restricts these to LAN/Tailnet peers so
    /// proactive bulk sync never rides the metered relay.
    public func sync(sessions: [SessionRef], candidates: [String]) async {
        for ref in sessions {
            if ref.archived {
                if await mirrorStore.has(ref.sessionID) { await mirrorStore.remove(ref.sessionID) }
                continue
            }
            guard activeCount < maxConcurrent, !inFlight.contains(ref.sessionID) else { continue }
            let held = await mirrorStore.heldSeq(ref.sessionID)
            let diverged = await epochDiverged(ref)
            if held > ref.lastSeq || diverged {
                // A revert/undo we missed (offline, or a transport that dropped the live
                // `.transcriptReverted` fan-out). Two detectors:
                // - `held > lastSeq`: the owner's tip regressed below what we hold — the revert
                //   dropped more than any regrowth added back.
                // - epoch mismatch: the owner's advertised `revertEpoch` differs from the one our
                //   mirror was synced at — catches the case tip arithmetic can't: the owner
                //   reverted and then *regrew to the same length or longer*, reusing our held
                //   seqs with different content.
                // Either way we can't just truncate our tail to the new tip: a revert resets the
                // canonical seq counter, so post-revert turns reuse the dropped seqs with
                // *different* content — our overlapping seqs may be stale. So drop the mirror and
                // refetch the owner's authoritative history from scratch. This is what makes undo
                // converge on a device that was offline when it happened; the live path
                // (`AppStore.applyPeerRevert`) handles connected devices.
                // Epoch-only heals are guarded to run once per advertised epoch while the owner
                // is unreachable (`attemptedHealEpochs`): a refetch from an old-build holder
                // can't confirm the epoch, and re-dropping every tick would loop.
                if held <= ref.lastSeq, let ownerEpoch = ref.revertEpoch,
                    attemptedHealEpochs[ref.sessionID] == ownerEpoch,
                    !candidates.contains(ref.ownerDeviceID)
                { continue }
                if let ownerEpoch = ref.revertEpoch {
                    attemptedHealEpochs[ref.sessionID] = ownerEpoch
                }
                let holders = orderedCandidates(ref: ref, candidates: candidates)
                inFlight.insert(ref.sessionID)
                activeCount += 1
                Task { await self.runHeal(ref, holders: holders) }
                continue
            }
            guard ref.lastSeq > held else { continue }
            let holders = orderedCandidates(ref: ref, candidates: candidates)
            guard !holders.isEmpty else { continue }
            // Reserve synchronously (no await between here and the spawn) so a second reconcile
            // tick can't double-start the same session or overshoot the concurrency cap.
            inFlight.insert(ref.sessionID)
            activeCount += 1
            Task { await self.runFetch(ref, held: held, holders: holders) }
        }
    }

    /// Whether a backfill is currently running for `id` — for tests/observability.
    public func isFetching(_ id: SessionID) -> Bool { inFlight.contains(id) }

    /// Promote one session for an interactive open. If background reconciliation is already
    /// fetching it, join that exact transfer. If it was skipped because all background slots were
    /// occupied, start it immediately in a foreground overflow slot. This deliberately may take
    /// `activeCount` one past `maxConcurrent`: user interaction outranks speculative mirroring,
    /// while repeated opens of the same transcript remain single-flight.
    public func prioritize(
        _ ref: SessionRef, candidates: [String]
    ) async {
        guard !ref.archived else { return }
        if inFlight.contains(ref.sessionID) {
            await waitForCompletion(ref.sessionID)
            return
        }

        let held = await mirrorStore.heldSeq(ref.sessionID)
        let diverged = await epochDiverged(ref)
        guard held > ref.lastSeq || diverged || ref.lastSeq > held else { return }
        let holders = orderedCandidates(ref: ref, candidates: candidates)
        guard !holders.isEmpty else { return }

        inFlight.insert(ref.sessionID)
        activeCount += 1
        if held > ref.lastSeq || diverged {
            Task { await self.runHeal(ref, holders: holders) }
        } else {
            Task { await self.runFetch(ref, held: held, holders: holders) }
        }
        await waitForCompletion(ref.sessionID)
    }

    private func waitForCompletion(_ id: SessionID) async {
        guard inFlight.contains(id) else { return }
        await withCheckedContinuation { continuation in
            completionWaiters[id, default: []].append(continuation)
        }
    }

    /// Whether the mirror for `ref` was synced at a different revert epoch than the owner now
    /// advertises — the tip-independent "we missed a revert" detector. No mirror or no advertised
    /// epoch (old owner) ⇒ false. A mirror with no recorded epoch (written before the field, or
    /// through an old-build holder that stripped it) counts as diverged only when the owner has
    /// ever reverted (`revertEpoch > 0`) — its history *may* include a rewrite we can't rule out;
    /// a never-reverted session's old mirror is fine as-is.
    private func epochDiverged(_ ref: SessionRef) async -> Bool {
        guard let ownerEpoch = ref.revertEpoch, await mirrorStore.has(ref.sessionID) else {
            return false
        }
        guard let heldEpoch = await mirrorStore.heldEpoch(ref.sessionID) else {
            return ownerEpoch > 0
        }
        return heldEpoch != ownerEpoch
    }

    // MARK: - Fetch

    private func runFetch(_ ref: SessionRef, held: UInt64, holders: [String]) async {
        for holder in holders {
            if await attempt(ref, held: held, holder: holder) { break }  // stop at the first that serves it
        }
        finish(ref.sessionID)
    }

    /// Heal a mirror that missed a revert (its cursor outran the owner's tip, or its revert
    /// epoch mismatches the owner's): drop the stale copy, then
    /// refetch the owner's full history from scratch (`held: 0`). The drop comes *first* so that if
    /// every holder is unreachable we're left holding nothing — a later on-demand open refetches
    /// fresh — rather than continuing to serve the reverted tail. A successful full fetch rewrites
    /// the mirror via `writeFull`; a failing one leaves it dropped.
    private func runHeal(_ ref: SessionRef, holders: [String]) async {
        await mirrorStore.remove(ref.sessionID)
        for holder in holders {
            if await attempt(ref, held: 0, holder: holder) { break }  // stop at the first that serves it
        }
        finish(ref.sessionID)
    }

    private func finish(_ id: SessionID) {
        inFlight.remove(id)
        activeCount = max(0, activeCount - 1)
        for waiter in completionWaiters.removeValue(forKey: id) ?? [] { waiter.resume() }
    }

    /// Drain one holder's fetch stream into a buffer and, on `complete`, commit it to the mirror.
    /// Returns true iff the mirror was updated (⇒ stop probing); false ⇒ try the next holder.
    private func attempt(_ ref: SessionRef, held: UInt64, holder: String) async -> Bool {
        let stream = await fetch(ref.sessionID, held, holder)
        var buffer: [AgentEvent] = []
        for await update in stream {
            switch update {
            case .chunk(let chunk):
                buffer.append(contentsOf: chunk.events)
            case .complete(let done):
                let header = done.headerJSON.flatMap { try? SessionHeader.decodeFromWire($0) }
                return await commit(
                    ref, held: held, holder: holder, header: header,
                    servedEpoch: done.revertEpoch, events: buffer)
            case .unavailable, .disconnected:
                return false
            }
        }
        return false  // stream ended without a terminal `complete`
    }

    private func commit(
        _ ref: SessionRef, held: UInt64, holder: String,
        header: SessionHeader?, servedEpoch: UInt64?, events: [AgentEvent]
    ) async -> Bool {
        // Verify the holder served the epoch the owner advertises before committing anything:
        // a mismatch means the holder's copy predates (or postdates — the fetch raced a fresh
        // revert) the transcript generation we're converging to, and stitching it in would plant
        // stale content at reused seqs. Reject ⇒ the caller probes the next holder; the raced
        // case self-corrects on the next reconcile tick against the new advertised epoch. An
        // unknown epoch (old-build holder: nil `servedEpoch`, and a re-encoded mirror header that
        // stripped the field) is accepted best-effort — `attemptedHealEpochs` keeps that from
        // looping, and an owner-served fetch always carries a definitive epoch.
        if let expected = ref.revertEpoch {
            if let servedEpoch, servedEpoch != expected { return false }
            if held == 0, let headerEpoch = header?.revertEpoch, headerEpoch != expected {
                return false
            }
        }
        do {
            if held == 0 {
                // A full fetch must carry the header so the mirror file's line 0 is faithful.
                guard let header else { return false }
                try await mirrorStore.writeFull(
                    ref.sessionID, sourceDeviceID: holder, header: header, events: events)
            } else {
                try await mirrorStore.append(ref.sessionID, events: events)
            }
            return true
        } catch {
            return false
        }
    }

    /// Owner first (freshest), then every other reachable peer (any may hold a mirror). `candidates`
    /// is already restricted to reachable/allowed transports, so an owner reachable only over a
    /// disallowed transport simply isn't in the list.
    private func orderedCandidates(ref: SessionRef, candidates: [String]) -> [String] {
        var order: [String] = []
        if candidates.contains(ref.ownerDeviceID) { order.append(ref.ownerDeviceID) }
        order.append(contentsOf: candidates.filter { $0 != ref.ownerDeviceID })
        return order
    }
}
