import Foundation
import Testing
import NucleicProtocol

@testable import NucleicCore

/// CARBON_SHARDING §15 phase 3 (wire half): the Carbon verbs served end-to-end through a real
/// `SyncHost`/`SyncClient` pair over an in-memory pipe — heads, per-id manifest replies, and
/// the §8.3 shard ack window (the server must never exceed 4 un-acked data frames).
@Suite("Carbon wire loopback", .serialized)
struct CarbonWireLoopbackTests {
    private let projectUUID = UUID(uuidString: "AB12CD34-EF56-4781-9ABC-DEF123456789")!
    private let key = try! CarbonCrypto.ProjectKey(bytes: Data((0..<32).map { UInt8($0) }))
    private var crypto: CarbonCrypto { CarbonCrypto(projectKey: key) }

    private func makeStore() throws -> CarbonShardStore {
        let root = FileManager.default.temporaryDirectory
            .appendingPathComponent("carbon-wire-\(UUID().uuidString)", isDirectory: true)
        try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true)
        return CarbonShardStore(root: root, keyProvider: { [key] _ in key })
    }

    /// Pair a Carbon-serving host with a raw client (manual verbs — so ack behavior is ours
    /// to control in the window test).
    private func makePaired(
        store: CarbonShardStore, lockManager: CarbonLockManager? = nil
    ) async throws -> (client: SyncClient, recorder: EventRecorder) {
        let bridge = FakeSyncBridge(
            capabilities: WireCapabilities(
                canModifyToolInput: true, allowAlwaysScopes: [], canServeCarbon: true),
            carbonStore: store, lockManager: lockManager)
        let hostIdentity = DeviceIdentity()
        let host = SyncHost(
            identity: hostIdentity, bridge: bridge, store: InMemoryPairedDeviceStore())
        let (clientCh, serverCh) = MemoryChannel.pair()
        try await host.start(listener: OneShotListener(serverCh))
        let payload = await host.beginPairing()
        let client = SyncClient(
            channel: clientCh, identity: DeviceIdentity(), hostStaticKey: payload.hostStaticKey,
            mode: .pair(secret: payload.pairingSecret), deviceID: "mirror-1",
            deviceLabel: "Mirror Mac")
        let recorder = EventRecorder()
        await recorder.consume(client.start())
        _ = await recorder.waitFor { if case .ready = $0 { return true } else { return false } }
        return (client, recorder)
    }

    @discardableResult
    private func seedManifest(
        _ store: CarbonShardStore, seq: UInt64, payloads: [String]
    ) async throws -> (id: String, manifest: CarbonManifest, shardIDs: [String]) {
        var refs: [CarbonManifest.ShardRef] = []
        for payload in payloads {
            let sealed = try crypto.seal(Data(payload.utf8))
            try await store.putShard(
                ciphertext: sealed.ciphertext, expectedID: sealed.id, projectUUID: projectUUID)
            refs.append(.init(id: sealed.id, byteCount: UInt64(sealed.ciphertext.count)))
        }
        let manifest = CarbonManifest(
            project: .init(uuid: projectUUID), stream: .history, sessionID: "s1",
            ownerDeviceID: "owner", ownerGeneration: 1, snapshotSeq: seq,
            payloadKind: .gitPack, shards: refs, createdAtMs: seq)
        let sealed = try manifest.sealed(with: crypto)
        try await store.putManifest(
            ciphertext: sealed.ciphertext, expectedID: sealed.id, projectUUID: projectUUID)
        try await store.advanceHead(manifest.head(manifestID: sealed.id))
        return (sealed.id, manifest, refs.map(\.id))
    }

    @Test func headsFetchRoundTrip() async throws {
        let store = try makeStore()
        let seeded = try await seedManifest(store, seq: 1, payloads: ["pack-1"])
        let (client, recorder) = try await makePaired(store: store)

        await client.send(.fetchCarbonHeads(
            CarbonHeadsFetch(requestID: "req-h", projectUUID: projectUUID)))
        let event = await recorder.waitFor {
            if case .carbonHeads(let reply) = $0 { return reply.requestID == "req-h" }
            return false
        }
        guard case .carbonHeads(let reply) = event else { return }
        #expect(reply.heads == [seeded.manifest.head(manifestID: seeded.id)])
    }

    @Test func manifestFetchAnswersPerID() async throws {
        let store = try makeStore()
        let seeded = try await seedManifest(store, seq: 1, payloads: ["pack-1"])
        let (client, recorder) = try await makePaired(store: store)
        let absent = String(repeating: "0", count: 64)

        await client.send(.fetchCarbonManifests(CarbonBlobsFetch(
            requestID: "req-m", projectUUID: projectUUID, ids: [seeded.id, absent])))
        let dataEvent = await recorder.waitFor {
            if case .carbonManifestData(let d) = $0 { return d.id == seeded.id }
            return false
        }
        guard case .carbonManifestData(let served) = dataEvent else { return }
        // The served blob is the verified ciphertext — decodes to the seeded manifest.
        #expect(try CarbonManifest.open(served.blob, with: crypto) == seeded.manifest)
        let miss = await recorder.waitFor {
            if case .carbonManifestUnavailable(let un) = $0 { return un.id == absent }
            return false
        }
        guard case .carbonManifestUnavailable(let unavailable) = miss else { return }
        #expect(unavailable.reason == .notHeld)
    }

    /// CARBON_SHARDING §17.9 (authority half): a remote session's acquire round-trips over
    /// the real wire — an interim `queued` decision when the waiter parks, the final
    /// `granted` once the contending holder releases, and a wire release that frees the path
    /// on the authority (carrying its fence heads).
    @Test func lockAcquireQueueGrantReleaseRoundTrip() async throws {
        let store = try makeStore()
        let lm = CarbonLockManager()
        let localA = SessionID(rawValue: "local-A")
        let remoteB = SessionID(rawValue: "remote-B")
        let dev = LockDomain("dev")
        #expect(await lm.acquire(localA, domain: dev, files: ["x"]) == .granted)
        let (client, recorder) = try await makePaired(store: store, lockManager: lm)

        // Contended acquire from the remote session: the authority answers `queued` first.
        await client.send(.carbonLockAcquire(CarbonLockRequest(
            requestID: "lk-1", projectUUID: projectUUID, domainRef: "dev",
            sessionID: remoteB.rawValue, paths: ["x"])))
        _ = await recorder.waitFor {
            if case .carbonLockDecision(let d) = $0 {
                return d.requestID == "lk-1" && d.outcome == .queued
            }
            return false
        }
        #expect(await lm.isWaiting(remoteB))

        // The holder's work lands → the parked wire waiter is granted (never both holders).
        await lm.release(localA, paths: ["x"])
        _ = await recorder.waitFor {
            if case .carbonLockDecision(let d) = $0 {
                return d.requestID == "lk-1" && d.outcome == .granted
            }
            return false
        }
        #expect(await lm.heldPaths(remoteB) == ["x"])
        #expect(await lm.heldPaths(localA).isEmpty)

        // Wire release (with the landing host's fence heads) frees the path.
        let fence = CarbonHead(
            key: CarbonStreamKey(projectUUID: projectUUID, stream: .history, sessionID: nil),
            manifestID: String(repeating: "b", count: 64), ownerGeneration: 1, snapshotSeq: 1)
        await client.send(.carbonLockRelease(CarbonLockRequest(
            requestID: "lk-2", projectUUID: projectUUID, domainRef: "dev",
            sessionID: remoteB.rawValue, paths: nil, requiredHeads: [fence])))
        var waited = 0
        while await !lm.heldPaths(remoteB).isEmpty, waited < 2000 {
            try? await Task.sleep(for: .milliseconds(5))
            waited += 5
        }
        #expect(await lm.heldPaths(remoteB).isEmpty)
    }

    /// The §8.3 discipline, observed from the wire: 10 shards requested, the server parks
    /// after 4 un-acked frames, acks release exactly as many more, and every shard arrives.
    @Test func shardServingHonorsTheAckWindow() async throws {
        let store = try makeStore()
        let seeded = try await seedManifest(
            store, seq: 1, payloads: (1...10).map { "shard-payload-\($0)" })
        let (client, recorder) = try await makePaired(store: store)

        await client.send(.fetchCarbonShards(CarbonBlobsFetch(
            requestID: "req-s", projectUUID: projectUUID, ids: seeded.shardIDs)))

        // Exactly the window arrives, then the server parks (no acks sent yet).
        var received: [String] = []
        for _ in 0..<CarbonFormat.shardAckWindow {
            let event = await recorder.waitFor {
                if case .carbonShardData = $0 { return true }
                return false
            }
            if case .carbonShardData(let d) = event { received.append(d.id) }
        }
        try await Task.sleep(nanoseconds: 200_000_000)
        #expect(await !recorder.hasBuffered {
            if case .carbonShardData = $0 { return true }
            return false
        })

        // Ack two ⇒ exactly two more; ack everything ⇒ the rest.
        await client.send(.carbonShardAck(
            CarbonShardAck(requestID: "req-s", ids: Array(received.prefix(2)))))
        for _ in 0..<2 {
            let event = await recorder.waitFor {
                if case .carbonShardData = $0 { return true }
                return false
            }
            if case .carbonShardData(let d) = event { received.append(d.id) }
        }
        try await Task.sleep(nanoseconds: 200_000_000)
        #expect(await !recorder.hasBuffered {
            if case .carbonShardData = $0 { return true }
            return false
        })

        await client.send(.carbonShardAck(CarbonShardAck(requestID: "req-s", ids: received)))
        while received.count < seeded.shardIDs.count {
            let event = await recorder.waitFor {
                if case .carbonShardData = $0 { return true }
                return false
            }
            if case .carbonShardData(let d) = event { received.append(d.id) }
        }
        #expect(Set(received) == Set(seeded.shardIDs))
    }
}
