import Foundation
import RastrilloNative
import RastrilloTestSupport
import XCTest

private func bytes(_ text: String) -> Data { Data(text.utf8) }

/// Answers `.corrupt` for chosen paths, removing them, as the sealed store does.
private final class CorruptingStore: BlobStore, @unchecked Sendable {
  let base: MemoryBlobStore
  let corrupt: Set<String>
  init(_ base: MemoryBlobStore, corrupt: Set<String>) { self.base = base; self.corrupt = corrupt }
  func write(_ data: Data, to path: String) throws { try base.write(data, to: path) }
  func read(_ path: String) throws -> Data? {
    if corrupt.contains(path) { try base.remove(path); throw BlobStoreError.corrupt(path) }
    return try base.read(path)
  }
  func list(_ directory: String) throws -> [String] { try base.list(directory) }
  func remove(_ path: String) throws { try base.remove(path) }
}

/// Answers `.keyLost` on the first read and removes everything, as the
/// sealed store does when the Keychain has lost the key.
private final class KeyLosingStore: BlobStore, @unchecked Sendable {
  let base: MemoryBlobStore
  init(_ base: MemoryBlobStore) { self.base = base }
  func write(_ data: Data, to path: String) throws { try base.write(data, to: path) }
  func read(_ path: String) throws -> Data? {
    guard try base.read(path) != nil else { return nil }
    try base.remove("")
    throw BlobStoreError.keyLost
  }
  func list(_ directory: String) throws -> [String] { try base.list(directory) }
  func remove(_ path: String) throws { try base.remove(path) }
}

final class OutboxTests: XCTestCase {
  func testEnqueueIsDurableOnReturnAndSeqResumesAfterTheHighest() async throws {
    let disk = MemoryBlobStore()
    let outbox = Outbox(store: disk, account: "a")
    let first = try await outbox.enqueue(lane: "doc:1", kind: "update", payload: bytes("one"))
    let second = try await outbox.enqueue(lane: "doc:1", kind: "update", payload: bytes("two"))
    XCTAssertEqual(disk.contents.keys.sorted(), ["outbox/0000000000000001.entry", "outbox/0000000000000002.entry"])

    let relaunched = Outbox(store: MemoryBlobStore(disk.contents), account: "a")
    let entries = await relaunched.entries(lane: "doc:1")
    XCTAssertEqual(entries.map(\.id), [first.id, second.id])
    XCTAssertEqual(entries.map(\.payload), [bytes("one"), bytes("two")])
    let third = try await relaunched.enqueue(lane: "doc:1", kind: "update", payload: bytes("three"))
    XCTAssertEqual(third.seq, 3, "seq is recovered as max + 1")
  }

  func testLanesAreFIFOAndIndependentAndBatchesAreCutByBytes() async throws {
    let outbox = Outbox(store: MemoryBlobStore(), account: "a")
    for index in 0..<4 { try await outbox.enqueue(lane: "doc:a", kind: "u", payload: Data(repeating: UInt8(index), count: 100)) }
    try await outbox.enqueue(lane: "doc:b", kind: "u", payload: bytes("b"))
    let batch = await outbox.sendable(lane: "doc:a", maxBytes: 250)
    XCTAssertEqual(batch.map(\.seq), [1, 2], "Cut by bytes, in seq order")
    let oversized = await outbox.sendable(lane: "doc:a", maxBytes: 10)
    XCTAssertEqual(oversized.map(\.seq), [1], "A batch always carries at least one entry")
    let other = await outbox.sendable(lane: "doc:b", maxBytes: nil)
    XCTAssertEqual(other.map(\.payload), [bytes("b")])
  }

  func testAHoldStopsItsOwnLaneOnly() async throws {
    let outbox = Outbox(store: MemoryBlobStore(), account: "a")
    let first = try await outbox.enqueue(lane: "doc:a", kind: "u", payload: bytes("1"))
    try await outbox.enqueue(lane: "doc:a", kind: "u", payload: bytes("2"))
    try await outbox.enqueue(lane: "doc:b", kind: "u", payload: bytes("3"))
    try await outbox.hold([first.id], Hold(.accessLost))
    let held = await outbox.sendable(lane: "doc:a", maxBytes: nil)
    XCTAssertEqual(held, [], "Nothing passes a hold: Docs updates depend on the ones before them")
    let other = await outbox.sendable(lane: "doc:b", maxBytes: nil)
    XCTAssertEqual(other.count, 1)
    let summary = await outbox.summary
    XCTAssertEqual(summary.lanes["doc:a"]?.held, 1)
    XCTAssertEqual(summary.lanes["doc:a"]?.pending, 1)
    XCTAssertEqual(summary.lanes["doc:a"]?.firstHold, .accessLost)
    try await outbox.release(first.id)
    let released = await outbox.sendable(lane: "doc:a", maxBytes: nil)
    XCTAssertEqual(released.count, 2)
  }

