diff --git a/README.md b/README.md index 85594de..a5928e6 100644 --- a/README.md +++ b/README.md @@ -162,8 +162,10 @@ This package supports optional remote storage sync to a PostgreSQL database: ```bash export HASNA_CONVERSATIONS_DATABASE_URL="" conversations storage status +conversations storage readiness conversations storage push conversations storage pull +conversations storage sync --tables cloud-runtime ``` Production storage for Hasna XYZ uses the `conversations` database on @@ -179,6 +181,25 @@ central rollback window closes. By default, sync only includes text-key/global tables to avoid local integer ID collisions across machines. +The default table group is `metadata`: projects, channels, channel membership, +channel subscriptions, presence, locks, graph edges, and feedback. + +Use `conversations storage readiness --json` to inspect the explicit runtime +path before any remote operation. CLI and MCP commands use local SQLite as the +live store by default. `--tables cloud-runtime` is an opt-in sync group that +adds `messages`, `message_read_receipts`, `channel_notification_reads`, +`message_mentions`, and `reactions`. Message rows are upserted by UUID, and +dependent read-state rows are translated through message UUIDs so local SQLite +integer ids do not become cross-machine identifiers. + +Digest cursors and local search continue to use SQLite message ids and FTS5. +The PostgreSQL migration creates a `tsvector` search index for remote readiness, +but live query commands do not switch to remote search unless a remote query +adapter is selected. Attachment files remain local under the conversations +attachments directory; `cloud-runtime` sync omits local attachment metadata and +file paths until an object-store migration exists. S3/AWS object storage, +production writes, secret creation, data migration, terraform apply, or +spend-affecting changes require a separate approval task. ## Data Directory diff --git a/src/cli/storage-contract.test.ts b/src/cli/storage-contract.test.ts index b1eb038..ded795b 100644 --- a/src/cli/storage-contract.test.ts +++ b/src/cli/storage-contract.test.ts @@ -30,6 +30,11 @@ describe("conversations storage CLI contract", () => { fallbackEnv: string; }; configured: boolean; + table_groups: { + cloudRuntime: string[]; + localOnly: string[]; + }; + message_uuid_duplicates: Array<{ uuid: string; count: number }>; }; expect(info.canonical).toEqual({ @@ -40,6 +45,35 @@ describe("conversations storage CLI contract", () => { fallbackEnv: "CONVERSATIONS_DATABASE_URL", }); expect(info.configured).toBe(false); + expect(info.table_groups.cloudRuntime).toContain("messages"); + expect(info.table_groups.cloudRuntime).toContain("message_read_receipts"); + expect(info.table_groups.localOnly).toContain("messages_fts"); + expect(info.message_uuid_duplicates).toEqual([]); + expect(stdout).not.toContain("postgres://"); + }); + + it("reports storage readiness without printing a URL", () => { + const result = Bun.spawnSync({ + cmd: ["bun", "src/cli/index.tsx", "storage", "readiness", "--json"], + cwd: join(import.meta.dir, "../.."), + env: { + ...process.env, + HASNA_CONVERSATIONS_DB_PATH: ":memory:", + CONVERSATIONS_DB_PATH: ":memory:", + HASNA_CONVERSATIONS_STORAGE_MODE: "local", + NO_COLOR: "1", + }, + }); + + expect(result.exitCode).toBe(0); + const stdout = new TextDecoder().decode(result.stdout); + const info = JSON.parse(stdout) as { + runtimePaths: Array<{ surface: string; status: string }>; + privacyAndMigrationGates: string[]; + }; + + expect(info.runtimePaths.find((path) => path.surface === "attachments")?.status).toBe("local_only"); + expect(info.privacyAndMigrationGates.join(" ")).toContain("Never print database URLs"); expect(stdout).not.toContain("postgres://"); }); @@ -66,6 +100,7 @@ describe("conversations storage CLI contract", () => { expect(source).toContain("registerStorageCommands"); expect(source).toContain('program.command("storage")'); + expect(source).toContain('storage.command("readiness")'); expect(source).not.toContain(retiredCloudCommand); expect(source).not.toContain(retiredCloudExport); }); diff --git a/src/cli/storage.ts b/src/cli/storage.ts index 80e3bc0..301d757 100644 --- a/src/cli/storage.ts +++ b/src/cli/storage.ts @@ -6,7 +6,9 @@ import { getStorageConfig, getStorageDatabaseUrl, getStoragePg, + getStorageReadiness, listConflicts, + listDuplicateMessageUuids, runStorageMigrations, storagePull, storagePush, @@ -33,16 +35,22 @@ function installStorageSubcommands(storage: Command): void { storage.command("status").description("Show remote storage config and sync state").option("--json", "Output as JSON").action((opts: { json?: boolean }) => { const config = getStorageConfig(); const canonical = getCanonicalConversationsRdsConfig(); + const readiness = getStorageReadiness(); const local = getDb(); const unresolved = listConflicts(local, { resolved: false }); const resolved = listConflicts(local, { resolved: true }); + const duplicateMessageUuids = listDuplicateMessageUuids(local); const info = { mode: config.mode, configured: Boolean(getStorageDatabaseUrl()), service: "conversations", canonical, tables: DEFAULT_STORAGE_TABLES, + table_groups: readiness.tableGroups, + runtime_paths: readiness.runtimePaths, + privacy_and_migration_gates: readiness.privacyAndMigrationGates, conflicts: { unresolved: unresolved.length, resolved: resolved.length }, + message_uuid_duplicates: duplicateMessageUuids, }; if (opts.json) { printJson(info); return; } console.log(`Mode: ${info.mode}`); @@ -52,8 +60,31 @@ function installStorageSubcommands(storage: Command): void { console.log(`Runtime secret path: ${canonical.runtimeSecretPath}`); console.log(`Database env: ${canonical.env} (fallback: ${canonical.fallbackEnv})`); console.log(`Remote storage configured: ${info.configured ? "yes" : "no"}`); - console.log(`Tables: ${info.tables.join(", ")}`); + console.log(`Default sync group: metadata (${info.tables.join(", ")})`); + console.log(`Cloud runtime group: cloud-runtime (${readiness.tableGroups.cloudRuntime.join(", ")})`); + console.log("Attachments: local files only; S3 object storage is an approval-gated follow-up."); console.log(`Sync conflicts: ${info.conflicts.unresolved} unresolved, ${info.conflicts.resolved} resolved`); + console.log(`Message UUID duplicates: ${duplicateMessageUuids.length}`); + }); + + storage.command("readiness").description("Show local/remote storage runtime readiness and migration gates").option("--json", "Output as JSON").action((opts: { json?: boolean }) => { + const readiness = getStorageReadiness(); + if (opts.json) { printJson(readiness); return; } + console.log(`Mode: ${readiness.mode}`); + console.log(`Remote storage configured: ${readiness.configured ? "yes" : "no"}`); + console.log(`Local SQLite: ${readiness.local.sqlite}`); + console.log(`Local attachments: ${readiness.local.attachments}`); + console.log(`Default sync group: metadata (${readiness.tableGroups.default.join(", ")})`); + console.log(`Cloud runtime group: cloud-runtime (${readiness.tableGroups.cloudRuntime.join(", ")})`); + for (const path of readiness.runtimePaths) { + console.log(`\n${path.surface}: ${path.status}`); + console.log(` local: ${path.local}`); + console.log(` remote: ${path.remote}`); + if (path.tables.length > 0) console.log(` tables: ${path.tables.join(", ")}`); + if (path.gates.length > 0) console.log(` gates: ${path.gates.join("; ")}`); + } + console.log("\nPrivacy and migration gates:"); + for (const gate of readiness.privacyAndMigrationGates) console.log(` - ${gate}`); }); storage.command("push").description("Push safe local conversations tables to remote PostgreSQL storage").option("--tables ", "Comma-separated table names").option("--json", "Output as JSON").action(async (opts: { tables?: string; json?: boolean }) => { diff --git a/src/lib/db.test.ts b/src/lib/db.test.ts index f657bba..f0dd485 100644 --- a/src/lib/db.test.ts +++ b/src/lib/db.test.ts @@ -2,6 +2,7 @@ import { describe, test, expect, beforeEach, afterEach } from "bun:test"; import { getDb, getDbPath, closeDb } from "./db"; import { createChannel, getChannel } from "./channels"; import { sendMessage } from "./messages"; +import { listDuplicateMessageUuids } from "./storage-sync"; import { unlinkSync } from "fs"; import { Database } from "bun:sqlite"; import { tmpdir } from "os"; @@ -95,6 +96,34 @@ describe("db", () => { expect(colNames).toContain("channel"); }); + test("keeps duplicate UUID preflight diagnosable instead of crashing on unique index creation", () => { + closeDb(); + const legacyDb = new Database(TEST_DB); + legacyDb.exec(` + CREATE TABLE messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + from_agent TEXT NOT NULL, + to_agent TEXT NOT NULL, + content TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%f', 'now')), + uuid TEXT + ); + INSERT INTO messages (session_id, from_agent, to_agent, content, uuid) + VALUES + ('dm:a:b', 'a', 'b', 'first', 'duplicate-uuid'), + ('dm:a:b', 'a', 'b', 'second', 'duplicate-uuid'); + `); + legacyDb.close(); + + const db = getDb(); + expect(listDuplicateMessageUuids(db)).toEqual([{ uuid: "duplicate-uuid", count: 2 }]); + const uniqueIndex = db.prepare("SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'idx_messages_uuid'").get(); + expect(uniqueIndex).toBeNull(); + const preflightIndex = db.prepare("SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'idx_messages_uuid_preflight'").get() as { name: string } | null; + expect(preflightIndex?.name).toBe("idx_messages_uuid_preflight"); + }); + test("migrates legacy channel messages before creating channel indexes", () => { closeDb(); const legacyDb = new Database(TEST_DB); diff --git a/src/lib/db.ts b/src/lib/db.ts index bc4e69b..27fb626 100644 --- a/src/lib/db.ts +++ b/src/lib/db.ts @@ -836,7 +836,27 @@ export function getDb(): Database { db.exec("ALTER TABLE messages ADD COLUMN uuid TEXT"); // Backfill existing rows with unique UUIDs db.exec("UPDATE messages SET uuid = lower(hex(randomblob(16))) WHERE uuid IS NULL"); + } + const missingUuidCount = db.get<{ count: number }>( + "SELECT COUNT(*) AS count FROM messages WHERE uuid IS NULL OR uuid = ''" + )?.count ?? 0; + if (missingUuidCount > 0) { + db.exec("UPDATE messages SET uuid = lower(hex(randomblob(16))) WHERE uuid IS NULL OR uuid = ''"); + } + const duplicateUuidCount = db.get<{ count: number }>(` + SELECT COUNT(*) AS count + FROM ( + SELECT uuid + FROM messages + WHERE uuid IS NOT NULL AND uuid <> '' + GROUP BY uuid + HAVING COUNT(*) > 1 + ) + `)?.count ?? 0; + if (duplicateUuidCount === 0) { db.exec("CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_uuid ON messages(uuid)"); + } else { + db.exec("CREATE INDEX IF NOT EXISTS idx_messages_uuid_preflight ON messages(uuid)"); } // Migrate agent_presence: add id, session_id, role, created_at columns diff --git a/src/lib/pg-migrations.test.ts b/src/lib/pg-migrations.test.ts index c59193f..978a3b9 100644 --- a/src/lib/pg-migrations.test.ts +++ b/src/lib/pg-migrations.test.ts @@ -62,8 +62,10 @@ describe("PG_MIGRATIONS", () => { const sql = PG_MIGRATIONS[0].toLowerCase(); const pgcrypto = sql.indexOf("create extension if not exists pgcrypto"); const uuidDefault = sql.indexOf("gen_random_uuid()"); + const uuidIndex = sql.indexOf("idx_messages_uuid"); expect(pgcrypto).toBeGreaterThan(-1); expect(uuidDefault).toBeGreaterThan(pgcrypto); + expect(uuidIndex).toBeGreaterThan(uuidDefault); }); test("legacy PostgreSQL spaces are imported before legacy storage is dropped", () => { diff --git a/src/lib/pg-migrations.ts b/src/lib/pg-migrations.ts index 2cc90f2..1dff407 100644 --- a/src/lib/pg-migrations.ts +++ b/src/lib/pg-migrations.ts @@ -86,6 +86,8 @@ export const PG_MIGRATIONS: string[] = [ read_at TEXT ); ALTER TABLE messages ADD COLUMN IF NOT EXISTS uuid TEXT DEFAULT gen_random_uuid()::text; + UPDATE messages SET uuid = gen_random_uuid()::text WHERE uuid IS NULL OR uuid = ''; + CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_uuid ON messages(uuid); ALTER TABLE messages ADD COLUMN IF NOT EXISTS channel TEXT; ALTER TABLE messages ADD COLUMN IF NOT EXISTS project_id TEXT; ALTER TABLE messages ADD COLUMN IF NOT EXISTS edited_at TEXT; diff --git a/src/lib/storage-sync.test.ts b/src/lib/storage-sync.test.ts index c4513c3..1dbbc07 100644 --- a/src/lib/storage-sync.test.ts +++ b/src/lib/storage-sync.test.ts @@ -5,20 +5,29 @@ import { CANONICAL_CONVERSATIONS_RDS_DATABASE, CANONICAL_CONVERSATIONS_RDS_SECRET_PATH, CONVERSATIONS_DATABASE_FALLBACK_ENV, + CLOUD_RUNTIME_STORAGE_TABLES, DEFAULT_STORAGE_TABLES, + STORAGE_CONFIG_PATH_ENV, STORAGE_DATABASE_ENV, + STORAGE_LOCAL_ONLY_TABLES, STORAGE_MODE_ENV, + detectAndLogConflicts, ensureConflictsTable, getCanonicalConversationsRdsConfig, getStorageConfig, getStorageDatabaseUrl, + getStorageReadiness, listConflicts, + listDuplicateMessageUuids, resolveTables, + syncPull, + syncPush, } from "./storage-sync.js"; import { ConversationsDatabase } from "./db.js"; const ENV_NAMES = [ ...STORAGE_DATABASE_ENV, + ...STORAGE_CONFIG_PATH_ENV, ...STORAGE_MODE_ENV, ["HASNA", "CONVERSATIONS", "CLOUD", "DATABASE", "URL"].join("_"), ["OPEN", "CONVERSATIONS", "CLOUD", "DATABASE", "URL"].join("_"), @@ -26,8 +35,113 @@ const ENV_NAMES = [ ["HASNA", "CONVERSATIONS", "CLOUD", "MODE"].join("_"), ["OPEN", "CONVERSATIONS", "CLOUD", "MODE"].join("_"), ["CONVERSATIONS", "CLOUD", "MODE"].join("_"), + "CONVERSATIONS_ATTACHMENTS_DIR", ] as const; +class SqliteRemoteAdapter { + constructor(private readonly db: ConversationsDatabase) {} + + async run(sql: string, ...params: unknown[]): Promise<{ changes: number }> { + const translated = sql + .replace(/\bGREATEST\(/g, "max(") + .replace(/\bLEAST\(/g, "min("); + return this.db.run(translated, ...params); + } + + async all(sql: string, ...params: unknown[]): Promise { + if (sql.includes("information_schema.columns")) { + const table = String(params[0]); + return this.db.all<{ name: string; type: string }>(`PRAGMA table_info("${table.replace(/"/g, '""')}")`) + .map((row) => ({ + column_name: row.name, + data_type: row.type.toLowerCase().includes("int") ? "integer" : "text", + })); + } + return this.db.all(sql, ...params); + } + + async get(sql: string, ...params: unknown[]): Promise { + return this.db.get(sql, ...params); + } + + async close(): Promise {} +} + +function asRemote(db: ConversationsDatabase) { + return new SqliteRemoteAdapter(db) as any; +} + +function createSyncDb(): ConversationsDatabase { + const db = new ConversationsDatabase(":memory:"); + db.exec(` + CREATE TABLE messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + uuid TEXT NOT NULL UNIQUE, + session_id TEXT NOT NULL, + from_agent TEXT NOT NULL, + to_agent TEXT NOT NULL, + channel TEXT, + content TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT '2026-01-01T00:00:00.000', + read_at TEXT, + reply_to INTEGER, + attachments TEXT + ); + CREATE TABLE channel_subscriptions ( + channel TEXT NOT NULL, + agent TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT '2026-01-01T00:00:00.000', + preview_chars INTEGER NOT NULL DEFAULT 140, + since_message_id INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (channel, agent) + ); + CREATE TABLE message_read_receipts ( + message_id INTEGER NOT NULL, + agent TEXT NOT NULL, + read_at TEXT NOT NULL, + PRIMARY KEY (message_id, agent) + ); + CREATE TABLE channel_notification_reads ( + agent TEXT NOT NULL, + message_id INTEGER NOT NULL, + read_at TEXT NOT NULL, + PRIMARY KEY (agent, message_id) + ); + CREATE TABLE reactions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + message_id INTEGER NOT NULL, + agent TEXT NOT NULL, + emoji TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(message_id, agent, emoji) + ); + CREATE TABLE message_mentions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + message_id INTEGER NOT NULL, + mentioned_agent TEXT NOT NULL, + from_agent TEXT NOT NULL, + channel TEXT, + notified_at TEXT, + created_at TEXT NOT NULL + ); + `); + return db; +} + +function insertMessage( + db: ConversationsDatabase, + uuid: string, + content: string, + options?: { replyTo?: number | null; attachments?: string | null }, +): number { + const row = db.get<{ id: number }>(` + INSERT INTO messages (uuid, session_id, from_agent, to_agent, channel, content, reply_to, attachments) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + RETURNING id + `, uuid, `channel:test`, "alice", "test", "test", content, options?.replyTo ?? null, options?.attachments ?? null); + return row!.id; +} + afterEach(() => { for (const name of ENV_NAMES) { delete process.env[name]; @@ -77,14 +191,50 @@ describe("conversations storage configuration", () => { it("returns default storage tables and rejects unsupported tables", () => { expect(resolveTables()).toEqual([...DEFAULT_STORAGE_TABLES]); + expect(resolveTables("metadata")).toEqual([...DEFAULT_STORAGE_TABLES]); + expect(resolveTables("cloud-runtime")).toEqual([...CLOUD_RUNTIME_STORAGE_TABLES]); + expect(resolveTables("messages,read-state")).toEqual([ + "messages", + "message_read_receipts", + "channel_notification_reads", + "message_mentions", + "reactions", + ]); expect(() => resolveTables("channels,missing")).toThrow("Unsupported conversations storage table"); }); + it("reports readiness without exposing database URL secrets", () => { + process.env["HASNA_CONVERSATIONS_DATABASE_URL"] = "postgres://example.invalid/conversations"; + process.env.CONVERSATIONS_ATTACHMENTS_DIR = "/tmp/custom-conversations-attachments"; + + const readiness = getStorageReadiness(); + + expect(readiness.configured).toBe(true); + expect(readiness.tableGroups.default).toEqual([...DEFAULT_STORAGE_TABLES]); + expect(readiness.tableGroups.cloudRuntime).toEqual([...CLOUD_RUNTIME_STORAGE_TABLES]); + expect(readiness.tableGroups.localOnly).toEqual([...STORAGE_LOCAL_ONLY_TABLES]); + expect(readiness.runtimePaths.map((path) => path.surface)).toEqual([ + "local-sqlite", + "remote-postgres", + "messages-and-sessions", + "read-state", + "search-and-digests", + "attachments", + "aws-production", + ]); + expect(readiness.runtimePaths.find((path) => path.surface === "attachments")?.remote) + .toContain("omits local attachment metadata"); + expect(readiness.local.attachments).toBe("/tmp/custom-conversations-attachments"); + expect(JSON.stringify(readiness)).not.toContain("postgres://"); + }); + it("exports storage helpers from the storage subpath source", async () => { const storage = await import("../storage.js"); expect(storage.DEFAULT_STORAGE_TABLES).toEqual(DEFAULT_STORAGE_TABLES); + expect(storage.CLOUD_RUNTIME_STORAGE_TABLES).toEqual(CLOUD_RUNTIME_STORAGE_TABLES); expect(storage.getStorageDatabaseUrl()).toBeNull(); + expect(storage.getStorageReadiness().tableGroups.cloudRuntime).toContain("messages"); expect(storage.PG_MIGRATIONS.length).toBeGreaterThan(0); expect(typeof storage.PgAdapterAsync).toBe("function"); }); @@ -113,4 +263,130 @@ describe("conversations storage configuration", () => { db.close(); } }); + + it("reports duplicate message UUIDs for migration preflight diagnostics", () => { + const db = new ConversationsDatabase(":memory:"); + try { + db.exec("CREATE TABLE messages (id INTEGER PRIMARY KEY AUTOINCREMENT, uuid TEXT)"); + db.run("INSERT INTO messages (uuid) VALUES (?)", "duplicate"); + db.run("INSERT INTO messages (uuid) VALUES (?)", "duplicate"); + db.run("INSERT INTO messages (uuid) VALUES (?)", "unique"); + + expect(listDuplicateMessageUuids(db)).toEqual([{ uuid: "duplicate", count: 2 }]); + } finally { + db.close(); + } + }); + + it("logs conflicts for composite-key agent presence rows", async () => { + const local = new ConversationsDatabase(":memory:"); + const remote = new ConversationsDatabase(":memory:"); + try { + for (const db of [local, remote]) { + db.exec(` + CREATE TABLE agent_presence ( + id TEXT NOT NULL DEFAULT '', + agent TEXT NOT NULL, + project_id TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'online', + created_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + PRIMARY KEY (agent, project_id) + ) + `); + } + local.run(` + INSERT INTO agent_presence (agent, project_id, status, created_at, last_seen_at) + VALUES (?, ?, ?, ?, ?) + `, "alice", "project-a", "online", "2026-01-01T00:00:00.000Z", "2026-01-01T00:00:00.000Z"); + remote.run(` + INSERT INTO agent_presence (agent, project_id, status, created_at, last_seen_at) + VALUES (?, ?, ?, ?, ?) + `, "alice", "project-a", "away", "2026-01-02T00:00:00.000Z", "2026-01-02T00:00:00.000Z"); + + const count = await detectAndLogConflicts(local, asRemote(remote), "agent_presence"); + const conflicts = listConflicts(local, { resolved: false }); + + expect(count).toBe(1); + expect(conflicts).toHaveLength(1); + expect(conflicts[0]?.pk).toBe("agent=alice|project_id=project-a"); + } finally { + local.close(); + remote.close(); + } + }); + + it("syncs message threads through UUIDs instead of raw reply_to ids", async () => { + const local = createSyncDb(); + const remote = createSyncDb(); + const target = createSyncDb(); + try { + insertMessage(remote, "remote-unrelated", "remote id padding"); + insertMessage(target, "target-unrelated-1", "target id padding 1"); + insertMessage(target, "target-unrelated-2", "target id padding 2"); + + const parentId = insertMessage(local, "parent-uuid", "parent"); + insertMessage(local, "child-uuid", "child", { + replyTo: parentId, + attachments: JSON.stringify([{ path: "/home/hasna/private.txt" }]), + }); + + const pushResult = await syncPush(local, asRemote(remote), { tables: ["messages"] }); + expect(pushResult[0].errors).toEqual([]); + + const remoteParent = remote.get<{ id: number; attachments: string | null }>( + "SELECT id, attachments FROM messages WHERE uuid = ?", + "parent-uuid", + )!; + const remoteChild = remote.get<{ id: number; reply_to: number; attachments: string | null }>( + "SELECT id, reply_to, attachments FROM messages WHERE uuid = ?", + "child-uuid", + )!; + expect(remoteChild.reply_to).toBe(remoteParent.id); + expect(remoteChild.reply_to).not.toBe(parentId); + expect(remoteParent.attachments).toBeNull(); + expect(remoteChild.attachments).toBeNull(); + + const pullResult = await syncPull(asRemote(remote), target, { tables: ["messages"] }); + expect(pullResult[0].errors).toEqual([]); + + const targetParent = target.get<{ id: number }>("SELECT id FROM messages WHERE uuid = ?", "parent-uuid")!; + const targetChild = target.get<{ id: number; reply_to: number }>("SELECT id, reply_to FROM messages WHERE uuid = ?", "child-uuid")!; + expect(targetChild.reply_to).toBe(targetParent.id); + expect(targetChild.reply_to).not.toBe(remoteParent.id); + } finally { + local.close(); + remote.close(); + target.close(); + } + }); + + it("keeps channel subscription cursors local during metadata sync", async () => { + const local = createSyncDb(); + const remote = createSyncDb(); + try { + local.run(` + INSERT INTO channel_subscriptions (channel, agent, preview_chars, since_message_id) + VALUES (?, ?, ?, ?) + `, "ops", "reader", 240, 999); + remote.run(` + INSERT INTO channel_subscriptions (channel, agent, preview_chars, since_message_id) + VALUES (?, ?, ?, ?) + `, "ops", "reader", 140, 3); + + const result = await syncPush(local, asRemote(remote), { tables: ["channel_subscriptions"] }); + expect(result[0].errors).toEqual([]); + + const row = remote.get<{ preview_chars: number; since_message_id: number }>( + "SELECT preview_chars, since_message_id FROM channel_subscriptions WHERE channel = ? AND agent = ?", + "ops", + "reader", + )!; + expect(row.preview_chars).toBe(240); + expect(row.since_message_id).toBe(3); + } finally { + local.close(); + remote.close(); + } + }); }); diff --git a/src/lib/storage-sync.ts b/src/lib/storage-sync.ts index 67be656..ebce14c 100644 --- a/src/lib/storage-sync.ts +++ b/src/lib/storage-sync.ts @@ -2,16 +2,11 @@ import { existsSync, readFileSync } from "fs"; import { homedir } from "os"; import { join } from "path"; import type { Database } from "./db.js"; -import { getDb } from "./db.js"; +import { getDb, getDbPath } from "./db.js"; import { PG_MIGRATIONS } from "./pg-migrations.js"; import { PgAdapterAsync } from "./remote-storage.js"; export const SYNC_EXCLUDED = new Set([ - "messages", - "reactions", - "message_read_receipts", - "message_mentions", - "channel_notification_reads", "tasks", "task_comments", "task_activity", @@ -34,10 +29,42 @@ export const DEFAULT_STORAGE_TABLES = [ "feedback", ] as const; -type SyncTable = (typeof DEFAULT_STORAGE_TABLES)[number]; +export const STORAGE_MESSAGE_TABLES = [ + "messages", + "message_read_receipts", + "channel_notification_reads", + "message_mentions", + "reactions", +] as const; + +export const CLOUD_RUNTIME_STORAGE_TABLES = [ + ...DEFAULT_STORAGE_TABLES, + ...STORAGE_MESSAGE_TABLES, +] as const; + +export const STORAGE_LOCAL_ONLY_TABLES = [ + "messages_fts", + "tasks_fts", + "_sync_conflicts", + "_migrations", + "sqlite_sequence", + "tasks", + "task_comments", + "task_activity", + "task_dependencies", +] as const; + +export const STORAGE_TABLE_GROUPS = { + metadata: DEFAULT_STORAGE_TABLES, + "cloud-runtime": CLOUD_RUNTIME_STORAGE_TABLES, + messages: ["messages"] as const, + "read-state": ["message_read_receipts", "channel_notification_reads", "message_mentions", "reactions"] as const, +} as const; + +type SyncTable = (typeof CLOUD_RUNTIME_STORAGE_TABLES)[number]; type Row = Record; -const PRIMARY_KEYS: Record = { +const PRIMARY_KEYS: Record = { projects: ["id"], channels: ["name"], channel_members: ["channel", "agent"], @@ -46,6 +73,7 @@ const PRIMARY_KEYS: Record = { resource_locks: ["resource_type", "resource_id", "lock_type"], graph_edges: ["from_type", "from_id", "to_type", "to_id", "relation"], feedback: ["id"], + messages: ["uuid"], }; const CONFLICT_TABLES = new Set(["channels", "projects", "agent_presence"]); @@ -82,8 +110,38 @@ export interface CanonicalConversationsRdsConfig { export interface StorageSyncResult { table: string; rowsRead: number; rowsWritten: number; errors: string[]; } export type SyncResult = StorageSyncResult; export interface SyncConflict { table: string; pk: string; local: Row; remote: Row; } +export interface MessageUuidDuplicate { uuid: string; count: number; } +export type StorageRuntimeStatus = "active" | "configured" | "not_configured" | "opt_in" | "local_only" | "manual_gate"; +export interface StorageRuntimePath { + surface: string; + status: StorageRuntimeStatus; + local: string; + remote: string; + tables: string[]; + gates: string[]; +} +export interface StorageReadiness { + mode: StorageMode; + configured: boolean; + canonical: CanonicalConversationsRdsConfig; + local: { + sqlite: string; + attachments: string; + }; + tableGroups: { + default: string[]; + cloudRuntime: string[]; + metadata: string[]; + messages: string[]; + readState: string[]; + localOnly: string[]; + }; + runtimePaths: StorageRuntimePath[]; + privacyAndMigrationGates: string[]; +} export const STORAGE_CONFIG_PATH = join(homedir(), ".hasna", "conversations", "storage", "config.json"); +export const STORAGE_CONFIG_PATH_ENV = ["HASNA_CONVERSATIONS_STORAGE_CONFIG", "CONVERSATIONS_STORAGE_CONFIG"] as const; export const STORAGE_DATABASE_ENV = [CANONICAL_CONVERSATIONS_DATABASE_ENV, CONVERSATIONS_DATABASE_FALLBACK_ENV] as const; export const STORAGE_MODE_ENV = ["HASNA_CONVERSATIONS_STORAGE_MODE", "CONVERSATIONS_STORAGE_MODE"] as const; @@ -97,6 +155,119 @@ export function getCanonicalConversationsRdsConfig(): CanonicalConversationsRdsC }; } +export function getStorageConfigPath(): string { + return firstEnv(STORAGE_CONFIG_PATH_ENV) ?? STORAGE_CONFIG_PATH; +} + +export function getStorageReadiness(): StorageReadiness { + const config = getStorageConfig(); + const configured = Boolean(getStorageDatabaseUrl()); + const canonical = getCanonicalConversationsRdsConfig(); + const attachmentsDir = process.env.CONVERSATIONS_ATTACHMENTS_DIR + || join(homedir(), ".hasna", "conversations", "attachments"); + + return { + mode: config.mode, + configured, + canonical, + local: { + sqlite: getDbPath(), + attachments: attachmentsDir, + }, + tableGroups: { + default: [...DEFAULT_STORAGE_TABLES], + cloudRuntime: [...CLOUD_RUNTIME_STORAGE_TABLES], + metadata: [...DEFAULT_STORAGE_TABLES], + messages: [...STORAGE_TABLE_GROUPS.messages], + readState: [...STORAGE_TABLE_GROUPS["read-state"]], + localOnly: [...STORAGE_LOCAL_ONLY_TABLES], + }, + runtimePaths: [ + { + surface: "local-sqlite", + status: "active", + local: "Primary runtime store for CLI, MCP, digest, search, cursors, and read state.", + remote: "Not used unless a storage command is invoked or a future remote runtime adapter is selected.", + tables: [...CLOUD_RUNTIME_STORAGE_TABLES], + gates: [], + }, + { + surface: "remote-postgres", + status: configured ? "configured" : "not_configured", + local: "Default storage sync keeps local SQLite authoritative for live commands.", + remote: `PostgreSQL sync uses ${canonical.env} or ${canonical.fallbackEnv}; default table group is metadata only.`, + tables: [...DEFAULT_STORAGE_TABLES], + gates: [ + "Set the database URL through the runtime environment or local config without printing the secret value.", + "Run storage migrate --dry-run before applying remote schema changes.", + ], + }, + { + surface: "messages-and-sessions", + status: "opt_in", + local: "Sessions and channel digests are derived from local messages ordered by message id/cursor.", + remote: "Use --tables cloud-runtime to sync message rows by UUID and keep remote integer ids machine-independent.", + tables: [...STORAGE_TABLE_GROUPS.messages], + gates: [ + "Back up local SQLite before first push or pull.", + "Verify no duplicate message UUIDs before cloud-runtime sync.", + ], + }, + { + surface: "read-state", + status: "opt_in", + local: "read_at, per-agent receipts, mentions, reactions, and channel notification reads remain local by default.", + remote: "cloud-runtime sync translates dependent read-state rows through message UUIDs instead of local integer ids.", + tables: [...STORAGE_TABLE_GROUPS["read-state"]], + gates: [ + "Sync messages before dependent read-state rows.", + "Run digest/read-state regression tests before cutover.", + ], + }, + { + surface: "search-and-digests", + status: "opt_in", + local: "SQLite FTS5 remains the default search path and digest cursor source.", + remote: "PostgreSQL migrations create a tsvector index for remote smoke/search readiness; live search still uses local SQLite unless a remote query adapter is selected.", + tables: ["messages", "messages_fts"], + gates: [ + "Rebuild or verify remote search_vector after imports.", + "Compare digest cursors before switching readers to remote-backed queries.", + ], + }, + { + surface: "attachments", + status: "local_only", + local: "Attachment files are copied under the local attachments directory; message rows store local attachment metadata.", + remote: "cloud-runtime sync omits local attachment metadata and file paths until an approved object-store migration exists.", + tables: ["messages.attachments"], + gates: [ + "Create a separate approval task for S3 bucket/object migration before syncing attachment bytes.", + "Do not expose attachment contents, local attachment paths, or secret values in diagnostics.", + ], + }, + { + surface: "aws-production", + status: "manual_gate", + local: "No production AWS mutation is required for local messaging.", + remote: `Canonical runtime target is ${canonical.cluster}/${canonical.database}; secret path is metadata only.`, + tables: [], + gates: [ + "No terraform apply, production migration, secret creation, spend increase, or data migration without explicit approval.", + "Run a read-only smoke before any write-enabled production command.", + ], + }, + ], + privacyAndMigrationGates: [ + "Never print database URLs, credentials, attachment contents, or secret values.", + "Keep default sync on metadata tables unless cloud-runtime is requested explicitly.", + "Back up local SQLite and attachment directories before first cross-machine push/pull.", + "Run storage status, storage migrate --dry-run, focused digest/search/read-state tests, and a read-only remote smoke before cutover.", + "Create a follow-up approval task for live AWS mutation, S3 object migration, or production data migration.", + ], + }; +} + function firstEnv(names: readonly string[]): string | null { for (const name of names) { const value = process.env[name]; @@ -114,9 +285,10 @@ function normalizeStorageMode(value?: string | null): StorageMode | null { export function getStorageConfig(): StorageConfig { const envMode = getStorageDatabaseUrlFromEnv() ? "remote" : "local"; - if (!existsSync(STORAGE_CONFIG_PATH)) return { mode: normalizeStorageMode(firstEnv(STORAGE_MODE_ENV)) ?? envMode, rds: {} }; + const configPath = getStorageConfigPath(); + if (!existsSync(configPath)) return { mode: normalizeStorageMode(firstEnv(STORAGE_MODE_ENV)) ?? envMode, rds: {} }; try { - const parsed = JSON.parse(readFileSync(STORAGE_CONFIG_PATH, "utf8")) as Partial; + const parsed = JSON.parse(readFileSync(configPath, "utf8")) as Partial; return { mode: normalizeStorageMode(firstEnv(STORAGE_MODE_ENV)) ?? normalizeStorageMode(parsed.mode) @@ -180,11 +352,27 @@ export function resolveTables(tables?: string[] | string): SyncTable[] { ? tables.split(",") : []; if (requested.length === 0) return [...DEFAULT_STORAGE_TABLES]; - const allowed = new Set(DEFAULT_STORAGE_TABLES); + const allowed = new Set(CLOUD_RUNTIME_STORAGE_TABLES); const clean = requested.map((table) => table.trim()).filter(Boolean); - const invalid = clean.filter((table) => !allowed.has(table)); + const expanded: string[] = []; + for (const table of clean) { + if (table === "default" || table === "metadata") { + expanded.push(...DEFAULT_STORAGE_TABLES); + continue; + } + if (table === "cloud-runtime" || table === "cloud_runtime") { + expanded.push(...CLOUD_RUNTIME_STORAGE_TABLES); + continue; + } + if (table === "read-state" || table === "read_state") { + expanded.push(...STORAGE_TABLE_GROUPS["read-state"]); + continue; + } + expanded.push(table); + } + const invalid = expanded.filter((table) => !allowed.has(table)); if (invalid.length > 0) throw new Error(`Unsupported conversations storage table(s): ${invalid.join(", ")}`); - return clean as SyncTable[]; + return [...new Set(expanded)] as SyncTable[]; } export async function storagePush(options?: { tables?: string[] | string }): Promise { @@ -228,6 +416,12 @@ export async function syncPull(remote: PgAdapterAsync, db: Database, options: { } async function pushTable(db: Database, remote: PgAdapterAsync, table: SyncTable): Promise { + if (table === "messages") return pushMessages(db, remote); + if (table === "message_read_receipts") return pushMessageReadReceipts(db, remote); + if (table === "channel_notification_reads") return pushChannelNotificationReads(db, remote); + if (table === "reactions") return pushReactions(db, remote); + if (table === "message_mentions") return pushMessageMentions(db, remote); + const result: StorageSyncResult = { table, rowsRead: 0, rowsWritten: 0, errors: [] }; try { if (!sqliteTableExists(db, table)) return result; @@ -235,7 +429,7 @@ async function pushTable(db: Database, remote: PgAdapterAsync, table: SyncTable) result.rowsRead = rows.length; if (rows.length === 0) return result; const remoteColumns = await getRemoteColumns(remote, table); - const columns = filterRemoteColumns(remoteColumns, Object.keys(rows[0]!)); + const columns = filterSyncColumns(table, filterRemoteColumns(remoteColumns, Object.keys(rows[0]!))); result.rowsWritten = await upsertPg(remote, table, columns, rows, remoteColumns); } catch (error) { result.errors.push(error instanceof Error ? error.message : String(error)); @@ -244,13 +438,19 @@ async function pushTable(db: Database, remote: PgAdapterAsync, table: SyncTable) } async function pullTable(remote: PgAdapterAsync, db: Database, table: SyncTable): Promise { + if (table === "messages") return pullMessages(remote, db); + if (table === "message_read_receipts") return pullMessageReadReceipts(remote, db); + if (table === "channel_notification_reads") return pullChannelNotificationReads(remote, db); + if (table === "reactions") return pullReactions(remote, db); + if (table === "message_mentions") return pullMessageMentions(remote, db); + const result: StorageSyncResult = { table, rowsRead: 0, rowsWritten: 0, errors: [] }; try { if (!sqliteTableExists(db, table)) return result; const rows = await remote.all(`SELECT * FROM ${quoteIdent(table)}`) as Row[]; result.rowsRead = rows.length; if (rows.length === 0) return result; - const columns = filterLocalColumns(db, table, Object.keys(rows[0]!)); + const columns = filterSyncColumns(table, filterLocalColumns(db, table, Object.keys(rows[0]!))); result.rowsWritten = upsertSqlite(db, table, columns, rows); } catch (error) { result.errors.push(error instanceof Error ? error.message : String(error)); @@ -286,11 +486,12 @@ export function ensureConflictsTable(db: Database): void { } } -export function detectConflicts(localRows: Row[], remoteRows: Row[], table: string, pk: string, tsCol = "created_at"): SyncConflict[] { - const remoteByPk = new Map(remoteRows.map((row) => [String(row[pk] ?? ""), row])); +export function detectConflicts(localRows: Row[], remoteRows: Row[], table: string, pk: string | string[], tsCol = "created_at"): SyncConflict[] { + const pkColumns = Array.isArray(pk) ? pk : [pk]; + const remoteByPk = new Map(remoteRows.map((row) => [rowKey(row, pkColumns), row])); const conflicts: SyncConflict[] = []; for (const local of localRows) { - const key = String(local[pk] ?? ""); + const key = rowKey(local, pkColumns); if (!key) continue; const remote = remoteByPk.get(key); if (!remote) continue; @@ -317,9 +518,327 @@ export function listConflicts(db: Database, options?: { resolved?: boolean }): R return db.all("SELECT * FROM _sync_conflicts WHERE resolved = ? ORDER BY detected_at DESC", options.resolved ? 1 : 0); } +export function listDuplicateMessageUuids(db: Database = getDb()): MessageUuidDuplicate[] { + try { + if (!sqliteTableExists(db, "messages")) return []; + return db.all(` + SELECT uuid, COUNT(*) AS count + FROM messages + WHERE uuid IS NOT NULL AND uuid <> '' + GROUP BY uuid + HAVING COUNT(*) > 1 + ORDER BY count DESC, uuid ASC + LIMIT 20 + `); + } catch { + return []; + } +} + +async function pushMessages(db: Database, remote: PgAdapterAsync): Promise { + const result: StorageSyncResult = { table: "messages", rowsRead: 0, rowsWritten: 0, errors: [] }; + try { + if (!sqliteTableExists(db, "messages")) return result; + const rows = db.all("SELECT * FROM messages WHERE uuid IS NOT NULL AND uuid <> ''"); + result.rowsRead = rows.length; + if (rows.length === 0) return result; + const remoteColumns = await getRemoteColumns(remote, "messages"); + const columns = filterSyncColumns("messages", filterRemoteColumns(remoteColumns, Object.keys(rows[0]!))); + result.rowsWritten = await upsertPg(remote, "messages", columns, rows, remoteColumns); + for (const row of rows) { + await updateRemoteReplyTo(db, remote, row); + } + } catch (error) { + result.errors.push(error instanceof Error ? error.message : String(error)); + } + return result; +} + +async function pullMessages(remote: PgAdapterAsync, db: Database): Promise { + const result: StorageSyncResult = { table: "messages", rowsRead: 0, rowsWritten: 0, errors: [] }; + try { + if (!sqliteTableExists(db, "messages")) return result; + const rows = await remote.all("SELECT * FROM messages WHERE uuid IS NOT NULL AND uuid <> ''") as Row[]; + result.rowsRead = rows.length; + if (rows.length === 0) return result; + const columns = filterSyncColumns("messages", filterLocalColumns(db, "messages", Object.keys(rows[0]!))); + result.rowsWritten = upsertSqlite(db, "messages", columns, rows); + for (const row of rows) { + await updateLocalReplyTo(remote, db, row); + } + } catch (error) { + result.errors.push(error instanceof Error ? error.message : String(error)); + } + return result; +} + +async function pushMessageReadReceipts(db: Database, remote: PgAdapterAsync): Promise { + const result: StorageSyncResult = { table: "message_read_receipts", rowsRead: 0, rowsWritten: 0, errors: [] }; + try { + if (!sqliteTableExists(db, "message_read_receipts") || !sqliteTableExists(db, "messages")) return result; + const rows = db.all(` + SELECT m.uuid AS message_uuid, r.agent, r.read_at + FROM message_read_receipts r + INNER JOIN messages m ON m.id = r.message_id + WHERE m.uuid IS NOT NULL AND m.uuid <> '' + `); + result.rowsRead = rows.length; + for (const row of rows) { + const messageId = await getRemoteMessageId(remote, row.message_uuid); + if (messageId === null) continue; + const write = await remote.run(` + INSERT INTO message_read_receipts (message_id, agent, read_at) + VALUES (?, ?, ?) + ON CONFLICT (message_id, agent) DO UPDATE SET + read_at = GREATEST(message_read_receipts.read_at, EXCLUDED.read_at) + `, messageId, row.agent, row.read_at); + result.rowsWritten += write.changes; + } + } catch (error) { + result.errors.push(error instanceof Error ? error.message : String(error)); + } + return result; +} + +async function pullMessageReadReceipts(remote: PgAdapterAsync, db: Database): Promise { + const result: StorageSyncResult = { table: "message_read_receipts", rowsRead: 0, rowsWritten: 0, errors: [] }; + try { + if (!sqliteTableExists(db, "message_read_receipts") || !sqliteTableExists(db, "messages")) return result; + const rows = await remote.all(` + SELECT m.uuid AS message_uuid, r.agent, r.read_at + FROM message_read_receipts r + INNER JOIN messages m ON m.id = r.message_id + WHERE m.uuid IS NOT NULL AND m.uuid <> '' + `) as Row[]; + result.rowsRead = rows.length; + const insert = db.prepare(` + INSERT INTO message_read_receipts (message_id, agent, read_at) + VALUES (?, ?, ?) + ON CONFLICT(message_id, agent) DO UPDATE SET + read_at = max(message_read_receipts.read_at, excluded.read_at) + `); + for (const row of rows) { + const messageId = getLocalMessageId(db, row.message_uuid); + if (messageId === null) continue; + result.rowsWritten += insert.run(messageId, row.agent, coerceForSqlite(row.read_at)).changes; + } + } catch (error) { + result.errors.push(error instanceof Error ? error.message : String(error)); + } + return result; +} + +async function pushChannelNotificationReads(db: Database, remote: PgAdapterAsync): Promise { + const result: StorageSyncResult = { table: "channel_notification_reads", rowsRead: 0, rowsWritten: 0, errors: [] }; + try { + if (!sqliteTableExists(db, "channel_notification_reads") || !sqliteTableExists(db, "messages")) return result; + const rows = db.all(` + SELECT m.uuid AS message_uuid, r.agent, r.read_at + FROM channel_notification_reads r + INNER JOIN messages m ON m.id = r.message_id + WHERE m.uuid IS NOT NULL AND m.uuid <> '' + `); + result.rowsRead = rows.length; + for (const row of rows) { + const messageId = await getRemoteMessageId(remote, row.message_uuid); + if (messageId === null) continue; + const write = await remote.run(` + INSERT INTO channel_notification_reads (agent, message_id, read_at) + VALUES (?, ?, ?) + ON CONFLICT (agent, message_id) DO UPDATE SET + read_at = GREATEST(channel_notification_reads.read_at, EXCLUDED.read_at) + `, row.agent, messageId, row.read_at); + result.rowsWritten += write.changes; + } + } catch (error) { + result.errors.push(error instanceof Error ? error.message : String(error)); + } + return result; +} + +async function pullChannelNotificationReads(remote: PgAdapterAsync, db: Database): Promise { + const result: StorageSyncResult = { table: "channel_notification_reads", rowsRead: 0, rowsWritten: 0, errors: [] }; + try { + if (!sqliteTableExists(db, "channel_notification_reads") || !sqliteTableExists(db, "messages")) return result; + const rows = await remote.all(` + SELECT m.uuid AS message_uuid, r.agent, r.read_at + FROM channel_notification_reads r + INNER JOIN messages m ON m.id = r.message_id + WHERE m.uuid IS NOT NULL AND m.uuid <> '' + `) as Row[]; + result.rowsRead = rows.length; + const insert = db.prepare(` + INSERT INTO channel_notification_reads (agent, message_id, read_at) + VALUES (?, ?, ?) + ON CONFLICT(agent, message_id) DO UPDATE SET + read_at = max(channel_notification_reads.read_at, excluded.read_at) + `); + for (const row of rows) { + const messageId = getLocalMessageId(db, row.message_uuid); + if (messageId === null) continue; + result.rowsWritten += insert.run(row.agent, messageId, coerceForSqlite(row.read_at)).changes; + } + } catch (error) { + result.errors.push(error instanceof Error ? error.message : String(error)); + } + return result; +} + +async function pushReactions(db: Database, remote: PgAdapterAsync): Promise { + const result: StorageSyncResult = { table: "reactions", rowsRead: 0, rowsWritten: 0, errors: [] }; + try { + if (!sqliteTableExists(db, "reactions") || !sqliteTableExists(db, "messages")) return result; + const rows = db.all(` + SELECT m.uuid AS message_uuid, r.agent, r.emoji, r.created_at + FROM reactions r + INNER JOIN messages m ON m.id = r.message_id + WHERE m.uuid IS NOT NULL AND m.uuid <> '' + `); + result.rowsRead = rows.length; + for (const row of rows) { + const messageId = await getRemoteMessageId(remote, row.message_uuid); + if (messageId === null) continue; + const write = await remote.run(` + INSERT INTO reactions (message_id, agent, emoji, created_at) + VALUES (?, ?, ?, ?) + ON CONFLICT (message_id, agent, emoji) DO UPDATE SET + created_at = LEAST(reactions.created_at, EXCLUDED.created_at) + `, messageId, row.agent, row.emoji, row.created_at); + result.rowsWritten += write.changes; + } + } catch (error) { + result.errors.push(error instanceof Error ? error.message : String(error)); + } + return result; +} + +async function pullReactions(remote: PgAdapterAsync, db: Database): Promise { + const result: StorageSyncResult = { table: "reactions", rowsRead: 0, rowsWritten: 0, errors: [] }; + try { + if (!sqliteTableExists(db, "reactions") || !sqliteTableExists(db, "messages")) return result; + const rows = await remote.all(` + SELECT m.uuid AS message_uuid, r.agent, r.emoji, r.created_at + FROM reactions r + INNER JOIN messages m ON m.id = r.message_id + WHERE m.uuid IS NOT NULL AND m.uuid <> '' + `) as Row[]; + result.rowsRead = rows.length; + const insert = db.prepare(` + INSERT INTO reactions (message_id, agent, emoji, created_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(message_id, agent, emoji) DO UPDATE SET + created_at = min(reactions.created_at, excluded.created_at) + `); + for (const row of rows) { + const messageId = getLocalMessageId(db, row.message_uuid); + if (messageId === null) continue; + result.rowsWritten += insert.run(messageId, row.agent, row.emoji, coerceForSqlite(row.created_at)).changes; + } + } catch (error) { + result.errors.push(error instanceof Error ? error.message : String(error)); + } + return result; +} + +async function pushMessageMentions(db: Database, remote: PgAdapterAsync): Promise { + const result: StorageSyncResult = { table: "message_mentions", rowsRead: 0, rowsWritten: 0, errors: [] }; + try { + if (!sqliteTableExists(db, "message_mentions") || !sqliteTableExists(db, "messages")) return result; + const rows = db.all(` + SELECT m.uuid AS message_uuid, mm.mentioned_agent, mm.from_agent, mm.channel, mm.notified_at, mm.created_at + FROM message_mentions mm + INNER JOIN messages m ON m.id = mm.message_id + WHERE m.uuid IS NOT NULL AND m.uuid <> '' + `); + result.rowsRead = rows.length; + for (const row of rows) { + const messageId = await getRemoteMessageId(remote, row.message_uuid); + if (messageId === null) continue; + const existing = await remote.get(` + SELECT mm.id + FROM message_mentions mm + WHERE mm.message_id = ? + AND mm.mentioned_agent = ? + AND mm.from_agent = ? + AND COALESCE(mm.channel, '') = COALESCE(?, '') + LIMIT 1 + `, messageId, row.mentioned_agent, row.from_agent, row.channel) as { id?: number | string } | null; + if (existing?.id !== undefined) { + if (row.notified_at) { + const write = await remote.run( + "UPDATE message_mentions SET notified_at = COALESCE(notified_at, ?) WHERE id = ?", + row.notified_at, + existing.id, + ); + result.rowsWritten += write.changes; + } + continue; + } + const write = await remote.run(` + INSERT INTO message_mentions (message_id, mentioned_agent, from_agent, channel, notified_at, created_at) + VALUES (?, ?, ?, ?, ?, ?) + `, messageId, row.mentioned_agent, row.from_agent, row.channel, row.notified_at, row.created_at); + result.rowsWritten += write.changes; + } + } catch (error) { + result.errors.push(error instanceof Error ? error.message : String(error)); + } + return result; +} + +async function pullMessageMentions(remote: PgAdapterAsync, db: Database): Promise { + const result: StorageSyncResult = { table: "message_mentions", rowsRead: 0, rowsWritten: 0, errors: [] }; + try { + if (!sqliteTableExists(db, "message_mentions") || !sqliteTableExists(db, "messages")) return result; + const rows = await remote.all(` + SELECT m.uuid AS message_uuid, mm.mentioned_agent, mm.from_agent, mm.channel, mm.notified_at, mm.created_at + FROM message_mentions mm + INNER JOIN messages m ON m.id = mm.message_id + WHERE m.uuid IS NOT NULL AND m.uuid <> '' + `) as Row[]; + result.rowsRead = rows.length; + const findMention = db.prepare(` + SELECT mm.id + FROM message_mentions mm + WHERE mm.message_id = ? + AND mm.mentioned_agent = ? + AND mm.from_agent = ? + AND COALESCE(mm.channel, '') = COALESCE(?, '') + LIMIT 1 + `); + const updateNotified = db.prepare("UPDATE message_mentions SET notified_at = COALESCE(notified_at, ?) WHERE id = ?"); + const insert = db.prepare(` + INSERT INTO message_mentions (message_id, mentioned_agent, from_agent, channel, notified_at, created_at) + VALUES (?, ?, ?, ?, ?, ?) + `); + for (const row of rows) { + const messageId = getLocalMessageId(db, row.message_uuid); + if (messageId === null) continue; + const existing = findMention.get(messageId, row.mentioned_agent, row.from_agent, row.channel) as { id?: number } | null; + if (existing?.id !== undefined) { + if (row.notified_at) { + result.rowsWritten += updateNotified.run(coerceForSqlite(row.notified_at), existing.id).changes; + } + continue; + } + result.rowsWritten += insert.run( + messageId, + row.mentioned_agent, + row.from_agent, + row.channel, + coerceForSqlite(row.notified_at), + coerceForSqlite(row.created_at), + ).changes; + } + } catch (error) { + result.errors.push(error instanceof Error ? error.message : String(error)); + } + return result; +} + export async function detectAndLogConflicts(local: Database, remote: PgAdapterAsync, table: string): Promise { if (!CONFLICT_TABLES.has(table)) return 0; - const pk = table === "channels" ? "name" : "id"; + const pk = PRIMARY_KEYS[table] ?? [table === "channels" ? "name" : "id"]; const localRows = local.all(`SELECT * FROM ${quoteIdent(table)}`); const remoteRows = await remote.all(`SELECT * FROM ${quoteIdent(table)}`) as Row[]; if (localRows.length === 0 || remoteRows.length === 0) return 0; @@ -354,6 +873,73 @@ function filterLocalColumns(db: Database, table: string, columns: string[]): str return columns.filter((column) => allowed.has(column)); } +function filterSyncColumns(table: string, columns: string[]): string[] { + if (table === "messages") { + return columns.filter((column) => column !== "id" && column !== "search_vector" && column !== "attachments" && column !== "reply_to"); + } + if (table === "channel_subscriptions") { + return columns.filter((column) => column !== "since_message_id"); + } + return columns; +} + +async function updateRemoteReplyTo(db: Database, remote: PgAdapterAsync, row: Row): Promise { + const messageUuid = row.uuid; + if (typeof messageUuid !== "string" || messageUuid.length === 0) return; + const childId = await getRemoteMessageId(remote, messageUuid); + if (childId === null) return; + + let remoteParentId: number | null = null; + const localParentId = Number(row.reply_to); + if (Number.isFinite(localParentId) && localParentId > 0) { + const parent = db.get<{ uuid: string }>("SELECT uuid FROM messages WHERE id = ?", localParentId); + if (parent?.uuid) { + remoteParentId = await getRemoteMessageId(remote, parent.uuid); + } + } + + await remote.run("UPDATE messages SET reply_to = ? WHERE id = ?", remoteParentId, childId); +} + +async function updateLocalReplyTo(remote: PgAdapterAsync, db: Database, row: Row): Promise { + const messageUuid = row.uuid; + if (typeof messageUuid !== "string" || messageUuid.length === 0) return; + const childId = getLocalMessageId(db, messageUuid); + if (childId === null) return; + + let localParentId: number | null = null; + const remoteParentId = Number(row.reply_to); + if (Number.isFinite(remoteParentId) && remoteParentId > 0) { + const parent = await remote.get("SELECT uuid FROM messages WHERE id = ?", remoteParentId) as { uuid?: string } | null; + if (parent?.uuid) { + localParentId = getLocalMessageId(db, parent.uuid); + } + } + + db.run("UPDATE messages SET reply_to = ? WHERE id = ?", localParentId, childId); +} + +async function getRemoteMessageId(remote: PgAdapterAsync, messageUuid: unknown): Promise { + if (typeof messageUuid !== "string" || messageUuid.length === 0) return null; + const row = await remote.get("SELECT id FROM messages WHERE uuid = ?", messageUuid) as { id?: number | string } | null; + if (!row || row.id === undefined || row.id === null) return null; + const id = Number(row.id); + return Number.isFinite(id) ? id : null; +} + +function getLocalMessageId(db: Database, messageUuid: unknown): number | null { + if (typeof messageUuid !== "string" || messageUuid.length === 0) return null; + const row = db.get<{ id: number }>("SELECT id FROM messages WHERE uuid = ?", messageUuid); + return row?.id ?? null; +} + +function rowKey(row: Row, columns: string[]): string { + const parts = columns.map((column) => String(row[column] ?? "")); + if (!parts.every((part) => part.length > 0)) return ""; + if (columns.length === 1) return parts[0]!; + return columns.map((column, index) => `${column}=${parts[index]}`).join("|"); +} + async function upsertPg(remote: PgAdapterAsync, table: SyncTable, columns: string[], rows: Row[], remoteColumns: Map): Promise { if (columns.length === 0) return 0; const primaryKeys = PRIMARY_KEYS[table]; diff --git a/src/mcp/tools/storage-contract.test.ts b/src/mcp/tools/storage-contract.test.ts index ffdc246..6aa996c 100644 --- a/src/mcp/tools/storage-contract.test.ts +++ b/src/mcp/tools/storage-contract.test.ts @@ -17,6 +17,7 @@ describe("conversations storage MCP contract", () => { expect(toolsSource).toContain('"conversations_storage_sync"'); expect(toolsSource).toContain('"conversations_storage_migrate"'); expect(toolsSource).toContain('"conversations_storage_feedback"'); + expect(toolsSource).toContain('"conversations_storage_readiness"'); expect(toolsSource.match(/await runStorageMigrations\(pg\)/g)?.length ?? 0).toBeGreaterThanOrEqual(3); expect(toolsSource).not.toContain(retiredTool("status")); expect(toolsSource).not.toContain(retiredTool("push")); diff --git a/src/mcp/tools/storage.test.ts b/src/mcp/tools/storage.test.ts index 786066f..28e03b2 100644 --- a/src/mcp/tools/storage.test.ts +++ b/src/mcp/tools/storage.test.ts @@ -3,39 +3,44 @@ import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { closeDb } from "../../lib/db.js"; -import { STORAGE_CONFIG_PATH } from "../../lib/storage-sync.js"; +import { getStorageConfigPath } from "../../lib/storage-sync.js"; import { registerStorageSyncTools } from "./storage.js"; import { existsSync, readFileSync, writeFileSync, mkdirSync, unlinkSync } from "fs"; import { tmpdir } from "os"; import { dirname, join } from "path"; -const STORAGE_CONFIG_DIR = dirname(STORAGE_CONFIG_PATH); +const TEST_STORAGE_CONFIG = join(tmpdir(), `conversations-test-storage-config-${Date.now()}.json`); // Save/restore real storage config let savedConfig: string | null = null; function saveStorageConfig(): void { - if (existsSync(STORAGE_CONFIG_PATH)) { - savedConfig = readFileSync(STORAGE_CONFIG_PATH, "utf-8"); + const storageConfigPath = getStorageConfigPath(); + if (existsSync(storageConfigPath)) { + savedConfig = readFileSync(storageConfigPath, "utf-8"); } } function restoreStorageConfig(): void { + const storageConfigPath = getStorageConfigPath(); + const storageConfigDir = dirname(storageConfigPath); if (savedConfig !== null) { - if (!existsSync(STORAGE_CONFIG_DIR)) { - mkdirSync(STORAGE_CONFIG_DIR, { recursive: true }); + if (!existsSync(storageConfigDir)) { + mkdirSync(storageConfigDir, { recursive: true }); } - writeFileSync(STORAGE_CONFIG_PATH, savedConfig, "utf-8"); + writeFileSync(storageConfigPath, savedConfig, "utf-8"); } else { - try { unlinkSync(STORAGE_CONFIG_PATH); } catch {} + try { unlinkSync(storageConfigPath); } catch {} } } function writeStorageConfig(config: Record): void { - if (!existsSync(STORAGE_CONFIG_DIR)) { - mkdirSync(STORAGE_CONFIG_DIR, { recursive: true }); + const storageConfigPath = getStorageConfigPath(); + const storageConfigDir = dirname(storageConfigPath); + if (!existsSync(storageConfigDir)) { + mkdirSync(storageConfigDir, { recursive: true }); } - writeFileSync(STORAGE_CONFIG_PATH, JSON.stringify(config, null, 2), "utf-8"); + writeFileSync(storageConfigPath, JSON.stringify(config, null, 2), "utf-8"); } const TEST_DB = join(tmpdir(), `conversations-test-storage-${Date.now()}.db`); @@ -43,6 +48,7 @@ let client: Client; let server: McpServer; beforeAll(async () => { + process.env.HASNA_CONVERSATIONS_STORAGE_CONFIG = TEST_STORAGE_CONFIG; saveStorageConfig(); process.env.CONVERSATIONS_DB_PATH = TEST_DB; closeDb(); @@ -66,6 +72,8 @@ afterAll(async () => { try { unlinkSync(TEST_DB); } catch {} try { unlinkSync(TEST_DB + "-wal"); } catch {} try { unlinkSync(TEST_DB + "-shm"); } catch {} + try { unlinkSync(TEST_STORAGE_CONFIG); } catch {} + delete process.env.HASNA_CONVERSATIONS_STORAGE_CONFIG; }); beforeEach(() => { @@ -89,6 +97,10 @@ describe("conversations_storage_status", () => { expect(text).toContain("Canonical RDS cluster: hasna-xyz-infra-apps-prod-postgres"); expect(text).toContain("Runtime secret path: hasna/xyz/opensource/conversations/prod/rds"); expect(text).toContain("PostgreSQL: skipped in local mode"); + expect(text).toContain("Default sync group: metadata"); + expect(text).toContain("Cloud runtime group: cloud-runtime"); + expect(text).toContain("Attachments: local files only"); + expect(text).toContain("Message UUID duplicates: 0"); }, 10000); test("reports conflict counts", async () => { @@ -99,6 +111,32 @@ describe("conversations_storage_status", () => { }, 20000); }); +// ---- conversations_storage_readiness ---- + +describe("conversations_storage_readiness", () => { + test("reports explicit local, remote, message, read-state, search, and attachment gates", async () => { + writeStorageConfig({ mode: "local" }); + const result = await client.callTool({ name: "conversations_storage_readiness", arguments: {} }) as any; + const text = getText(result); + const readiness = JSON.parse(text) as { + configured: boolean; + tableGroups: { cloudRuntime: string[]; localOnly: string[] }; + runtimePaths: Array<{ surface: string; status: string; remote: string }>; + privacyAndMigrationGates: string[]; + }; + + expect(readiness.configured).toBe(false); + expect(readiness.tableGroups.cloudRuntime).toContain("messages"); + expect(readiness.tableGroups.cloudRuntime).toContain("message_read_receipts"); + expect(readiness.tableGroups.localOnly).toContain("messages_fts"); + expect(readiness.runtimePaths.map((path) => path.surface)).toContain("attachments"); + expect(readiness.runtimePaths.find((path) => path.surface === "attachments")?.status).toBe("local_only"); + expect(readiness.runtimePaths.find((path) => path.surface === "search-and-digests")?.remote).toContain("tsvector"); + expect(readiness.privacyAndMigrationGates.join(" ")).toContain("Never print database URLs"); + expect(text).not.toContain("postgres://"); + }, 10000); +}); + // ---- conversations_storage_push ---- describe("conversations_storage_push", () => { diff --git a/src/mcp/tools/storage.ts b/src/mcp/tools/storage.ts index 6105f61..f0c7e73 100644 --- a/src/mcp/tools/storage.ts +++ b/src/mcp/tools/storage.ts @@ -8,7 +8,9 @@ import { getStorageConfig, getStorageDatabaseUrl, getStoragePg, + getStorageReadiness, listConflicts, + listDuplicateMessageUuids, resolveTables, runStorageMigrations, saveFeedback, @@ -28,6 +30,7 @@ export function registerStorageSyncTools(server: McpServer): void { try { const config = getStorageConfig(); const canonical = getCanonicalConversationsRdsConfig(); + const readiness = getStorageReadiness(); const lines = [ `Mode: ${config.mode}`, "Service: conversations", @@ -36,6 +39,9 @@ export function registerStorageSyncTools(server: McpServer): void { `Runtime secret path: ${canonical.runtimeSecretPath}`, `Database env: ${canonical.env} (fallback: ${canonical.fallbackEnv})`, `RDS Host: ${config.rds.host || (getStorageDatabaseUrl() ? "(env database url)" : "(not configured)")}`, + `Default sync group: metadata (${DEFAULT_STORAGE_TABLES.join(", ")})`, + `Cloud runtime group: cloud-runtime (${readiness.tableGroups.cloudRuntime.join(", ")})`, + "Attachments: local files only; S3 object storage is an approval-gated follow-up.", ]; if (config.mode === "local" && !getStorageDatabaseUrl()) { @@ -57,6 +63,7 @@ export function registerStorageSyncTools(server: McpServer): void { const unresolved = listConflicts(local, { resolved: false }); const resolved = listConflicts(local, { resolved: true }); lines.push(`Sync conflicts: ${unresolved.length} unresolved, ${resolved.length} resolved`); + lines.push(`Message UUID duplicates: ${listDuplicateMessageUuids(local).length}`); } catch {} return { content: [{ type: "text", text: lines.join("\n") }] }; @@ -66,6 +73,20 @@ export function registerStorageSyncTools(server: McpServer): void { }, ); + server.tool( + "conversations_storage_readiness", + "Show local SQLite, remote PostgreSQL, message/read-state, digest/search, attachment, and production migration readiness without exposing secrets", + {}, + async () => { + try { + const readiness = getStorageReadiness(); + return { content: [{ type: "text", text: JSON.stringify(readiness, null, 2) }] }; + } catch (e) { + return { content: [{ type: "text", text: formatError(e) }], isError: true }; + } + }, + ); + server.tool( "conversations_storage_push", "Push local conversations data to remote PostgreSQL storage. Detects conflicts and syncs safe text-key tables.", diff --git a/src/storage.ts b/src/storage.ts index e417291..ad40a75 100644 --- a/src/storage.ts +++ b/src/storage.ts @@ -5,18 +5,26 @@ export { CANONICAL_CONVERSATIONS_RDS_CLUSTER, CANONICAL_CONVERSATIONS_RDS_DATABASE, CANONICAL_CONVERSATIONS_RDS_SECRET_PATH, + CLOUD_RUNTIME_STORAGE_TABLES, CONVERSATIONS_DATABASE_FALLBACK_ENV, DEFAULT_STORAGE_TABLES, + STORAGE_LOCAL_ONLY_TABLES, + STORAGE_MESSAGE_TABLES, + STORAGE_TABLE_GROUPS, STORAGE_CONFIG_PATH, + STORAGE_CONFIG_PATH_ENV, STORAGE_DATABASE_ENV, STORAGE_MODE_ENV, SYNC_EXCLUDED, ensureConflictsTable, getCanonicalConversationsRdsConfig, getStorageConfig, + getStorageConfigPath, getStorageDatabaseUrl, getStoragePg, + getStorageReadiness, listConflicts, + listDuplicateMessageUuids, listPgTables, listSqliteTables, resolveTables, @@ -29,7 +37,11 @@ export { type StorageConfig, type CanonicalConversationsRdsConfig, type StorageMode, + type StorageReadiness, + type StorageRuntimePath, + type StorageRuntimeStatus, type StorageSyncResult, type SyncConflict, type SyncResult, + type MessageUuidDuplicate, } from "./lib/storage-sync.js";