feat: add bounded postgres data migration stack
This commit is contained in:
@@ -1,4 +1,55 @@
|
|||||||
services:
|
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:
|
api:
|
||||||
build:
|
build:
|
||||||
context: .
|
context: .
|
||||||
@@ -155,3 +206,6 @@ services:
|
|||||||
networks:
|
networks:
|
||||||
nginx_proxy_manager_default:
|
nginx_proxy_manager_default:
|
||||||
external: true
|
external: true
|
||||||
|
|
||||||
|
volumes:
|
||||||
|
postgres_data:
|
||||||
|
|||||||
@@ -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.
|
||||||
@@ -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); });
|
||||||
Reference in New Issue
Block a user