  func testReplaceKeepsTheOldestSeqAndTakesANewID() async throws {
    let disk = MemoryBlobStore()
    let outbox = Outbox(store: disk, account: "a")
    var ids: [UUID] = []
    for text in ["a", "b", "c"] { ids.append(try await outbox.enqueue(lane: "doc:1", kind: "update", payload: bytes(text)).id) }
    let after = try await outbox.enqueue(lane: "doc:1", kind: "update", payload: bytes("d"))
    let merged = try await outbox.replace(Array(ids[0...2]), with: bytes("abc"))
    XCTAssertEqual(merged.seq, 1)
    XCTAssertFalse(ids.contains(merged.id), "The merged entry gets a new id")
    XCTAssertEqual(merged.lane, "doc:1")
    XCTAssertEqual(merged.kind, "update")
    let lane = await outbox.entries(lane: "doc:1")
    XCTAssertEqual(lane.map(\.payload), [bytes("abc"), bytes("d")], "Lane order is unchanged")
    XCTAssertEqual(lane.last?.id, after.id)
    XCTAssertEqual(disk.contents.keys.sorted(), ["outbox/0000000000000001.entry", "outbox/0000000000000004.entry"])
  }

  func testReplaceRefusesInFlightHeldGappedAndCrossLaneRuns() async throws {
    let outbox = Outbox(store: MemoryBlobStore(), account: "a")
    var a: [UUID] = []
    for index in 0..<4 {
      a.append(try await outbox.enqueue(lane: "doc:a", kind: "u", payload: Data(repeating: 1, count: 10)).id)
      // Another lane interleaves in seq; that does not break a run of doc:a.
      try await outbox.enqueue(lane: "doc:b", kind: "u", payload: bytes("b\(index)"))
    }
    let b = await outbox.entries(lane: "doc:b").map(\.id)
    guard b.count == 4 else { return XCTFail("Expected four durable entries in lane b") }
    let inFlight = try await outbox.beginSend(lane: "doc:a", maxBytes: 15)
    XCTAssertEqual(inFlight.map(\.id), [a[0]])

    await assertThrows(OutboxError.inFlight) { try await outbox.replace([a[0], a[1]], with: Data()) }
    await assertThrows(OutboxError.notContiguous) { try await outbox.replace([a[1], a[3]], with: Data()) }
    await assertThrows(OutboxError.notContiguous) { try await outbox.replace([a[1], b[1]], with: Data()) }
    try await outbox.hold([a[3]], Hold(.tooLarge))
    await assertThrows(OutboxError.held) { try await outbox.replace([a[2], a[3]], with: Data()) }

    let merged = try await outbox.replace([a[1], a[2]], with: bytes("merged"))
    XCTAssertEqual(merged.seq, 3, "A run of doc:a skips over doc:b's seqs")
    try await outbox.ack([a[0]])
    let lane = await outbox.entries(lane: "doc:a")
    XCTAssertEqual(lane.map(\.id), [merged.id, a[3]], "Acking the in-flight head never touches the merged entry")
  }

  func testACrashBetweenReplaceStepsLeavesDuplicatesNeverAGap() async throws {
    let base = MemoryBlobStore()
    let setup = Outbox(store: base, account: "a")
    var ids: [UUID] = []
    for text in ["a", "b", "c", "d"] { ids.append(try await setup.enqueue(lane: "doc:1", kind: "u", payload: bytes(text)).id) }

    let faulting = FaultingBlobStore(MemoryBlobStore(base.contents))
    let outbox = Outbox(store: faulting, account: "a")
    try await outbox.load()
    faulting.crash(after: 1)   // step 1 (the write) happens; the first remove does not
    do {
      try await outbox.replace(Array(ids[0...2]), with: bytes("abc"))
      XCTFail("Expected the simulated crash")
    } catch is SimulatedCrash {}

    let relaunched = Outbox(store: faulting.committed, account: "a")
    let lane = await relaunched.entries(lane: "doc:1")
    XCTAssertEqual(lane.map(\.payload), [bytes("abc"), bytes("b"), bytes("c"), bytes("d")],
                   "Duplicates only: the merged head, then what it merged, then the rest in order")
  }

