import Foundation
import IrdP2PCrypto
import IrdP2PSignaling

public struct RoomOptions: Sendable {
    /// How long an unverified peer may stay silent before it is ignored as a stranger.
    public var helloTimeout: Duration = .seconds(10)
    /// App-level `ping` cadence on the relay socket.
    public var keepaliveInterval: Duration = .seconds(20)
    /// Bound on bytes held in partially reassembled messages per peer.
    public var maxInflightChunkBytes: Int = 8 << 20
    /// How long a partially reassembled message may wait for its remaining chunks.
    public var maxChunkAge: Duration = .seconds(60)
    /// Open WebRTC data channels and prefer them over the relay.
    public var preferDataChannel = true
    /// Caps the WebRTC mesh: only the peers nearest by XOR distance on the peer id get a
    /// data channel, the rest stay relayed. `nil` connects to everyone.
    public var maxDirectPeers: Int? = nil
    /// Keep gathering ICE candidates after the first set, so a peer connection survives
    /// network changes. `false` gathers once: fewer TURN allocations and less radio use.
    public var continualGathering: Bool = true
    /// Pace relay frames against the node's budgets (disable only in tests).
    public var pacing = true

    public init() {}
}

/// Diagnostic hook: the SDK's own tests attach a sink to see the room's internal decisions.
enum RoomTrace {
    nonisolated(unsafe) static var sink: (@Sendable (String) -> Void)?
    static func log(_ message: @autoclosure () -> String) { if let sink { sink(message()) } }
}

