nucleic/Sources/NucleicTailnet/FDFrameChannel.swift

import Foundation
import NucleicProtocol

/// A `FrameChannel` over a raw, already-connected file descriptor. The Tailnet transport's
/// dial/accept calls hand back one end of a socketpair that the embedded tsnet node pumps
/// to/from the tailnet TCP stream (libtailscale: "a pipe(2) on which you can use read(2),
/// write(2), and close(2)"), so plain DispatchIO gives us a full-duplex byte stream. Framing
/// matches `LANChannel`/`NWFrameChannel`: `FrameAccumulator` de-frames inbound bytes,
/// `WireFraming` length-prefixes outbound frames.
///
/// TailscaleKit-independent on purpose — it works over any connected fd, which is also what
/// makes it unit-testable with an ordinary socketpair.
public final class FDFrameChannel: FrameChannel, @unchecked Sendable {
    public let peerDescription: String
    private let io: DispatchIO
    private let queue = DispatchQueue(label: "nucleic.tailnet.channel")
    private let accumulator = FrameAccumulator()
    private let stream: AsyncStream<Data>
    private let continuation: AsyncStream<Data>.Continuation

    /// Wrap a connected, bidirectional fd. The channel takes ownership: the fd is closed
    /// exactly once, when the channel closes (or fails).
    public init(fd: Int32, peerDescription: String) {
        self.peerDescription = peerDescription
        var cont: AsyncStream<Data>.Continuation!
        stream = AsyncStream(bufferingPolicy: .unbounded) { cont = $0 }
        continuation = cont
        io = DispatchIO(type: .stream, fileDescriptor: fd, queue: queue) { _ in
            Darwin.close(fd)
        }
        io.setLimit(lowWater: 1)
        receiveLoop()
    }

    public func frames() -> AsyncStream<Data> { stream }

    public func send(_ frame: Data) {
        let framed = WireFraming.frame(frame)
        let dispatchData = framed.withUnsafeBytes { DispatchData(bytes: $0) }
        io.write(offset: 0, data: dispatchData, queue: queue) { [weak self] done, _, error in
            if done, error != 0 { self?.teardown() }
        }
    }

    public func close() { teardown() }

    private func teardown() {
        io.close(flags: .stop)
        continuation.finish()
    }

    private func receiveLoop() {
        io.read(offset: 0, length: Int.max, queue: queue) { [weak self] done, data, error in
            guard let self else { return }
            if let data, !data.isEmpty {
                if let frames = try? self.accumulator.push(Data(data)) {
                    for frame in frames { self.continuation.yield(frame) }
                } else {
                    // Oversize/garbage frame → drop the connection (same as LANChannel).
                    self.teardown()
                    return
                }
            }
            if done {
                // EOF or error — either way the peer is gone.
                self.teardown()
            }
        }
    }
}