remove :storage-ksqlite (conversation/message) and related tests; decouple schema from journal

2026-09-22 16:01:05 +03:00
parent 639c7d1748
commit 84f5fd84f3
39 changed files with 1569 additions and 616 deletions
@@ -14,31 +14,67 @@ import kotlinx.coroutines.withContext
 /**
 * ksqlite-реализация [MutableJournalStore] (append-only audit log).
 *
- * Структура — копия [pw.binom.agentik.storage.ksqlite.KsqliteMessageStore]
- * из `:storage-ksqlite`, но:
- *   - лежит в собственном модуле `:journal-ksqlite`;
- *   - реализует переименованный [MutableJournalStore] (раньше был
- *     `MutableMessageStore`, теперь главный класс — `JournalStore` /
- *     `MutableJournalStore`); сам тип записи [MessageRecord] не
- *     переименовывался.
+ * Миграция завершена: это ЕДИНСТВЕННЫЙ класс для message-таблицы. Параллельная
+ * копия `:storage-ksqlite/KsqliteMessageStore.kt` удалена вместе со своим
+ * тестом (consumer `KsqliteStores.assemble()` уже мигрировал на этот класс).
+ *
+ * ## Lifecycle соединения
+ *
+ * Три формы конструктора с разной семантикой владения:
+ *  - `KsqliteJournalStore(connection)` — внешнее соединение, store НЕ закрывает
+ *    его в [close]. Для shared-connection bundles (`KsqliteStores.assemble`),
+ *    где один connection используется многими store'ами и закрывается bundle'ом.
+ *  - `KsqliteJournalStore(path)` — открывает файловое соединение, закрывает
+ *    его в [close].
+ *  - `KsqliteJournalStore.memory(name)` — открывает in-memory соединение,
+ *    закрывает его в [close].
+ *
+ * ## Миграция
+ *
+ * [Schema.migrate] прогоняется ВСЕГДА при конструировании — это idempotent
+ * (CREATE TABLE / INDEX IF NOT EXISTS), так что лишних эффектов нет ни в
+ * standalone-форме, ни в shared-connection bundle'е, где несколько store'ов
+ * прогоняют миграцию одной и той же схемы по очереди.
 *
 * Prepared statements (insert / list / clear) препарируются один раз в
- * конструкторе и закрываются в [close]. Без этого GC финалайзеры каждого
- * StmtHolder'а пытаются `sqlite3_finalize` stmt, чей parent connection уже
- * закрыт → SIGSEGV в `pthread_mutex_lock` (см. [pw.binom.db.ksqlite.StmtHolder]).
+ * конструкторе и закрываются в [close] ДО закрытия owned connection. Без этого
+ * GC финалайзеры каждого StmtHolder'а пытаются `sqlite3_finalize` stmt, чей
+ * parent connection уже закрыт → SIGSEGV в `pthread_mutex_lock`
+ * (см. [pw.binom.db.ksqlite.StmtHolder]).
 *
 * `payloadJson` хранит JSON-сериализованные kind-specific поля. encoding
 * helpers (`encodeRecord` / `toMessageRecord` / `CallPayload` / ...) лежат
 * в [MessageCodecs.kt] рядом.
- *
- * ВНИМАНИЕ: `:storage-ksqlite/KsqliteMessageStore.kt` остаётся на диске —
- * это копия, не замена. Не удалять старый файл; миграция consumers'ов —
- * отдельно.
 */
