import Foundation
import NucleicProtocol

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

/// The Windows `SandboxEngine` (docs/WINDOWS_PORT.md §3.2): the same mechanism surface the
/// Apple-containerization `ContainerEngine` provides on macOS, implemented over the broker
/// RPC. One wslc *session* per channel hosts every Nucleic container as siblings; per-spec
/// resources map to the session, mounts map to `ContainerVolume`s, and rootfs/kernel/vminitd
/// concepts vanish (WSL supplies kernel + init; wslc owns image storage).
public actor WslcContainerEngine: SandboxEngine {
    private let client: WslcBrokerClient

    /// Session-level knobs (wslc resources are per-session, not per-container — §3.2).
    private let sessionName: String
    private let sessionDataDir: String
    private let sessionCPUs: Int?
    private let sessionMemoryMB: Int64?

    private var sessionGateway: String?
    private var lastSpec: [String: ContainerSpec] = [:]
    /// Cold starts in flight, coalesced per name (same race as the macOS engine's `starting`).
    private var starting: [String: Task<(name: String, hostGateway: String), Error>] = [:]
    private var downloadProgress: ContainerDownloadProgress?

    public init(
        client: WslcBrokerClient,
        sessionCPUs: Int? = nil, sessionMemoryMB: Int64? = nil
    ) {
        self.client = client
        self.sessionName = "nucleic" + ContainerManager.channelSuffix
        let localAppData = ProcessInfo.processInfo.environment["LOCALAPPDATA"]
            ?? NSTemporaryDirectory()
        self.sessionDataDir = (localAppData as NSString)
            .appendingPathComponent("Nucleic\(ContainerManager.channelSuffix)/wslc")
        self.sessionCPUs = sessionCPUs
        self.sessionMemoryMB = sessionMemoryMB
    }

    // MARK: - Progress plumbing (fed by the client's image.pullProgress notifications)

    /// Fold a broker pull-progress notification into the shared download-progress surface.
    /// Installed as the `WslcBrokerClient.onPullProgress` sink by the hostd composition.
    public func notePullProgress(ref: String, status: String, current: Int64, total: Int64) {
        downloadProgress = status == "done"
            ? nil
            : ContainerDownloadProgress(phase: .image, receivedBytes: current, totalBytes: total)
    }

    public func currentDownloadProgress() async -> ContainerDownloadProgress? { downloadProgress }

    // MARK: - Session

    private func ensureSession() async throws -> String {
        if let sessionGateway { return sessionGateway }
        var params: [String: JSONValue] = [
            "name": .string(sessionName),
            "dataDir": .string(sessionDataDir),
        ]
        if let sessionCPUs { params["cpu"] = .number(Double(sessionCPUs)) }
        if let sessionMemoryMB { params["memoryMB"] = .number(Double(sessionMemoryMB)) }
        let result = try await client.call("session.ensure", .object(params))
        guard let gateway = result["gateway"]?.stringValue else {
            throw ContainerError.gatewayUnavailable
        }
        sessionGateway = gateway
        return gateway
    }

    /// A broker reattach invalidates the cached gateway (the new session may re-derive it).
    public func brokerReattached() {
        sessionGateway = nil
    }

    /// The wslc session's WSL-facing gateway, ensuring the session on first use — known
    /// before any container exists, which is what lets backends bind the approval server and
    /// bake its endpoint into the spec env ahead of `ensureRunning` (docs/WINDOWS_PORT.md §5).
    public func sessionHostGateway() async throws -> String? {
        try await ensureSession()
    }

    // MARK: - Lifecycle

    public func ensureRunning(_ spec: ContainerSpec) async throws
        -> (name: String, hostGateway: String)
    {
        lastSpec[spec.name] = spec
        if let inFlight = starting[spec.name] { return try await inFlight.value }
        let task = Task<(name: String, hostGateway: String), Error> {
            defer { starting[spec.name] = nil }
            return try await performStart(spec)
        }
        starting[spec.name] = task
        return try await task.value
    }

    private func performStart(_ spec: ContainerSpec) async throws
        -> (name: String, hostGateway: String)
    {
        let gateway = try await ensureSession()

        let state = try await containerState(spec.name)
        if state == "running" { return (spec.name, gateway) }

        if state == "absent" {
            try await ensureImage(spec.image)
            // narOS init (docs/NAROS.md §5) with the control bridge supervised; the bridge's
            // TCP dial target (NUCLEIC_CONTROL_HOST/PORT — §5) arrives via `spec.env` from
            // the approval-server plumbing of work item 7.
            var env = spec.env
            env["NAROS_BRIDGE"] = "1"
            let volumes: [JSONValue] = spec.mounts.map {
                .object([
                    "host": .string($0.host), "guest": .string($0.container),
                    "ro": .bool($0.readOnly),
                ])
            }
            try await callMapped("container.create", .object([
                "name": .string(spec.name),
                "image": .string(spec.image),
                "volumes": .array(volumes),
                "env": .object(env.mapValues { .string($0) }),
                "initArgv": .array([.string("/usr/sbin/naros-init")]),
            ]))
        }
        try await callMapped("container.start", .object(["name": .string(spec.name)]))
        return (spec.name, gateway)
    }

    /// Pull `image` unless the session already has it (progress rides the notification sink).
    private func ensureImage(_ image: String) async throws {
        let inspected = try await client.call(
            "image.inspect", .object(["ref": .string(image)]))
        if inspected["image"] != nil, inspected["image"] != .null { return }
        var params: [String: JSONValue] = ["ref": .string(image)]
        if let auth = Self.registryAuth(for: image) {
            params["auth"] = .object([
                "username": .string(auth.username), "password": .string(auth.password),
            ])
        }
        try await callMapped("image.pull", .object(params))
    }

    /// Basic-auth credentials for the private sandbox registry — same resolution rule as the
    /// macOS engine's `registryAuth(for:)` (NUCLEIC_REGISTRY_TOKEN env, else the app GitHub
    /// token for ghcr.io; scoped to the host that serves `ProjectSandbox.defaultImage`).
    static func registryAuth(for reference: String) -> (username: String, password: String)? {
        let env = ProcessInfo.processInfo.environment
        let segments = reference.split(separator: "/").map(String.init)
        let host = segments.first ?? ""
        let privateHost = ProjectSandbox.defaultImage.split(separator: "/").first.map(String.init)
            ?? "ghcr.io"
        guard host == privateHost else { return nil }
        let token: String?
        if let t = env["NUCLEIC_REGISTRY_TOKEN"], !t.isEmpty {
            token = t
        } else if privateHost == "ghcr.io", let t = GitHubCredentialStore.loadToken(), !t.isEmpty {
            token = t
        } else {
            token = nil
        }
        guard let token else { return nil }
        let username = env["NUCLEIC_REGISTRY_USER"].flatMap { $0.isEmpty ? nil : $0 }
            ?? ContainerServiceSettings.registryUsername
            ?? (segments.count >= 2 ? segments[1] : nil)
            ?? "x-access-token"
        return (username, token)
    }

    public func exec(
        name: String, workdir: String, env: [String: String], argv: [String],
        uid: Int? = nil, gid: Int? = nil, memoryLimitBytes: UInt64? = nil
    ) async throws -> any ProcessHandle {
        // Per-exec memory ceilings need a cgroup wrapper inside the container (§3.2); the
        // session-level MemoryMB cap is the enforced bound until then.
        var params: [String: JSONValue] = [
            "container": .string(name),
            "argv": .array(argv.map { .string($0) }),
            "env": .object(env.mapValues { .string($0) }),
            "cwd": .string(workdir),
            "tty": .bool(false),
        ]
        if let uid { params["uid"] = .number(Double(uid)) }
        if let gid { params["gid"] = .number(Double(gid)) }
        let result = try await callMapped("proc.exec", .object(params))
        guard let procId = result["procId"]?.intValue else {
            throw ContainerError.startFailed("broker returned no procId")
        }
        return await WslcProcessHandle(client: client, procId: Int64(procId))
    }

    public func runCapturing(
        name: String, argv: [String], env: [String: String] = [:], uid: Int = 0, gid: Int = 0
    ) async -> (exitCode: Int32, output: String)? {
        guard (try? await containerState(name)) == "running" else { return nil }
        let handle: any ProcessHandle
        do {
            handle = try await exec(
                name: name, workdir: "/", env: env, argv: argv, uid: uid, gid: gid)
        } catch {
            return (exitCode: -1, output: "\(error)")
        }
        func drain(_ stream: AsyncThrowingStream<Data, Error>) async -> [String] {
            var acc: [String] = []
            do {
                for try await line in stream where acc.count < 500 {
                    if let s = String(data: line, encoding: .utf8) { acc.append(s) }
                }
            } catch {}
            return acc
        }
        async let out = drain(handle.stdoutLines)
        async let err = drain(handle.stderrLines)
        let code = await handle.wait()
        let lines = await (out + err)
        return (exitCode: code, output: lines.suffix(60).joined(separator: "\n"))
    }

    public func stop(name: String) async {
        _ = try? await callMapped(
            "container.stop",
            .object(["name": .string(name), "signal": .number(15), "graceMs": .number(5000)]))
    }

    @discardableResult
    public func restart(name: String) async -> Bool {
        guard let spec = lastSpec[name] else {
            await stop(name: name)
            return false
        }
        await stop(name: name)
        return (try? await ensureRunning(spec)) != nil
    }

    /// Networking is scoped to the wslc session on Windows, so the equivalent of replacing macOS's
    /// shared vmnet reference is terminating that idle session. Its data directory persists; the
    /// next `ensureRunning` creates a fresh session/network and reuses the stored container state.
    public func resetNetwork() async {
        _ = try? await client.call("session.terminate")
        sessionGateway = nil
    }

    @discardableResult
    public func remove(name: String) async -> Bool {
        await stop(name: name)
        do {
            try await callMapped(
                "container.delete", .object(["name": .string(name), "force": .bool(true)]))
            return true
        } catch {
            return (try? await containerState(name)) == "absent"
        }
    }

    public func isRunning(_ name: String) async -> Bool {
        (try? await containerState(name)) == "running"
    }

    public func containerExists(_ name: String) async -> Bool {
        let state = (try? await containerState(name)) ?? "absent"
        return state != "absent"
    }

    public func list() async -> [String] {
        guard let result = try? await client.call("container.list") else { return [] }
        return (result["containers"]?.arrayValue ?? []).compactMap { $0["name"]?.stringValue }
    }

    private func containerState(_ name: String) async throws -> String {
        let result = try await callMapped("container.state", .object(["name": .string(name)]))
        return result["state"]?.stringValue ?? "absent"
    }

    // MARK: - Resource usage

    public func sampleResourceUsage(name: String) async -> ContainerResourceSample? {
        // Same shape as the macOS engine: two cumulative-CPU reads around a short sleep,
        // normalized to 0…100 (the broker reports cgroup v2 counters).
        guard let s1 = await stats(name) else { return nil }
        let t1 = DispatchTime.now().uptimeNanoseconds
        try? await Task.sleep(nanoseconds: 200_000_000)
        guard let s2 = await stats(name) else { return nil }
        let t2 = DispatchTime.now().uptimeNanoseconds
        let windowUsec = max(1.0, Double(t2 &- t1) / 1000.0)
        let deltaUsec = max(0, Double(s2.cpuUsageUsec) - Double(s1.cpuUsageUsec))
        return ContainerResourceSample(
            cpuPercent: deltaUsec / windowUsec * 100,
            memoryUsedBytes: UInt64(max(0, s2.memoryUsedBytes)),
            memoryTotalBytes: UInt64(max(0, s2.memoryLimitBytes)))
    }

    private struct Stats {
        let cpuUsageUsec: Int
        let memoryUsedBytes: Int
        let memoryLimitBytes: Int
        let oomKills: Int?
    }

    private func stats(_ name: String) async -> Stats? {
        guard let result = try? await client.call(
            "container.stats", .object(["name": .string(name)])),
            let stats = result["stats"], stats != .null
        else { return nil }
        return Stats(
            cpuUsageUsec: stats["cpuUsageUsec"]?.intValue ?? 0,
            memoryUsedBytes: stats["memoryUsedBytes"]?.intValue ?? 0,
            memoryLimitBytes: stats["memoryLimitBytes"]?.intValue ?? 0,
            oomKills: stats["oomKills"]?.intValue)
    }

    // MARK: - Health

    public func probeControlPlane(name: String) async -> ContainerEngine.ControlPlaneProbe {
        // Identical in-guest probe to the macOS engine: the control plane is the same
        // loopback-bridge contract on both platforms (docs/WINDOWS_PORT.md §5).
        let script = """
            const net = require('net');
            let done = false;
            const finish = (code, msg) => { if (done) return; done = true; console.log(msg); process.exit(code); };
            const s = net.connect(\(ContainerSpec.controlBridgePort), '127.0.0.1');
            s.on('connect', () => { s.write('GET /mcp HTTP/1.1\\r\\nHost: nucleic\\r\\nConnection: close\\r\\n\\r\\n'); });
            s.on('data', () => finish(0, 'RESPONSIVE'));
            s.on('error', (e) => finish(2, 'REFUSED:' + (e && e.code)));
            setTimeout(() => finish(3, 'SILENT'), 3000);
            """
        guard let result = await runCapturing(name: name, argv: ["node", "-e", script]) else {
            return .probeFailed("container isn't running")
        }
        switch result.exitCode {
        case 0: return .responsive
        case 2: return .bridgeUnreachable(result.output)
        case 3: return .unresponsive
        default: return .probeFailed("exit \(result.exitCode): \(result.output)")
        }
    }

    public func diagnoseKill(name: String) async -> ContainerKillDiagnosis {
        guard (try? await containerState(name)) == "running" else {
            return ContainerKillDiagnosis(containerRunning: false, oomKills: nil, sample: nil)
        }
        guard let stats = await stats(name) else {
            return ContainerKillDiagnosis(
                containerRunning: true, probeUnanswered: true, oomKills: nil, sample: nil)
        }
        return ContainerKillDiagnosis(
            containerRunning: true,
            oomKills: stats.oomKills,
            sample: ContainerResourceSample(
                cpuPercent: 0,
                memoryUsedBytes: UInt64(max(0, stats.memoryUsedBytes)),
                memoryTotalBytes: UInt64(max(0, stats.memoryLimitBytes))))
    }

    // MARK: - Disk / image maintenance

    public func reconcileDisk(keepNames: Set<String>) async {
        // wslc state is service-backed (no per-container rootfs clones to GC): drop every
        // container this channel owns that isn't in the keep set.
        for name in await list()
        where ContainerManager.ownsContainer(named: name) && !keepNames.contains(name) {
            _ = await remove(name: name)
        }
    }

    public func removeDefaultRootfs() async {
        _ = try? await client.call(
            "image.delete", .object(["ref": .string(ProjectSandbox.defaultImage)]))
    }

    public func checkDefaultImageUpdate() async -> ImageUpdateStatus {
        // M2: manifest-digest HEAD against the registry, compared to the session's stored
        // digest (`image.inspect`) — the macOS `checkDefaultImageUpdate` logic relocated
        // behind the broker. Until then report idle (never a false "update available").
        .idle
    }

    public func reclaimMemoryNow() async {
        // No per-container balloon on WSL (§3.2): WSL's autoreclaim owns steady-state memory.
        // Best-effort immediate assist: drop page caches in each running container.
        for name in await list() where ContainerManager.ownsContainer(named: name) {
            _ = await runCapturing(
                name: name,
                argv: ["sh", "-c", "sync; echo 1 > /proc/sys/vm/drop_caches 2>/dev/null || true"],
                env: [:], uid: 0, gid: 0)
        }
    }

    // MARK: - Error mapping

    /// Map broker RPC failures onto the shared `ContainerError` surface so call sites see the
    /// same error taxonomy as the macOS engine.
    @discardableResult
    private func callMapped(_ method: String, _ params: JSONValue?) async throws -> JSONValue {
        do {
            return try await client.call(method, params)
        } catch let error as WslcBrokerClient.BrokerLost {
            throw ContainerError.unavailable(error.description)
        } catch let JSONRPCConnection.RPCError.server(_, message, kind) {
            // Prefer the broker's own `data.kind` — it knows why it failed, whereas inferring from
            // the method name is a guess that was sometimes wrong (a `proc.exec` refused because
            // the image cannot drop privileges is not `notRunning`).
            switch kind {
            case "not_found", "not_running": throw ContainerError.notRunning(message)
            case "image_pull_failed": throw ContainerError.imagePullFailed(message)
            // Both mean the sandbox as a whole is not usable right now, which is what
            // `.unavailable` says. `session_exists` additionally implies a manual
            // `wsl --shutdown` when D13 Tier 1 could not clear it — the message carries that.
            case "wslc_unavailable", "session_exists": throw ContainerError.unavailable(message)
            // `unsupported` means this broker build cannot serve the call as asked — no pty, a
            // signal outside wslc's six, an image whose setpriv cannot drop uid. Not "impossible":
            // each has a route that is deprioritized until unavoidable. What matters here is only
            // that RETRYING will not help, so the caller should surface it instead of backing off.
            // `ContainerError` has no case carrying that distinction and `.startFailed` reads as
            // transient; mapped here with the reason preserved in the message. A dedicated
            // `.unsupported` case is the right fix and needs the Darwin definition plus its
            // LinuxSupport mirror to move together, so it is not slipped in from the Windows side.
            case "unsupported": throw ContainerError.startFailed(message)
            case "already_exists", "start_failed": throw ContainerError.startFailed(message)
            default:
                // Older broker, or a failure it did not classify: fall back to the method.
                switch method {
                case "image.pull": throw ContainerError.imagePullFailed(message)
                case "proc.exec": throw ContainerError.notRunning(message)
                default: throw ContainerError.startFailed(message)
                }
            }
        }
    }
}
