126 lines
3.6 KiB
TypeScript
126 lines
3.6 KiB
TypeScript
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;
|
|
};
|