import Foundation

/// One queued change, durable on disk until the server has it.
public struct OutboxEntry: Codable, Sendable, Identifiable, Equatable {
  /// Stable for the entry's life; apps may send it as an idempotency key.
  public let id: UUID
  /// Per account and monotonic, recovered as max + 1 at load. It orders a lane.
  public let seq: UInt64
  /// The FIFO scope, such as "doc:<key>". It never changes.
  public let lane: String
  /// App-defined.
  public let kind: String
  /// Opaque app bytes.
  public let payload: Data
  public let createdAt: Date
  /// Failed sends so far. Counted in memory; written with the next rewrite.
  public var attempts: Int
  /// nil means sendable.
  public var hold: Hold?

  public init(id: UUID, seq: UInt64, lane: String, kind: String, payload: Data, createdAt: Date, attempts: Int = 0, hold: Hold? = nil) {
    self.id = id; self.seq = seq; self.lane = lane; self.kind = kind; self.payload = payload
    self.createdAt = createdAt; self.attempts = attempts; self.hold = hold
  }
}

/// Why an entry stopped: the person has to decide what happens to it.
public struct Hold: Codable, Sendable, Equatable {
  public enum Reason: Codable, Sendable, Equatable {
    case conflict(Data)
    case rejected(String)
    case accessLost
    case tooLarge
  }
  public let reason: Reason
  /// Every v1 hold stops its lane. A hold that lets later entries pass
  /// arrives with Sheets, whose writes to different cells do not depend on
  /// each other; Docs' updates do, so it never needs one.
  public var blocksLane: Bool { true }
  public init(_ reason: Reason) { self.reason = reason }
}

/// Per-lane counts, for status and for the app's own caps.
public struct OutboxSummary: Sendable, Equatable {
  public struct Lane: Sendable, Equatable {
    /// Entries not held, including any in flight.
    public var pending = 0
    public var held = 0
    public var inFlight = 0
    /// Payload bytes of every entry in the lane, held or not.
    public var bytes = 0
    public var firstHold: Hold.Reason?
    public init() {}
  }
  public var lanes: [String: Lane] = [:]
  /// The store could not be read. Counts are the last ones known, which is
  /// never the same thing as "nothing pending".
  public var unavailable = false
  /// Unsent entries were lost to a lost key or a corrupt file. Report it
  /// once, then call `Outbox.acknowledgeLoss()`.
  public var lostUnsent = false

  public init() {}

  public var pending: Int { lanes.values.reduce(0) { $0 + $1.pending } }
  public var held: Int { lanes.values.reduce(0) { $0 + $1.held } }
  public var inFlight: Int { lanes.values.reduce(0) { $0 + $1.inFlight } }
  /// Lanes with anything in them: the default "N" of "Offline · N pending".
  public var pendingLanes: Int { lanes.values.filter { $0.pending + $0.held > 0 }.count }
  public var isEmpty: Bool { lanes.values.allSatisfy { $0.pending + $0.held == 0 } }

  /// The same summary, narrowed to one lane.
  public func only(_ lane: String) -> OutboxSummary {
    var copy = self
    copy.lanes = lanes[lane].map { [lane: $0] } ?? [:]
    return copy
  }
}

public enum OutboxError: Error, Sendable, Equatable {
  /// `replace` was asked to merge an entry that is being sent. The batch's
  /// `ack` would otherwise delete the merged file, and with it content whose
  /// own files the merge had already removed.
  case inFlight
  /// `replace` needs a run of entries that are next to each other in one lane.
  case notContiguous
  /// `replace` never merges a held entry: it is waiting for a person.
  case held
  case unknownEntry(UUID)
  case duplicateEntry(UUID)
}