-class KsqliteJournalStore internal constructor(
+class KsqliteJournalStore private constructor(
    private val connection: SQLiteConnection,
+    private val ownsConnection: Boolean,
 ) : MutableJournalStore {

+    /**
+     * Открывает файловое соединение через [SQLiteConnection.open] и берёт на
+     * себя его закрытие в [close]. Для standalone использования, когда у
+     * store'а нет bundle'а-владельца connection'а.
+     */
+    constructor(path: String) : this(
+        connection = SQLiteConnection.open(path = path),
+        ownsConnection = true,
+    )
+
+    /**
+     * Внешнее соединение — store НЕ закрывает его в [close]. Для
+     * shared-connection bundles (`KsqliteStores.assemble`), где один
+     * connection используется многими store'ами и закрывается bundle'ом.
+     */
+    constructor(connection: SQLiteConnection) : this(
+        connection = connection,
+        ownsConnection = false,
+    )
+
+    init {
+        Schema.migrate(connection)
+    }
+
    private val mutex = Mutex()
    private val json = Json { ignoreUnknownKeys = true }

@@ -94,13 +130,15 @@ class KsqliteJournalStore internal constructor(
            listStmt.bindLong(4, offset.toLong())
            val out = mutableListOf<MessageRecord>()
            listStmt.executeQuery().use { rs ->
-                while (rs.next()) out.add(rs.toMessageRecord(json))
+                while (rs.next()) {
+                    out.add(rs.toMessageRecord(json))
+                }
            }
            out
        }
    }

-    internal suspend fun clear(conversationId: String): Unit = withContext(Dispatchers.Default) {
+    override suspend fun clear(conversationId: String): Unit = withContext(Dispatchers.Default) {
        mutex.withLock {
            clearStmt.reset()
            clearStmt.clearBindings()
@@ -113,5 +151,21 @@ class KsqliteJournalStore internal constructor(
        insertStmt.close()
        listStmt.close()
        clearStmt.close()
+        if (ownsConnection) {
+            connection.close()
+        }
+    }
+
+    companion object {
+        /**
+         * Открывает in-memory соединение через [SQLiteConnection.memory] и
+         * берёт на себя его закрытие в [close]. Удобно для тестов и ephemeral
+         * runtime.
+         */
+        fun memory(name: String? = null) =
+            KsqliteJournalStore(
+                connection = SQLiteConnection.memory(name),
+                ownsConnection = true,
+            )
    }
 }
@@ -0,0 +1,251 @@
+package pw.binom.agentik.journal.ksqlite
+
+import kotlin.time.Clock
+import kotlin.time.Instant
+import pw.binom.agentik.journal.ConversationRecord
+import pw.binom.agentik.journal.MutableConversationStore
+import pw.binom.db.ksqlite.SQLiteConnection
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
+
+/**
+ * ksqlite-реализация [MutableConversationStore]. Схема таблицы `conversation` живёт
+ * в [Schema] (миграция через PRAGMA user_version) — этот класс только
+ * готовит и выполняет SQL, ссылаясь на `Schema.COL_*` / `Schema.TABLE_*`.
+ *
+ * Каскадное удаление связанных данных (message + working_memory) делает
+ * владелец lifecycle диалога (см. ChatAgent.deleteConversation) — этот
+ * store знает только про свою таблицу.
+ *
+ * ## Lifecycle соединения
+ *
+ * Семантика владения connection'ом идентична [KsqliteJournalStore]:
+ *  - `KsqliteMutableConversationStore(connection)` — внешнее соединение,
+ *    store НЕ закрывает его в [close] (используется shared-connection
+ *    bundle'ом `KsqliteStores.assemble`).
+ *  - `KsqliteMutableConversationStore(path)` — открывает файловое соединение,
+ *    закрывает его в [close].
+ *  - `KsqliteMutableConversationStore.memory(name)` — in-memory, закрывает
+ *    в [close].
+ */
+class KsqliteMutableConversationStore private constructor(
+    private val connection: SQLiteConnection,
+    private val ownsConnection: Boolean,
+) : MutableConversationStore {
+
+    /**
+     * Открывает файловое соединение через [SQLiteConnection.open] и берёт на
+     * себя его закрытие в [close]. Для standalone использования, когда у
+     * store'а нет bundle'а-владельца connection'а.
+     */
+    constructor(path: String) : this(
+        connection = SQLiteConnection.open(path = path),
+        ownsConnection = true,
+    )
+
+    /**
+     * Внешнее соединение — store НЕ закрывает его в [close]. Для
+     * shared-connection bundles (`KsqliteStores.assemble`).
+     */
+    constructor(connection: SQLiteConnection) : this(
+        connection = connection,
+        ownsConnection = false,
+    )
+
+    init {
+        Schema.migrate(connection)
+    }
+
+    private val mutex = Mutex()
+
+    // pre-prepare всех statement'ов — аналогично KsqliteMessageStore (см.
+    // KDoc там — почему GC-finalize на StmtHolder'е роняет JVM, если stmt
+    // живёт после закрытия connection).
+    private val existsStmt = connection.prepare(
+        "SELECT 1 FROM ${Schema.TABLE_CONVERSATION} WHERE ${Schema.COL_ID} = ?"
+    )
+    private val updateStmt = connection.prepare(
+        """
+        UPDATE ${Schema.TABLE_CONVERSATION}
+        SET ${Schema.COL_TITLE} = ?, ${Schema.COL_IS_TEMPORAL} = ?, ${Schema.COL_UPDATED_AT} = ?
+        WHERE ${Schema.COL_ID} = ?
+        """.trimIndent()
+    )
+    private val insertStmt = connection.prepare(
+        """
+        INSERT INTO ${Schema.TABLE_CONVERSATION}
+            (${Schema.COL_ID}, ${Schema.COL_TITLE}, ${Schema.COL_IS_TEMPORAL},
+             ${Schema.COL_CREATED_AT}, ${Schema.COL_UPDATED_AT})
+        VALUES (?, ?, ?, ?, ?)
+        """.trimIndent()
+    )
+    private val getStmt = connection.prepare(
+        """
+        SELECT ${Schema.COL_ID}, ${Schema.COL_TITLE}, ${Schema.COL_IS_TEMPORAL},
+               ${Schema.COL_CREATED_AT}, ${Schema.COL_UPDATED_AT}
+        FROM ${Schema.TABLE_CONVERSATION}
+        WHERE ${Schema.COL_ID} = ?
+        """.trimIndent()
+    )
+    private val deleteStmt = connection.prepare(
+        "DELETE FROM ${Schema.TABLE_CONVERSATION} WHERE ${Schema.COL_ID} = ?"
+    )
+    private val listStmt = connection.prepare(
+        """
+        SELECT ${Schema.COL_ID}, ${Schema.COL_TITLE}, ${Schema.COL_IS_TEMPORAL},
+               ${Schema.COL_CREATED_AT}, ${Schema.COL_UPDATED_AT}
+        FROM ${Schema.TABLE_CONVERSATION}
+        WHERE ${Schema.COL_IS_TEMPORAL} = 0
+        ORDER BY ${Schema.COL_UPDATED_AT} DESC
+        LIMIT ? OFFSET ?
+        """.trimIndent()
+    )
+    private val renameStmt = connection.prepare(
+        """
+        UPDATE ${Schema.TABLE_CONVERSATION}
+        SET ${Schema.COL_TITLE} = ?, ${Schema.COL_UPDATED_AT} = ?
+        WHERE ${Schema.COL_ID} = ?
+        """.trimIndent()
+    )
+    private val renameUpdatedAtStmt = connection.prepare(
+        "SELECT ${Schema.COL_UPDATED_AT} FROM ${Schema.TABLE_CONVERSATION} WHERE ${Schema.COL_ID} = ?"
+    )
+    private val touchStmt = connection.prepare(
+        """
+        UPDATE ${Schema.TABLE_CONVERSATION}
+        SET ${Schema.COL_UPDATED_AT} = ?
+        WHERE ${Schema.COL_ID} = ?
+        """.trimIndent()
+    )
+
+    override suspend fun upsert(record: ConversationRecord): Unit = withContext(Dispatchers.Default) {
+        mutex.withLock {
+            val exists = execExists(record.id)
+            if (exists) {
+                updateStmt.reset()
+                updateStmt.clearBindings()
+                val t = record.title
+                if (t != null) updateStmt.bindText(1, t) else updateStmt.bindNull(1)
+                updateStmt.bindInt(2, if (record.isTemporal) 1 else 0)
+                updateStmt.bindLong(3, record.updatedAt.toEpochMilliseconds())
+                updateStmt.bindText(4, record.id)
+                updateStmt.executeUpdate()
+            } else {
+                insertStmt.reset()
+                insertStmt.clearBindings()
+                insertStmt.bindText(1, record.id)
+                val title = record.title
+                if (title != null) insertStmt.bindText(2, title) else insertStmt.bindNull(2)
+                insertStmt.bindInt(3, if (record.isTemporal) 1 else 0)
+                insertStmt.bindLong(4, record.createdAt.toEpochMilliseconds())
+                insertStmt.bindLong(5, record.updatedAt.toEpochMilliseconds())
+                insertStmt.executeUpdate()
+            }
+        }
+    }
+
+    override suspend fun get(id: String): ConversationRecord? = withContext(Dispatchers.Default) {
+        mutex.withLock {
+            getStmt.reset()
+            getStmt.clearBindings()
+            getStmt.bindText(1, id)
+            getStmt.executeQuery().use { rs ->
+                if (rs.next()) rs.toRecord() else null
+            }
+        }
+    }
+
+    override suspend fun delete(id: String): Boolean = withContext(Dispatchers.Default) {
+        mutex.withLock {
+            // Проверяем существование через raw query, НЕ через get() — get() тоже
+            // берёт mutex (не реентрант), что привело бы к deadlock.
+            if (!execExists(id)) return@withContext false
+            deleteStmt.reset()
+            deleteStmt.clearBindings()
+            deleteStmt.bindText(1, id)
+            deleteStmt.executeUpdate()
+            true
+        }
+    }
+
+    override suspend fun list(offset: Int, limit: Int): List<ConversationRecord> = withContext(Dispatchers.Default) {
+        mutex.withLock {
+            listStmt.reset()
+            listStmt.clearBindings()
+            listStmt.bindLong(1, limit.toLong())
+            listStmt.bindLong(2, offset.toLong())
+            val result = mutableListOf<ConversationRecord>()
+            listStmt.executeQuery().use { rs ->
+                while (rs.next()) result.add(rs.toRecord())
+            }
+            result
+        }
+    }
+
+    override suspend fun rename(id: String, title: String?): Instant? = withContext(Dispatchers.Default) {
+        mutex.withLock {
+            val nowMs = Clock.System.now().toEpochMilliseconds()
+            renameStmt.reset()
+            renameStmt.clearBindings()
+            if (title != null) renameStmt.bindText(1, title) else renameStmt.bindNull(1)
+            renameStmt.bindLong(2, nowMs)
+            renameStmt.bindText(3, id)
+            renameStmt.executeUpdate()
+
+            renameUpdatedAtStmt.reset()
+            renameUpdatedAtStmt.clearBindings()
+            renameUpdatedAtStmt.bindText(1, id)
+            renameUpdatedAtStmt.executeQuery().use { rs ->
+                if (rs.next()) Instant.fromEpochMilliseconds(rs.getLong(0)!!) else null
+            }
+        }
+    }
+
+    override suspend fun touch(id: String, now: Instant): Unit = withContext(Dispatchers.Default) {
+        mutex.withLock {
+            touchStmt.reset()
+            touchStmt.clearBindings()
+            touchStmt.bindLong(1, now.toEpochMilliseconds())
+            touchStmt.bindText(2, id)
+            touchStmt.executeUpdate()
+        }
+    }
+
+    override fun close() {
+        existsStmt.close()
+        updateStmt.close()
+        insertStmt.close()
+        getStmt.close()
+        deleteStmt.close()
+        listStmt.close()
+        renameStmt.close()
+        renameUpdatedAtStmt.close()
+        touchStmt.close()
+        if (ownsConnection) connection.close()
+    }
+
+    companion object {
+        fun memory(name: String? = null): KsqliteMutableConversationStore =
+            KsqliteMutableConversationStore(
+                connection = SQLiteConnection.memory(name),
+                ownsConnection = true,
+            )
+    }
+
+    private fun execExists(id: String): Boolean {
+        existsStmt.reset()
+        existsStmt.clearBindings()
+        existsStmt.bindText(1, id)
+        existsStmt.executeQuery().use { rs -> return rs.next() }
+    }
+
+    private fun pw.binom.db.ksqlite.SQLiteResultSet.toRecord(): ConversationRecord = ConversationRecord(
+        id = getText(0)!!,
+        title = getText(1),
+        isTemporal = (getInt(2) ?: 0) != 0,
+        createdAt = Instant.fromEpochMilliseconds(getLong(3)!!),
+        updatedAt = Instant.fromEpochMilliseconds(getLong(4)!!),
+    )
+}
@@ -5,32 +5,51 @@ import pw.binom.db.ksqlite.SQLiteConnection
 /**
 * Имена таблиц/колонок/индексов для ksqlite-бэкенда `:journal-api`.
 *
- * Минимум — только то, что относится к `message` (append-only audit log).
- * Остальные таблицы агента (`conversation`, `working_memory`, `reflection`)
- * живут в других ksqlite-модулях.
+ * Владеет двумя таблицами:
+ *  - `conversation` — реестр диалогов агента (см. ConversationRecord);
+ *  - `message` — append-only audit log сообщений диалогов.
+ *
+ * `working_memory` и `reflection` живут в других ksqlite-модулях.
 *
 * Все DDL/DML в этом модуле должны ссылаться на эти константы — никаких
 * хардкоженных литералов в `prepare("SELECT ... FROM foo ...")` в store'е.
 */
-internal object Schema {
+object Schema {

    /** Версия схемы модуля. Увеличивать при ЛЮБОМ изменении DDL. */
    const val CURRENT_VERSION: Int = 1

-    // ───── Таблица ─────
+    // ───── Таблицы ─────
+    const val TABLE_CONVERSATION = "conversation"
    const val TABLE_MESSAGE = "message"

-    // ───── Колонки ─────
+    // ───── Колонки conversation ─────
    const val COL_ID = "id"
+    const val COL_TITLE = "title"
+    const val COL_IS_TEMPORAL = "is_temporal"
+    const val COL_CREATED_AT = "created_at"
+    const val COL_UPDATED_AT = "updated_at"
+
+    // ───── Колонки message ─────
    const val COL_CONVERSATION_ID = "conversation_id"
    const val COL_KIND = "kind"
    const val COL_PAYLOAD_JSON = "payload_json"
-    const val COL_CREATED_AT = "created_at"

    // ───── Индексы ─────
+    const val IDX_CONV_UPDATED = "idx_conv_updated"
    const val IDX_MSG_CONV = "idx_msg_conv"

-    private val v1Ddl = """
+    private val v1ConversationDdl = """
+        CREATE TABLE IF NOT EXISTS $TABLE_CONVERSATION (
+            $COL_ID          TEXT NOT NULL PRIMARY KEY,
+            $COL_TITLE       TEXT,
+            $COL_IS_TEMPORAL INTEGER NOT NULL DEFAULT 0,
+            $COL_CREATED_AT  INTEGER NOT NULL,
+            $COL_UPDATED_AT  INTEGER NOT NULL
+        );
+    """
+
+    private val v1MessageDdl = """
        CREATE TABLE IF NOT EXISTS $TABLE_MESSAGE (
            $COL_ID              TEXT NOT NULL PRIMARY KEY,
            $COL_CONVERSATION_ID TEXT NOT NULL,
@@ -38,54 +57,43 @@ internal object Schema {
            $COL_PAYLOAD_JSON    TEXT NOT NULL,
            $COL_CREATED_AT      INTEGER NOT NULL
        );
-    """.trimIndent()
+    """

    private val v1IndexesDdl = """
+        CREATE INDEX IF NOT EXISTS $IDX_CONV_UPDATED
+            ON $TABLE_CONVERSATION($COL_UPDATED_AT DESC);
+
+        -- Главный hot-path индекс для list/сообщений: фильтр по conv +
+        -- сортировка по created_at (используется list(), cascade-clear, etc.)
        CREATE INDEX IF NOT EXISTS $IDX_MSG_CONV
            ON $TABLE_MESSAGE($COL_CONVERSATION_ID, $COL_CREATED_AT);
-    """.trimIndent()
+    """

    /**
     * Прогоняет миграцию схемы до [CURRENT_VERSION] на пустой или существующей БД.
     *
-     * Версия хранится в `PRAGMA user_version` (стандартный SQLite-механизм,
-     * 32-bit int в заголовке БД — без своей таблицы). Каждая миграция —
-     * блок DDL под номером `fromV+1`, выполняется в транзакции. Если миграция
-     * упадёт посередине — `ROLLBACK` оставит БД на предыдущей версии.
+     * Гарантии:
+     *  - идемпотентность: `CREATE TABLE/INDEX IF NOT EXISTS` — безопасно на
+     *    уже-мигрированной БД;
+     *  - атомарность: каждая миграция в BEGIN/COMMIT — упал посреди →
+     *    ROLLBACK оставит БД консистентной.
     *
-     * Идемпотентен: повторный вызов на уже мигрированной БД — no-op.
+     **NOTE**: в сплит-мире (4 ksqlite-модуля, каждый владеет своей таблицей)
+     * user_version как gate перестал работать — два модуля ставят его в 1,
+     * второй вызов short-circuit'ит. Поэтому migrate() просто прогоняет DDL
+     * idempotently; координация multi-module миграций — ответственность
+     * вызывающего (см. `KsqliteStores.open()` в `:storage-ksqlite`).
     */
    fun migrate(conn: SQLiteConnection) {
-        val current = readUserVersion(conn)
-        if (current >= CURRENT_VERSION) return
-
        conn.exec("BEGIN")
        try {
-            if (current < 1) {
-                conn.exec(v1Ddl)
-                conn.exec(v1IndexesDdl)
-            }
-            // future: if (current < 2) { conn.exec(v2Ddl) }
-            writeUserVersion(conn, CURRENT_VERSION)
+            conn.exec(v1ConversationDdl)
+            conn.exec(v1MessageDdl)
+            conn.exec(v1IndexesDdl)
            conn.exec("COMMIT")
        } catch (t: Throwable) {
            runCatching { conn.exec("ROLLBACK") }
            throw t
        }
    }
-
-    private fun readUserVersion(conn: SQLiteConnection): Int {
-        conn.prepare("PRAGMA user_version").use { stmt ->
-            stmt.executeQuery().use { rs ->
-                if (rs.next()) return rs.getLong(0)?.toInt() ?: 0
-            }
-        }
-        return 0
-    }
-
-    private fun writeUserVersion(conn: SQLiteConnection, version: Int) {
-        // SQLite PRAGMA с literal-аргументом нельзя параметризовать через `?`,
-        // поэтому собираем SQL строкой (значение контролируемое, не user input).
-        conn.exec("PRAGMA user_version = $version")
-    }
 }
@@ -13,15 +13,9 @@ import kotlin.test.assertEquals
 import kotlin.time.Instant

 /**
- * Тесты для [KsqliteJournalStore] — точная копия
- * `KsqliteMessageStoreTest` из `:storage-ksqlite`, с переименованием типов
- * (`MessageStore` → `JournalStore`) и автономной фикстурой (in-memory
- * SQLiteConnection + Schema.migrate).
- *
- * Тест `testClearRemovesByConversation` из оригинала использовал
- * `stores.conversations.delete(...)` (cascade через `KsqliteStores`) — здесь
- * он заменён на прямой вызов `store.clear(...)`, потому что `:journal-ksqlite`
- * автономен и не знает про ConversationStore.
+ * Тесты для [KsqliteJournalStore]. Автономная фикстура: in-memory
+ * SQLiteConnection + конструктор `KsqliteJournalStore(connection)` — store сам
+ * прогоняет `Schema.migrate` в init, явный вызов не нужен.
 */
 class KsqliteJournalStoreTest {

@@ -31,7 +25,6 @@ class KsqliteJournalStoreTest {
    @BeforeTest
    fun setup() {
        conn = SQLiteConnection.memory("journal-${kotlin.random.Random.nextLong()}")
-        Schema.migrate(conn)
        store = KsqliteJournalStore(conn)
    }