From 61978acb0208681919eb6eed2ae33dee0ce94074 Mon Sep 17 00:00:00 2001 From: Brandon Wolfgang Date: Thu, 22 Oct 2026 10:20:32 -0500 Subject: [PATCH] feat: add PostgreSQL document persistence - Configure a shared PostgreSQL connection pool with connection and idle timeouts - Add repository functions to create documents and store embedded chunks - Persist chunks and mark documents as completed within a transaction - Roll back failed chunk storage operations while preserving the original error - Load environment variables from .env in development and production commands --- package.json | 4 +-- src/db.ts | 14 ++++++++ src/document-repository.ts | 74 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 2 deletions(-) create mode 100644 src/db.ts create mode 100644 src/document-repository.ts diff --git a/package.json b/package.json index 800bcfd..10238f9 100644 --- a/package.json +++ b/package.json @@ -5,9 +5,9 @@ "main": "index.js", "scripts": { "test": "tsx --test src/**/*.test.ts", - "dev": "tsx watch src/server.ts", + "dev": "node --env-file=.env --import tsx/esm --watch src/server.ts", "build": "tsc -p tsconfig.json", - "start": "node dist/server.js" + "start": "node --env-file=.env dist/server.js" }, "keywords": [], "author": "", diff --git a/src/db.ts b/src/db.ts new file mode 100644 index 0000000..19e8724 --- /dev/null +++ b/src/db.ts @@ -0,0 +1,14 @@ +import pg from "pg"; + +const { Pool } = pg; + +export const pool = new Pool({ + connectionString: process.env.DATABASE_URL, + connectionTimeoutMillis: 5_000, + idleTimeoutMillis: 30_000, + max: 10, +}); + +pool.on("error", (error) => { + console.error("Unexpected PostgreSQL pool error:", error); +}); diff --git a/src/document-repository.ts b/src/document-repository.ts new file mode 100644 index 0000000..c420287 --- /dev/null +++ b/src/document-repository.ts @@ -0,0 +1,74 @@ +import { pool } from "./db.js"; +import type { TextChunk } from "./chunking.js"; + +export interface NewDocument { + title: string; + source?: string; +} + +export interface StoredChunk extends TextChunk { + embedding: number[]; +} + +export const createDocument = async ( + document: NewDocument, +): Promise => { + const result = await pool.query<{ id: string }>( + `INSERT INTO documents (title, source, status) + VALUES ($1, $2, 'processing') + RETURNING id`, + [document.title, document.source ?? null], + ); + + const id = result.rows[0]?.id; + + if (!id) { + throw new Error("Failed to create document"); + } + + return id; +}; + +export const storeDocumentChunks = async ( + documentId: string, + chunks: StoredChunk[], +): Promise => { + const client = await pool.connect(); + + try { + await client.query("BEGIN"); + + for (const chunk of chunks) { + await client.query( + `INSERT INTO document_chunks + (document_id, chunk_index, content, embedding) + VALUES ($1, $2, $3, $4::vector)`, + [ + documentId, + chunk.chunkIndex, + chunk.content, + `[${chunk.embedding.join(",")}]`, + ], + ); + } + + await client.query( + `UPDATE documents + SET status = 'completed' + WHERE id = $1`, + [documentId], + ); + + await client.query("COMMIT"); + } catch (error) { + try { + await client.query("ROLLBACK"); + } catch { + // Preserve the original error. + } + + throw error; + } finally { + client.release(); + } +};