/// Serializes disk mutations and send reservations for one account. Memory is
/// changed after each successful disk operation, including partial batch acks.
public actor Outbox {
  public nonisolated let account: String
  public static let directory = "outbox"
  private let store: BlobStore
  private var values: [OutboxEntry] = []
  private var flying: Set<UUID> = []
  private var loaded = false
  private var unavailable = false
  private var lost = false
  private var next: UInt64 = 1
  private var subscribers: [UUID: AsyncStream<OutboxSummary>.Continuation] = [:]
  public init(store: BlobStore, account: String) { self.store = store; self.account = account }
  private func path(_ seq: UInt64) -> String { "outbox/" + String(format: "%016llx", seq) + ".entry" }
  public func load() throws {
    guard !loaded || unavailable else { return }
    var recovered: [OutboxEntry] = []
    do {
      let names = try store.list(Self.directory)
      for name in names where name.hasSuffix(".entry") {
        let file = Self.directory + "/" + name
        do {
          guard let data = try store.read(file) else { continue }
          guard let entry = try? JSONDecoder().decode(OutboxEntry.self, from: data), path(entry.seq) == file,
                !recovered.contains(where: { $0.id == entry.id }) else {
            try store.remove(file); lost = true; continue
          }
          recovered.append(entry)
        } catch BlobStoreError.corrupt { lost = true }
      }
      values = recovered.sorted { $0.seq < $1.seq }
      next = max(next, (values.last?.seq ?? 0) + 1)
      loaded = true; unavailable = false; notify()
    } catch BlobStoreError.keyLost {
      values = []; flying = []; loaded = true; unavailable = false; lost = true; notify()
    } catch { unavailable = true; notify(); throw error }
  }
  private func mutate<T>(_ body: () throws -> T) throws -> T {
    do { let result = try body(); unavailable = false; notify(); return result }
    catch { unavailable = true; notify(); throw error }
  }
  private func persist(_ entry: OutboxEntry) throws { try store.write(JSONEncoder().encode(entry), to: path(entry.seq)) }
  @discardableResult
  public func enqueue(lane: String, kind: String, payload: Data, id: UUID = UUID()) throws -> OutboxEntry {
    try load()
    guard !values.contains(where: { $0.id == id }) else { throw OutboxError.duplicateEntry(id) }
    return try mutate {
      let entry = OutboxEntry(id: id, seq: next, lane: lane, kind: kind, payload: payload, createdAt: Date())
      try persist(entry); values.append(entry); next += 1; return entry
    }
  }
  /// A last-known snapshot, including held entries. On an unavailable store
  /// this may be stale, just like summary (whose unavailable flag is set).
  /// Use it for display/deduplication hints only; replay must first call load()
  /// successfully. Every mutation and beginSend independently require load().
  public func entries(lane: String, excludingInFlight: Bool = false) -> [OutboxEntry] {
    try? load()
    return values.filter { $0.lane == lane && (!excludingInFlight || !flying.contains($0.id)) }
  }
  /// Fails closed to an empty batch when loading is unavailable.
  public func sendable(lane: String, maxBytes: Int? = nil) -> [OutboxEntry] {
    try? load()
    guard !unavailable else { return [] }
    var result: [OutboxEntry] = []; var bytes = 0
    for entry in values where entry.lane == lane {
      if entry.hold != nil { break }
      // Compaction may use the suffix after a reserved batch, but beginSend
      // below never sends it before the earlier batch finishes.
      if flying.contains(entry.id) { continue }
      if let maxBytes, !result.isEmpty, bytes + entry.payload.count > maxBytes { break }
      result.append(entry); bytes += entry.payload.count
    }
    return result
  }
  public func beginSend(lane: String, maxBytes: Int? = nil) throws -> [OutboxEntry] {
    try load()
    guard flying.isEmpty else { return [] }
    let batch = sendable(lane: lane, maxBytes: maxBytes)
    if !batch.isEmpty { flying.formUnion(batch.map(\.id)); notify() }
    return batch
  }
  public func endSend(_ ids: [UUID], failed: Bool = true) {
    for id in ids {
      flying.remove(id)
      if failed, let index = values.firstIndex(where: { $0.id == id }) { values[index].attempts += 1 }
    }
    notify()
  }
  @discardableResult
  public func replace(_ ids: [UUID], with payload: Data) throws -> OutboxEntry {
    try load()
    guard !ids.isEmpty, Set(ids).count == ids.count else { throw OutboxError.notContiguous }
    if ids.contains(where: { flying.contains($0) }) { throw OutboxError.inFlight }
    let run = try ids.map { id in
      guard let entry = values.first(where: { $0.id == id }) else { throw OutboxError.unknownEntry(id) }
      return entry
    }.sorted { $0.seq < $1.seq }
    let first = run[0]
    let lane = values.filter { $0.lane == first.lane }
    guard run.allSatisfy({ $0.lane == first.lane && $0.kind == first.kind }),
          let start = lane.firstIndex(where: { $0.id == first.id }),
          Array(lane.dropFirst(start).prefix(run.count)).map(\.id) == run.map(\.id) else { throw OutboxError.notContiguous }
    let sendableIDs = Set(sendable(lane: first.lane).map(\.id))
    guard run.allSatisfy({ $0.hold == nil && sendableIDs.contains($0.id) }) else { throw OutboxError.held }
    return try mutate {
      let merged = OutboxEntry(id: UUID(), seq: first.seq, lane: first.lane, kind: first.kind, payload: payload, createdAt: first.createdAt)
      try persist(merged)
      values[values.firstIndex(where: { $0.id == first.id })!] = merged
      for entry in run.dropFirst() {
        try store.remove(path(entry.seq)); values.removeAll { $0.id == entry.id }
      }
      return merged
    }
  }
  public func ack(_ ids: [UUID]) throws {
    try load()
    try mutate {
      for id in ids {
        guard let entry = values.first(where: { $0.id == id }) else { continue }
        try store.remove(path(entry.seq)); values.removeAll { $0.id == id }; flying.remove(id)
      }
    }
  }
  public func hold(_ ids: [UUID], _ hold: Hold) throws {
    try load()
    try mutate {
      for id in ids {
        guard let index = values.firstIndex(where: { $0.id == id }) else { continue }
        var entry = values[index]; entry.hold = hold
        try persist(entry); values[index] = entry; flying.remove(id)
      }
    }
  }
  public func release(_ id: UUID) throws {
    try load()
    guard let index = values.firstIndex(where: { $0.id == id }) else { throw OutboxError.unknownEntry(id) }
    try mutate { var entry = values[index]; entry.hold = nil; try persist(entry); values[index] = entry }
  }
  public func discard(lane: String) throws { try load(); try ack(values.filter { $0.lane == lane }.map(\.id)) }
  public func purge() throws {
    try load()
    try mutate { try store.remove(Self.directory); values = []; flying = [] }
  }
  public func acknowledgeLoss() { lost = false; notify() }
  private var currentSummary: OutboxSummary {
    var result = OutboxSummary(); result.unavailable = unavailable; result.lostUnsent = lost
    for entry in values {
      var lane = result.lanes[entry.lane] ?? OutboxSummary.Lane()
      if let hold = entry.hold { lane.held += 1; if lane.firstHold == nil { lane.firstHold = hold.reason } }
      else { lane.pending += 1 }
      if flying.contains(entry.id) { lane.inFlight += 1 }
      lane.bytes += entry.payload.count; result.lanes[entry.lane] = lane
    }
    return result
  }
  /// Last-known counts; unavailable marks retained counts as non-authoritative.
  public var summary: OutboxSummary { try? load(); return currentSummary }
  private func notify() { for subscriber in subscribers.values { subscriber.yield(currentSummary) } }
  private func unsubscribe(_ id: UUID) { subscribers[id] = nil }
  public var changes: AsyncStream<OutboxSummary> {
    try? load()
    let (stream, continuation) = AsyncStream<OutboxSummary>.makeStream(bufferingPolicy: .bufferingNewest(1))
    let id = UUID(); subscribers[id] = continuation; continuation.yield(currentSummary)
    continuation.onTermination = { [weak self] _ in Task { await self?.unsubscribe(id) } }
    return stream
  }
}