import Foundation
import Testing

@testable import NucleicCore
import NucleicProtocol

/// CARBON_SHARDING §15 phase 3 (coordinator half): 3-member loopback convergence through
/// partition/rejoin, the honest-head rule (never advance without the shards), stale-generation
/// fencing, the relay payload policy, and §5.3 durability accounting — all against in-memory
/// member fakes; the wire adapter binds the same closures to real verbs.
@Suite("Carbon sync coordinator", .serialized)
struct CarbonSyncCoordinatorTests {
    private let projectUUID = UUID(uuidString: "5A6B7C8D-9E0F-4123-8456-789ABCDEF012")!
    private let key = try! CarbonCrypto.ProjectKey(bytes: Data((0..<32).map { UInt8($0) }))
    private var crypto: CarbonCrypto { CarbonCrypto(projectKey: key) }

    /// Mutable mesh conditions the transport closures consult — partitions and transport
    /// class per peer, toggled mid-test.
    private final class Mesh: @unchecked Sendable {
        private let lock = NSLock()
        private var reachable: Set<String> = []
        private var relayOnly: Set<String> = []
        var stores: [String: CarbonShardStore] = [:]

        func setReachable(_ peers: Set<String>) {
            lock.lock()
            reachable = peers
            lock.unlock()
        }

        func setRelayOnly(_ peers: Set<String>) {
            lock.lock()
            relayOnly = peers
            lock.unlock()
        }

        func isReachable(_ peer: String) -> Bool {
            lock.lock()
            defer { lock.unlock() }
            return reachable.contains(peer)
        }

        func transportClass(_ peer: String) -> CarbonSyncCoordinator.TransportClass {
            lock.lock()
            defer { lock.unlock() }
            return relayOnly.contains(peer) ? .relay : .direct
        }
    }

    private func makeMember(
        _ name: String, mesh: Mesh, root: URL, keyKnown: Bool = true,
        relayPayloadLimit: UInt64 = CarbonSyncCoordinator.defaultRelayPayloadLimit
    ) -> (store: CarbonShardStore, coordinator: CarbonSyncCoordinator) {
        let held = keyKnown ? key : nil
        let store = CarbonShardStore(
            root: root.appendingPathComponent(name, isDirectory: true), keyProvider: { _ in held })
        mesh.stores[name] = store
        let transport = CarbonSyncCoordinator.PeerTransport(
            fetchHeads: { [weak mesh] peer, project in
                guard let mesh, mesh.isReachable(peer), let peerStore = mesh.stores[peer]
                else { return nil }
                let heads = await peerStore.heads(projectUUID: project ?? self.projectUUID)
                return Array(heads.values)
            },
            fetchManifests: { [weak mesh] peer, project, ids in
                guard let mesh, mesh.isReachable(peer), let peerStore = mesh.stores[peer]
                else { return [:] }
                var out: [String: Data] = [:]
                for id in ids {
                    if let blob = await peerStore.manifestBlob(id: id, projectUUID: project) {
                        out[id] = blob
                    }
                }
                return out
            },
            fetchShards: { [weak mesh] peer, project, ids in
                guard let mesh, mesh.isReachable(peer), let peerStore = mesh.stores[peer]
                else { return [:] }
                var out: [String: Data] = [:]
                for id in ids {
                    if let blob = await peerStore.shard(id: id, projectUUID: project) {
                        out[id] = blob
                    }
                }
                return out
            },
            transportClass: { [weak mesh] peer in mesh?.transportClass(peer) ?? .direct })
        let coordinator = CarbonSyncCoordinator(
            store: store, keyProvider: { _ in held }, transport: transport,
            relayPayloadLimit: relayPayloadLimit)
        return (store, coordinator)
    }

    private func tempRoot() throws -> URL {
        let url = FileManager.default.temporaryDirectory
            .appendingPathComponent("carbon-sync-\(UUID().uuidString)", isDirectory: true)
        try FileManager.default.createDirectory(at: url, withIntermediateDirectories: true)
        return url
    }

    /// Extend the owner's history chain by one manifest (one shard) and return its head.
    @discardableResult
    private func produce(
        on store: CarbonShardStore, seq: UInt64, gen: UInt64 = 1, parent: String? = nil,
        payload: String? = nil
    ) async throws -> (head: CarbonHead, manifestID: String) {
        let sealed = try crypto.seal(Data((payload ?? "pack-\(gen)-\(seq)").utf8))
        try await store.putShard(
            ciphertext: sealed.ciphertext, expectedID: sealed.id, projectUUID: projectUUID)
        let manifest = CarbonManifest(
            project: .init(uuid: projectUUID), stream: .history, sessionID: "s1",
            ownerDeviceID: "owner", ownerGeneration: gen, snapshotSeq: seq,
            parentManifest: parent, payloadKind: .gitPack,
            shards: [.init(id: sealed.id, byteCount: UInt64(sealed.ciphertext.count))],
            createdAtMs: seq)
        let sealedManifest = try manifest.sealed(with: crypto)
        try await store.putManifest(
            ciphertext: sealedManifest.ciphertext, expectedID: sealedManifest.id,
            projectUUID: projectUUID)
        let head = manifest.head(manifestID: sealedManifest.id)
        try await store.advanceHead(head)
        return (head, sealedManifest.id)
    }

