import Foundation
import NucleicProtocol
#if canImport(Darwin)
import Darwin
#elseif canImport(Musl)
import Musl
#elseif canImport(Glibc)
import Glibc
#endif

/// Every member's on-disk holder of Carbon shards and manifests (docs/CARBON_SHARDING.md §7)
/// — a content-addressed sibling of `TranscriptMirrorStore`. Blobs are ciphertext, write-once,
/// verified against their ID **on every write and on every read** (§7.2): a bit-rotted holder
/// must degrade to "doesn't have it," never poison the mesh.
///
/// Layout (spec §7.1):
///
///     <root>/<projectUUID>/
///       objects/<aa>/<shardID>        aa = first two hex chars; write-once ciphertext
///       objects/quarantine/           blobs that failed verify-on-read
///       manifests/<aa>/<manifestID>   ciphertext CBOR; write-once
///       manifests/quarantine/
///       heads.json                    this member's head per stream (IDs/seqs only — no
///                                     content, no secrets, so plaintext JSON is fine)
///
/// Writes are atomic: temp file in the destination directory, `fsync`, then `rename(2)`.
/// A `put` of an existing ID is a no-op success (idempotent). Eviction of a project is a
/// directory drop, mirroring `TranscriptMirrorStore.removeSource`.
public actor CarbonShardStore {
    /// Resolves a project's master key (from the credential vault / escrow). Blob storage and
    /// serving never need it; it is required only to *decode* manifests (`manifest(id:)`) and
    /// to mark reachable shards during GC. A missing key makes GC retain everything — safe.
    public typealias KeyProvider = @Sendable (UUID) -> CarbonCrypto.ProjectKey?

    private let root: URL
    private let keyProvider: KeyProvider
    /// heads.json contents per project, loaded lazily and kept authoritative in memory.
    private var headsCache: [UUID: [CarbonStreamKey: CarbonHead]] = [:]
    /// Last computed ``usage(project:now:)`` per project, with when it was walked — see
    /// ``usageCacheTTL``. Purely a rate limiter on the directory walk; never a source of truth.
    private var usageCache: [UUID: (usage: CarbonStoreUsage, computedAt: Date)] = [:]

    public init(root: URL, keyProvider: @escaping KeyProvider) {
        self.root = root
        self.keyProvider = keyProvider
    }

    // MARK: - Shards

    /// Store one shard blob. Verifies `SHA256(ciphertext) == expectedID` before anything
    /// touches disk; rejects on mismatch (spec §7.2). Idempotent for an already-held ID.
    public func putShard(ciphertext: Data, expectedID: String, projectUUID: UUID) throws {
        try put(ciphertext, expectedID: expectedID, in: objectsDir(projectUUID))
    }

    /// A verified shard blob, or nil when absent — including when the on-disk copy fails
    /// verification, in which case it is quarantined so this holder honestly stops claiming it.
    public func shard(id: String, projectUUID: UUID) -> Data? {
        verifiedRead(id: id, in: objectsDir(projectUUID))
    }

    /// Whether a verified-on-write shard blob is present (no read/verify — presence only).
    public func hasShard(id: String, projectUUID: UUID) -> Bool {
        FileManager.default.fileExists(atPath: blobURL(id, in: objectsDir(projectUUID)).path)
    }

    // MARK: - Manifests

    public func putManifest(ciphertext: Data, expectedID: String, projectUUID: UUID) throws {
        try put(ciphertext, expectedID: expectedID, in: manifestsDir(projectUUID))
    }

    /// The verified ciphertext of a manifest (what a mirror serves to a peer — no key needed).
    public func manifestBlob(id: String, projectUUID: UUID) -> Data? {
        verifiedRead(id: id, in: manifestsDir(projectUUID))
    }

    /// Decrypt + decode a held manifest. nil when not held (or quarantined on failed verify);
    /// throws `.keyMissing` when the blob is held but this member has no project key.
    public func manifest(id: String, projectUUID: UUID) throws -> CarbonManifest? {
        guard let blob = verifiedRead(id: id, in: manifestsDir(projectUUID)) else { return nil }
        guard let key = keyProvider(projectUUID) else {
            throw CarbonStoreError.keyMissing(projectUUID)
        }
        return try CarbonManifest.open(blob, with: CarbonCrypto(projectKey: key))
    }

    /// Shard IDs referenced by `manifest` that this store does not hold — what a mirror
    /// fetches before it may advance its head (the honest-head rule, spec §8.2 step 3).
    public func missingShards(for manifest: CarbonManifest) -> [String] {
        let dir = objectsDir(manifest.project.uuid)
        return manifest.shards.map(\.id).filter {
            !FileManager.default.fileExists(atPath: blobURL($0, in: dir).path)
        }
    }

    // MARK: - Heads

    public func heads(projectUUID: UUID) -> [CarbonStreamKey: CarbonHead] {
        loadHeads(projectUUID)
    }

    /// Advance a stream head, enforcing the §10 fencing rule with no exceptions: a head moves
    /// only to a strictly higher `(ownerGeneration, snapshotSeq)`. A stale-generation manifest
    /// may be *stored* (as a chain ancestor) but can never become a head.
    public func advanceHead(_ head: CarbonHead) throws {
        var heads = loadHeads(head.key.projectUUID)
        if let current = heads[head.key] {
            let advances =
                head.ownerGeneration > current.ownerGeneration
                || (head.ownerGeneration == current.ownerGeneration
                    && head.snapshotSeq > current.snapshotSeq)
            guard advances else {
                throw CarbonStoreError.staleHead(current: current, attempted: head)
            }
        }
        heads[head.key] = head
        headsCache[head.key.projectUUID] = heads
        try persistHeads(heads, projectUUID: head.key.projectUUID)
    }

    // MARK: - GC (spec §7.3 — mirror-autonomous, pure local rules, no delete verbs anywhere)

    public func collectGarbage(
        project: UUID, policy: CarbonRetention
    ) throws -> CarbonGCReport {
        let fm = FileManager.default
        var report = CarbonGCReport()
        let heads = loadHeads(project)

        // Without the key we cannot tell which shards are reachable — retain everything.
        // (Unless the whole project is unretained, where reachability is irrelevant.)
        let key = keyProvider(project)
        if key == nil && !policy.projectUnretained { return report }

        var retainedManifestIDs = Set<String>()
        var markedShardIDs = Set<String>()

        if !policy.projectUnretained, let key {
            let crypto = CarbonCrypto(projectKey: key)
            // Decrypt every held manifest and bucket into chains.
            var chains: [CarbonStreamKey: [(id: String, manifest: CarbonManifest)]] = [:]
            for url in blobFiles(in: manifestsDir(project)) {
                guard let blob = try? Data(contentsOf: url),
                    CarbonCrypto.blobID(blob) == url.lastPathComponent,
                    let m = try? CarbonManifest.open(blob, with: crypto)
                else { continue }  // unreadable ⇒ handled by the sweep's age floor below
                let chainKey = CarbonStreamKey(
                    projectUUID: project, stream: m.stream, sessionID: m.sessionID)
                chains[chainKey, default: []].append((url.lastPathComponent, m))
            }

            for (chainKey, entries) in chains {
                let dropped = chainKey.sessionID.map { policy.droppedSessionIDs.contains($0) } ?? false
                if dropped { continue }  // rule 3: ended session past grace ⇒ chain unretained
                switch chainKey.stream {
                case .worktreeSnapshot:
                    // Rule 1: last K of the highest observed generation; lower-generation
                    // chains ride along until the session ends (they anchor thin-pack bases).
                    let maxGen = entries.map(\.manifest.ownerGeneration).max() ?? 0
                    let current = entries.filter { $0.manifest.ownerGeneration == maxGen }
                        .sorted { $0.manifest.snapshotSeq < $1.manifest.snapshotSeq }
                    let kept =
                        current.suffix(max(policy.snapshotManifestsPerChain, 1))
                        + entries.filter { $0.manifest.ownerGeneration < maxGen }
                    for e in kept { retainedManifestIDs.insert(e.id) }
                    for e in kept { for s in e.manifest.shards { markedShardIDs.insert(s.id) } }
                case .history, .nativeTranscript, .lock:
                    // Entire chains while the session/project is retained.
                    for e in entries { retainedManifestIDs.insert(e.id) }
                    for e in entries { for s in e.manifest.shards { markedShardIDs.insert(s.id) } }
                }
            }

            // Belt-and-braces: a current head's manifest (and its shards) is never collected,
            // even if a policy quirk excluded it — a gossiped head must stay honest.
            for head in heads.values {
                if retainedManifestIDs.contains(head.manifestID) { continue }
                retainedManifestIDs.insert(head.manifestID)
                if let m = try? CarbonManifest.open(
                    Data(contentsOf: blobURL(head.manifestID, in: manifestsDir(project))),
                    with: CarbonCrypto(projectKey: key))
                {
                    for s in m.shards { markedShardIDs.insert(s.id) }
                }
            }
        }

        // Sweep, oldest-first is irrelevant — only the age floor matters: never delete a blob
        // younger than the floor (protects shards that landed before their manifest).
        let cutoff = Date().addingTimeInterval(-policy.sweepMinAge)
        func sweep(_ dir: URL, retained: Set<String>, counting: WritableKeyPath<CarbonGCReport, Int>) {
            for url in blobFiles(in: dir) {
                guard !retained.contains(url.lastPathComponent) else { continue }
                guard let mtime = try? fm.attributesOfItem(atPath: url.path)[.modificationDate]
                    as? Date, mtime < cutoff
                else { continue }
                let size = (try? fm.attributesOfItem(atPath: url.path)[.size] as? Int) ?? 0
                if (try? fm.removeItem(at: url)) != nil {
                    report[keyPath: counting] += 1
                    report.bytesReclaimed += size
                }
            }
        }
        sweep(manifestsDir(project), retained: retainedManifestIDs, counting: \.manifestsRemoved)
        sweep(objectsDir(project), retained: markedShardIDs, counting: \.shardsRemoved)
        return report
    }

    // MARK: - Usage (spec §12 — what the Carbon Viewer reports)

    /// How long a computed usage tally is served before the directories are walked again.
    /// The viewer polls faster than this on purpose: heads and peer lag stay live, while the
    /// blob counts — which move slowly and cost a full directory walk on a store holding tens
    /// of thousands of blobs — refresh on their own, slower cadence.
    static let usageCacheTTL: TimeInterval = 10

    /// Blob counts + bytes on disk for one project. A directory walk, never a decrypt: this
    /// needs no project key, so a keyless mirror can still report honestly what it is holding.
    ///
    /// Recomputed from the filesystem at most every ``usageCacheTTL``, so a viewer polling on a
    /// short cadence can't starve the producer of this actor — and because each refresh is a
    /// fresh walk rather than a running total, the figures can never drift from what's on disk.
    public func usage(project: UUID, now: Date = Date()) -> CarbonStoreUsage {
        if let cached = usageCache[project],
            now.timeIntervalSince(cached.computedAt) < Self.usageCacheTTL
        {
            return cached.usage
        }
        let objects = tally(objectsDir(project))
        let manifests = tally(manifestsDir(project))
        let usage = CarbonStoreUsage(
            shardCount: objects.live,
            manifestCount: manifests.live,
            quarantinedCount: objects.quarantined + manifests.quarantined,
            bytes: objects.bytes + manifests.bytes)
        usageCache[project] = (usage, now)
        return usage
    }

    /// Recursive count/size of a fan-out directory, splitting `quarantine/` out from the live
    /// blobs — quarantined bytes are still on disk, but this holder has stopped claiming them
    /// (§7.2), so they must not inflate the "held" counts.
    private func tally(_ dir: URL) -> (live: Int, quarantined: Int, bytes: UInt64) {
        let keys: [URLResourceKey] = [.fileSizeKey, .isRegularFileKey]
        guard let walker = FileManager.default.enumerator(
            at: dir, includingPropertiesForKeys: keys, options: [.skipsHiddenFiles])
        else { return (0, 0, 0) }
        var live = 0
        var quarantined = 0
        var bytes: UInt64 = 0
        for case let url as URL in walker {
            guard let values = try? url.resourceValues(forKeys: Set(keys)),
                values.isRegularFile == true,
                !url.lastPathComponent.hasPrefix(".tmp-")
            else { continue }
            bytes += UInt64(values.fileSize ?? 0)
            if url.deletingLastPathComponent().lastPathComponent == "quarantine" {
                quarantined += 1
            } else {
                live += 1
            }
        }
        return (live, quarantined, bytes)
    }

    /// Every project this store holds anything for (its top-level directories) — how a
    /// serve-all-heads request enumerates without a separate registry.
    public func knownProjects() -> [UUID] {
        let dirs = (try? FileManager.default.contentsOfDirectory(
            at: root, includingPropertiesForKeys: nil, options: [.skipsHiddenFiles])) ?? []
        return dirs.compactMap { UUID(uuidString: $0.lastPathComponent) }
    }

    /// Drop everything held for a project — a directory drop (spec §7.1).
    public func evict(project: UUID) throws {
        headsCache[project] = nil
        usageCache[project] = nil  // the directory is about to go; never serve its old tally
        let dir = projectDir(project)
        if FileManager.default.fileExists(atPath: dir.path) {
            try FileManager.default.removeItem(at: dir)
        }
    }

    // MARK: - Verified blob IO

    private func put(_ ciphertext: Data, expectedID: String, in dir: URL) throws {
        let actual = CarbonCrypto.blobID(ciphertext)
        guard actual == expectedID else {
            throw CarbonStoreError.idMismatch(expected: expectedID, actual: actual)
        }
        let dst = blobURL(expectedID, in: dir)
        if FileManager.default.fileExists(atPath: dst.path) { return }  // idempotent
        try atomicWrite(ciphertext, to: dst)
    }

    private func verifiedRead(id: String, in dir: URL) -> Data? {
        let url = blobURL(id, in: dir)
        guard let data = try? Data(contentsOf: url) else { return nil }
        guard CarbonCrypto.blobID(data) == id else {
            quarantine(url, in: dir)
            return nil
        }
        return data
    }

    private func quarantine(_ url: URL, in dir: URL) {
        let qDir = dir.appendingPathComponent("quarantine", isDirectory: true)
        try? FileManager.default.createDirectory(at: qDir, withIntermediateDirectories: true)
        let dst = qDir.appendingPathComponent(url.lastPathComponent)
        try? FileManager.default.removeItem(at: dst)
        try? FileManager.default.moveItem(at: url, to: dst)
    }

    /// Temp-in-destination-directory + fsync + `rename(2)`: a crash leaves either the old
    /// state or the new file, never a partial blob under its content address.
    private func atomicWrite(_ data: Data, to url: URL) throws {
        let dir = url.deletingLastPathComponent()
        try FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true)
        let tmp = dir.appendingPathComponent(".tmp-\(UUID().uuidString)")
        try data.write(to: tmp)
        #if os(Windows)
        // Windows cannot use the POSIX recipe: there is no `fsync` (durability rides
        // `MOVEFILE_WRITE_THROUGH`), and the CRT's `rename` **fails** when the destination exists
        // rather than replacing it — which for a content-addressed store would turn every
        // idempotent re-write of an already-present blob into an error. `MoveFileExW` with
        // REPLACE_EXISTING is the atomic-replace primitive, and `Data.write` has already flushed
        // the file's own contents.
        do {
            try WindowsAtomicMove.replace(from: tmp.path, to: url.path)
        } catch {
            try? FileManager.default.removeItem(at: tmp)
            throw CarbonStoreError.io("\(error)")
        }
        #else
        let fd = open(tmp.path, O_RDONLY)
        if fd >= 0 {
            fsync(fd)
            close(fd)
        }
        guard rename(tmp.path, url.path) == 0 else {
            try? FileManager.default.removeItem(at: tmp)
            throw CarbonStoreError.io(String(cString: strerror(errno)))
        }
        #endif
    }

    // MARK: - Layout

    private func projectDir(_ uuid: UUID) -> URL {
        root.appendingPathComponent(uuid.uuidString, isDirectory: true)
    }

    private func objectsDir(_ uuid: UUID) -> URL {
        projectDir(uuid).appendingPathComponent("objects", isDirectory: true)
    }

    private func manifestsDir(_ uuid: UUID) -> URL {
        projectDir(uuid).appendingPathComponent("manifests", isDirectory: true)
    }

    private func blobURL(_ id: String, in dir: URL) -> URL {
        dir.appendingPathComponent(String(id.prefix(2)), isDirectory: true)
            .appendingPathComponent(id)
    }

    /// All blob files under a fan-out directory, skipping `quarantine/` and temp residue.
    private func blobFiles(in dir: URL) -> [URL] {
        let fm = FileManager.default
        let fanouts = (try? fm.contentsOfDirectory(
            at: dir, includingPropertiesForKeys: nil, options: [.skipsHiddenFiles])) ?? []
        return fanouts.filter { $0.lastPathComponent != "quarantine" }.flatMap { sub in
            (try? fm.contentsOfDirectory(
                at: sub, includingPropertiesForKeys: nil, options: [.skipsHiddenFiles])) ?? []
        }.filter { !$0.lastPathComponent.hasPrefix(".tmp-") }
    }

    // MARK: - Heads persistence

    private func headsURL(_ uuid: UUID) -> URL {
        projectDir(uuid).appendingPathComponent("heads.json")
    }

    private func loadHeads(_ uuid: UUID) -> [CarbonStreamKey: CarbonHead] {
        if let cached = headsCache[uuid] { return cached }
        var map: [CarbonStreamKey: CarbonHead] = [:]
        if let data = try? Data(contentsOf: headsURL(uuid)),
            let list = try? JSONDecoder().decode([CarbonHead].self, from: data)
        {
            for head in list { map[head.key] = head }
        }
        headsCache[uuid] = map
        return map
    }

    private func persistHeads(_ heads: [CarbonStreamKey: CarbonHead], projectUUID: UUID) throws {
        let encoder = JSONEncoder()
        encoder.outputFormatting = [.sortedKeys]
        let list = heads.values.sorted {
            ($0.key.stream.rawValue, $0.key.sessionID ?? "")
                < ($1.key.stream.rawValue, $1.key.sessionID ?? "")
        }
        try atomicWrite(try encoder.encode(list), to: headsURL(projectUUID))
    }
}