  func testAckDiscardAndPurgeRemoveFiles() async throws {
    let disk = MemoryBlobStore()
    let outbox = Outbox(store: disk, account: "a")
    let one = try await outbox.enqueue(lane: "doc:a", kind: "u", payload: bytes("1"))
    try await outbox.enqueue(lane: "doc:b", kind: "u", payload: bytes("2"))
    try await outbox.enqueue(lane: "doc:c", kind: "u", payload: bytes("3"))
    try await outbox.ack([one.id, UUID()])
    try await outbox.discard(lane: "doc:b")
    XCTAssertEqual(disk.contents.count, 1)
    try await outbox.purge()
    XCTAssertEqual(disk.contents.count, 0)
    let summary = await outbox.summary
    XCTAssertTrue(summary.isEmpty)
  }

  func testAnUnavailableStoreIsNeverAnEmptyStore() async throws {
    let base = MemoryBlobStore()
    try await Outbox(store: base, account: "a").enqueue(lane: "doc:1", kind: "u", payload: bytes("kept"))
    let faulting = FaultingBlobStore(base)
    faulting.unavailable = true
    let outbox = Outbox(store: faulting, account: "a")
    await assertThrows(BlobStoreError.unavailable) { try await outbox.load() }
    await assertThrows(BlobStoreError.unavailable) { try await outbox.enqueue(lane: "doc:1", kind: "u", payload: Data()) }
    let summary = await outbox.summary
    XCTAssertTrue(summary.unavailable)
    XCTAssertEqual(base.contents.count, 1, "Nothing was deleted while locked")
    XCTAssertFalse(faulting.operations.contains { if case .remove = $0 { return true } else { return false } })

    faulting.unavailable = false
    try await outbox.load()
    let lane = await outbox.entries(lane: "doc:1")
    XCTAssertEqual(lane.map(\.payload), [bytes("kept")])
    let after = await outbox.summary
    XCTAssertFalse(after.unavailable)
  }

  func testOneCorruptEntryIsRemovedAloneAndReported() async throws {
    let base = MemoryBlobStore()
    let setup = Outbox(store: base, account: "a")
    for text in ["1", "2", "3"] { try await setup.enqueue(lane: "doc:1", kind: "u", payload: bytes(text)) }
    let outbox = Outbox(store: CorruptingStore(base, corrupt: ["outbox/0000000000000002.entry"]), account: "a")
    let lane = await outbox.entries(lane: "doc:1")
    XCTAssertEqual(lane.map(\.payload), [bytes("1"), bytes("3")])
    XCTAssertEqual(base.contents.count, 2, "Only the corrupt file went")
    var summary = await outbox.summary
    XCTAssertTrue(summary.lostUnsent)
    await outbox.acknowledgeLoss()
    summary = await outbox.summary
    XCTAssertFalse(summary.lostUnsent, "Reported once")
  }

  func testALostKeyEmptiesTheOutboxAndSaysSoOnce() async throws {
    let base = MemoryBlobStore()
    try await Outbox(store: base, account: "a").enqueue(lane: "doc:1", kind: "u", payload: bytes("gone"))
    let outbox = Outbox(store: KeyLosingStore(base), account: "a")
    try await outbox.load()
    let summary = await outbox.summary
    XCTAssertTrue(summary.isEmpty)
    XCTAssertTrue(summary.lostUnsent)
    try await outbox.enqueue(lane: "doc:1", kind: "u", payload: bytes("new"))
    let lane = await outbox.entries(lane: "doc:1")
    XCTAssertEqual(lane.map(\.payload), [bytes("new")])
  }

  func testChangesDeliverTheCurrentSummaryThenEachChange() async throws {
    let outbox = Outbox(store: MemoryBlobStore(), account: "a")
    let changes = await outbox.changes
    var iterator = changes.makeAsyncIterator()
    let first = await iterator.next()
    XCTAssertEqual(first?.pending, 0)
    try await outbox.enqueue(lane: "doc:1", kind: "u", payload: bytes("x"))
    let second = await iterator.next()
    XCTAssertEqual(second?.pending, 1)
  }

