import Foundation
import Testing
import NucleicProtocol

@testable import NucleicCore

/// Mesh casting over the real loopback pipe (Noise + CBOR): subscribe → catch-up serve,
/// live fanout gating, and echo suppression.
@Suite struct CastSyncTests {
    private let originA = String(repeating: "aa", count: 32)

    private func makeCast(_ seq: UInt64, origin: String? = nil, channel: String = "activity") -> WireCast {
        WireCast(
            originHostID: origin ?? originA, channel: channel, castSeq: seq, mode: .stream,
            typeID: "t.v1", payload: Data([UInt8(truncatingIfNeeded: seq)]),
            sentAt: Date(timeIntervalSince1970: 1_700_000_000))
    }

    private func castingBridge() -> FakeSyncBridge {
        FakeSyncBridge(capabilities: WireCapabilities(
            canModifyToolInput: true, allowAlwaysScopes: [], canCast: true))
    }

    /// Pair one client (optionally cast-capable) against a casting host.
    private func makePaired(
        bridge: FakeSyncBridge, deviceID: String, canCast: Bool
    ) async throws -> (host: SyncHost, client: SyncClient, recorder: EventRecorder) {
        let host = SyncHost(
            identity: DeviceIdentity(), 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: deviceID,
            deviceLabel: "Device \(deviceID)",
            clientCaps: canCast ? WireClientCapabilities(mesh: 1, canCast: true) : nil)
        let recorder = EventRecorder()
        await recorder.consume(client.start())
        _ = await recorder.waitFor { if case .ready = $0 { return true } else { return false } }
        return (host, client, recorder)
    }

    @Test func castSubscribeServesCatchUpWithTheClientCursor() async throws {
        let bridge = castingBridge()
        let batch = CastCatchUp(
            originHostID: originA, channel: "activity", mode: .stream, batchIndex: 0,
            casts: [makeCast(5), makeCast(6)], complete: true, tipSeq: 6)
        await bridge.setCastCatchUp(
            tips: [CastTip(originHostID: originA, channel: "activity", mode: .stream, tipSeq: 6, floorSeq: 1)],
            batches: ["\(originA)/activity": [batch]])
        let (host, client, recorder) = try await makePaired(
            bridge: bridge, deviceID: "iphone-1", canCast: true)

        await client.send(.castSubscribe(CastSubscribe(
            cursors: [CastCursor(originHostID: originA, channel: "activity", afterSeq: 4)])))
        let event = await recorder.waitFor {
            if case .castCatchUp = $0 { return true } else { return false }
        }
        guard case .castCatchUp(let received) = event else {
            Issue.record("no castCatchUp"); return
        }
        #expect(received == batch)
        // The handler threaded the client's cursor through to the bridge.
        let requests = await poll { await bridge.castCatchUpRequests }
        #expect(requests.first?.afterSeq == 4)

        // Live fanout reaches the subscriber.
        await bridge.push(.casts([makeCast(7)]))
        let live = await recorder.waitFor {
            if case .casts = $0 { return true } else { return false }
        }
        guard case .casts(let liveBatch) = live else { Issue.record("no live casts"); return }
        #expect(liveBatch.casts.map(\.castSeq) == [7])
        await host.stop()
    }

    @Test func liveCastsSkipNonSubscribedAndNonCapableClients() async throws {
        let bridge = castingBridge()
        // Capable but never subscribed…
        let (host, client, recorder) = try await makePaired(
            bridge: bridge, deviceID: "iphone-2", canCast: true)
        await bridge.push(.casts([makeCast(1)]))
        // …flush the pipe with a ping round-trip, then assert nothing cast-shaped arrived.
        await client.send(.listSessions)
        _ = await recorder.waitFor { if case .sessionList = $0 { return true } else { return false } }
        let buffered_recorder_casts = await recorder.hasBuffered { (event: SyncClient.Event) in
            if case .casts = event { return true } else { return false }
        }
        #expect(!buffered_recorder_casts)

        // Subscribed but not cast-capable (a pre-cast client that somehow sent the verb is
        // impossible — it wouldn't know the tag — so model the gate with capability off).
        let bridge2 = castingBridge()
        let (host2, client2, recorder2) = try await makePaired(
            bridge: bridge2, deviceID: "iphone-3", canCast: false)
        await client2.send(.castSubscribe(CastSubscribe()))
        await bridge2.push(.casts([makeCast(2)]))
        await client2.send(.listSessions)
        _ = await recorder2.waitFor { if case .sessionList = $0 { return true } else { return false } }
        let buffered_recorder2_casts = await recorder2.hasBuffered { (event: SyncClient.Event) in
            if case .casts = event { return true } else { return false }
        }
        #expect(!buffered_recorder2_casts)
        await host.stop()
        await host2.stop()
    }

