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;
};