feat(postgres): add PG source manager and query scheduler (Task 11)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
2c9d218211
commit
c922762dfb
152
display/middleware/src/modules/postgres/query-scheduler.ts
Normal file
152
display/middleware/src/modules/postgres/query-scheduler.ts
Normal file
@ -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<Omit<CreateQueryData, 'source_id'>>;
|
||||
|
||||
const timers = new Map<number, NodeJS.Timeout>();
|
||||
|
||||
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<void> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
188
display/middleware/src/modules/postgres/source-manager.ts
Normal file
188
display/middleware/src/modules/postgres/source-manager.ts
Normal file
@ -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<CreateSourceData>;
|
||||
|
||||
const pools = new Map<number, Pool>();
|
||||
|
||||
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<Pool> {
|
||||
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<boolean> {
|
||||
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<void> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user