/// One joined room: verified peers, end-to-end encrypted messaging over data
/// channels with the relay as fallback, and a stream of events.
public actor Room {
    public nonisolated let roomId: RoomId
    public nonisolated let peerId: PeerId

    private var crypto: RoomCrypto
    private let relay: RelaySession
    private let identity: Identity?
    private let appSalt: String
    private let options: RoomOptions
    private let clock: any P2PClock
    private let peerConnections: any PeerConnectionFactory
    private let ice: @Sendable () async -> IceConfiguration
    private let maxFrameBytes: Int
    private let bus = EventBus<RoomEvent>()

    private var peers: [PeerId: PeerState] = [:]
    private var strangers: Set<PeerId> = []
    private var relayRoster: Set<PeerId> = []
    private var relayTask: Task<Void, Never>?
    private var lifecycle: AppLifecycleObserver?
    private var closed = false
    private var joinedOnce = false

    struct PeerState {
        var verified = false
        var helloSent = false
        var relayPresent = true
        var transport: PeerTransport = .relayed
        var identity: PeerIdentity?
        var pc: (any PeerConnectionHandle)?
        var pcTask: Task<Void, Never>?
        var pcGeneration = 0
        var remoteDescriptionSet = false
        var pendingIce: [(String, String?, Int?)] = []
        var replaySig = ReplayWindow()
        var replayMsg = ReplayWindow()
        var assembler: ChunkAssembler
        var helloTimer: Task<Void, Never>?
        var restartTask: Task<Void, Never>?
        var restartAttempt = 0
        var negotiating = false
        var budgetDeferred = false
    }

    init(crypto: RoomCrypto, relay: RelaySession, identity: Identity?, appSalt: String, options: RoomOptions,
         clock: any P2PClock, peerConnections: any PeerConnectionFactory, ice: @escaping @Sendable () async -> IceConfiguration,
         maxFrameBytes: Int, observeLifecycle: Bool) {
        roomId = crypto.roomId
        peerId = crypto.peerId
        self.crypto = crypto
        self.relay = relay
        self.identity = identity
        self.appSalt = appSalt
        self.options = options
        self.clock = clock
        self.peerConnections = peerConnections
        self.ice = ice
        self.maxFrameBytes = maxFrameBytes
        if observeLifecycle {
            // set after init completes, see start()
        }
    }

    private var tracePrefix: String { "\(roomId.string.prefix(6))/\(peerId.hex.prefix(4))" }

    // MARK: - public surface

    /// A fresh stream of events for this subscriber (unbounded buffer: drain it).
    /// It starts with the current `.status` and `.roster`, then continues live.
    public nonisolated var events: AsyncStream<RoomEvent> { bus.stream() }

    /// Verified members.
    public var peers_: [PeerId] { peers.filter { $0.value.verified }.map(\.key).sorted() }
    public var members: [PeerId] { peers_ }

    public func transport(of peer: PeerId) -> PeerTransport? { peers[peer].flatMap { $0.verified ? $0.transport : nil } }
    public func identity(of peer: PeerId) -> PeerIdentity? { peers[peer]?.identity }

    /// Sends to one verified peer: over its data channel when open, else through the relay.
    public func send(to peer: PeerId, _ message: Message) async throws {
        guard !closed else { throw P2PError.closed }
        guard let state = peers[peer], state.verified else { throw P2PError.peerUnknown(peer) }
        if state.transport == .direct, let pc = state.pc, options.preferDataChannel {
            do { try await sendDirect(message, to: peer, pc: pc) }
            catch { try await sendRelayed(message, to: peer) }   // the channel closed mid-send
        } else {
            try await sendRelayed(message, to: peer)
        }
    }

    public func send<T: Encodable>(to peer: PeerId, json value: T) async throws {
        try await send(to: peer, .json(try JSONValue(encoding: value)))
    }

    /// Sends to every verified peer; failures for individual peers are swallowed.
    public func broadcast(_ message: Message) async {
        for p in peers_ { try? await send(to: p, message) }
    }

    public func broadcast<T: Encodable>(json value: T) async throws {
        await broadcast(.json(try JSONValue(encoding: value)))
    }

    /// The app came to the foreground or the network changed: reconnect now, re-hello
    /// silent peers and retry lost data channels.
    public func wake() async {
        guard !closed else { return }
        await relay.wake()
        strangers.removeAll()
        for id in relayRoster where peers[id] == nil || !peers[id]!.verified { await ensurePeer(id) }
        for (id, state) in peers where state.verified && state.transport == .relayed && state.relayPresent {
            peers[id]?.restartTask?.cancel()
            peers[id]?.restartTask = nil
            peers[id]?.restartAttempt = 0
            await maybeNegotiate(id, restart: true)
        }
    }

    /// Leaves the room and closes every connection; the room cannot be rejoined.
    public func leave() async {
        guard !closed else { return }
        closed = true
        lifecycle?.cancel()
        relayTask?.cancel()
        for (id, var s) in peers {
            s.helloTimer?.cancel(); s.restartTask?.cancel(); s.pcTask?.cancel()
            if let pc = s.pc { await pc.close() }
            s.pc = nil
            peers[id] = s
        }
        peers.removeAll()
        await relay.leave()
        bus.sendSticky("status", .status(.closed(reason: "left")))
        bus.finish()
    }

    // MARK: - start

    func start(observeLifecycle: Bool) async throws {
        try await relay.start()
        relayTask = Task { [weak self] in
            guard let self else { return }
            for await event in self.relay.events {
                await self.handle(relay: event)
            }
        }
        if observeLifecycle {
            lifecycle = AppLifecycleObserver { [weak self] in await self?.wake() }
        }
    }

    // MARK: - relay events

    private func handle(relay event: RelayEvent) async {
        guard !closed else { return }
        switch event {
        case .joined:
            bus.sendSticky("status", .status(.joined))
        case .roster(let list):
            let fresh = Set(list)
            // peers gone from the roster while we were away
            for (id, s) in peers where !fresh.contains(id) && s.relayPresent {
                await relayLeft(id)
            }
            relayRoster = fresh
            for id in list { await ensurePeer(id) }
            if joinedOnce {
                // rejoin: re-hello everyone so peers with a fresh session re-verify us
                for id in list where peers[id]?.verified == true { await sendHello(to: id) }
            }
            joinedOnce = true
            bus.sendSticky("roster", .roster(peers_))
        case .peerJoined(let id):
            relayRoster.insert(id)
            strangers.remove(id)
            if var s = peers[id] {
                s.relayPresent = true
                peers[id] = s
                await sendHello(to: id)
            } else {
                await ensurePeer(id)
            }
        case .peerLeft(let id):
            relayRoster.remove(id)
            await relayLeft(id)
        case .signal(let from, let payload):
            await handleSignal(from: from, payload: payload)
        case .relayError(let code):
            bus.send(.relayError(code))
        case .reconnecting(let attempt, _):
            bus.sendSticky("status", .status(.reconnecting(attempt: attempt)))
        case .closed(let reason):
            if !closed {
                closed = true
                for (_, s) in peers { s.helloTimer?.cancel(); s.restartTask?.cancel(); s.pcTask?.cancel(); if let pc = s.pc { await pc.close() } }
                peers.removeAll()
                bus.sendSticky("status", .status(.closed(reason: reason)))
                bus.finish()
            }
        }
    }

    private func relayLeft(_ id: PeerId) async {
        guard var s = peers[id] else { return }
        s.relayPresent = false
        if s.verified && s.transport == .direct {
            // still talking directly; keep the peer until the data channel dies
            peers[id] = s
            return
        }
        await removePeer(id, announce: s.verified)
    }

    private func removePeer(_ id: PeerId, announce: Bool) async {
        guard let s = peers.removeValue(forKey: id) else { return }
        s.helloTimer?.cancel(); s.restartTask?.cancel(); s.pcTask?.cancel()
        if let pc = s.pc { await pc.close() }
        if announce { bus.send(.peerLeft(id)) }
        if s.verified { await budgetFreed() }
    }

    /// Knows about `id` and makes sure a hello is on its way (unless it is a known stranger).
    private func ensurePeer(_ id: PeerId) async {
        guard id != peerId, !strangers.contains(id) else { return }
        if peers[id] == nil {
            peers[id] = PeerState(assembler: ChunkAssembler(maxBytes: options.maxInflightChunkBytes, maxAge: options.maxChunkAge))
        }
        guard peers[id]?.helloSent == false else { return }
        await sendHello(to: id)
        armHelloTimer(id)
    }

    private func armHelloTimer(_ id: PeerId) {
        peers[id]?.helloTimer?.cancel()
        let timeout = options.helloTimeout
        peers[id]?.helloTimer = Task { [weak self] in
            guard (try? await self?.clock.sleep(for: timeout)) != nil else { return }
            await self?.helloTimedOut(id)
        }
    }

    private func helloTimedOut(_ id: PeerId) async {
        guard let s = peers[id], !s.verified else { return }
        strangers.insert(id)
        await removePeer(id, announce: false)
    }

    private func sendHello(to id: PeerId) async {
        var block: IdentityBlock?
        if let identity {
            block = IdentityBlock(pub: identity.publicKeyBase64URL, sig: identity.assert(appSalt: appSalt, roomId: roomId, peerId: peerId),
                                  name: identity.name, hue: identity.hue, glyph: identity.glyph)
        }
        peers[id]?.helloSent = true
        await sendInner(.hello(block), to: id)
    }

    // MARK: - signaling path

    private func sendInner(_ inner: InnerMessage, to id: PeerId) async {
        let env = crypto.sealSignal(inner.json, to: id)
        do { try await relay.send(signal: env.json, to: id) } catch {
            // frameTooLarge/notJoined: nothing more to do here; callers that care use sendRelayed
            RoomTrace.log("\(tracePrefix) sendInner \(inner.json["kind"]?.stringValue ?? "?") to \(id.hex.prefix(4)) FAILED: \(error)")
        }
    }

    private func handleSignal(from: PeerId, payload: JSONValue) async {
        guard from != peerId, let env = Envelope(json: payload) else { RoomTrace.log("\(tracePrefix) drop non-envelope from \(from.hex.prefix(4))"); return }
        guard let plain = crypto.openSignal(env, from: from) else { RoomTrace.log("\(tracePrefix) drop undecryptable from \(from.hex.prefix(4))"); return }   // not a member: drop silently
        RoomTrace.log("\(tracePrefix) got \(plain["kind"]?.stringValue ?? "?") from \(from.hex.prefix(4))")
        if peers[from] == nil {
            strangers.remove(from)
            peers[from] = PeerState(assembler: ChunkAssembler(maxBytes: options.maxInflightChunkBytes, maxAge: options.maxChunkAge))
        }
        guard peers[from]!.replaySig.accept(iv: env.iv) else { return }
        await verified(from)
        guard let inner = InnerMessage(json: plain) else { return }
        await handle(inner: inner, from: from, viaDataChannel: false)
    }

    /// The first decrypted envelope proves membership.
    private func verified(_ id: PeerId) async {
        guard var s = peers[id] else { return }
        if !s.helloSent {
            peers[id] = s
            await sendHello(to: id)
            s = peers[id]!
        }
        guard !s.verified else { return }
        s.verified = true
        s.helloTimer?.cancel()
        s.helloTimer = nil
        peers[id] = s
        bus.send(.peerJoined(id))
        bus.send(.peerReady(id, .relayed))
        await maybeNegotiate(id)
    }

    private func handle(inner: InnerMessage, from: PeerId, viaDataChannel: Bool) async {
        switch inner {
        case .hello(let block):
            if let block, let fp = Identity.verifyAssertion(appSalt: appSalt, roomId: roomId, peerId: from, publicKeyBase64URL: block.pub, signature: block.sig) {
                let ident = PeerIdentity(publicKeyBase64URL: block.pub, fingerprint: fp, name: block.name.map { String($0.prefix(32)) }, hue: block.hue.flatMap { (0..<360).contains($0) ? $0 : nil }, glyph: block.glyph.map { String($0.prefix(1)) })
                if peers[from]?.identity != ident {
                    peers[from]?.identity = ident
                    bus.send(.peerIdentity(from, ident))
                }
            }
        case .offer(let sdp): await handleOffer(sdp, from: from)
        case .answer(let sdp): await handleAnswer(sdp, from: from)
        case .ice(let candidate, let mid, let index): await handleIce(candidate, mid, index, from: from)
        case .restart:
            guard isOfferer(for: from) else { break }
            // the answerer gave up on our offer (or refused it, see handleOffer): it is not in flight any more
            if peers[from]?.transport == .relayed { peers[from]?.negotiating = false }
            scheduleRestart(from)
        case .msg(let payload): bus.send(.message(from: from, .json(payload)))
        case .bin(let data): bus.send(.message(from: from, .data(data)))
        case .chunk(let id, let index, let count, let enc, let data):
            guard var s = peers[from] else { return }
            let now = clock.now
            let expired = s.assembler.sweep(now: now)
            let result = s.assembler.add(id: id, index: index, count: count, encoding: enc, data: data, now: now)
            peers[from] = s
            for drop in expired { bus.send(.assemblyDropped(peer: from, id: drop.id, reason: drop.reason)) }
            switch result {
            case .complete(.bin, let whole): bus.send(.message(from: from, .data(whole)))
            case .complete(.json, let whole):
                if let v = try? JSONValue.parse(whole), let m = InnerMessage(json: v), case .msg(let payload) = m {
                    bus.send(.message(from: from, .json(payload)))
                }
            case .dropped(let reason): bus.send(.assemblyDropped(peer: from, id: id, reason: reason))
            case .incomplete: break
            }
        case .ping, .unknown: break
        }
    }

    // MARK: - WebRTC negotiation (lower peer id offers; higher answers)

    private func isOfferer(for peer: PeerId) -> Bool { peerId < peer }

    /// `maxDirectPeers`: whether `peer` is within the budget (see `DirectBudget`), judged on
    /// the verified roster. The budget only gates NEW negotiations: a data channel that is
    /// already open when a roster change pushes its peer out of the budget is never torn
    /// down — it lives until it dies naturally, and only then is not renegotiated. A
    /// roster-size threshold ("mesh off above N members") would instead drop every
    /// connection at member N+1 and flap around it.
    private func directAllowed(_ peer: PeerId) -> Bool {
        guard let limit = options.maxDirectPeers else { return true }
        return DirectBudget.allows(peerId, peer, members: Set(peers_), limit: limit)
    }

    /// A verified peer left: peers the budget kept relayed may be within it now.
    private func budgetFreed() async {
        guard options.maxDirectPeers != nil, !closed else { return }
        for (id, s) in peers where s.budgetDeferred && s.restartTask == nil { await maybeNegotiate(id) }
    }

    /// Every path that offers (verification, the restart ladder, `wake()`) passes through the budget here.
    private func maybeNegotiate(_ id: PeerId, restart: Bool = false) async {
        guard options.preferDataChannel, isOfferer(for: id), var s = peers[id], s.verified, s.relayPresent else { return }
        guard directAllowed(id) else {
            s.budgetDeferred = true
            // an offer the answerer refused while the rosters disagreed is abandoned; an open channel stays
            let abandoned = s.transport == .relayed && s.pc != nil
            if abandoned { s.negotiating = false }
            peers[id] = s
            if abandoned { await tearDownPC(id) }
            return
        }
        s.budgetDeferred = false
        if s.transport == .direct && !restart { return }
        if s.negotiating { return }
        s.negotiating = true
        peers[id] = s
        await tearDownPC(id)
        do {
            let pc = try await peerConnections.makePeerConnection(ice: await ice(), continualGathering: options.continualGathering)
            attach(pc, to: id)
            let sdp = try await pc.createOffer(iceRestart: restart)
            await sendInner(.offer(sdp: sdp), to: id)
        } catch {
            peers[id]?.negotiating = false
            scheduleRestart(id)
        }
    }

    private func handleOffer(_ sdp: String, from: PeerId) async {
        guard options.preferDataChannel, !isOfferer(for: from), peers[from]?.verified == true else {
            RoomTrace.log("\(tracePrefix) handleOffer guard failed: prefer=\(options.preferDataChannel) offerer=\(isOfferer(for: from)) verified=\(peers[from]?.verified ?? false)")
            return
        }
        // The budget gates answering too, or an offerer outside it would connect anyway. The rule
        // is symmetric, so with agreeing rosters no such offer is ever sent; one arrives only while
        // the rosters momentarily disagree (a join or leave one side has not seen yet). Refusing is
        // safe: no connection forms, and `restart` tells the offerer its offer is dead, so it retries
        // on its restart ladder — by then the rosters agree and it either stays quiet or is answered.
        guard directAllowed(from) else {
            RoomTrace.log("\(tracePrefix) handleOffer refused: \(from.hex.prefix(4)) is outside the direct budget")
            await sendInner(.restart, to: from)
            return
        }
        await tearDownPC(from)
        do {
            let pc = try await peerConnections.makePeerConnection(ice: await ice(), continualGathering: options.continualGathering)
            attach(pc, to: from)
            try await pc.setRemoteDescription(.offer, sdp: sdp)
            peers[from]?.remoteDescriptionSet = true
            await flushIce(from)
            let answer = try await pc.createAnswer()
            await sendInner(.answer(sdp: answer), to: from)
        } catch {
            RoomTrace.log("\(tracePrefix) handleOffer failed: \(error)")
            await tearDownPC(from)
        }
    }

    private func handleAnswer(_ sdp: String, from: PeerId) async {
        guard isOfferer(for: from), let pc = peers[from]?.pc, peers[from]?.remoteDescriptionSet == false else { return }
        do {
            try await pc.setRemoteDescription(.answer, sdp: sdp)
            peers[from]?.remoteDescriptionSet = true
            await flushIce(from)
        } catch {
            peers[from]?.negotiating = false
            scheduleRestart(from)
        }
    }

    private func handleIce(_ candidate: String?, _ mid: String?, _ index: Int?, from: PeerId) async {
        guard let candidate, var s = peers[from] else { return }   // end-of-candidates carries nothing to add
        if s.remoteDescriptionSet, let pc = s.pc {
            try? await pc.addIceCandidate(candidate: candidate, sdpMid: mid, sdpMLineIndex: index)
        } else {
            if s.pendingIce.count < 64 { s.pendingIce.append((candidate, mid, index)) }
            peers[from] = s
        }
    }

    private func flushIce(_ id: PeerId) async {
        guard var s = peers[id], let pc = s.pc else { return }
        let pending = s.pendingIce
        s.pendingIce.removeAll()
        peers[id] = s
        for (c, m, i) in pending { try? await pc.addIceCandidate(candidate: c, sdpMid: m, sdpMLineIndex: i) }
    }

    private func attach(_ pc: any PeerConnectionHandle, to id: PeerId) {
        guard var s = peers[id] else { return }
        s.pcGeneration += 1
        let generation = s.pcGeneration
        s.pc = pc
        s.remoteDescriptionSet = false
        s.pendingIce.removeAll()
        s.pcTask = Task { [weak self] in
            for await event in pc.events {
                guard let self else { return }
                await self.handle(pc: event, from: id, generation: generation)
            }
        }
        peers[id] = s
    }

    private func tearDownPC(_ id: PeerId) async {
        guard var s = peers[id] else { return }
        s.pcTask?.cancel()
        s.pcTask = nil
        let old = s.pc
        s.pc = nil
        s.remoteDescriptionSet = false
        s.pendingIce.removeAll()
        if s.transport == .direct {
            s.transport = .relayed
            peers[id] = s
            if s.verified { bus.send(.peerReady(id, .relayed)) }
        } else {
            peers[id] = s
        }
        if let old { await old.close() }
    }

    private func handle(pc event: PeerConnectionEvent, from id: PeerId, generation: Int) async {
        guard var s = peers[id], s.pcGeneration == generation else { RoomTrace.log("\(tracePrefix) pc event \(event) for \(id.hex.prefix(4)) ignored (gen \(generation) vs \(peers[id]?.pcGeneration ?? -1))"); return }
        RoomTrace.log("\(tracePrefix) pc event \(event) from \(id.hex.prefix(4))")
        switch event {
        case .iceCandidate(let c, let mid, let index):
            await sendInner(.ice(candidate: c, sdpMid: mid, sdpMLineIndex: index), to: id)
        case .iceGatheringComplete:
            await sendInner(.ice(candidate: nil, sdpMid: nil, sdpMLineIndex: nil), to: id)
        case .dataChannelOpen:
            s.transport = .direct
            s.negotiating = false
            s.restartAttempt = 0
            s.restartTask?.cancel()
            s.restartTask = nil
            peers[id] = s
            if s.verified { bus.send(.peerReady(id, .direct)) }
        case .dataChannelClosed, .connectionState(.failed), .connectionState(.closed), .connectionState(.disconnected):
            let wasDirect = s.transport == .direct
            s.transport = .relayed
            s.negotiating = false
            peers[id] = s
            if !s.relayPresent {
                await removePeer(id, announce: s.verified)
                return
            }
            if wasDirect, s.verified { bus.send(.peerReady(id, .relayed)) }
            if isOfferer(for: id) { scheduleRestart(id) }
        case .connectionState:
            break
        case .message(let data, let isBinary):
            await receiveDirect(data, isBinary: isBinary, from: id)
        case .bufferedAmountLow:
            break
        }
    }

    /// Schedules the next rung of the ICE-restart ladder. Idempotent while a restart
    /// is already pending: a lost data channel reports several events (closed, failed).
    private func scheduleRestart(_ id: PeerId) {
        guard var s = peers[id], s.verified, s.relayPresent, s.restartTask == nil else { return }
        let ladder = Backoff.iceRestartLadder
        guard s.restartAttempt < ladder.count else { return }   // idle until wake()
        let delay = ladder[s.restartAttempt]
        s.restartAttempt += 1
        s.restartTask = Task { [weak self] in
            guard (try? await self?.clock.sleep(for: delay)) != nil else { return }
            await self?.restartFired(id)
        }
        peers[id] = s
    }

    private func restartFired(_ id: PeerId) async {
        peers[id]?.restartTask = nil
        await maybeNegotiate(id, restart: true)
    }

    // MARK: - data path

    private func sendDirect(_ message: Message, to id: PeerId, pc: any PeerConnectionHandle) async throws {
        switch message {
        case .data(let data) where data.count <= 16_000:
            try await pc.send(data: crypto.sealMessageBinary(data))
        case .data(let data):
            for chunk in Chunker.split(data, encoding: .bin, sliceSize: Chunker.dataChannelSliceSize) {
                try await pc.send(text: crypto.sealMessage(chunk.json).json.serialized)
            }
        case .json(let payload):
            let inner = InnerMessage.msg(payload)
            let bytes = inner.json.serializedData
            if bytes.count <= Chunker.dataChannelSliceSize {
                try await pc.send(text: crypto.sealMessage(inner.json).json.serialized)
            } else {
                for chunk in Chunker.split(bytes, encoding: .json, sliceSize: Chunker.dataChannelSliceSize) {
                    try await pc.send(text: crypto.sealMessage(chunk.json).json.serialized)
                }
            }
        }
    }

    private func sendRelayed(_ message: Message, to id: PeerId) async throws {
        let slice = Chunker.relaySliceSize(maxFrameBytes: maxFrameBytes)
        switch message {
        case .data(let data) where data.count <= slice:
            try await relay.send(signal: crypto.sealSignal(InnerMessage.bin(data).json, to: id).json, to: id)
        case .data(let data):
            for chunk in Chunker.split(data, encoding: .bin, sliceSize: slice) {
                try await relay.send(signal: crypto.sealSignal(chunk.json, to: id).json, to: id)
            }
        case .json(let payload):
            let inner = InnerMessage.msg(payload)
            let bytes = inner.json.serializedData
            if bytes.count <= slice {
                try await relay.send(signal: crypto.sealSignal(inner.json, to: id).json, to: id)
            } else {
                for chunk in Chunker.split(bytes, encoding: .json, sliceSize: slice) {
                    try await relay.send(signal: crypto.sealSignal(chunk.json, to: id).json, to: id)
                }
            }
        }
    }

    private func receiveDirect(_ data: Data, isBinary: Bool, from id: PeerId) async {
        guard peers[id]?.verified == true else { return }
        if isBinary {
            guard data.count >= BinaryEnvelopePrefix.minimumLength else { return }
            let iv = Data(data[data.startIndex + 1 ..< data.startIndex + 13])
            guard peers[id]!.replayMsg.accept(iv: iv), let plain = crypto.openMessageBinary(data, from: id) else { return }
            bus.send(.message(from: id, .data(plain)))
        } else {
            guard let v = try? JSONValue.parse(data), let env = Envelope(json: v),
                  peers[id]!.replayMsg.accept(iv: env.iv), let plain = crypto.openMessage(env, from: id),
                  let inner = InnerMessage(json: plain) else { return }
            await handle(inner: inner, from: id, viaDataChannel: true)
        }
    }
}