import Foundation

/// The initiator-side sync engine (the phone). Mirrors the host's `ConnectionHandler`: sends
/// the plaintext preamble, runs the Noise handshake (XXpsk0 to pair, IK to reconnect),
/// exchanges hello/welcome, then surfaces every `HostMsg` as an `Event` and accepts
/// `ClientMsg`s via `send`. Transport-agnostic over `FrameChannel`, so the iOS app wires it
/// to NWConnection and tests wire it to an in-memory pipe.
public actor SyncClient {
    public enum Mode: Sendable {
        case pair(secret: Data)   // XXpsk0, first-time pairing
        case reconnect            // IK against the pinned host static
    }

    public enum Event: Sendable {
        case connecting
        case ready(Welcome)
        case sessionList([SessionSummary])
        case snapshot(SessionSnapshot)
        case events(EventBatch)
        case approvalRequested(ApprovalRequest)
        case approvalResolved(ApprovalResolved)
        case sessionUpdated(SessionSummary)
        case dashboard(DashboardSnapshot)
        case sessionDiff(WireSessionDiff)
        /// The host's mesh peer list, answering `ClientMsg.listPeers` (mesh P4).
        case peerList([PeerSummary])
        /// The host's mesh roster (mesh "join") — live members + revocation tombstones. The
        /// initiator merges it: a Mac auto-pins unknown mac members, a phone adds them to its
        /// paired-hosts registry, and both drop tombstoned devices.
        case meshRoster(MeshRosterPush)
        // Session transfer replies (mesh P5) — the source drives a transfer as a client of the
        // destination and observes the destination's answers through these events.
        case transferAccept(TransferAccept)
        case transferReject(TransferReject)
        case transferReady(String)
        case transferCommitted(String)
        case transferChunkAck(TransferChunkAck)
        // Full-transcript mesh sync replies — the requester drives a fetch as a client of the
        // holder and observes the holder's answers through these events (keyed by requestID).
        case transcriptChunk(TranscriptChunk)
        case transcriptFetchComplete(TranscriptFetchComplete)
        case transcriptUnavailable(TranscriptUnavailable)
        // Carbon shard replication replies + push (docs/CARBON_SHARDING.md §8.1) — a mirror
        // drives its fetches as a client of the holder and observes the answers here (keyed by
        // requestID); `carbonHeadAdvanced` is the owner's unsolicited push.
        case carbonHeads(CarbonHeadsReply)
        case carbonHeadAdvanced(CarbonHead)
        case carbonManifestData(CarbonBlobData)
        case carbonManifestUnavailable(CarbonBlobUnavailable)
        case carbonShardData(CarbonBlobData)
        case carbonShardUnavailable(CarbonBlobUnavailable)
        // Carbon lock verbs (CARBON_SHARDING §17.5): the authority's arbitration answer for a
        // remote acquire, and its unsolicited push on every lock-ledger append.
        case carbonLockDecision(CarbonLockDecisionMsg)
        case carbonLockLedgerAdvanced(CarbonHead)
        /// The owner pushed the current collapsed Bash summaries for a subscribed session (mesh
        /// session sync) — a viewer drops each line onto its matching card.
        case toolSummaries(ToolSummaryBatch)
        /// A subscribed session's transcript was truncated in place (undo/revert) — the viewer
        /// drops every event past `throughSeq` so its transcript shrinks to match the owner's.
        case transcriptReverted(TranscriptReverted)
        /// The host issued (or refreshed) this device's relay credential (mesh P2) — persist
        /// it so the relay is dialable as a fallback on the next reconnect.
        case relayMembership(WireRelayMembership)
        /// A pairing/join code a connected Mac minted at this client's request
        /// (`ClientMsg.requestPairingCode`) — the `nucleic://pair?d=…` string to show as a QR /
        /// copyable code. `nil` when the host couldn't mint one right now.
        case pairingCode(String?)
        /// A Mac is trying to join and needs allow/deny, forwarded here so this client can approve
        /// the join (`ClientMsg.respondMacPair`) instead of walking to the accepting Mac.
        case macPairRequested(WireMacPairRequest)
        /// The pending Mac-pair (by `deviceID`) was answered/withdrawn — dismiss the prompt.
        case macPairResolved(String, Bool)
        /// The host delegated one intelligence call here (runner §5) — run it on the local AFM
        /// queue and answer `ClientMsg.intelligenceResult` with the same id. Pushed only when
        /// this client advertised `WireClientCapabilities.canProvideIntelligence`.
        case intelligenceRequest(WireIntelligenceRequest)
        /// The host lacks credentials it needs to run agents (runner §6) — a holding device
        /// seals them to the carried key and answers `ClientMsg.credentialProvision`.
        case credentialNeeded(WireCredentialNeed)
        /// The host rotated a credential (it holds the refresh lease) and mirrors the new
        /// revision, sealed to this device's sealing key — merge into the local vault.
        case credentialUpdate(SealedCredentialEnvelope)
        /// Credential kinds were deleted mesh-wide (key deletion from any device) — clear the
        /// matching local copies and record the tombstones (`deletedAt`-monotonic) so this
        /// device's own manifest stops offering them.
        case credentialRevoked([CredentialTombstone])
        /// The outcome of this client's `createProject` (CLOUD_RUNTIME §4.3), keyed by its
        /// requestID — success (project id + name) or a human-readable failure.
        case projectCreated(WireProjectCreated)
        /// The peer shared its runner-pool credential (COVALENCE_RUNNER item 4) — persist the
        /// newest revision so this device manages the same pool.
        case runnerPoolCredential(WireRunnerPoolCredential)
        /// The definitive outcome of a mesh-dispatched `startChat` this client sent (mesh
        /// dispatch), correlated by its `requestID`.
        case chatStarted(WireChatStarted)
        /// A live batch of mesh casts (this client sent `castSubscribe`) — dedup on
        /// `(originHostID, channel, castSeq)` and apply.
        case casts(CastBatch)
        /// Catch-up for one (origin, channel) of the cast feed — apply batches in order,
        /// honoring `reset`, and advance the cursor to `tipSeq` on `complete`.
        case castCatchUp(CastCatchUp)
        /// The account-level synced settings changed (`HostMsg.settings`) — replace the local
        /// mirror. The connect-time value arrives inside `ready(Welcome)`'s `Welcome.settings`.
        case settings(SyncedSettings)
        /// The consent challenge for a remote agent sign-in this client began
        /// (`ClientMsg.agentLoginBegin`, docs/REMOTE_AGENT_LOGIN.md): open the authorize URL in
        /// a browser and capture the redirect as described.
        case agentLoginChallenge(WireAgentLoginChallenge)
        /// The definitive outcome of this client's sign-in attempt, keyed by requestID.
        case agentLoginResult(WireAgentLoginResult)
        /// The host's per-provider install/auth state (and which providers it can broker a
        /// sign-in for) — pushed post-hello and on every change.
        case agentAuthStatus([WireProviderAuthStatus])
        /// The host answered this client's `directOffer` (Covalence direct, SYNC §3.3) with its
        /// own candidates/NAT/token — start punching.
        case directAnswer(DirectPunchOffer)
        /// The host's QUIC listener is up on the selected candidate — dial the direct leg now.
        case directGo
        /// The host declined the direct upgrade — stay on relay, back off.
        case directDecline(String)
        case wireError(WireError)
        case failed(String)
        case closed
    }

    private let channel: any FrameChannel
    private let identity: DeviceIdentity
    private let hostStaticKey: Data
    private let mode: Mode
    private let deviceID: String
    private let deviceLabel: String
    private let scopeClaim: DeviceScope
    private let pushToken: String?
    private let pushTopic: String?
    private let releaseChannel: ReleaseChannel?
    private let deviceKind: String?
    private let clientCaps: WireClientCapabilities?
    private let addresses: PeerAddresses?

    private var session: SecureSession?
    private var emit: AsyncStream<Event>.Continuation?

    // Inbound FIFO (same pattern as the host handler).
    private var pending: [Data] = []
    private var waiter: CheckedContinuation<Data?, Never>?
    private var inboundFinished = false

    // Application-level keepalive. The transport layers can't all detect a silently-dead link:
    // LAN has TCP keepalive and relay has a WebSocket ping, but the tailnet channel is a raw
    // socketpair fd pumped by the embedded node — a half-open WireGuard tunnel never delivers an
    // EOF, so the fd stays "open", `messageLoop` blocks forever, and the app is stranded in a
    // phantom `.connected` state that only a full re-pair clears. Driving the wire's existing
    // ping/pong (the host answers a `.ping` with a `.pong`) and enforcing a deadline on *any*
    // inbound frame gives every transport a liveness probe: a missed deadline closes the channel,
    // which surfaces as `.closed` and lets the host connection reconnect.
    private static let pingInterval: Duration = .seconds(20)
    private static let livenessDeadline: Duration = .seconds(55)
    private var lastInboundAt: ContinuousClock.Instant?
    private var keepaliveTask: Task<Void, Never>?

    public init(
        channel: any FrameChannel, identity: DeviceIdentity, hostStaticKey: Data,
        mode: Mode, deviceID: String, deviceLabel: String,
        scopeClaim: DeviceScope = .control, pushToken: String? = nil,
        pushTopic: String? = nil,
        releaseChannel: ReleaseChannel? = nil,
        deviceKind: String? = nil, clientCaps: WireClientCapabilities? = nil,
        addresses: PeerAddresses? = nil
    ) {
        self.channel = channel
        self.identity = identity
        self.hostStaticKey = hostStaticKey
        self.mode = mode
        self.deviceID = deviceID
        self.deviceLabel = deviceLabel
        self.scopeClaim = scopeClaim
        self.pushToken = pushToken
        self.pushTopic = pushTopic
        self.releaseChannel = releaseChannel
        self.deviceKind = deviceKind
        self.clientCaps = clientCaps
        self.addresses = addresses
    }

    /// The session's static public key once established — pin this at pairing.
    public func hostKey() -> Data? { session?.remoteStaticKey }

    /// The established session's channel-binding hash (nil before the handshake completes) —
    /// what the direct-path upgrade derives the multipath mux's control key from
    /// (`MultipathFrameChannel.bindingKey(handshakeHash:)`, SYNC §3.3).
    public func handshakeHash() -> Data? { session?.handshakeHash }

    public func start() -> AsyncStream<Event> {
        AsyncStream { continuation in
            self.emit = continuation
            Task { await self.run() }
        }
    }

    public func send(_ msg: ClientMsg) {
        guard let session else { return }
        if let frame = try? session.seal(msg) { channel.send(frame) }
    }

    public func disconnect() { keepaliveTask?.cancel(); keepaliveTask = nil; channel.close() }

    // MARK: - Run

    private func run() async {
        startPump()
        emit?.yield(.connecting)
        do {
            let secure = try await handshake()
            session = secure
            try await helloWelcome()
            startKeepalive()
            try await messageLoop()
        } catch let wire as WireError {
            emit?.yield(.wireError(wire))
        } catch {
            emit?.yield(.failed("\(error)"))
        }
        keepaliveTask?.cancel(); keepaliveTask = nil
        emit?.yield(.closed)
        emit?.finish()
        channel.close()
    }

    // MARK: - Keepalive / liveness

    /// Start probing the link once the session is up: ping on an interval, and close the channel if
    /// nothing at all has arrived within the deadline (a half-open transport). Runs on the actor.
    private func startKeepalive() {
        lastInboundAt = ContinuousClock.now
        keepaliveTask?.cancel()
        keepaliveTask = Task { [weak self] in
            while !Task.isCancelled {
                try? await Task.sleep(for: Self.pingInterval)
                guard !Task.isCancelled else { break }
                let dead = await self?.probeLiveness() ?? true
                if dead { break }
            }
        }
    }

    /// One keepalive tick: if the deadline lapsed with no inbound frame the transport is dead in all
    /// but name — close it (→ `.closed` → reconnect). Otherwise nudge the host; its `.pong` (or any
    /// frame) refreshes the deadline, and the ping doubles as a NAT-binding keepalive. Returns
    /// whether the link was torn down.
    private func probeLiveness() -> Bool {
        if let last = lastInboundAt, last.duration(to: ContinuousClock.now) > Self.livenessDeadline {
            channel.close()
            return true
        }
        send(.ping)
        return false
    }

    /// Foreground liveness probe (the fast-reconnect path). iOS suspends a backgrounded app, which
    /// silently kills the socket — but no `.closed`/`.failed` is delivered while the app is frozen,
    /// so the connection can wake still believing it's live. Rather than wait out the 55s keepalive
    /// deadline, the client can ask this: ping now, then report whether the host answers (any inbound
    /// frame — the host replies to `.ping` with `.pong`) before `deadline`. Returns as soon as a
    /// frame arrives (≈1 RTT on a live LAN link), or `false` once the deadline lapses on a dead one.
    public func probeAlive(within deadline: Duration) async -> Bool {
        guard session != nil else { return false }
        let mark = ContinuousClock.now
        send(.ping)
        let step = Duration.milliseconds(50)
        var waited = Duration.zero
        while waited < deadline {
            try? await Task.sleep(for: step)
            if let last = lastInboundAt, last > mark { return true }
            waited += step
        }
        return false
    }

    private func handshake() async throws -> SecureSession {
        let psk: Data?
        let handshakeMode: HandshakeMode
        switch mode {
        case .pair(let secret): psk = secret; handshakeMode = .pair
        case .reconnect: psk = nil; handshakeMode = .reconnect
        }

        // Plaintext preamble so the host routes to the right pattern.
        let preamble = HandshakePreamble(mode: handshakeMode, deviceID: deviceID)
        channel.send(try CBOREncoder().encode(preamble))

        let driver = NoiseSession.makeInitiator(
            identity: identity, hostStaticKey: hostStaticKey, psk: psk)
        while !driver.isComplete {
            if driver.isMyTurnToWrite {
                channel.send(try driver.write())
            } else {
                guard let frame = await nextFrame() else { throw WireError(code: .unauthorized, message: "handshake closed") }
                _ = try driver.read(frame)
            }
        }
        return try driver.session()
    }

    private func helloWelcome() async throws {
        let hello = Hello(
            deviceID: deviceID, deviceLabel: deviceLabel,
            scopeClaim: scopeClaim, pushToken: pushToken, pushTopic: pushTopic,
            channel: releaseChannel,
            deviceKind: deviceKind, clientCaps: clientCaps, addresses: addresses)
        send(.hello(hello))
        guard let frame = await nextFrame() else { throw WireError(code: .unauthorized, message: "closed before welcome") }
        guard let session else { throw WireError(code: .malformed, message: "no session") }
        let msg = try session.open(HostMsg.self, from: frame)
        switch msg {
        case .welcome(let welcome):
            // Confirm the host's channel from this side too. The host has already gated the
            // hello, so an incompatible host normally replies with `.error` below; this also
            // covers a host that predates the host-side gate.
            if let mine = releaseChannel, let theirs = welcome.channel, !mine.isCompatible(with: theirs) {
                throw WireError(
                    code: .channelMismatch,
                    message: "host is a \(theirs.rawValue) build; this remote is \(mine.rawValue)")
            }
            emit?.yield(.ready(welcome))
        case .error(let wire):
            // Surface the host's rejection (e.g. a channel mismatch) instead of masking it.
            throw wire
        default:
            throw WireError(code: .malformed, message: "expected welcome")
        }
    }

    private func messageLoop() async throws {
        while let frame = await nextFrame() {
            guard let session else { break }
            guard let msg = try? session.open(HostMsg.self, from: frame) else { continue }
            switch msg {
            case .welcome(let w): emit?.yield(.ready(w))
            case .sessionList(let l): emit?.yield(.sessionList(l))
            case .snapshot(let s): emit?.yield(.snapshot(s))
            case .events(let b): emit?.yield(.events(b))
            case .approvalRequested(let r): emit?.yield(.approvalRequested(r))
            case .approvalResolved(let r): emit?.yield(.approvalResolved(r))
            case .sessionUpdated(let s): emit?.yield(.sessionUpdated(s))
            case .dashboard(let d): emit?.yield(.dashboard(d))
            case .sessionDiff(let d): emit?.yield(.sessionDiff(d))
            case .peerList(let peers): emit?.yield(.peerList(peers))
            case .meshRoster(let roster): emit?.yield(.meshRoster(roster))
            case .transferAccept(let a): emit?.yield(.transferAccept(a))
            case .transferReject(let r): emit?.yield(.transferReject(r))
            case .transferReady(let id): emit?.yield(.transferReady(id))
            case .transferCommitted(let id): emit?.yield(.transferCommitted(id))
            case .transferChunkAck(let ack): emit?.yield(.transferChunkAck(ack))
            case .transcriptChunk(let chunk): emit?.yield(.transcriptChunk(chunk))
            case .transcriptFetchComplete(let done): emit?.yield(.transcriptFetchComplete(done))
            case .transcriptUnavailable(let un): emit?.yield(.transcriptUnavailable(un))
            case .carbonHeads(let reply): emit?.yield(.carbonHeads(reply))
            case .carbonHeadAdvanced(let head): emit?.yield(.carbonHeadAdvanced(head))
            case .carbonManifestData(let data): emit?.yield(.carbonManifestData(data))
            case .carbonManifestUnavailable(let un): emit?.yield(.carbonManifestUnavailable(un))
            case .carbonShardData(let data): emit?.yield(.carbonShardData(data))
            case .carbonShardUnavailable(let un): emit?.yield(.carbonShardUnavailable(un))
            case .carbonLockDecision(let d): emit?.yield(.carbonLockDecision(d))
            case .carbonLockLedgerAdvanced(let head): emit?.yield(.carbonLockLedgerAdvanced(head))
            case .toolSummaries(let b): emit?.yield(.toolSummaries(b))
            case .transcriptReverted(let r): emit?.yield(.transcriptReverted(r))
            case .relayMembership(let m): emit?.yield(.relayMembership(m))
            case .pairingCode(let qr): emit?.yield(.pairingCode(qr))
            case .macPairRequested(let req): emit?.yield(.macPairRequested(req))
            case .macPairResolved(let id, let ok): emit?.yield(.macPairResolved(id, ok))
            case .intelligenceRequest(let req): emit?.yield(.intelligenceRequest(req))
            case .credentialNeeded(let need): emit?.yield(.credentialNeeded(need))
            case .credentialUpdate(let envelope): emit?.yield(.credentialUpdate(envelope))
            case .credentialRevoked(let tombstones): emit?.yield(.credentialRevoked(tombstones))
            case .projectCreated(let outcome): emit?.yield(.projectCreated(outcome))
            case .runnerPoolCredential(let credential): emit?.yield(.runnerPoolCredential(credential))
            case .chatStarted(let outcome): emit?.yield(.chatStarted(outcome))
            case .casts(let batch): emit?.yield(.casts(batch))
            case .castCatchUp(let catchUp): emit?.yield(.castCatchUp(catchUp))
            case .settings(let s): emit?.yield(.settings(s))
            case .agentLoginChallenge(let challenge): emit?.yield(.agentLoginChallenge(challenge))
            case .agentLoginResult(let result): emit?.yield(.agentLoginResult(result))
            case .agentAuthStatus(let statuses): emit?.yield(.agentAuthStatus(statuses))
            case .directAnswer(let offer): emit?.yield(.directAnswer(offer))
            case .directGo: emit?.yield(.directGo)
            case .directDecline(let reason): emit?.yield(.directDecline(reason))
            case .error(let e): emit?.yield(.wireError(e))
            case .pong, .unknown: break
            }
        }
    }

    // MARK: - Inbound FIFO

    private func feed(_ frame: Data) {
        lastInboundAt = ContinuousClock.now   // any inbound frame is proof of life — refresh the keepalive deadline
        if let w = waiter { waiter = nil; w.resume(returning: frame) }
        else { pending.append(frame) }
    }
    private func finishInbound() {
        inboundFinished = true
        if let w = waiter { waiter = nil; w.resume(returning: nil) }
    }
    private func nextFrame() async -> Data? {
        if !pending.isEmpty { return pending.removeFirst() }
        if inboundFinished { return nil }
        return await withCheckedContinuation { waiter = $0 }
    }
    private func startPump() {
        let frames = channel.frames()
        Task { [weak self] in
            for await frame in frames { await self?.feed(frame) }
            await self?.finishInbound()
        }
    }
}
