import Foundation
import RastrilloNative
import RastrilloTestSupport
import XCTest
@testable import RastrilloApple

@MainActor private func settle(until condition: () async -> Bool, seconds: Double = 3) async {
  let deadline = Date().addingTimeInterval(seconds)
  while !(await condition()) && Date() < deadline { try? await Task.sleep(nanoseconds: 5_000_000) }
}

/// A send that POSTs each entry's payload, in order, as one request per batch.
@MainActor private func poster(_ calls: @escaping @MainActor (String, [OutboxEntry]) -> Void = { _, _ in }) -> OutboxDrainer.Send {
  { lane, entries, transport in
    calls(lane, entries)
    let body = entries.map { String(decoding: $0.payload, as: UTF8.self) }.joined(separator: ",")
    let reply = try await transport.responseDetails("/\(lane)", method: "POST", body: Data(body.utf8))
    try reply.requireSuccess()
    return Dictionary(uniqueKeysWithValues: entries.map { ($0.id, .sent) })
  }
}

@MainActor final class OutboxDrainerTests: XCTestCase {
  private func connectivity(_ status: NativePath.Status = .satisfied) async -> (NativeConnectivity, ScriptedPathSource) {
    let source = ScriptedPathSource(status)
    let connectivity = NativeConnectivity(source: source)
    await settle { connectivity.path?.status == status }
    return (connectivity, source)
  }

  private func outbox(_ lanes: [(String, String)]) async throws -> Outbox {
    let outbox = Outbox(store: MemoryBlobStore(), account: "a")
    for (lane, text) in lanes { try await outbox.enqueue(lane: lane, kind: "u", payload: Data(text.utf8)) }
    return outbox
  }

  private func bodies(_ transport: ScriptedTransport) -> [String] {
    transport.committed.map { String(decoding: $0.body ?? Data(), as: UTF8.self) }
  }

