Infrastructure build to support third-party app integrations

backend/db/notes.dao.ts

@@ -0,0 +1,125 @@
+import QueryStream from 'pg-query-stream';
+import { pipeline } from 'node:stream/promises';
+import { Readable } from 'node:stream';
+import { query, getPool } from './index.js';
+import { batch } from '../lib/streams.js';
+import type { Note, NoteInput } from '../types/domain.js';
+
+const SELECT_NOTES = 'SELECT id, text, x, y, author, color FROM notes ORDER BY id';
+
+// Postgres caps a statement at 65535 bind parameters; seven columns per note
+// leaves 9362 as the hard ceiling.
+const INSERT_BATCH_SIZE = 1000;
+
+const INSERT_COLUMNS = 7;
+
+export const getAllNotes = async (): Promise<Note[]> => {
+    const { rows } = await query<Note>(SELECT_NOTES);
+    return rows;
+};
+
+/**
+ * Streams every note as an object-mode Readable. The pooled client is released on end, error, or
+ * destruction by consumer.
+ */
+export const streamAllNotes = async (): Promise<Readable> => {
+    const client = await getPool().connect();
+
+    let released = false;
+    const release = () => {
+        if (released) return;
+        released = true;
+        client.release();
+    };
+
+    try {
+        const rows = client.query(new QueryStream(SELECT_NOTES)) as unknown as Readable;
+        rows.once('end', release);
+        rows.once('error', release);
+        rows.once('close', release);
+        return rows;
+    } catch (err) {
+        release();
+        throw err;
+    }
+};
+
+export const getNoteById = async (id: string): Promise<Note | null> => {
+    const { rows } = await query<Note>(
+        'SELECT id, text, x, y, author, color FROM notes WHERE id = $1',
+        [id]
+    );
+    return rows[0] || null;
+};
+
+export const createNote = async (note: NoteInput): Promise<Note> => {
+    const { rows } = await query<Note>(
+        `INSERT INTO notes (id, text, x, y, author, color, source_meta)
+     VALUES ($1, $2, $3, $4, $5, $6, $7)
+     RETURNING id, text, x, y, author, color`,
+        [
+            note.id,
+            note.text,
+            note.x ?? 0,
+            note.y ?? 0,
+            note.author,
+            note.color ?? 'yellow',
+            JSON.stringify(note.sourceMeta ?? {}),
+        ]
+    );
+    return rows[0];
+};
+
+const insertNoteBatch = async (notes: NoteInput[]): Promise<Note[]> => {
+    const values: unknown[] = [];
+    const placeholders: string[] = [];
+
+    notes.forEach((note, i) => {
+        const offset = i * INSERT_COLUMNS;
+        const slots = Array.from(
+            { length: INSERT_COLUMNS },
+            (_, col) => `$${offset + col + 1}`
+        );
+        placeholders.push(`(${slots.join(', ')})`);
+        values.push(
+            note.id,
+            note.text,
+            note.x ?? 0,
+            note.y ?? 0,
+            note.author,
+            note.color ?? 'yellow',
+            JSON.stringify(note.sourceMeta ?? {})
+        );
+    });
+
+    // Redelivered webhooks can carry a note that already landed; skipping the
+    // conflict keeps ingestion idempotent at the row level as well.
+    const { rows } = await query<Note>(
+        `INSERT INTO notes (id, text, x, y, author, color, source_meta)
+     VALUES ${placeholders.join(', ')}
+     ON CONFLICT (id) DO NOTHING
+     RETURNING id, text, x, y, author, color`,
+        values
+    );
+    return rows;
+};
+
+export const createNotes = async (notes: NoteInput[]): Promise<Note[]> => {
+    if (!notes || notes.length === 0) {
+        return [];
+    }
+
+    const inserted: Note[] = [];
+
+    await pipeline(
+        Readable.from(notes, { objectMode: true }),
+        batch<NoteInput>(INSERT_BATCH_SIZE),
+        async (batches: AsyncIterable<NoteInput[]>) => {
+            for await (const chunk of batches) {
+                inserted.push(...await insertNoteBatch(chunk));
+            }
+        }
+    );
+
+    return inserted;
+};