openclaw/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatTranscriptCache.swift

import Foundation
import OSLog
import SQLite3
#if os(iOS)
import UIKit
#endif

private let cacheLogger = Logger(subsystem: "ai.openclaw", category: "OpenClawChatTranscriptCache")

/// Read-only offline cache seam for chat sessions and transcripts.
///
/// The cache only pre-paints cold opens and covers offline browsing; connected
/// reads always come from the gateway and replace cached content wholesale.
/// Implementations must scope all rows to a single gateway identity so
/// transcripts never leak across paired gateways.
public protocol OpenClawChatTranscriptCache: Sendable {
    func loadSessions() async -> [OpenClawChatSessionEntry]
    func loadTranscript(sessionKey: String) async -> [OpenClawChatMessage]
    func storeSessions(_ sessions: [OpenClawChatSessionEntry]) async
    func storeTranscript(sessionKey: String, messages: [OpenClawChatMessage]) async
}

/// One durable queued chat command (text only in v1). `id` is the client UUID
/// that becomes the transport idempotency key on flush, so at-least-once
/// delivery stays safe across retries and app restarts.
///
/// Naming mirrors the watch-side `QueuedCommand` shape (WatchChatCoordinator)
/// so the two queues can merge into one owner later.
public struct OpenClawChatOutboxCommand: Hashable, Sendable, Identifiable {
    public enum Status: String, Sendable {
        case queued
        case sending
        case failed
    }

    public let id: String
    public let sessionKey: String
    public let text: String
    /// Thinking level captured when the command was queued, so a later flush
    /// never borrows the setting of whichever session is visible then.
    public let thinking: String
    /// Seconds since 1970; flush order is strictly ascending `createdAt`.
    public let createdAt: Double
    public var status: Status
    public var retryCount: Int
    public var lastError: String?

    public init(
        id: String,
        sessionKey: String,
        text: String,
        thinking: String,
        createdAt: Double,
        status: Status,
        retryCount: Int,
        lastError: String?)
    {
        self.id = id
        self.sessionKey = sessionKey
        self.text = text
        self.thinking = thinking
        self.createdAt = createdAt
        self.status = status
        self.retryCount = retryCount
        self.lastError = lastError
    }
}

/// Durable offline outbox for chat commands, scoped to one gateway identity
/// exactly like the transcript cache. Implementations persist queued sends so
/// they survive app restarts and flush on reconnect.
public protocol OpenClawChatCommandOutbox: Sendable {
    /// Returns false when the queue is full (`maxQueuedCommands`) or storage
    /// is unavailable; callers surface that instead of dropping text silently.
    func enqueueCommand(_ command: OpenClawChatOutboxCommand) async -> Bool
    /// Gateway-scoped rows in `createdAt` order. Applies the staleness gate:
    /// queued rows older than `outboxCommandMaxAge` become failed("expired")
    /// so reconnect never sends stale commands silently.
    func loadCommands() async -> [OpenClawChatOutboxCommand]
    /// Crash safety: rows stuck in 'sending' from a previous process revert
    /// to 'queued'; the idempotency key makes the re-send safe. Returns false
    /// when the store was unreachable (for example Complete file protection
    /// while the device is locked) so callers can retry recovery later.
    @discardableResult
    func recoverInterruptedSends() async -> Bool
    /// Claims a row for sending. Returns false when the row no longer exists
    /// (deleted mid-flush), so the flush skips it instead of sending stale text.
    @discardableResult
    func markCommandSending(id: String) async -> Bool
    func markCommandQueued(id: String, retryCount: Int, lastError: String?) async
    func markCommandFailed(id: String, retryCount: Int, lastError: String?) async
    /// Explicit user retry: reset attempts and refresh `createdAt` so an
    /// expired row can send again (retry is new intent, so it also moves the
    /// command to the queue tail rather than replaying its old position).
    func markCommandRetried(id: String) async
    func deleteCommand(id: String) async
}