  func testThrowingLaneDoesNotStarveOtherLanesAndEventuallyHolds() async throws {
    struct Broken: Error {}
    let (connectivity, _) = await connectivity()
    let outbox = try await outbox([("a", "broken"), ("b", "good"), ("c", "good")])
    let transport = ScriptedTransport([])
    var calls: [String] = []
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity) { lane, entries, _ in
      calls.append(lane)
      if lane == "a" { throw Broken() }
      return Dictionary(uniqueKeysWithValues: entries.map { ($0.id, .sent) })
    }
    await drainer.drain()
    XCTAssertEqual(calls, ["a", "b", "c"])
    let remaining = await outbox.summary
    XCTAssertEqual(remaining.lanes.count, 1)
    await drainer.drain(); await drainer.drain()
    let held = await outbox.entries(lane: "a")
    XCTAssertEqual(held.first?.hold, Hold(.rejected("send")))
    await drainer.drain()
    XCTAssertEqual(calls.filter { $0 == "a" }.count, 3, "Three consecutive throws hold the lane instead of spinning")
  }

  func testConcurrentDrainWaitsForTheReservedBatch() async throws {
    let (connectivity, _) = await connectivity()
    let outbox = try await outbox([("doc:a", "1")])
    let transport = ScriptedTransport([])
    var release: CheckedContinuation<Void, Never>?
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity) { _, entries, _ in
      await withCheckedContinuation { release = $0 }
      return Dictionary(uniqueKeysWithValues: entries.map { ($0.id, .sent) })
    }
    let first = Task { await drainer.drain() }
    await settle { release != nil }
    var finished = false
    let second = Task { let outcome = await drainer.drain(); finished = true; return outcome }
    try await Task.sleep(for: .milliseconds(30))
    XCTAssertFalse(finished, "A concurrent close-save cannot report idle before acknowledgement")
    release?.resume()
    let outcomes = await (first.value, second.value)
    XCTAssertEqual(outcomes.0, .idle); XCTAssertEqual(outcomes.1, .idle)
    let summary = await outbox.summary
    XCTAssertTrue(summary.isEmpty)
  }

  func testUnavailableStoreKeepsTheLastKnownStatusUntilUnlock() async throws {
    let (connectivity, _) = await connectivity(.unsatisfied)
    let disk = FaultingBlobStore()
    let outbox = Outbox(store: disk, account: "a")
    try await outbox.enqueue(lane: "doc:a", kind: "update", payload: Data([1]))
    let drainer = OutboxDrainer(outbox: outbox, transport: { ScriptedTransport([]) }, connectivity: connectivity, send: poster())
    await drainer.drain()
    XCTAssertEqual(drainer.status(), .offline(pending: 1))
    disk.unavailable = true
    // Force an I/O operation even after a successful initial load.
    do { try await outbox.enqueue(lane: "doc:a", kind: "update", payload: Data([2])); XCTFail("Locked store must refuse enqueue") }
    catch BlobStoreError.unavailable { }
    await drainer.drain()
    XCTAssertEqual(drainer.status(), .offline(pending: 1), "Unavailable keeps the known status; it is not lost")
    XCTAssertEqual(drainer.status(localFailure: .cannotSaveOnDevice), .failed(pending: 1, reason: .cannotSaveOnDevice), "The app still reports undurable input explicitly")
    disk.unavailable = false
    await drainer.drain()
    XCTAssertEqual(drainer.status(), .offline(pending: 1))
    XCTAssertEqual(disk.committed.contents.count, 1)
  }

  func testDrainsEachLaneInOrderAndAcknowledges() async throws {
    let (connectivity, _) = await connectivity()
    let transport = ScriptedTransport([.respond(status: 204), .respond(status: 204)])
    let outbox = try await outbox([("doc:a", "1"), ("doc:b", "x"), ("doc:a", "2")])
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    let outcome = await drainer.drain()
    XCTAssertEqual(outcome, .idle)
    XCTAssertEqual(transport.committed.map(\.path), ["/doc:a", "/doc:b"])
    XCTAssertEqual(bodies(transport), ["1,2", "x"], "One batch per lane, FIFO")
    let summary = await outbox.summary
    XCTAssertTrue(summary.isEmpty)
    XCTAssertEqual(drainer.status(), .saved)
  }

  func testBatchesAreCutByBytes() async throws {
    let (connectivity, _) = await connectivity()
    let transport = ScriptedTransport([.respond(status: 204), .respond(status: 204)])
    let outbox = try await outbox([("doc:a", "aaaa"), ("doc:a", "bbbb"), ("doc:a", "cccc")])
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, maxBatchBytes: 8, send: poster())
    await drainer.drain()
    XCTAssertEqual(bodies(transport), ["aaaa,bbbb", "cccc"])
  }

  func testAPathDownMakesNoAttemptAndWaits() async throws {
    let (connectivity, _) = await connectivity(.unsatisfied)
    let transport = ScriptedTransport([])
    let outbox = try await outbox([("doc:a", "1")])
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    let outcome = await drainer.drain()
    XCTAssertEqual(outcome, .waitingForNetwork)
    XCTAssertEqual(transport.calls.count, 0, "An offline phone makes no attempts")
    XCTAssertEqual(drainer.status(), .offline(pending: 1))
  }

  func testOfflineMidSendWithThePathDownWaitsForTheNetwork() async throws {
    let (connectivity, source) = await connectivity()
    let outbox = try await outbox([("doc:a", "1")])
    let transport = ScriptedTransport(responder: { _ in
      source.send(.unsatisfied)
      return .offline(.notConnectedToInternet)
    })
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    await settle { connectivity.isPathSatisfied }
    let first = await drainer.drain()
    await settle { !connectivity.isPathSatisfied }
    // Whether the path update landed before or after the classification,
    // the next drain must not try again while the path is down.
    let second = await drainer.drain()
    XCTAssertEqual(second, .waitingForNetwork, "first was \(first)")
    XCTAssertEqual(transport.calls.count, 1)
  }

  /// M2: a lost response or timeout while the path stays up backs off and
  /// retries the same batch; it does not wait for a path change.
  func testALostResponseOnAWorkingPathBacksOffAndResendsTheSameBatch() async throws {
    let (connectivity, _) = await connectivity()
    let transport = ScriptedTransport([.commitThenLoseResponse(), .respond(status: 204)])
    let outbox = try await outbox([("doc:a", "1"), ("doc:a", "2")])
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    let first = await drainer.drain()
    guard case .backingOff(let delay) = first else { return XCTFail("Expected a backoff, got \(first)") }
    XCTAssertGreaterThan(delay, 0)
    XCTAssertEqual(drainer.phase, .backingOff)
    XCTAssertEqual(drainer.status(), .retrying(pending: 1))
    let kept = await outbox.entries(lane: "doc:a")
    XCTAssertEqual(kept.count, 2, "Nothing is acknowledged that the drainer did not see acknowledged")
    let outcome1 = await drainer.drain()
    XCTAssertEqual(outcome1, .idle)
    XCTAssertEqual(bodies(transport), ["1,2", "1,2"], "The same batch, sent again")
  }

  func testAnEndedSessionStopsEveryLaneAndHoldsNothing() async throws {
    let (connectivity, _) = await connectivity()
    let transport = ScriptedTransport([.respond(status: 401), .respond(status: 204), .respond(status: 204)])
    let outbox = try await outbox([("doc:a", "1"), ("doc:b", "2")])
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    let outcome2 = await drainer.drain()
    XCTAssertEqual(outcome2, .sessionEnded)
    XCTAssertEqual(transport.calls.count, 1, "Every lane stops")
    let summary = await outbox.summary
    XCTAssertEqual(summary.held, 0, "Never hold or drop on 401")
    XCTAssertEqual(summary.pending, 2)
    XCTAssertEqual(drainer.status(), .needsSignIn(pending: 2))
    let outcome3 = await drainer.drain()
    XCTAssertEqual(outcome3, .sessionEnded, "It stays stopped until signed in again")
    drainer.sessionRestored()
    let outcome4 = await drainer.drain()
    XCTAssertEqual(outcome4, .idle)
    XCTAssertEqual(transport.calls.count, 3)
  }

  func testNoSessionIsAnEndedSession() async throws {
    let (connectivity, _) = await connectivity()
    let outbox = try await outbox([("doc:a", "1")])
    let drainer = OutboxDrainer(outbox: outbox, transport: { nil }, connectivity: connectivity, send: poster())
    let outcome5 = await drainer.drain()
    XCTAssertEqual(outcome5, .sessionEnded)
  }

  func testRefusalsHoldTheirLaneAndOtherLanesKeepDraining() async throws {
    let (connectivity, _) = await connectivity()
    let transport = ScriptedTransport { call in
      switch call.path {
      case "/doc:gone": return .respond(status: 403)
      case "/doc:big": return .respond(status: 413)
      case "/doc:bad": return .respond(status: 400)
      default: return .respond(status: 204)
      }
    }
    let outbox = try await outbox([("doc:gone", "1"), ("doc:big", "2"), ("doc:bad", "3"), ("doc:ok", "4")])
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    let outcome6 = await drainer.drain()
    XCTAssertEqual(outcome6, .idle)
    let summary = await outbox.summary
    XCTAssertEqual(summary.lanes["doc:gone"]?.firstHold, .accessLost)
    XCTAssertEqual(summary.lanes["doc:big"]?.firstHold, .tooLarge)
    XCTAssertEqual(summary.lanes["doc:bad"]?.firstHold, .rejected("400"))
    XCTAssertNil(summary.lanes["doc:ok"], "Other lanes are not stopped by a hold")
    XCTAssertEqual(transport.calls.count, 4, "A held lane is not retried")
    XCTAssertEqual(drainer.status(lane: "doc:gone"), .failed(pending: 1, reason: .held(.accessLost)))
    XCTAssertEqual(drainer.status(lane: "doc:ok"), .saved)
  }

  func testRateLimitingIsAFloorOnTheDelay() async throws {
    let (connectivity, _) = await connectivity()
    let transport = ScriptedTransport([.respond(status: 429, headers: ["retry-after": "30"])])
    let outbox = try await outbox([("doc:a", "1")])
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    let outcome = await drainer.drain()
    guard case .rateLimited(let delay) = outcome else { return XCTFail("Expected rate limiting, got \(outcome)") }
    XCTAssertGreaterThanOrEqual(delay, 30)
  }

  func testServerErrorsBackOff() async throws {
    let (connectivity, _) = await connectivity()
    let transport = ScriptedTransport([.respond(status: 502)])
    let outbox = try await outbox([("doc:a", "1")])
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    let backedOff = await drainer.drain()
    guard case .backingOff = backedOff else { return XCTFail("Expected a backoff") }
    let summary = await outbox.summary
    XCTAssertEqual(summary.pending, 1)
    XCTAssertEqual(summary.held, 0)
  }

  /// H4: the send went through but the store locked before the ack. The
  /// batch stays queued and goes again; it is never lost.
  func testALockedStoreDuringTheAckIsARetry() async throws {
    let (connectivity, _) = await connectivity()
    let disk = FaultingBlobStore()
    let outbox = Outbox(store: disk, account: "a")
    try await outbox.enqueue(lane: "doc:a", kind: "u", payload: Data("1".utf8))
    let transport = ScriptedTransport([.respond(status: 204), .respond(status: 204)])
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity,
                                send: poster { _, _ in disk.unavailable = true })
    let backedOff = await drainer.drain()
    guard case .backingOff = backedOff else { return XCTFail("Expected a retry") }
    disk.unavailable = false
    let kept = await outbox.entries(lane: "doc:a")
    XCTAssertEqual(kept.count, 1)
    let again = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    let outcome7 = await again.drain()
    XCTAssertEqual(outcome7, .idle)
    XCTAssertEqual(transport.committed.count, 2)
  }

  func testAnUnavailableStoreWaitsAndDeletesNothing() async throws {
    let (connectivity, _) = await connectivity()
    let disk = FaultingBlobStore()
    let setup = Outbox(store: disk, account: "a")
    try await setup.enqueue(lane: "doc:a", kind: "u", payload: Data("1".utf8))
    disk.unavailable = true
    let outbox = Outbox(store: disk, account: "a")
    let transport = ScriptedTransport([])
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    let outcome = await drainer.drain()
    XCTAssertTrue(outcome == .storeUnavailable || outcome == .idle, "\(outcome)")
    XCTAssertEqual(transport.calls.count, 0)
    disk.unavailable = false
    XCTAssertEqual(disk.committed.contents.count, 1)
  }

  func testAppDispositionsRetryHoldAndSkip() async throws {
    let (connectivity, _) = await connectivity()
    let outbox = try await outbox([("doc:a", "1"), ("doc:a", "2"), ("doc:a", "3")])
    let transport = ScriptedTransport([])
    var round = 0
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity) { _, entries, _ in
      round += 1
      // Round 1: only the first could be sent (a merge fallback); the rest
      // are left out and go back without penalty.
      if round == 1 { return [entries[0].id: .sent] }
      return [entries[0].id: .hold(Hold(.rejected("no")))]
    }
    let outcome8 = await drainer.drain()
    XCTAssertEqual(outcome8, .idle)
    XCTAssertEqual(round, 2)
    let lane = await outbox.entries(lane: "doc:a")
    XCTAssertEqual(lane.map { String(decoding: $0.payload, as: UTF8.self) }, ["2", "3"])
    XCTAssertEqual(lane.first?.hold, Hold(.rejected("no")))
    XCTAssertEqual(lane.last?.attempts, 0, "A left-out entry is not a failed attempt")
  }

  /// The background loop: an enqueue kicks it, and a restored path wakes it
  /// with no timer running in between.
  func testTheLoopSendsOnKickAndOnReconnect() async throws {
    let (connectivity, source) = await connectivity()
    let transport = ScriptedTransport { _ in .respond(status: 204) }
    let outbox = Outbox(store: MemoryBlobStore(), account: "a")
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster())
    drainer.start()
    defer { drainer.stop() }
    try await outbox.enqueue(lane: "doc:a", kind: "u", payload: Data("1".utf8))
    drainer.kick()
    await settle { transport.committed.count == 1 }
    XCTAssertEqual(transport.committed.count, 1)

    source.send(.unsatisfied)
    await settle { !connectivity.isPathSatisfied }
    try await outbox.enqueue(lane: "doc:a", kind: "u", payload: Data("2".utf8))
    drainer.kick()
    try await Task.sleep(nanoseconds: 100_000_000)
    XCTAssertEqual(transport.calls.count, 1, "Nothing is attempted while the path is down")
    await settle { drainer.status() == .offline(pending: 1) }
    XCTAssertEqual(drainer.status(), .offline(pending: 1))
    source.send(.satisfied)
    await settle { transport.committed.count == 2 }
    XCTAssertEqual(bodies(transport), ["1", "2"])
    await settle { drainer.summary.isEmpty }
    XCTAssertEqual(drainer.status(), .saved)
  }
  func testAnAppRetryDoesNotStarveOtherLanes() async throws {
    let (connectivity, _) = await connectivity()
    let outbox = try await outbox([("doc:a-draft", "1"), ("doc:z-ready", "2")])
    let transport = ScriptedTransport([])
    var sent: [String] = []
    let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity) { lane, entries, _ in
      sent.append(lane)
      return Dictionary(uniqueKeysWithValues: entries.map { ($0.id, lane == "doc:a-draft" ? .retry : .sent) })
    }
    let result = await drainer.drain()
    guard case .backingOff = result else { return XCTFail("The draft still needs a retry") }
    XCTAssertEqual(sent, ["doc:a-draft", "doc:z-ready"])
    let remaining = await outbox.entries(lane: "doc:z-ready")
    XCTAssertTrue(remaining.isEmpty, "An unattached draft must not block another document")
  }

}