diff --git a/docker-compose.yml b/docker-compose.yml index e3f9200..ddd138e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,4 +1,55 @@ services: + postgres: + image: pgvector/pgvector:pg16 + environment: + POSTGRES_DB: "${POSTGRES_DB:-duriin}" + POSTGRES_USER: "${POSTGRES_USER:-duriin}" + POSTGRES_PASSWORD: "${POSTGRES_PASSWORD:?Set POSTGRES_PASSWORD in .env before starting PostgreSQL}" + command: + - postgres + - -c + - shared_buffers=128MB + - -c + - work_mem=4MB + - -c + - maintenance_work_mem=64MB + - -c + - effective_cache_size=256MB + - -c + - max_connections=40 + volumes: + - postgres_data:/var/lib/postgresql/data + mem_limit: 512m + cpus: "0.50" + restart: unless-stopped + healthcheck: + test: ["CMD-SHELL", "pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB"] + interval: 10s + timeout: 5s + retries: 12 + networks: + - nginx_proxy_manager_default + + postgres-migrate: + profiles: [migration] + build: + context: . + provenance: false + command: node scripts/migrate-sqlite-to-postgres.js + env_file: .env + volumes: + - ./config.json:/app/config.json:ro + - ./data:/data:ro + environment: + DATABASE_URL: "postgresql://${POSTGRES_USER:-duriin}:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB:-duriin}" + SQLITE_ARCHIVE_PATH: /data/archive.sqlite + SQLITE_INTELLIGENCE_PATH: /data/intelligence.sqlite + depends_on: + postgres: + condition: service_healthy + networks: + - nginx_proxy_manager_default + api: build: context: . @@ -155,3 +206,6 @@ services: networks: nginx_proxy_manager_default: external: true + +volumes: + postgres_data: diff --git a/docs/postgres-migration.md b/docs/postgres-migration.md new file mode 100644 index 0000000..18e0954 --- /dev/null +++ b/docs/postgres-migration.md @@ -0,0 +1,23 @@ +# PostgreSQL migration + +The Compose stack includes PostgreSQL with a hard 512 MB container limit and +conservative in-server memory settings. Set `POSTGRES_PASSWORD` in the server +`.env`, then start it with `docker compose up -d postgres`. + +Run the resumable logical copy with: + +```sh +docker compose --profile migration run --rm postgres-migrate +``` + +It copies archive data to the `archive` schema and autonomy/intelligence data +to the `intelligence` schema. Each table is count-verified and progress is +persisted in each schema's `_migration_progress` table, so restarting resumes +at the last committed batch. SQLite vec0 internal tables are derived indexes; +their canonical embedding data is migrated from `article_embedding_store`. + +The application is intentionally not switched by this migration alone: it has +SQLite-specific synchronous query and transaction code in the autonomy workers +and admin routes. Pointing those processes at PostgreSQL before that code is +ported would make the live system unreliable. Keep SQLite mounted until a +separate verified runtime cutover is complete. diff --git a/scripts/migrate-sqlite-to-postgres.js b/scripts/migrate-sqlite-to-postgres.js new file mode 100644 index 0000000..21a7fd4 --- /dev/null +++ b/scripts/migrate-sqlite-to-postgres.js @@ -0,0 +1,112 @@ +#!/usr/bin/env node +/* + * Resumable logical migration for Duriin's two SQLite databases. + * + * It deliberately copies tables, not SQLite files: PostgreSQL receives usable + * relational data in `archive` and `intelligence` schemas. SQLite vec0 + * implementation tables are indexes, not source-of-truth data; the canonical + * article_embedding_store is copied and can be used to rebuild pgvector later. + */ +const Database = require('better-sqlite3'); +const { Pool } = require('pg'); + +const archivePath = process.env.SQLITE_ARCHIVE_PATH || '/data/archive.sqlite'; +const intelligencePath = process.env.SQLITE_INTELLIGENCE_PATH || '/data/intelligence.sqlite'; +const connectionString = process.env.DATABASE_URL; +if (!connectionString) throw new Error('DATABASE_URL is required'); + +const pool = new Pool({ connectionString, max: 2 }); +const BATCH_SIZE = Math.max(1, Number(process.env.POSTGRES_MIGRATION_BATCH_SIZE) || 200); +const DERIVED_SQLITE_TABLES = new Set([ + 'article_embeddings', 'article_embeddings_chunks', 'article_embeddings_info', + 'article_embeddings_rowids', 'article_embeddings_vector_chunks00', +]); + +function quote(name) { return `"${String(name).replaceAll('"', '""')}"`; } +function pgType(sqliteType) { + const type = String(sqliteType || '').toUpperCase(); + if (type.includes('INT')) return 'BIGINT'; + if (type.includes('REAL') || type.includes('FLOA') || type.includes('DOUB')) return 'DOUBLE PRECISION'; + if (type.includes('BLOB')) return 'BYTEA'; + return 'TEXT'; +} + +async function ensureSchema(client, schema) { + await client.query(`CREATE SCHEMA IF NOT EXISTS ${quote(schema)}`); + await client.query(`CREATE TABLE IF NOT EXISTS ${quote(schema)}.${quote('_migration_progress')} ( + table_name TEXT PRIMARY KEY, source_rows BIGINT NOT NULL, copied_rows BIGINT NOT NULL, + completed_at TIMESTAMPTZ, updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + )`); +} + +function sourceTables(sqlite) { + return sqlite.prepare(`SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name`).all() + .map((row) => row.name).filter((name) => !DERIVED_SQLITE_TABLES.has(name)); +} + +async function createTable(client, schema, sqlite, table) { + const columns = sqlite.prepare(`PRAGMA table_info(${quote(table)})`).all(); + if (!columns.length) return null; + const primary = columns.filter((column) => column.pk).sort((a, b) => a.pk - b.pk).map((column) => quote(column.name)); + const definitions = columns.map((column) => `${quote(column.name)} ${pgType(column.type)}${column.notnull ? ' NOT NULL' : ''}`); + if (primary.length) definitions.push(`PRIMARY KEY (${primary.join(', ')})`); + await client.query(`CREATE TABLE IF NOT EXISTS ${quote(schema)}.${quote(table)} (${definitions.join(', ')})`); + return columns; +} + +async function copyTable(client, schema, sqlite, table, columns) { + const sourceRows = Number(sqlite.prepare(`SELECT COUNT(*) AS count FROM ${quote(table)}`).get().count); + const progress = await client.query(`SELECT copied_rows, source_rows, completed_at FROM ${quote(schema)}.${quote('_migration_progress')} WHERE table_name=$1`, [table]); + const prior = progress.rows[0]; + if (prior?.completed_at && Number(prior.source_rows) === sourceRows) { + console.log(`[migrate] ${schema}.${table}: already verified (${sourceRows})`); + return; + } + // Tables with an integer primary key are copied by key. Other tables use a + // deterministic row offset and are still restartable at batch boundaries. + const key = columns.find((column) => column.pk === 1 && /INT/i.test(column.type)); + let cursor = key ? Number(prior?.copied_rows || 0) : 0; + const names = columns.map((column) => column.name); + const insert = `INSERT INTO ${quote(schema)}.${quote(table)} (${names.map(quote).join(', ')}) VALUES (${names.map((_, i) => `$${i + 1}`).join(', ')}) ON CONFLICT DO NOTHING`; + console.log(`[migrate] ${schema}.${table}: ${sourceRows} source rows`); + while (true) { + const rows = key + ? sqlite.prepare(`SELECT ${names.map(quote).join(', ')} FROM ${quote(table)} WHERE ${quote(key.name)} > ? ORDER BY ${quote(key.name)} LIMIT ?`).all(cursor, BATCH_SIZE) + : sqlite.prepare(`SELECT ${names.map(quote).join(', ')} FROM ${quote(table)} LIMIT ? OFFSET ?`).all(BATCH_SIZE, cursor); + if (!rows.length) break; + await client.query('BEGIN'); + try { + for (const row of rows) await client.query(insert, names.map((name) => row[name] ?? null)); + cursor = key ? Number(rows[rows.length - 1][key.name]) : cursor + rows.length; + await client.query(`INSERT INTO ${quote(schema)}.${quote('_migration_progress')} (table_name, source_rows, copied_rows, updated_at) + VALUES ($1,$2,$3,NOW()) ON CONFLICT (table_name) DO UPDATE SET source_rows=EXCLUDED.source_rows, copied_rows=EXCLUDED.copied_rows, updated_at=NOW()`, [table, sourceRows, cursor]); + await client.query('COMMIT'); + } catch (error) { await client.query('ROLLBACK'); throw error; } + process.stdout.write(`\r[migrate] ${schema}.${table}: ${Math.min(cursor, sourceRows)}/${sourceRows}`); + } + const target = await client.query(`SELECT COUNT(*)::bigint AS count FROM ${quote(schema)}.${quote(table)}`); + if (Number(target.rows[0].count) < sourceRows) throw new Error(`${schema}.${table}: target count is short`); + await client.query(`UPDATE ${quote(schema)}.${quote('_migration_progress')} SET completed_at=NOW(), source_rows=$2, copied_rows=$2 WHERE table_name=$1`, [table, sourceRows]); + console.log(`\r[migrate] ${schema}.${table}: verified ${sourceRows}`); +} + +async function migrateDatabase(client, schema, file) { + const sqlite = new Database(file, { readonly: true }); + try { + await ensureSchema(client, schema); + for (const table of sourceTables(sqlite)) { + const columns = await createTable(client, schema, sqlite, table); + if (columns) await copyTable(client, schema, sqlite, table, columns); + } + } finally { sqlite.close(); } +} + +(async () => { + const client = await pool.connect(); + try { + await client.query('CREATE EXTENSION IF NOT EXISTS vector'); + await migrateDatabase(client, 'archive', archivePath); + await migrateDatabase(client, 'intelligence', intelligencePath); + console.log('[migrate] SQLite logical data verified in PostgreSQL. SQLite remains the live source until application cutover.'); + } finally { client.release(); await pool.end(); } +})().catch((error) => { console.error('[migrate] fatal:', error); process.exit(1); });