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
This commit is contained in:
parent
dae610d2b2
commit
61978acb02
3 changed files with 90 additions and 2 deletions
|
|
@ -5,9 +5,9 @@
|
||||||
"main": "index.js",
|
"main": "index.js",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
"test": "tsx --test src/**/*.test.ts",
|
"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",
|
"build": "tsc -p tsconfig.json",
|
||||||
"start": "node dist/server.js"
|
"start": "node --env-file=.env dist/server.js"
|
||||||
},
|
},
|
||||||
"keywords": [],
|
"keywords": [],
|
||||||
"author": "",
|
"author": "",
|
||||||
|
|
|
||||||
14
src/db.ts
Normal file
14
src/db.ts
Normal file
|
|
@ -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);
|
||||||
|
});
|
||||||
74
src/document-repository.ts
Normal file
74
src/document-repository.ts
Normal file
|
|
@ -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<string> => {
|
||||||
|
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<void> => {
|
||||||
|
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();
|
||||||
|
}
|
||||||
|
};
|
||||||
Loading…
Add table
Reference in a new issue