import Foundation
import Testing

@testable import NucleicCore

@Suite("MergeQueue — autoship ordering & serialization")
struct MergeQueueTests {
    private func request(_ session: String, project: String = "p", target: String = "main") -> ShipRequest {
        ShipRequest(
            sessionID: SessionID(rawValue: session), projectID: ProjectID(rawValue: project),
            target: GitRef(target), strategy: .squash)
    }

    /// Drains a (pre-subscribed) update stream, returning the terminal updates in arrival
    /// order once `count` of them have been seen. The stream must be obtained *before*
    /// enqueuing; its unbounded buffer holds any updates emitted before draining starts.
    private func drainTerminals(_ stream: AsyncStream<ShipUpdate>, count: Int) async -> [ShipUpdate] {
        var out: [ShipUpdate] = []
        for await update in stream where update.status.isTerminal {
            out.append(update)
            if out.count == count { break }
        }
        return out
    }

    // Records max observed concurrency and the order integrators started.
    private actor ConcurrencyProbe {
        private(set) var current = 0
        private(set) var peak = 0
        private(set) var startOrder: [String] = []
        func enter(_ id: String) { current += 1; peak = max(peak, current); startOrder.append(id) }
        func leave() { current -= 1 }
    }

    @Test func sameTargetRunsStrictlySerialInFifoOrder() async throws {
        let queue = MergeQueue()
        let probe = ConcurrencyProbe()
        await queue.setIntegrator { req in
            await probe.enter(req.sessionID.rawValue)
            try? await Task.sleep(for: .milliseconds(40))
            await probe.leave()
            return .merged(commit: req.sessionID.rawValue)
        }
        let stream = await queue.updates()

        #expect(await queue.enqueue(request("s1")))
        #expect(await queue.enqueue(request("s2")))
        #expect(await queue.enqueue(request("s3")))

        let terminals = await drainTerminals(stream, count: 3)
        #expect(await probe.peak == 1)  // never two merges into the same target at once
        #expect(await probe.startOrder == ["s1", "s2", "s3"])  // FIFO
        #expect(terminals.map(\.sessionID.rawValue) == ["s1", "s2", "s3"])
    }

    @Test func differentTargetsRunConcurrently() async throws {
        let queue = MergeQueue()
        let probe = ConcurrencyProbe()
        await queue.setIntegrator { req in
            await probe.enter(req.sessionID.rawValue)
            try? await Task.sleep(for: .milliseconds(40))
            await probe.leave()
            return .merged(commit: req.sessionID.rawValue)
        }
        let stream = await queue.updates()

        #expect(await queue.enqueue(request("s1", target: "main")))
        #expect(await queue.enqueue(request("s2", target: "dev")))
        #expect(await queue.enqueue(request("s3", target: "release")))

        _ = await drainTerminals(stream, count: 3)
        #expect(await probe.peak >= 2)  // distinct targets overlap
    }

    // Lets a test hold an integrator in-flight until released.
    private actor Gate {
        private var waiters: [CheckedContinuation<Void, Never>] = []
        private var opened = false
        func wait() async {
            if opened { return }
            await withCheckedContinuation { waiters.append($0) }
        }
        func open() {
            opened = true
            for w in waiters { w.resume() }
            waiters.removeAll()
        }
    }

    @Test func dedupesAnInFlightSession() async throws {
        let queue = MergeQueue()
        let gate = Gate()
        await queue.setIntegrator { req in
            await gate.wait()
            return .merged(commit: req.sessionID.rawValue)
        }
        let stream = await queue.updates()

        let first = await queue.enqueue(request("s1"))
        let dup = await queue.enqueue(request("s1"))  // same session, still live → ignored
        #expect(first)
        #expect(!dup)
        #expect(await queue.isLive(SessionID(rawValue: "s1")))

        await gate.open()
        let terminals = await drainTerminals(stream, count: 1)
        #expect(terminals.count == 1)
        #expect(terminals[0].sessionID.rawValue == "s1")
        #expect(!(await queue.isLive(SessionID(rawValue: "s1"))))  // slot freed after merge
    }

    @Test func reportsFullLifecycleStatusesAndConflict() async throws {
        let queue = MergeQueue()
        await queue.setIntegrator { _ in .conflicted(["a.txt"]) }

        // Subscribe before enqueuing; the stream buffers so no early update is lost.
        let stream = await queue.updates()
        #expect(await queue.enqueue(request("s1")))

        var seen: [ShipStatus] = []
        for await update in stream {
            seen.append(update.status)
            if update.status.isTerminal { break }
        }
        #expect(seen.first == .queued(position: 0))
        #expect(seen.contains(.merging))
        #expect(seen.last == .conflicted(["a.txt"]))
    }

    @Test func missingIntegratorReportsFailure() async throws {
        let queue = MergeQueue()  // no integrator attached
        let stream = await queue.updates()
        #expect(await queue.enqueue(request("s1")))

        var terminal: ShipStatus?
        for await update in stream where update.status.isTerminal {
            terminal = update.status
            break
        }
        guard case .failed = terminal else {
            Issue.record("expected .failed with no integrator, got \(String(describing: terminal))")
            return
        }
    }
}