    private var chainKey: CarbonStreamKey {
        CarbonStreamKey(projectUUID: projectUUID, stream: .history, sessionID: "s1")
    }

    // MARK: - Convergence

    @Test func threeMemberConvergenceAndFallthrough() async throws {
        let root = try tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let mesh = Mesh()
        let owner = makeMember("owner", mesh: mesh, root: root)
        let a = makeMember("a", mesh: mesh, root: root)
        let b = makeMember("b", mesh: mesh, root: root)
        mesh.setReachable(["owner", "a", "b"])

        // Owner produces a 3-deep chain; A converges via reconcile (backfill — no special mode).
        var parent: String?
        var head = CarbonHead(key: chainKey, manifestID: "", ownerGeneration: 1, snapshotSeq: 0)
        for seq: UInt64 in 1...3 {
            let produced = try await produce(on: owner.store, seq: seq, parent: parent)
            parent = produced.manifestID
            head = produced.head
        }
        await a.coordinator.reconcile(projectUUID: projectUUID, peers: ["owner"])
        #expect(await a.store.heads(projectUUID: projectUUID)[chainKey] == head)
        #expect(try await a.store.manifest(id: head.manifestID, projectUUID: projectUUID) != nil)

        // Owner goes dark; B hears only A's head push and falls through to A for everything.
        mesh.setReachable(["a", "b"])
        await b.coordinator.observeHeadAdvanced(head, from: "a", candidates: ["owner"])
        #expect(await b.store.heads(projectUUID: projectUUID)[chainKey] == head)
        // B's copy is complete and verifiable — every shard held.
        let manifest = try #require(
            try await b.store.manifest(id: head.manifestID, projectUUID: projectUUID))
        #expect(await b.store.missingShards(for: manifest).isEmpty)
    }

    @Test func partitionThenRejoinConverges() async throws {
        let root = try tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let mesh = Mesh()
        let owner = makeMember("owner", mesh: mesh, root: root)
        let a = makeMember("a", mesh: mesh, root: root)
        mesh.setReachable(["owner", "a"])

        let first = try await produce(on: owner.store, seq: 1)
        await a.coordinator.observeHeadAdvanced(first.head, from: "owner")
        #expect(await a.store.heads(projectUUID: projectUUID)[chainKey] == first.head)

        // A partitions; the owner keeps producing.
        mesh.setReachable([])
        let second = try await produce(on: owner.store, seq: 2, parent: first.manifestID)
        let third = try await produce(on: owner.store, seq: 3, parent: second.manifestID)
        await a.coordinator.observeHeadAdvanced(third.head, from: "owner")  // push lost to partition
        #expect(await a.store.heads(projectUUID: projectUUID)[chainKey] == first.head)

        // Rejoin: heads are state, so one reconcile heals the whole gap.
        mesh.setReachable(["owner", "a"])
        await a.coordinator.reconcile(projectUUID: projectUUID, peers: ["owner"])
        #expect(await a.store.heads(projectUUID: projectUUID)[chainKey] == third.head)
        #expect(
            try await a.store.manifest(id: second.manifestID, projectUUID: projectUUID) != nil)
    }

    // MARK: - Honest heads + fencing

    @Test func headNeverAdvancesWithoutTheShards() async throws {
        let root = try tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let mesh = Mesh()
        let owner = makeMember("owner", mesh: mesh, root: root)
        let a = makeMember("a", mesh: mesh, root: root)
        mesh.setReachable(["owner", "a"])

        let produced = try await produce(on: owner.store, seq: 1)
        // Sabotage: the owner "serves" the manifest but its shard vanished (disk loss etc.).
        let manifest = try #require(
            try await owner.store.manifest(id: produced.manifestID, projectUUID: projectUUID))
        let shardID = manifest.shards[0].id
        let shardFile = root.appendingPathComponent("owner", isDirectory: true)
            .appendingPathComponent(projectUUID.uuidString, isDirectory: true)
            .appendingPathComponent("objects", isDirectory: true)
            .appendingPathComponent(String(shardID.prefix(2)), isDirectory: true)
            .appendingPathComponent(shardID)
        try FileManager.default.removeItem(at: shardFile)

        await a.coordinator.observeHeadAdvanced(produced.head, from: "owner")
        // Manifest may be cached locally, but the HEAD must not have advanced (honest claim).
        #expect(await a.store.heads(projectUUID: projectUUID)[chainKey] == nil)

        // The shard reappears (owner re-produced it) ⇒ the next pass completes and advances.
        let sealed = try crypto.seal(Data("pack-1-1".utf8))
        try await owner.store.putShard(
            ciphertext: sealed.ciphertext, expectedID: sealed.id, projectUUID: projectUUID)
        await a.coordinator.observeHeadAdvanced(produced.head, from: "owner")
        #expect(await a.store.heads(projectUUID: projectUUID)[chainKey] == produced.head)
    }

    @Test func staleGenerationHeadIsIgnored() async throws {
        let root = try tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let mesh = Mesh()
        let owner = makeMember("owner", mesh: mesh, root: root)
        let stale = makeMember("stale", mesh: mesh, root: root)
        let a = makeMember("a", mesh: mesh, root: root)
        mesh.setReachable(["owner", "stale", "a"])

        // A converges to the gen-2 chain (a resuscitated owner), then the stale gen-1 owner
        // returns from a partition still pushing its old head.
        let gen2 = try await produce(on: owner.store, seq: 1, gen: 2)
        await a.coordinator.reconcile(projectUUID: projectUUID, peers: ["owner"])
        #expect(await a.store.heads(projectUUID: projectUUID)[chainKey] == gen2.head)

        let gen1 = try await produce(on: stale.store, seq: 99, gen: 1, payload: "stale-pack")
        await a.coordinator.observeHeadAdvanced(gen1.head, from: "stale")
        #expect(await a.store.heads(projectUUID: projectUUID)[chainKey] == gen2.head)
    }

    @Test func keylessMemberNeverMirrors() async throws {
        let root = try tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let mesh = Mesh()
        let owner = makeMember("owner", mesh: mesh, root: root)
        let keyless = makeMember("keyless", mesh: mesh, root: root, keyKnown: false)
        mesh.setReachable(["owner", "keyless"])

        let produced = try await produce(on: owner.store, seq: 1)
        await keyless.coordinator.observeHeadAdvanced(produced.head, from: "owner")
        #expect(await keyless.store.heads(projectUUID: projectUUID)[chainKey] == nil)
    }

    // MARK: - Relay policy (§8.4)

    @Test func oversizePayloadDefersOnRelayThenSyncsOnDirect() async throws {
        let root = try tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let mesh = Mesh()
        let owner = makeMember("owner", mesh: mesh, root: root)
        // A tiny relay limit so an ordinary payload counts as bulk.
        let a = makeMember("a", mesh: mesh, root: root, relayPayloadLimit: 16)
        mesh.setReachable(["owner", "a"])
        mesh.setRelayOnly(["owner"])

        let produced = try await produce(
            on: owner.store, seq: 1, payload: "a payload comfortably over sixteen bytes")
        await a.coordinator.observeHeadAdvanced(produced.head, from: "owner")
        #expect(await a.store.heads(projectUUID: projectUUID)[chainKey] == nil)
        #expect(await a.coordinator.chainsAwaitingDirectLink().contains(chainKey))

        // A direct path opens ⇒ the deferred chain completes.
        mesh.setRelayOnly([])
        await a.coordinator.reconcile(projectUUID: projectUUID, peers: ["owner"])
        #expect(await a.store.heads(projectUUID: projectUUID)[chainKey] == produced.head)
        #expect(await a.coordinator.chainsAwaitingDirectLink().isEmpty)
    }

    // MARK: - Durability accounting (§5.3)

    @Test func durabilityDegradesHonestly() async throws {
        let root = try tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let mesh = Mesh()
        let owner = makeMember("owner", mesh: mesh, root: root)
        let a = makeMember("a", mesh: mesh, root: root)
        let b = makeMember("b", mesh: mesh, root: root)
        mesh.setReachable(["owner", "a", "b"])
        let replicaSet: Set<String> = ["a", "b"]

        let produced = try await produce(on: owner.store, seq: 1)
        #expect(
            await owner.coordinator.durability(of: produced.head, replicaSet: replicaSet)
                == .notProtected(reason: "no replica holds it yet"))
        #expect(
            await owner.coordinator.durability(of: produced.head, replicaSet: [])
                == .notProtected(reason: "no eligible replica members"))

        // A converges and the owner learns of it (head exchange) ⇒ reduced redundancy.
        await a.coordinator.reconcile(projectUUID: projectUUID, peers: ["owner"])
        await owner.coordinator.reconcile(projectUUID: projectUUID, peers: ["a"])
        #expect(
            await owner.coordinator.durability(of: produced.head, replicaSet: replicaSet)
                == .reducedRedundancy(reason: "one replica holds it"))

        // B too ⇒ mesh-durable: protected ×3 (both replicas + the owner).
        await b.coordinator.reconcile(projectUUID: projectUUID, peers: ["owner"])
        await owner.coordinator.reconcile(projectUUID: projectUUID, peers: ["b"])
        #expect(
            await owner.coordinator.durability(of: produced.head, replicaSet: replicaSet)
                == .protected(holders: 3))
    }
}