    @Test func echoSuppressionNeverReturnsAPeersOwnCasts() async throws {
        let bridge = castingBridge()
        // The "peer" client's deviceID IS the cast origin (a Mac dialed into us).
        let (host, peerClient, peerRecorder) = try await makePaired(
            bridge: bridge, deviceID: originA, canCast: true)
        await peerClient.send(.castSubscribe(CastSubscribe()))
        await bridge.push(.casts([makeCast(3)]))
        await peerClient.send(.listSessions)
        _ = await peerRecorder.waitFor { if case .sessionList = $0 { return true } else { return false } }
        let buffered_peerRecorder_casts = await peerRecorder.hasBuffered { (event: SyncClient.Event) in
            if case .casts = event { return true } else { return false }
        }
        #expect(!buffered_peerRecorder_casts)

        // And catch-up likewise skips the peer's own origin.
        await bridge.setCastCatchUp(
            tips: [CastTip(originHostID: originA, channel: "activity", mode: .stream, tipSeq: 3, floorSeq: 1)],
            batches: ["\(originA)/activity": [CastCatchUp(
                originHostID: originA, channel: "activity", mode: .stream, batchIndex: 0,
                casts: [makeCast(3)], complete: true, tipSeq: 3)]])
        await peerClient.send(.castSubscribe(CastSubscribe()))
        await peerClient.send(.listSessions)
        _ = await peerRecorder.waitFor { if case .sessionList = $0 { return true } else { return false } }
        let buffered_peerRecorder_castCatchUp = await peerRecorder.hasBuffered { (event: SyncClient.Event) in
            if case .castCatchUp = event { return true } else { return false }
        }
        #expect(!buffered_peerRecorder_castCatchUp)
        await host.stop()
    }

    @Test func channelFilterLimitsBothLegs() async throws {
        let bridge = castingBridge()
        await bridge.setCastCatchUp(
            tips: [
                CastTip(originHostID: originA, channel: "activity", mode: .stream, tipSeq: 2, floorSeq: 1),
                CastTip(originHostID: originA, channel: "messages", mode: .stream, tipSeq: 2, floorSeq: 1),
            ],
            batches: [
                "\(originA)/activity": [CastCatchUp(
                    originHostID: originA, channel: "activity", mode: .stream, batchIndex: 0,
                    casts: [makeCast(1)], complete: true, tipSeq: 2)],
                "\(originA)/messages": [CastCatchUp(
                    originHostID: originA, channel: "messages", mode: .stream, batchIndex: 0,
                    casts: [makeCast(1, channel: "messages")], complete: true, tipSeq: 2)],
            ])
        let (host, client, recorder) = try await makePaired(
            bridge: bridge, deviceID: "iphone-4", canCast: true)
        await client.send(.castSubscribe(CastSubscribe(channels: ["activity"])))

        let event = await recorder.waitFor {
            if case .castCatchUp = $0 { return true } else { return false }
        }
        guard case .castCatchUp(let received) = event else { Issue.record("no catch-up"); return }
        #expect(received.channel == "activity")

        // Live: only the subscribed channel passes the filter.
        await bridge.push(.casts([makeCast(3, channel: "messages"), makeCast(3)]))
        let live = await recorder.waitFor { if case .casts = $0 { return true } else { return false } }
        guard case .casts(let batch) = live else { Issue.record("no live casts"); return }
        #expect(batch.casts.map(\.channel) == ["activity"])
        // And the messages catch-up never arrived.
        let buffered_recorder_castCatchUp = await recorder.hasBuffered { (event: SyncClient.Event) in
            if case .castCatchUp = event { return true } else { return false }
        }
        #expect(!buffered_recorder_castCatchUp)
        await host.stop()
    }
}
