import Foundation
import NucleicProtocol

// Compiled ONLY on Windows (this directory is excluded elsewhere — Package.swift).

/// Adapts one broker-managed in-container process (`proc.exec`) onto the ``ProcessHandle``
/// contract — the Windows mirror of `ContainerizedProcessHandle`: broker `proc.stdout` /
/// `proc.stderr` notifications feed the same ``LineSplitter`` framing a host `ChildProcess`
/// uses, `writeLine` frames NDJSON onto `proc.stdin`, `wait()` resolves on `proc.exit`, and
/// signals ride `proc.signal` (real POSIX signals — wslc delivers them in-guest, §4.3).
final class WslcProcessHandle: ProcessHandle, @unchecked Sendable {
    private let client: WslcBrokerClient
    private let procId: Int64
    private let activity = ActivityClock()

    let stdoutLines: AsyncThrowingStream<Data, Error>
    let stderrLines: AsyncThrowingStream<Data, Error>
    private let stdoutCont: AsyncThrowingStream<Data, Error>.Continuation
    private let stderrCont: AsyncThrowingStream<Data, Error>.Continuation
    private let exitTask: Task<Int32, Never>
    private let exitCont: AsyncStream<Int32>.Continuation

    /// Not a host pid: the broker keys the process by its minted id. Never signaled directly.
    var processID: Int32 { Int32(truncatingIfNeeded: procId) }
    var lastActivityNanos: UInt64 { activity.lastNanos }
    var supportsHostCPUSampling: Bool { false }

    /// Built by ``WslcContainerEngine/exec`` after the broker answered `proc.exec`; installs
    /// its notification sinks with the client before returning, so no early output is lost
    /// (the broker only emits after the exec response is enqueued — same ordered pipe).
    init(client: WslcBrokerClient, procId: Int64) async {
        self.client = client
        self.procId = procId

        let (outStream, outCont) = AsyncThrowingStream<Data, Error>.makeStream()
        let (errStream, errCont) = AsyncThrowingStream<Data, Error>.makeStream()
        stdoutLines = outStream
        stderrLines = errStream
        stdoutCont = outCont
        stderrCont = errCont

        let (exitStream, exitCont) = AsyncStream<Int32>.makeStream()
        self.exitCont = exitCont
        exitTask = Task {
            for await code in exitStream { return code }
            return -1  // stream finished without a code: broker lost
        }

        let outSplitter = LineSplitter()
        let errSplitter = LineSplitter()
        let activity = self.activity
        await client.register(procId: procId, sinks: .init(
            stdout: { chunk in
                activity.bump()
                for line in outSplitter.push(chunk) { outCont.yield(line) }
            },
            stderr: { chunk in
                activity.bump()
                for line in errSplitter.push(chunk) { errCont.yield(line) }
            },
            exit: { code in
                if let rest = outSplitter.drain() { outCont.yield(rest) }
                if let rest = errSplitter.drain() { errCont.yield(rest) }
                outCont.finish()
                errCont.finish()
                exitCont.yield(code)
                exitCont.finish()
            }))
    }

    func writeLine(_ data: Data) throws {
        let client = client
        let procId = procId
        Task {
            _ = try? await client.call(
                "proc.stdin",
                .object([
                    "procId": .number(Double(procId)),
                    "b64": .string(data.base64EncodedString()),
                ]))
        }
    }

    func closeStdin() {
        let client = client
        let procId = procId
        Task {
            _ = try? await client.call(
                "proc.closeStdin", .object(["procId": .number(Double(procId))]))
        }
    }

    func sendSignal(_ sig: Int32) {
        let client = client
        let procId = procId
        Task {
            _ = try? await client.call(
                "proc.signal",
                .object(["procId": .number(Double(procId)), "sig": .number(Double(sig))]))
        }
    }

    /// Resize the guest tty (Terminal panel; pipe-mode processes ignore it).
    func resize(cols: Int, rows: Int) {
        let client = client
        let procId = procId
        Task {
            _ = try? await client.call(
                "proc.resize",
                .object([
                    "procId": .number(Double(procId)),
                    "cols": .number(Double(cols)), "rows": .number(Double(rows)),
                ]))
        }
    }

    func wait() async -> Int32 {
        await exitTask.value
    }

    /// Broker lost mid-run: finish the streams so a blocked consumer is released — the same
    /// recovery seam `ContainerizedProcessHandle.forceCloseStreams()` serves on macOS.
    func forceCloseStreams() {
        stdoutCont.finish()
        stderrCont.finish()
        exitCont.yield(-1)
        exitCont.finish()
    }
}
