From c922762dfb415c6ef2480f8839045edf5d6a38fc Mon Sep 17 00:00:00 2001 From: Christian Mueller Date: Mon, 6 Apr 2026 12:56:28 +0200 Subject: [PATCH] feat(postgres): add PG source manager and query scheduler (Task 11) Co-Authored-By: Claude Sonnet 4.6 --- .../src/modules/postgres/query-scheduler.ts | 152 ++++++++++++++ .../src/modules/postgres/source-manager.ts | 188 ++++++++++++++++++ 2 files changed, 340 insertions(+) create mode 100644 display/middleware/src/modules/postgres/query-scheduler.ts create mode 100644 display/middleware/src/modules/postgres/source-manager.ts diff --git a/display/middleware/src/modules/postgres/query-scheduler.ts b/display/middleware/src/modules/postgres/query-scheduler.ts new file mode 100644 index 0000000..a1becdb --- /dev/null +++ b/display/middleware/src/modules/postgres/query-scheduler.ts @@ -0,0 +1,152 @@ +import { getDb } from '../../db/index'; +import { getPool } from './source-manager'; +import { getDatapointByQueryId, updateDatapointValue } from '../datapoints/datapoint-manager'; + +export interface PgQuery { + id: number; + source_id: number; + name: string; + query: string; + interval_sec: number; + result_column: string; + last_result: string | null; + last_run_at: string | null; +} + +export interface CreateQueryData { + source_id: number; + name: string; + query: string; + interval_sec?: number; + result_column: string; +} + +export type UpdateQueryData = Partial>; + +const timers = new Map(); + +export function getAllQueries(sourceId: number): PgQuery[] { + const db = getDb(); + return db + .prepare('SELECT * FROM pg_queries WHERE source_id = ? ORDER BY id ASC') + .all(sourceId) as PgQuery[]; +} + +export function getQueryById(id: number): PgQuery | undefined { + const db = getDb(); + return db.prepare('SELECT * FROM pg_queries WHERE id = ?').get(id) as PgQuery | undefined; +} + +export function createQuery(data: CreateQueryData): PgQuery { + const db = getDb(); + const result = db.run( + `INSERT INTO pg_queries (source_id, name, query, interval_sec, result_column) + VALUES (?, ?, ?, ?, ?)`, + [ + data.source_id, + data.name, + data.query, + data.interval_sec ?? 60, + data.result_column, + ] + ); + const id = result.lastInsertRowid as number; + return db.prepare('SELECT * FROM pg_queries WHERE id = ?').get(id) as PgQuery; +} + +export function updateQuery(id: number, data: UpdateQueryData): PgQuery | undefined { + const db = getDb(); + const fields: string[] = []; + const values: unknown[] = []; + + if (data.name !== undefined) { fields.push('name = ?'); values.push(data.name); } + if (data.query !== undefined) { fields.push('query = ?'); values.push(data.query); } + if (data.interval_sec !== undefined) { fields.push('interval_sec = ?'); values.push(data.interval_sec); } + if (data.result_column !== undefined) { fields.push('result_column = ?'); values.push(data.result_column); } + + if (fields.length === 0) return getQueryById(id); + + values.push(id); + db.run(`UPDATE pg_queries SET ${fields.join(', ')} WHERE id = ?`, values); + return getQueryById(id); +} + +export function deleteQuery(id: number): boolean { + const db = getDb(); + stopQueryTimer(id); + const result = db.run('DELETE FROM pg_queries WHERE id = ?', [id]); + return (result.changes as number) > 0; +} + +async function executeQuery(queryId: number): Promise { + const query = getQueryById(queryId); + if (!query) return; + + const pool = getPool(query.source_id); + if (!pool) { + console.warn(`No pool for PG source ${query.source_id}, skipping query ${queryId}`); + return; + } + + try { + const result = await pool.query(query.query); + const row = result.rows[0]; + if (!row) return; + + const rawValue = row[query.result_column]; + if (rawValue === undefined || rawValue === null) return; + + const valueStr = String(rawValue); + + const db = getDb(); + db.run( + `UPDATE pg_queries SET last_result = ?, last_run_at = datetime('now') WHERE id = ?`, + [valueStr, queryId] + ); + + const dp = getDatapointByQueryId(queryId); + if (dp) { + updateDatapointValue(dp.id, valueStr); + } + } catch (err) { + console.error(`Query ${queryId} execution error:`, (err as Error).message); + } +} + +export function startQueryTimer(queryId: number): void { + const query = getQueryById(queryId); + if (!query) return; + + // Stop existing timer if any + stopQueryTimer(queryId); + + // Execute immediately + void executeQuery(queryId); + + const intervalMs = query.interval_sec * 1000; + const timer = setInterval(() => { + void executeQuery(queryId); + }, intervalMs); + + timers.set(queryId, timer); +} + +export function stopQueryTimer(queryId: number): void { + const timer = timers.get(queryId); + if (timer) { + clearInterval(timer); + timers.delete(queryId); + } +} + +export function startAllQueryTimers(): void { + const db = getDb(); + const queries = db.prepare('SELECT * FROM pg_queries').all() as PgQuery[]; + + for (const query of queries) { + const pool = getPool(query.source_id); + if (pool) { + startQueryTimer(query.id); + } + } +} diff --git a/display/middleware/src/modules/postgres/source-manager.ts b/display/middleware/src/modules/postgres/source-manager.ts new file mode 100644 index 0000000..0bb8d61 --- /dev/null +++ b/display/middleware/src/modules/postgres/source-manager.ts @@ -0,0 +1,188 @@ +import { Pool } from 'pg'; +import { getDb } from '../../db/index'; + +export interface PgSource { + id: number; + name: string; + host: string; + port: number; + database_name: string; + username: string; + password: string; + use_tls: number; + enabled: number; + status: string; + created_at: string; +} + +export interface CreateSourceData { + name: string; + host: string; + port?: number; + database_name: string; + username: string; + password: string; + use_tls?: number; + enabled?: number; +} + +export type UpdateSourceData = Partial; + +const pools = new Map(); + +export function getAllSources(): PgSource[] { + const db = getDb(); + return db.prepare('SELECT * FROM pg_sources ORDER BY id ASC').all() as PgSource[]; +} + +export function getSourceById(id: number): PgSource | undefined { + const db = getDb(); + return db.prepare('SELECT * FROM pg_sources WHERE id = ?').get(id) as PgSource | undefined; +} + +export function createSource(data: CreateSourceData): PgSource { + const db = getDb(); + const result = db.run( + `INSERT INTO pg_sources (name, host, port, database_name, username, password, use_tls, enabled) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + [ + data.name, + data.host, + data.port ?? 5432, + data.database_name, + data.username, + data.password, + data.use_tls ?? 0, + data.enabled ?? 1, + ] + ); + const id = result.lastInsertRowid as number; + return db.prepare('SELECT * FROM pg_sources WHERE id = ?').get(id) as PgSource; +} + +export function updateSource(id: number, data: UpdateSourceData): PgSource | undefined { + const db = getDb(); + const fields: string[] = []; + const values: unknown[] = []; + + if (data.name !== undefined) { fields.push('name = ?'); values.push(data.name); } + if (data.host !== undefined) { fields.push('host = ?'); values.push(data.host); } + if (data.port !== undefined) { fields.push('port = ?'); values.push(data.port); } + if (data.database_name !== undefined) { fields.push('database_name = ?'); values.push(data.database_name); } + if (data.username !== undefined) { fields.push('username = ?'); values.push(data.username); } + if (data.password !== undefined) { fields.push('password = ?'); values.push(data.password); } + if (data.use_tls !== undefined) { fields.push('use_tls = ?'); values.push(data.use_tls); } + if (data.enabled !== undefined) { fields.push('enabled = ?'); values.push(data.enabled); } + + if (fields.length === 0) return getSourceById(id); + + values.push(id); + db.run(`UPDATE pg_sources SET ${fields.join(', ')} WHERE id = ?`, values); + return getSourceById(id); +} + +export function deleteSource(id: number): boolean { + const db = getDb(); + disconnectSource(id); + const result = db.run('DELETE FROM pg_sources WHERE id = ?', [id]); + return (result.changes as number) > 0; +} + +function setStatus(id: number, status: string): void { + const db = getDb(); + db.run('UPDATE pg_sources SET status = ? WHERE id = ?', [status, id]); +} + +export async function connectSource(id: number): Promise { + const source = getSourceById(id); + if (!source) throw new Error(`PG source ${id} not found`); + + // Disconnect existing pool + if (pools.has(id)) { + await pools.get(id)!.end(); + pools.delete(id); + } + + const pool = new Pool({ + host: source.host, + port: source.port, + database: source.database_name, + user: source.username, + password: source.password, + ssl: source.use_tls ? { rejectUnauthorized: false } : undefined, + max: 5, + idleTimeoutMillis: 30000, + connectionTimeoutMillis: 10000, + }); + + // Test connection + try { + const client = await pool.connect(); + client.release(); + setStatus(id, 'connected'); + pools.set(id, pool); + return pool; + } catch (err) { + setStatus(id, 'error'); + await pool.end(); + throw err; + } +} + +export function disconnectSource(id: number): void { + const pool = pools.get(id); + if (pool) { + pool.end().catch((err) => console.error(`Error closing PG pool ${id}:`, err)); + pools.delete(id); + setStatus(id, 'disconnected'); + } +} + +export function getPool(id: number): Pool | undefined { + return pools.get(id); +} + +export async function testSource(data: { + host: string; + port?: number; + database_name: string; + username: string; + password: string; + use_tls?: number; +}): Promise { + const pool = new Pool({ + host: data.host, + port: data.port ?? 5432, + database: data.database_name, + user: data.username, + password: data.password, + ssl: data.use_tls ? { rejectUnauthorized: false } : undefined, + connectionTimeoutMillis: 5000, + max: 1, + }); + + try { + const client = await pool.connect(); + client.release(); + await pool.end(); + return true; + } catch { + await pool.end().catch(() => {}); + return false; + } +} + +export async function connectAllEnabledSources(): Promise { + const db = getDb(); + const sources = db + .prepare('SELECT * FROM pg_sources WHERE enabled = 1') + .all() as PgSource[]; + + for (const source of sources) { + try { + await connectSource(source.id); + } catch (err) { + console.error(`Failed to connect PG source ${source.id}:`, (err as Error).message); + } + } +}