import Foundation

// MARK: - Autoship merge queue
//
// Autoship lets a session merge its own branch into the project's default branch the
// moment the agent finishes its work (see AppStore.shipIfCompleted). Many sessions run
// in parallel, so their ship requests must land one-at-a-time per target branch — never
// concurrently — or two `git checkout target` + merge sequences would race on the same
// main working copy.
//
// The git-level race is *already* impossible: GitWorktreeManager is an actor (structural
// ops serialized) and `integrate` merges against the live target tip and never
// force-updates, so a stale branch can't overwrite newer work and a conflict is aborted
// with the branch left intact. This queue layers the *observable* guarantees on top:
//   • FIFO ordering and fairness across sessions targeting the same branch,
//   • strict per-target serialization (different targets/repos still run in parallel),
//   • per-session dedupe so a re-fired `.completed` turn can't double-enqueue,
//   • a single place that reports queued / merging / merged / conflicted / failed.
//
// The queue owns *ordering*; the injected integrator closure owns the *git work*. That
// split keeps the queue pure and unit-testable with no git dependency.

/// A request to merge one session's branch into a target branch.
public struct ShipRequest: Sendable, Equatable {
    public let sessionID: SessionID
    public let projectID: ProjectID
    public let target: GitRef
    public let strategy: IntegrationStrategy

    public init(sessionID: SessionID, projectID: ProjectID, target: GitRef, strategy: IntegrationStrategy) {
        self.sessionID = sessionID
        self.projectID = projectID
        self.target = target
        self.strategy = strategy
    }

    /// Serialization key: requests sharing a key are processed strictly one-at-a-time.
    var targetKey: TargetKey { TargetKey(projectID: projectID, target: target) }
}

struct TargetKey: Hashable, Sendable {
    let projectID: ProjectID
    let target: GitRef
}

/// The lifecycle of a ship request, fanned out on `MergeQueue.updates`.
public enum ShipStatus: Sendable, Equatable {
    /// Waiting behind `position` other requests for the same target (0 = next up).
    case queued(position: Int)
    /// The merge is running now.
    case merging
    /// Landed cleanly; `commit` is the new target tip.
    case merged(commit: String)
    /// Aborted on conflict — branch intact, nothing overwritten. Paths are the clashes.
    case conflicted([String])
    /// The integrator threw (e.g. the main tree was dirty). Branch intact.
    case failed(String)
    /// Nothing to do (session no longer eligible, or empty diff).
    case skipped(String)

    /// A status that ends a request's life in the queue (frees the per-session slot).
    var isTerminal: Bool {
        switch self {
        case .queued, .merging: return false
        case .merged, .conflicted, .failed, .skipped: return true
        }
    }
}

/// One status change for a session, delivered on the `updates` stream.
public struct ShipUpdate: Sendable, Equatable {
    public let sessionID: SessionID
    public let status: ShipStatus
    public init(sessionID: SessionID, status: ShipStatus) {
        self.sessionID = sessionID
        self.status = status
    }
}

/// Serializes autoship merges per target branch and reports their progress.
///
/// Fail-safe: with no integrator attached (tests, early startup) an enqueue is reported
/// `.failed` rather than silently dropped, so a wiring gap is visible, never a lost merge.
public actor MergeQueue {
    /// Does the actual merge for one request. Injected by `AppStore` after construction
    /// so the queue carries no git/SessionController dependency.
    public typealias Integrator = @Sendable (ShipRequest) async -> ShipStatus

    private var integrator: Integrator?

    /// Pending requests per target, in FIFO order. The head of each array is in flight
    /// (or about to be) once that target is draining.
    private var pending: [TargetKey: [ShipRequest]] = [:]
    /// Targets with a live drain task, so we never start a second worker for one target.
    private var draining: Set<TargetKey> = []
    /// Sessions currently queued or merging — the dedupe set.
    private var liveSessions: Set<SessionID> = []

    private var subscribers: [UUID: AsyncStream<ShipUpdate>.Continuation] = [:]

    public init(integrator: Integrator? = nil) {
        self.integrator = integrator
    }

    public func setIntegrator(_ integrator: @escaping Integrator) {
        self.integrator = integrator
    }

    /// Live fan-out of status changes. Each subscriber gets every update appended after
    /// it subscribes (mirrors `SessionController.subscribe()`).
    public func updates() -> AsyncStream<ShipUpdate> {
        let id = UUID()
        return AsyncStream { continuation in
            subscribers[id] = continuation
            continuation.onTermination = { [weak self] _ in
                Task { await self?.removeSubscriber(id) }
            }
        }
    }

    private func removeSubscriber(_ id: UUID) {
        subscribers[id] = nil
    }

    /// Enqueue a request. Idempotent per session: a session already queued or merging is
    /// ignored (returns `false`), so a repeated `.completed` turn can't double-ship.
    /// Returns `true` if the request was accepted.
    @discardableResult
    public func enqueue(_ request: ShipRequest) -> Bool {
        guard !liveSessions.contains(request.sessionID) else { return false }
        liveSessions.insert(request.sessionID)
        let key = request.targetKey
        pending[key, default: []].append(request)
        emit(request.sessionID, .queued(position: pending[key]!.count - 1))
        if !draining.contains(key) {
            draining.insert(key)
            Task { await self.drain(key) }
        }
        return true
    }

    /// True if the session has a request queued or in flight (used by callers to avoid
    /// re-enqueuing). Cheap, actor-isolated read.
    public func isLive(_ sessionID: SessionID) -> Bool { liveSessions.contains(sessionID) }

    /// Drain one target's FIFO, one request at a time, until empty.
    private func drain(_ key: TargetKey) async {
        defer { draining.remove(key) }
        while let request = pending[key]?.first {
            emit(request.sessionID, .merging)
            let status: ShipStatus
            if let integrator {
                status = await integrator(request)
            } else {
                status = .failed("merge queue has no integrator attached")
            }
            // Pop the head only now — after the integrator returns — so the live target
            // tip the *next* request merges against already reflects this merge.
            pending[key]?.removeFirst()
            if pending[key]?.isEmpty == true { pending[key] = nil }
            liveSessions.remove(request.sessionID)
            emit(request.sessionID, status)
        }
    }

    private func emit(_ sessionID: SessionID, _ status: ShipStatus) {
        let update = ShipUpdate(sessionID: sessionID, status: status)
        for continuation in subscribers.values { continuation.yield(update) }
    }
}