import Foundation
import GRDB
import NucleicProtocol

/// Mesh-cast persistence (`CastPersisting`) over the `cast_event` table (v27). Every method is
/// one write transaction, so seq assignment, compaction, and retention can't tear.
extension GRDBMetadataStore: CastPersisting {
    public func publishCast(
        origin: String, channel: String, mode: CastMode, key: String?,
        typeID: String, payload: Data, deleted: Bool, sentAt: Date
    ) async throws -> WireCast {
        try await writer.write { db in
            let tip = try UInt64.fetchOne(
                db,
                sql: "SELECT MAX(cast_seq) FROM cast_event WHERE origin_host_id = ? AND channel = ?",
                arguments: [origin, channel]) ?? 0
            let cast = WireCast(
                originHostID: origin, channel: channel, castSeq: tip + 1, mode: mode,
                key: key, typeID: typeID, payload: payload, deleted: deleted, sentAt: sentAt)
            try Self.insert(cast, into: db, orIgnore: false)
            try Self.compactAndSweep(db, origin: origin, channel: channel, mode: mode, key: key)
            return cast
        }
    }

    public func ingestCasts(_ casts: [WireCast]) async throws -> [WireCast] {
        guard !casts.isEmpty else { return [] }
        return try await writer.write { db in
            var inserted: [WireCast] = []
            for cast in casts {
                try Self.insert(cast, into: db, orIgnore: true)
                if db.changesCount > 0 { inserted.append(cast) }
            }
            // Compact/sweep once per (origin, channel, key) actually touched.
            var seen = Set<String>()
            for cast in inserted {
                let sweepKey = "\(cast.originHostID)\u{1}\(cast.channel)\u{1}\(cast.key ?? "")"
                guard seen.insert(sweepKey).inserted else { continue }
                try Self.compactAndSweep(
                    db, origin: cast.originHostID, channel: cast.channel,
                    mode: cast.mode, key: cast.key)
            }
            return inserted
        }
    }

    public func replaceCasts(origin: String, channel: String, with casts: [WireCast]) async throws {
        try await writer.write { db in
            try db.execute(
                sql: "DELETE FROM cast_event WHERE origin_host_id = ? AND channel = ?",
                arguments: [origin, channel])
            for cast in casts where cast.originHostID == origin && cast.channel == channel {
                try Self.insert(cast, into: db, orIgnore: true)
            }
        }
    }

    public func castTips() async throws -> [CastTip] {
        try await writer.read { db in
            let rows = try Row.fetchAll(db, sql: """
                SELECT origin_host_id, channel, MIN(mode) AS mode,
                       MAX(cast_seq) AS tip, MIN(cast_seq) AS floor
                FROM cast_event GROUP BY origin_host_id, channel
                """)
            return rows.map { row in
                CastTip(
                    originHostID: row["origin_host_id"], channel: row["channel"],
                    mode: CastMode(rawValue: row["mode"]),
                    tipSeq: row["tip"], floorSeq: row["floor"])
            }
        }
    }

    public func castsAfter(
        origin: String, channel: String, afterSeq: UInt64
    ) async throws -> [WireCast] {
        try await writer.read { db in
            try Row.fetchAll(db, sql: """
                SELECT * FROM cast_event
                WHERE origin_host_id = ? AND channel = ? AND cast_seq > ?
                ORDER BY cast_seq ASC
                """, arguments: [origin, channel, afterSeq]).map(Self.cast(from:))
        }
    }

    public func castStateSnapshot(origin: String, channel: String) async throws -> [WireCast] {
        // Write-time compaction keeps exactly the latest row per key, so the snapshot is just
        // everything held (tombstones included — a late joiner must drop the key too).
        try await castsAfter(origin: origin, channel: channel, afterSeq: 0)
    }

    public func removeCasts(origin: String) async throws {
        try await writer.write { db in
            try db.execute(
                sql: "DELETE FROM cast_event WHERE origin_host_id = ?", arguments: [origin])
        }
    }

    // MARK: - Internals

    private static func insert(_ cast: WireCast, into db: Database, orIgnore: Bool) throws {
        try db.execute(sql: """
            INSERT \(orIgnore ? "OR IGNORE " : "")INTO cast_event
              (origin_host_id, channel, cast_seq, mode, key, type_id, payload, deleted,
               sent_at, received_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, arguments: [
                cast.originHostID, cast.channel, cast.castSeq, cast.mode.rawValue, cast.key,
                cast.typeID, cast.payload, cast.deleted, cast.sentAt, Date(),
            ])
    }

    private static func cast(from row: Row) -> WireCast {
        WireCast(
            originHostID: row["origin_host_id"], channel: row["channel"],
            castSeq: row["cast_seq"], mode: CastMode(rawValue: row["mode"]),
            key: row["key"], typeID: row["type_id"], payload: row["payload"],
            deleted: row["deleted"], sentAt: row["sent_at"])
    }

    /// State: keep only the newest row per compaction key. Stream: enforce the channel policy's
    /// count/age caps. Unknown channels use the cast's own stamped mode with the registry's
    /// conservative default caps.
    private static func compactAndSweep(
        _ db: Database, origin: String, channel: String, mode: CastMode, key: String?
    ) throws {
        let policy = CastChannelRegistry.policy(for: channel)
        let effectiveMode = policy.mode ?? mode
        if effectiveMode == .state {
            if let key {
                try db.execute(sql: """
                    DELETE FROM cast_event
                    WHERE origin_host_id = ? AND channel = ? AND key = ?
                      AND cast_seq < (SELECT MAX(cast_seq) FROM cast_event
                                      WHERE origin_host_id = ? AND channel = ? AND key = ?)
                    """, arguments: [origin, channel, key, origin, channel, key])
            }
        } else {
            // Count cap: keep the newest `maxEvents`.
            try db.execute(sql: """
                DELETE FROM cast_event
                WHERE origin_host_id = ? AND channel = ?
                  AND cast_seq <= (SELECT MAX(cast_seq) FROM cast_event
                                   WHERE origin_host_id = ? AND channel = ?) - ?
                """, arguments: [origin, channel, origin, channel, policy.maxEvents])
            // Age cap, on receipt time (local clock only — never the origin's `sentAt`).
            try db.execute(sql: """
                DELETE FROM cast_event
                WHERE origin_host_id = ? AND channel = ? AND received_at < ?
                """, arguments: [origin, channel, Date().addingTimeInterval(-policy.maxAge)])
        }
    }
}