  /// Random enqueues, sends, acks, compactions and crashes. After every
  /// relaunch: every entry whose enqueue returned and was never acked is
  /// still there, in order, and each lane's content is a
  /// duplicates-allowed replay of what was enqueued.
  func testRandomCrashesNeverLoseOrReorderAnEntry() async throws {
    var generator = SeededGenerator(seed: 0x6b33)
    for _ in 0..<150 {
      var disk = MemoryBlobStore()
      var faulting = FaultingBlobStore(disk)
      var outbox = Outbox(store: faulting, account: "a")
      var expected: [String: [String]] = [:]   // lane -> symbols not yet acked, in order
      var acked: Set<String> = []
      var symbol = 0
      for _ in 0..<40 {
        let lane = ["doc:a", "doc:b"][Int.random(in: 0...1, using: &generator)]
        if Int.random(in: 0..<12, using: &generator) == 0 { faulting.crash(after: Int.random(in: 0...2, using: &generator)) }
        do {
          switch Int.random(in: 0..<10, using: &generator) {
          case 0..<5:
            symbol += 1
            let text = "s\(symbol)"
            try await outbox.enqueue(lane: lane, kind: "u", payload: bytes(text))
            expected[lane, default: []].append(text)
          case 5..<7:
            let batch = try await outbox.beginSend(lane: lane, maxBytes: 8)
            guard !batch.isEmpty else { continue }
            // The server has the batch before the ack starts: a crash part
            // way through the ack loses nothing the server lacks.
            for entry in batch {
              let text = String(decoding: entry.payload, as: UTF8.self)
              for part in text.split(separator: "+") { acked.insert(String(part)) }
            }
            expected[lane] = expected[lane, default: []].filter { !acked.contains($0) }
            try await outbox.ack(batch.map(\.id))
          case 7..<9:
            let run = await outbox.sendable(lane: lane, maxBytes: nil)
            guard run.count >= 2 else { continue }
            let take = Array(run.prefix(Int.random(in: 2...run.count, using: &generator)))
            let joined = take.map { String(decoding: $0.payload, as: UTF8.self) }.joined(separator: "+")
            try await outbox.replace(take.map(\.id), with: bytes(joined))
          default:
            let batch = try await outbox.beginSend(lane: lane, maxBytes: nil)
            await outbox.endSend(batch.map(\.id))
          }
        } catch is SimulatedCrash {
          disk = faulting.committed
          faulting = FaultingBlobStore(disk)
          outbox = Outbox(store: faulting, account: "a")
        }
        // The invariant, checked against a fresh reading of the disk.
        let reread = Outbox(store: MemoryBlobStore(faulting.committed.contents), account: "a")
        for (lane, symbols) in expected {
          let present = await reread.entries(lane: lane).flatMap {
            String(decoding: $0.payload, as: UTF8.self).split(separator: "+").map(String.init)
          }
          let unacked = present.filter { !acked.contains($0) }
          var seen: [String] = []
          for item in unacked where seen.last != item && !seen.contains(item) { seen.append(item) }
          XCTAssertEqual(seen, symbols, "lane \(lane): lost or reordered an entry")
        }
      }
    }
  }
}

struct SeededGenerator: RandomNumberGenerator {
  private var state: UInt64
  init(seed: UInt64) { state = seed }
  mutating func next() -> UInt64 {
    state &+= 0x9E3779B97F4A7C15
    var z = state
    z = (z ^ (z >> 30)) &* 0xBF58476D1CE4E5B9
    z = (z ^ (z >> 27)) &* 0x94D049BB133111EB
    return z ^ (z >> 31)
  }
}

func assertThrows<E: Error & Equatable>(_ expected: E, file: StaticString = #filePath, line: UInt = #line,
                                        _ body: () async throws -> some Any) async {
  do {
    _ = try await body()
    XCTFail("Expected \(expected)", file: file, line: line)
  } catch let error as E {
    XCTAssertEqual(error, expected, file: file, line: line)
  } catch {
    XCTFail("Expected \(expected), got \(error)", file: file, line: line)
  }
}