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
        /// 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, archived: Bool, ownerDeviceID: String) {
            self.sessionID = sessionID
            self.lastSeq = lastSeq
            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

    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, 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)
            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) }

    // 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
        }
        inFlight.remove(ref.sessionID)
        activeCount -= 1
    }

    /// 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, 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?, events: [AgentEvent]
    ) async -> Bool {
        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
    }
}