/// What a GC pass may treat as unretained (spec §7.3). Rules 3–4 (session end + grace,
/// ex-replica grace) are decided by the caller — the store never interprets session state.
public struct CarbonRetention: Sendable {
    /// Rule 1: K snapshot manifests kept per (sessionID, highest observed generation).
    public var snapshotManifestsPerChain: Int
    /// Rule 3: sessions whose end-of-life grace has expired — their chains are unretained.
    public var droppedSessionIDs: Set<String>
    /// Rule 4: this member left the project's replica set and the 14-day grace has expired.
    public var projectUnretained: Bool
    /// Rule 5: never delete a blob younger than this (protects shards that landed before
    /// their manifest). Default 48 h.
    public var sweepMinAge: TimeInterval

    public init(
        snapshotManifestsPerChain: Int = 8, droppedSessionIDs: Set<String> = [],
        projectUnretained: Bool = false, sweepMinAge: TimeInterval = 48 * 3600
    ) {
        self.snapshotManifestsPerChain = snapshotManifestsPerChain
        self.droppedSessionIDs = droppedSessionIDs
        self.projectUnretained = projectUnretained
        self.sweepMinAge = sweepMinAge
    }
}

/// What one project occupies in a member's store (spec §12's read-only view). Counts are of
/// *blobs held*, so a quarantined blob is tallied apart from the live ones even though its
/// bytes are still on disk — `bytes` is the honest disk footprint, `shardCount` the honest claim.
public struct CarbonStoreUsage: Sendable, Equatable {
    public var shardCount: Int
    public var manifestCount: Int
    public var quarantinedCount: Int
    /// Total bytes across shards, manifests, and quarantine (excludes `heads.json`, which is
    /// bookkeeping, not held content).
    public var bytes: UInt64

    public init(
        shardCount: Int = 0, manifestCount: Int = 0, quarantinedCount: Int = 0, bytes: UInt64 = 0
    ) {
        self.shardCount = shardCount
        self.manifestCount = manifestCount
        self.quarantinedCount = quarantinedCount
        self.bytes = bytes
    }

    /// Nothing held at all — the viewer's genuine empty state, as opposed to the hardcoded one
    /// it used to show unconditionally.
    public var isEmpty: Bool { shardCount == 0 && manifestCount == 0 && quarantinedCount == 0 }
}

public struct CarbonGCReport: Sendable, Equatable {
    public var manifestsRemoved = 0
    public var shardsRemoved = 0
    public var bytesReclaimed = 0

    public init() {}
}

public enum CarbonStoreError: Error {
    case idMismatch(expected: String, actual: String)
    case keyMissing(UUID)
    case staleHead(current: CarbonHead, attempted: CarbonHead)
    case io(String)
}
