import Foundation
import RastrilloNative

struct PersistedReplica: Codable, Sendable {
    let format: Int
    let sequence: Int64
    let guid: String
    let savedAt: Date
    let state: Data
}

/// One document's durability boundary. The writer and editor share this
/// watermark, but only a successful outbox write may advance it.
@MainActor final class DocsPersistence {
    let store: BlobStore
    let outbox: Outbox
    let key: String
    let replica: CRDTReplica
    private(set) var commandsDurable = 0
    private var active = true
    let cache: DocsReplicaCache
    private let generation: Int
    var lane: String { "doc:" + key }
    var path: String { "cache/docs/" + key + "/replica" }
    init(store: BlobStore, outbox: Outbox, key: String, replica: CRDTReplica, cache: DocsReplicaCache? = nil) {
        self.store = store; self.outbox = outbox; self.key = key; self.replica = replica
        self.cache = cache ?? DocsReplicaCache(store: store)
        generation = self.cache.generation(key)
    }
    func reopen(fallback: Data, sequence: Int64) async throws {
        try await outbox.load()
        let entries = await outbox.entries(lane: lane)
        let saved = try store.read(path).map { try JSONDecoder().decode(PersistedReplica.self, from: $0) }
        if let saved, saved.format != 1 { throw BlobStoreError.unavailable }
        _ = try await replica.open(update: saved?.state ?? fallback)
        await replica.setSequence(saved?.sequence ?? sequence)
        // Replay includes held entries: the editor must show everything kept
        // here, even when it cannot currently send it.
        for entry in entries where entry.kind == "update" { _ = try await replica.apply(entry.payload) }
        commandsDurable = 0
    }
    func enqueue(_ update: Data) async throws {
        try await outbox.enqueue(lane: lane, kind: "update", payload: update)
        commandsDurable += 1
    }
    @discardableResult func persist() async throws -> Bool {
        let captured = try await replica.persistentState()
        // No suspension between checking the watermark and the atomic write.
        // A writer running between command and enqueue must defer its capture.
        guard active, captured.commandsApplied == commandsDurable else { return false }
        let saved = PersistedReplica(format: 1, sequence: captured.sequence, guid: captured.guid, savedAt: Date(), state: captured.state)
        return try await cache.commit(saved, key: key, generation: generation) {
            self.active && captured.commandsApplied == self.commandsDurable
        }
    }
    func compact() async throws {
        let entries = await outbox.sendable(lane: lane, maxBytes: nil)
        guard entries.count > 256 || entries.reduce(0, { $0 + $1.payload.count }) > 1024 * 1024 else { return }
        var batch: [OutboxEntry] = []; var bytes = 0
        for entry in entries {
            if entry.kind != "update" { if batch.isEmpty { continue }; break }
            if !batch.isEmpty && bytes + entry.payload.count > 512 * 1024 { break }
            batch.append(entry); bytes += entry.payload.count
        }
        guard batch.count > 1 else { return }
        let merged = try await replica.mergeUpdates(batch.map(\.payload))
        guard merged.count <= 512 * 1024 else { return }
        // A send can reserve the batch while mergeUpdates suspends. Refusing
        // that compaction is expected; it must never delete the reservation.
        do { try await outbox.replace(batch.map(\.id), with: merged) }
        catch OutboxError.inFlight { }
        catch OutboxError.unknownEntry { }
    }
    func discard() async throws {
        active = false
        cache.invalidate(key)
        // Remove the replica first: a crash may replay lane entries again,
        // but can never leave local text whose lane has already been removed.
        try store.remove(path)
        try await outbox.discard(lane: lane)
    }
}