/// SQLite-backed transcript cache for one gateway identity. Owners should use
/// one database file per gateway so reset can physically remove that gateway's
/// cached transcript bytes without disturbing other paired gateways; queries
/// are additionally scoped by `gatewayID` as a defensive belt.
///
/// The cache is disposable: any open, schema, or decode mismatch drops the
/// affected state and rebuilds silently. There are no migrations. The command
/// outbox shares this database; a drop also clears queued commands, which is
/// acceptable because a queue that predates a schema change is stale anyway.
public actor OpenClawChatSQLiteTranscriptCache: OpenClawChatTranscriptCache, OpenClawChatCommandOutbox {
    /// Bounds keep the cache small: enough for a recently-used session picker
    /// and a full first screen of transcript, not a durable archive.
    public static let maxCachedSessions = 50
    public static let maxCachedTranscripts = 50
    public static let maxCachedMessagesPerSession = 200
    /// Outbox bounds: refuse enqueue beyond this many rows per gateway, and
    /// expire queued commands instead of sending them after two days offline.
    public static let maxQueuedCommands = 50
    public static let outboxCommandMaxAge: TimeInterval = 48 * 60 * 60
    /// Machine-readable `lastError` set by the staleness gate.
    public static let outboxExpiredError = "expired"
    /// v2 adds the outbox_commands table; older shapes drop-and-rebuild.
    static let schemaVersion: Int32 = 2

    /// Owns the raw sqlite handle so it closes on release without needing an
    /// isolated actor deinit (OpaquePointer is not Sendable).
    private final class Connection: @unchecked Sendable {
        let raw: OpaquePointer

        init(raw: OpaquePointer) {
            self.raw = raw
        }

        deinit {
            sqlite3_close_v2(self.raw)
        }
    }

    private let databaseURL: URL
    private let gatewayID: String
    private var db: Connection?
    private var isRetired = false
    /// After a failed drop-and-rebuild the cache becomes a no-op instead of
    /// erroring the chat surface; a fresh launch retries from scratch.
    private var isBroken = false

    public init(databaseURL: URL, gatewayID: String) {
        self.databaseURL = databaseURL
        self.gatewayID = gatewayID
    }

    /// Startup-only cleanup, before any cache actor can own an open handle.
    public static func removeDatabaseFiles(at databaseURL: URL) {
        let fm = FileManager.default
        try? fm.removeItem(at: databaseURL)
        for suffix in ["-wal", "-shm", "-journal"] {
            try? fm.removeItem(at: URL(fileURLWithPath: databaseURL.path + suffix))
        }
    }

    // MARK: - OpenClawChatTranscriptCache

    public func loadSessions() async -> [OpenClawChatSessionEntry] {
        guard !self.isRetired else { return [] }
        guard let db = await self.handle() else { return [] }
        guard let payload = self.selectPayload(
            db,
            sql: "SELECT payload FROM cached_sessions WHERE gateway_id = ?1",
            bindings: [self.gatewayID])
        else {
            return []
        }
        guard let decoded = try? JSONDecoder().decode(
            [OpenClawChatSessionEntry].self,
            from: Data(payload.utf8))
        else {
            // Decode mismatch means a stale/foreign shape: drop the row silently.
            self.execute(db, sql: "DELETE FROM cached_sessions WHERE gateway_id = ?1", bindings: [self.gatewayID])
            return []
        }
        return decoded
    }

    public func loadTranscript(sessionKey: String) async -> [OpenClawChatMessage] {
        guard !self.isRetired else { return [] }
        guard let db = await self.handle() else { return [] }
        guard let payload = self.selectPayload(
            db,
            sql: "SELECT payload FROM cached_transcripts WHERE gateway_id = ?1 AND session_key = ?2",
            bindings: [self.gatewayID, sessionKey])
        else {
            return []
        }
        guard let decoded = try? JSONDecoder().decode(
            [OpenClawChatMessage].self,
            from: Data(payload.utf8))
        else {
            self.execute(
                db,
                sql: "DELETE FROM cached_transcripts WHERE gateway_id = ?1 AND session_key = ?2",
                bindings: [self.gatewayID, sessionKey])
            return []
        }
        return decoded
    }

    public func storeSessions(_ sessions: [OpenClawChatSessionEntry]) async {
        guard !self.isRetired else { return }
        guard let db = await self.handle() else { return }
        let bounded = Self.boundedSessions(sessions)
        guard !bounded.isEmpty else {
            self.execute(db, sql: "DELETE FROM cached_sessions WHERE gateway_id = ?1", bindings: [self.gatewayID])
            return
        }
        guard let payload = Self.encodeJSON(bounded) else { return }
        self.execute(
            db,
            sql: """
            INSERT OR REPLACE INTO cached_sessions(gateway_id, payload, updated_at)
            VALUES (?1, ?2, ?3)
            """,
            bindings: [self.gatewayID, payload, Date().timeIntervalSince1970])
    }

    public func storeTranscript(sessionKey: String, messages: [OpenClawChatMessage]) async {
        guard !self.isRetired else { return }
        guard let db = await self.handle() else { return }
        let bounded = Self.cacheableMessages(messages)
        guard !bounded.isEmpty else {
            // An emptied live transcript must also empty the cache, or the next
            // cold open would ghost-paint messages the gateway no longer has.
            self.execute(
                db,
                sql: "DELETE FROM cached_transcripts WHERE gateway_id = ?1 AND session_key = ?2",
                bindings: [self.gatewayID, sessionKey])
            return
        }
        guard let payload = Self.encodeJSON(bounded) else { return }
        self.execute(
            db,
            sql: """
            INSERT OR REPLACE INTO cached_transcripts(gateway_id, session_key, payload, updated_at)
            VALUES (?1, ?2, ?3, ?4)
            """,
            bindings: [self.gatewayID, sessionKey, payload, Date().timeIntervalSince1970])
        // rowid tie-breaks equal timestamps: INSERT OR REPLACE mints a fresh
        // rowid, so the most recently written transcript always survives.
        self.execute(
            db,
            sql: """
            DELETE FROM cached_transcripts WHERE gateway_id = ?1 AND session_key NOT IN (
                SELECT session_key FROM cached_transcripts WHERE gateway_id = ?1
                ORDER BY updated_at DESC, rowid DESC LIMIT \(Self.maxCachedTranscripts)
            )
            """,
            bindings: [self.gatewayID])
    }

    public func retire() async {
        // A queued write then either finishes before retirement or becomes a
        // no-op. Closing the handle lets the owner delete the whole cache file.
        self.isRetired = true
        self.db = nil
    }

    // MARK: - OpenClawChatCommandOutbox

    public func enqueueCommand(_ command: OpenClawChatOutboxCommand) async -> Bool {
        guard !self.isRetired, let db = await self.handle() else { return false }
        let count = self.selectInt(
            db,
            sql: "SELECT COUNT(*) FROM outbox_commands WHERE gateway_id = ?1",
            bindings: [self.gatewayID]) ?? 0
        // Bound the queue per gateway across all statuses so failed rows also
        // count: the user must clear them before queueing more.
        guard count < Self.maxQueuedCommands else { return false }
        return self.execute(
            db,
            sql: """
            INSERT INTO outbox_commands(
                client_uuid, gateway_id, session_key, text, thinking, created_at, status, retry_count, last_error
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, '')
            """,
            bindings: [
                command.id,
                self.gatewayID,
                command.sessionKey,
                command.text,
                command.thinking,
                command.createdAt,
                command.status.rawValue,
                command.retryCount,
            ])
    }

    public func loadCommands() async -> [OpenClawChatOutboxCommand] {
        guard !self.isRetired, let db = await self.handle() else { return [] }
        // Staleness gate: a command queued 48h ago is more likely wrong than
        // wanted; surface it as failed("expired") instead of sending it.
        self.execute(
            db,
            sql: """
            UPDATE outbox_commands SET status = 'failed', last_error = ?3
            WHERE gateway_id = ?1 AND status = 'queued' AND created_at < ?2
            """,
            bindings: [
                self.gatewayID,
                Date().timeIntervalSince1970 - Self.outboxCommandMaxAge,
                Self.outboxExpiredError,
            ])

        var statement: OpaquePointer?
        let sql = """
        SELECT client_uuid, session_key, text, thinking, created_at, status, retry_count, last_error
        FROM outbox_commands WHERE gateway_id = ?1
        ORDER BY created_at ASC, id ASC
        """
        guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else { return [] }
        defer { sqlite3_finalize(statement) }
        guard self.bind(statement, bindings: [self.gatewayID]) else { return [] }

        var commands: [OpenClawChatOutboxCommand] = []
        while sqlite3_step(statement) == SQLITE_ROW {
            guard let id = sqlite3_column_text(statement, 0),
                  let sessionKey = sqlite3_column_text(statement, 1),
                  let text = sqlite3_column_text(statement, 2)
            else { continue }
            let thinking = sqlite3_column_text(statement, 3).map { String(cString: $0) } ?? ""
            let statusRaw = sqlite3_column_text(statement, 5).map { String(cString: $0) } ?? ""
            let lastError = sqlite3_column_text(statement, 7).map { String(cString: $0) } ?? ""
            commands.append(
                OpenClawChatOutboxCommand(
                    id: String(cString: id),
                    sessionKey: String(cString: sessionKey),
                    text: String(cString: text),
                    thinking: thinking,
                    createdAt: sqlite3_column_double(statement, 4),
                    // Unknown status means a foreign writer; treating it as
                    // queued is safe because the idempotency key dedupes.
                    status: OpenClawChatOutboxCommand.Status(rawValue: statusRaw) ?? .queued,
                    retryCount: Int(sqlite3_column_int64(statement, 6)),
                    lastError: lastError.isEmpty ? nil : lastError))
        }
        return commands
    }

    @discardableResult
    public func recoverInterruptedSends() async -> Bool {
        guard !self.isRetired, let db = await self.handle() else { return false }
        return self.execute(
            db,
            sql: "UPDATE outbox_commands SET status = 'queued' WHERE gateway_id = ?1 AND status = 'sending'",
            bindings: [self.gatewayID])
    }

    @discardableResult
    public func markCommandSending(id: String) async -> Bool {
        guard !self.isRetired, let db = await self.handle() else { return false }
        let updated = self.execute(
            db,
            sql: "UPDATE outbox_commands SET status = 'sending' WHERE gateway_id = ?1 AND client_uuid = ?2",
            bindings: [self.gatewayID, id])
        // Zero changed rows means the command was deleted while this claim
        // was queued; the caller must not send it.
        return updated && sqlite3_changes(db) > 0
    }

    public func markCommandQueued(id: String, retryCount: Int, lastError: String?) async {
        await self.updateCommandStatus(id: id, status: "queued", retryCount: retryCount, lastError: lastError)
    }

    public func markCommandFailed(id: String, retryCount: Int, lastError: String?) async {
        await self.updateCommandStatus(id: id, status: "failed", retryCount: retryCount, lastError: lastError)
    }

    public func markCommandRetried(id: String) async {
        guard !self.isRetired, let db = await self.handle() else { return }
        // Fresh createdAt: without it the staleness gate would immediately
        // re-expire a retried row that sat offline past the 48h bound.
        self.execute(
            db,
            sql: """
            UPDATE outbox_commands SET status = 'queued', retry_count = 0, last_error = '', created_at = ?3
            WHERE gateway_id = ?1 AND client_uuid = ?2
            """,
            bindings: [self.gatewayID, id, Date().timeIntervalSince1970])
    }

    public func deleteCommand(id: String) async {
        guard !self.isRetired, let db = await self.handle() else { return }
        self.execute(
            db,
            sql: "DELETE FROM outbox_commands WHERE gateway_id = ?1 AND client_uuid = ?2",
            bindings: [self.gatewayID, id])
    }

    private func updateCommandStatus(id: String, status: String, retryCount: Int, lastError: String?) async {
        guard !self.isRetired, let db = await self.handle() else { return }
        self.execute(
            db,
            sql: """
            UPDATE outbox_commands SET status = ?3, retry_count = ?4, last_error = ?5
            WHERE gateway_id = ?1 AND client_uuid = ?2
            """,
            bindings: [self.gatewayID, id, status, retryCount, lastError ?? ""])
    }

    // MARK: - Cached shapes

    /// Text rows only in v1: strip attachment/binary payloads and tool
    /// arguments so the cache never persists base64 blobs or large payloads.
    static func cacheableMessages(_ messages: [OpenClawChatMessage]) -> [OpenClawChatMessage] {
        messages.suffix(self.maxCachedMessagesPerSession).map { message in
            OpenClawChatMessage(
                id: message.id,
                role: message.role,
                content: message.content.map { item in
                    OpenClawChatMessageContent(
                        type: item.type,
                        text: item.text,
                        thinking: item.thinking,
                        thinkingSignature: nil,
                        mimeType: item.mimeType,
                        fileName: item.fileName,
                        content: nil,
                        id: item.id,
                        name: item.name,
                        arguments: nil)
                },
                timestamp: message.timestamp,
                idempotencyKey: message.idempotencyKey,
                toolCallId: message.toolCallId,
                toolName: message.toolName,
                usage: message.usage,
                stopReason: message.stopReason,
                errorMessage: message.errorMessage)
        }
    }

    static func boundedSessions(_ sessions: [OpenClawChatSessionEntry]) -> [OpenClawChatSessionEntry] {
        guard sessions.count > self.maxCachedSessions else { return sessions }
        return Array(
            sessions
                .sorted { ($0.updatedAt ?? 0) > ($1.updatedAt ?? 0) }
                .prefix(self.maxCachedSessions))
    }

    private static func encodeJSON(_ value: some Encodable) -> String? {
        guard let data = try? JSONEncoder().encode(value) else { return nil }
        return String(bytes: data, encoding: .utf8)
    }

    // MARK: - Connection lifecycle

    private func handle() async -> OpaquePointer? {
        guard !self.isRetired else { return nil }
        if let db { return db.raw }
        if self.isBroken { return nil }
        #if os(iOS)
        // Complete protection intentionally makes the cache unavailable while
        // locked. Treat that as a temporary miss, never as corruption.
        guard await self.isProtectedDataAvailable(), !self.isRetired else { return nil }
        #endif
        if let opened = self.openConnection() {
            self.db = Connection(raw: opened)
            return opened
        }
        #if os(iOS)
        guard await self.isProtectedDataAvailable(), !self.isRetired else { return nil }
        #endif
        // Cache is disposable: on any open/schema failure drop the file
        // (and SQLite sidecars) and rebuild once, silently.
        self.removeDatabaseFiles()
        if let reopened = self.openConnection() {
            self.db = Connection(raw: reopened)
            return reopened
        }
        cacheLogger.error("chat transcript cache unavailable; continuing without offline cache")
        self.isBroken = true
        return nil
    }

    #if os(iOS)
    private func isProtectedDataAvailable() async -> Bool {
        await MainActor.run { UIApplication.shared.isProtectedDataAvailable }
    }
    #endif

    private func openConnection() -> OpaquePointer? {
        let fm = FileManager.default
        try? fm.createDirectory(
            at: self.databaseURL.deletingLastPathComponent(),
            withIntermediateDirectories: true)
        var opened: OpaquePointer?
        var flags = SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX
        #if os(iOS)
        // Apply Complete protection through the SQLite VFS so auxiliary files
        // receive the same class as the main transcript database.
        flags |= SQLITE_OPEN_FILEPROTECTION_COMPLETE
        #endif
        guard sqlite3_open_v2(self.databaseURL.path, &opened, flags, nil) == SQLITE_OK, let opened else {
            sqlite3_close_v2(opened)
            return nil
        }
        guard let version = self.readUserVersion(opened) else {
            sqlite3_close_v2(opened)
            return nil
        }
        if version == 0 {
            guard self.createSchema(opened) else {
                sqlite3_close_v2(opened)
                return nil
            }
        } else if version != Self.schemaVersion {
            // Unknown schema: no migrations by design, force drop-and-rebuild.
            sqlite3_close_v2(opened)
            return nil
        }
        #if os(iOS)
        // Upgrade a database created by an older build to the stricter class.
        try? fm.setAttributes(
            [.protectionKey: FileProtectionType.complete],
            ofItemAtPath: self.databaseURL.path)
        #endif
        return opened
    }

    private func readUserVersion(_ db: OpaquePointer) -> Int32? {
        var statement: OpaquePointer?
        guard sqlite3_prepare_v2(db, "PRAGMA user_version", -1, &statement, nil) == SQLITE_OK else {
            return nil
        }
        defer { sqlite3_finalize(statement) }
        guard sqlite3_step(statement) == SQLITE_ROW else { return nil }
        return sqlite3_column_int(statement, 0)
    }

    private func createSchema(_ db: OpaquePointer) -> Bool {
        let statements = [
            """
            CREATE TABLE IF NOT EXISTS cached_sessions(
                gateway_id TEXT NOT NULL PRIMARY KEY,
                payload TEXT NOT NULL,
                updated_at REAL NOT NULL
            )
            """,
            """
            CREATE TABLE IF NOT EXISTS cached_transcripts(
                gateway_id TEXT NOT NULL,
                session_key TEXT NOT NULL,
                payload TEXT NOT NULL,
                updated_at REAL NOT NULL,
                PRIMARY KEY(gateway_id, session_key)
            )
            """,
            // last_error uses '' for "none" so every column binds non-null.
            // rowid `id` breaks created_at ties so flush order stays stable.
            """
            CREATE TABLE IF NOT EXISTS outbox_commands(
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                client_uuid TEXT NOT NULL UNIQUE,
                gateway_id TEXT NOT NULL,
                session_key TEXT NOT NULL,
                text TEXT NOT NULL,
                thinking TEXT NOT NULL DEFAULT '',
                created_at REAL NOT NULL,
                status TEXT NOT NULL,
                retry_count INTEGER NOT NULL DEFAULT 0,
                last_error TEXT NOT NULL DEFAULT ''
            )
            """,
            "PRAGMA user_version = \(Self.schemaVersion)",
        ]
        for sql in statements {
            guard sqlite3_exec(db, sql, nil, nil, nil) == SQLITE_OK else { return false }
        }
        return true
    }

    // MARK: - Statement helpers

    @discardableResult
    private func execute(_ db: OpaquePointer, sql: String, bindings: [Any]) -> Bool {
        var statement: OpaquePointer?
        guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else {
            cacheLogger.error("cache statement prepare failed")
            return false
        }
        defer { sqlite3_finalize(statement) }
        guard self.bind(statement, bindings: bindings) else { return false }
        return sqlite3_step(statement) == SQLITE_DONE
    }

    private func selectInt(_ db: OpaquePointer, sql: String, bindings: [Any]) -> Int? {
        var statement: OpaquePointer?
        guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else { return nil }
        defer { sqlite3_finalize(statement) }
        guard self.bind(statement, bindings: bindings) else { return nil }
        guard sqlite3_step(statement) == SQLITE_ROW else { return nil }
        return Int(sqlite3_column_int64(statement, 0))
    }

    private func selectPayload(_ db: OpaquePointer, sql: String, bindings: [Any]) -> String? {
        var statement: OpaquePointer?
        guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else { return nil }
        defer { sqlite3_finalize(statement) }
        guard self.bind(statement, bindings: bindings) else { return nil }
        guard sqlite3_step(statement) == SQLITE_ROW else { return nil }
        guard let text = sqlite3_column_text(statement, 0) else { return nil }
        return String(cString: text)
    }

    private func bind(_ statement: OpaquePointer?, bindings: [Any]) -> Bool {
        // SQLITE_TRANSIENT: sqlite copies the buffer before the Swift string dies.
        let transient = unsafeBitCast(-1, to: sqlite3_destructor_type.self)
        for (offset, value) in bindings.enumerated() {
            let index = Int32(offset + 1)
            let result: Int32 = switch value {
            case let text as String:
                sqlite3_bind_text(statement, index, text, -1, transient)
            case let int as Int:
                sqlite3_bind_int64(statement, index, Int64(int))
            case let real as Double:
                sqlite3_bind_double(statement, index, real)
            default:
                SQLITE_MISUSE
            }
            guard result == SQLITE_OK else { return false }
        }
        return true
    }

    private func removeDatabaseFiles() {
        self.db = nil
        Self.removeDatabaseFiles(at: self.databaseURL)
    }
}