diff --git a/apps/web/src/app/api/sql-client/connect/route.ts b/apps/web/src/app/api/sql-client/connect/route.ts index 61f2cd6b..147e61cd 100644 --- a/apps/web/src/app/api/sql-client/connect/route.ts +++ b/apps/web/src/app/api/sql-client/connect/route.ts @@ -1,7 +1,6 @@ import { requireBackendSession } from "@/lib/require-backend-session"; import { NextResponse } from "next/server"; -import { Pool as PgPool } from "pg"; -import mysql from "mysql2/promise"; +import { getSqlPool, releaseSqlPool, SqlDbType } from "@/lib/sql-client-pool"; export async function POST(request: Request) { const authError = await requireBackendSession(request); @@ -14,37 +13,29 @@ export async function POST(request: Request) { return NextResponse.json({ error: "type, host, and database are required" }, { status: 400 }); } - if (type === "postgresql") { - const pool = new PgPool({ + if (type === "postgresql" || type === "mysql" || type === "mariadb") { + const handle = await getSqlPool({ + type: type as SqlDbType, host, - port: port || 5432, + port: Number(port) || 0, database, - user: username, + username, password, - ssl: ssl ? { rejectUnauthorized: false } : false, - connectionTimeoutMillis: 10000, + ssl: Boolean(ssl), }); - const client = await pool.connect(); - const res = await client.query("SELECT version()"); - client.release(); - await pool.end(); - return NextResponse.json({ success: true, version: res.rows[0]?.version }); - } - if (type === "mysql" || type === "mariadb") { - const conn = await mysql.createConnection({ - host, - port: port || 3306, - database, - user: username, - password, - ssl: ssl ? { rejectUnauthorized: false } : undefined, - connectTimeout: 10000, - }); - const [rows] = await conn.execute("SELECT VERSION() as version"); - await conn.end(); - const version = (rows as { version: string }[])[0]?.version; - return NextResponse.json({ success: true, version }); + try { + if (handle.pg) { + const res = await handle.pg.query("SELECT version()"); + return NextResponse.json({ success: true, version: res.rows[0]?.version }); + } + + const [rows] = await handle.mysql!.query("SELECT VERSION() as version"); + const version = (rows as { version: string }[])[0]?.version; + return NextResponse.json({ success: true, version }); + } finally { + releaseSqlPool(handle.key); + } } return NextResponse.json({ error: `Unsupported database type: ${type}` }, { status: 400 }); diff --git a/apps/web/src/app/api/sql-client/query/route.ts b/apps/web/src/app/api/sql-client/query/route.ts index 45cad313..6afcc697 100644 --- a/apps/web/src/app/api/sql-client/query/route.ts +++ b/apps/web/src/app/api/sql-client/query/route.ts @@ -1,109 +1,89 @@ import { requireBackendSession } from "@/lib/require-backend-session"; import { NextResponse } from "next/server"; -import { Pool as PgPool } from "pg"; -import mysql from "mysql2/promise"; +import { getSqlPool, releaseSqlPool, SqlDbType } from "@/lib/sql-client-pool"; +import { splitSqlStatements } from "@/lib/sql-split"; const MAX_ROWS = 5000; -const TIMEOUT_MS = 30000; -export async function POST(request: Request) { - const authError = await requireBackendSession(request); - if (authError) return authError; - - try { - const { type, host, port, database, username, password, ssl, query, limit } = - await request.json(); - - if (!type || !host || !database || !query) { - return NextResponse.json( - { error: "type, host, database, and query are required" }, - { status: 400 } - ); - } - - const rowLimit = Math.min(Number(limit) || 500, MAX_ROWS); - const start = Date.now(); - - if (type === "postgresql") { - const pool = new PgPool({ - host, - port: port || 5432, - database, - user: username, - password, - ssl: ssl ? { rejectUnauthorized: false } : false, - connectionTimeoutMillis: TIMEOUT_MS, - statement_timeout: TIMEOUT_MS, - }); - - const client = await pool.connect(); - try { - // Split statements by semicolon and execute each, returning last result - const statements = query - .split(";") - .map((s: string) => s.trim()) - .filter(Boolean); - - let result = null; - for (const stmt of statements) { - result = await client.query(stmt); - } - - const elapsed = Date.now() - start; - const rows = result?.rows?.slice(0, rowLimit) ?? []; - const columns = result?.fields?.map((f: { name: string }) => f.name) ?? []; - const rowCount = result?.rowCount ?? rows.length; - - return NextResponse.json({ rows, columns, rowCount, executionTime: elapsed }); - } finally { - client.release(); - await pool.end(); - } - } - - if (type === "mysql" || type === "mariadb") { - const conn = await mysql.createConnection({ - host, - port: port || 3306, - database, - user: username, - password, - ssl: ssl ? { rejectUnauthorized: false } : undefined, - connectTimeout: TIMEOUT_MS, - multipleStatements: true, - }); - - try { - const [rawRows, rawFields] = await conn.execute(query); - const elapsed = Date.now() - start; +interface StatementResult { + rows: Record[]; + columns: string[]; + rowCount: number; +} - // multipleStatements may return arrays of result sets - const isMulti = Array.isArray(rawRows) && Array.isArray(rawRows[0]); - const rows = isMulti - ? (((rawRows as unknown) as unknown[][]).at(-1) as Record[]) ?? [] - : (rawRows as Record[]); +export async function POST(request: Request) { + const authError = await requireBackendSession(request); + if (authError) return authError; + + try { + const { type, host, port, database, username, password, ssl, query, limit } = + await request.json(); + + if (!type || !host || !database || !query) { + return NextResponse.json( + { error: "type, host, database, and query are required" }, + { status: 400 } + ); + } - const fields = isMulti - ? (((rawFields as unknown) as unknown[][]).at(-1) as { name: string }[]) ?? [] - : (rawFields as { name: string }[]); + const rowLimit = Math.min(Number(limit) || 500, MAX_ROWS); + const start = Date.now(); + const statements = splitSqlStatements(query); + if (statements.length === 0) { + return NextResponse.json({ error: "No SQL statement provided" }, { status: 400 }); + } - const slicedRows = Array.isArray(rows) ? rows.slice(0, rowLimit) : []; - const columns = Array.isArray(fields) ? fields.map((f) => f.name) : []; + const handle = await getSqlPool({ + type: type as SqlDbType, + host, + port: Number(port) || 0, + database, + username, + password, + ssl: Boolean(ssl), + }); - return NextResponse.json({ - rows: slicedRows, - columns, - rowCount: Array.isArray(rows) ? rows.length : 0, - executionTime: elapsed, - }); - } finally { - await conn.end(); - } + try { + const results: StatementResult[] = []; + + for (const stmt of statements) { + // ponytail: cooperative abort between statements only. True mid-statement + // cancel needs pg_cancel_backend / conn.destroy(); add if long queries need killing. + if (request.signal.aborted) throw new Error("Query aborted"); + + if (handle.pg) { + const r = await handle.pg.query(stmt); + const rows = (r.rows ?? []) as Record[]; + results.push({ + rows: rows.slice(0, rowLimit), + columns: r.fields?.map((f: { name: string }) => f.name) ?? [], + rowCount: r.rowCount ?? rows.length, + }); + } else if (handle.mysql) { + const [rawRows, rawFields] = await handle.mysql.query(stmt); + const rows = Array.isArray(rawRows) ? (rawRows as Record[]) : []; + const fields = Array.isArray(rawFields) ? (rawFields as { name: string }[]) : []; + results.push({ + rows: rows.slice(0, rowLimit), + columns: fields.map((f) => f.name), + rowCount: rows.length, + }); } - - return NextResponse.json({ error: `Unsupported database type: ${type}` }, { status: 400 }); - } catch (error: unknown) { - const message = error instanceof Error ? error.message : String(error); - return NextResponse.json({ error: message }, { status: 500 }); + } + + const last = results[results.length - 1] ?? { rows: [], columns: [], rowCount: 0 }; + return NextResponse.json({ + rows: last.rows, + columns: last.columns, + rowCount: last.rowCount, + executionTime: Date.now() - start, + results, + }); + } finally { + releaseSqlPool(handle.key); } + } catch (error: unknown) { + const message = error instanceof Error ? error.message : String(error); + return NextResponse.json({ error: message }, { status: 500 }); + } } diff --git a/apps/web/src/app/api/sql-client/tables/route.ts b/apps/web/src/app/api/sql-client/tables/route.ts index 8b52931d..9839b65a 100644 --- a/apps/web/src/app/api/sql-client/tables/route.ts +++ b/apps/web/src/app/api/sql-client/tables/route.ts @@ -1,7 +1,6 @@ import { requireBackendSession } from "@/lib/require-backend-session"; import { NextResponse } from "next/server"; -import { Pool as PgPool } from "pg"; -import mysql from "mysql2/promise"; +import { getSqlPool, releaseSqlPool, SqlDbType } from "@/lib/sql-client-pool"; export async function POST(request: Request) { const authError = await requireBackendSession(request); @@ -14,92 +13,80 @@ export async function POST(request: Request) { return NextResponse.json({ error: "type, host, and database are required" }, { status: 400 }); } - if (type === "postgresql") { - const pool = new PgPool({ + if (type === "postgresql" || type === "mysql" || type === "mariadb") { + const handle = await getSqlPool({ + type: type as SqlDbType, host, - port: port || 5432, + port: Number(port) || 0, database, - user: username, + username, password, - ssl: ssl ? { rejectUnauthorized: false } : false, - connectionTimeoutMillis: 10000, + ssl: Boolean(ssl), }); - const client = await pool.connect(); - const tablesRes = await client.query(` - SELECT - t.table_schema AS schema, - t.table_name AS name, - t.table_type AS type, - ( - SELECT COUNT(*)::int - FROM information_schema.columns c - WHERE c.table_schema = t.table_schema AND c.table_name = t.table_name - ) AS column_count - FROM information_schema.tables t - WHERE t.table_schema NOT IN ('pg_catalog', 'information_schema') - ORDER BY t.table_schema, t.table_name - `); - - const columnsRes = await client.query(` - SELECT - c.table_schema AS schema, - c.table_name, - c.column_name, - c.data_type, - c.is_nullable, - c.column_default, - c.ordinal_position - FROM information_schema.columns c - WHERE c.table_schema NOT IN ('pg_catalog', 'information_schema') - ORDER BY c.table_schema, c.table_name, c.ordinal_position - `); - - client.release(); - await pool.end(); - - return NextResponse.json({ tables: tablesRes.rows, columns: columnsRes.rows }); - } + try { + if (handle.pg) { + const tablesRes = await handle.pg.query(` + SELECT + t.table_schema AS schema, + t.table_name AS name, + t.table_type AS type, + ( + SELECT COUNT(*)::int + FROM information_schema.columns c + WHERE c.table_schema = t.table_schema AND c.table_name = t.table_name + ) AS column_count + FROM information_schema.tables t + WHERE t.table_schema NOT IN ('pg_catalog', 'information_schema') + ORDER BY t.table_schema, t.table_name + `); - if (type === "mysql" || type === "mariadb") { - const conn = await mysql.createConnection({ - host, - port: port || 3306, - database, - user: username, - password, - ssl: ssl ? { rejectUnauthorized: false } : undefined, - connectTimeout: 10000, - }); + const columnsRes = await handle.pg.query(` + SELECT + c.table_schema AS schema, + c.table_name, + c.column_name, + c.data_type, + c.is_nullable, + c.column_default, + c.ordinal_position + FROM information_schema.columns c + WHERE c.table_schema NOT IN ('pg_catalog', 'information_schema') + ORDER BY c.table_schema, c.table_name, c.ordinal_position + `); - const [tables] = await conn.execute(` - SELECT - TABLE_SCHEMA AS \`schema\`, - TABLE_NAME AS name, - TABLE_TYPE AS type, - TABLE_ROWS AS row_estimate - FROM information_schema.TABLES - WHERE TABLE_SCHEMA = DATABASE() - ORDER BY TABLE_NAME - `); + return NextResponse.json({ tables: tablesRes.rows, columns: columnsRes.rows }); + } - const [columns] = await conn.execute(` - SELECT - TABLE_SCHEMA AS \`schema\`, - TABLE_NAME AS table_name, - COLUMN_NAME AS column_name, - DATA_TYPE AS data_type, - IS_NULLABLE AS is_nullable, - COLUMN_DEFAULT AS column_default, - ORDINAL_POSITION AS ordinal_position - FROM information_schema.COLUMNS - WHERE TABLE_SCHEMA = DATABASE() - ORDER BY TABLE_NAME, ORDINAL_POSITION - `); + const [tables] = await handle.mysql!.query(` + SELECT + TABLE_SCHEMA AS \`schema\`, + TABLE_NAME AS name, + TABLE_TYPE AS type, + TABLE_ROWS AS row_estimate + FROM information_schema.TABLES + WHERE TABLE_SCHEMA = DATABASE() + ORDER BY TABLE_NAME + `); - await conn.end(); + const [columns] = await handle.mysql!.query(` + SELECT + TABLE_SCHEMA AS \`schema\`, + TABLE_NAME AS table_name, + COLUMN_NAME AS column_name, + DATA_TYPE AS data_type, + IS_NULLABLE AS is_nullable, + COLUMN_DEFAULT AS column_default, + ORDINAL_POSITION AS ordinal_position + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + ORDER BY TABLE_NAME, ORDINAL_POSITION + `); - return NextResponse.json({ tables, columns }); + return NextResponse.json({ tables, columns }); + } finally { + releaseSqlPool(handle.key); + } } return NextResponse.json({ error: `Unsupported database type: ${type}` }, { status: 400 }); diff --git a/apps/web/src/lib/__tests__/sql-client-pool.test.ts b/apps/web/src/lib/__tests__/sql-client-pool.test.ts new file mode 100644 index 00000000..7c491aca --- /dev/null +++ b/apps/web/src/lib/__tests__/sql-client-pool.test.ts @@ -0,0 +1,24 @@ +import { poolKey, SqlConnParams } from "@/lib/sql-client-pool"; + +const base: SqlConnParams = { + type: "postgresql", + host: "localhost", + port: 5432, + database: "app", + username: "u", + password: "p", + ssl: false, +}; + +describe("poolKey", () => { + it("is stable for identical params", () => { + expect(poolKey(base)).toBe(poolKey({ ...base })); + }); + + it("differs when any connection field differs", () => { + expect(poolKey(base)).not.toBe(poolKey({ ...base, database: "other" })); + expect(poolKey(base)).not.toBe(poolKey({ ...base, port: 5433 })); + expect(poolKey(base)).not.toBe(poolKey({ ...base, password: "p2" })); + expect(poolKey(base)).not.toBe(poolKey({ ...base, ssl: true })); + }); +}); diff --git a/apps/web/src/lib/__tests__/sql-split.test.ts b/apps/web/src/lib/__tests__/sql-split.test.ts new file mode 100644 index 00000000..9d3f2445 --- /dev/null +++ b/apps/web/src/lib/__tests__/sql-split.test.ts @@ -0,0 +1,59 @@ +import { splitSqlStatements } from "@/lib/sql-split"; + +describe("splitSqlStatements", () => { + it("splits simple statements", () => { + expect(splitSqlStatements("SELECT 1; SELECT 2")).toEqual(["SELECT 1", "SELECT 2"]); + }); + + it("ignores semicolons inside single-quoted strings", () => { + expect(splitSqlStatements("SELECT 'a;b'; SELECT 2")).toEqual(["SELECT 'a;b'", "SELECT 2"]); + }); + + it("handles escaped single quotes ('')", () => { + expect(splitSqlStatements("SELECT 'it''s;fine'; SELECT 2")).toEqual([ + "SELECT 'it''s;fine'", + "SELECT 2", + ]); + }); + + it("ignores semicolons in line comments", () => { + expect(splitSqlStatements("SELECT 1 -- a;b\n; SELECT 2")).toEqual([ + "SELECT 1 -- a;b", + "SELECT 2", + ]); + }); + + it("ignores semicolons in block comments", () => { + expect(splitSqlStatements("SELECT 1 /* a;b */; SELECT 2")).toEqual([ + "SELECT 1 /* a;b */", + "SELECT 2", + ]); + }); + + it("ignores semicolons inside dollar-quoted bodies", () => { + const sql = "CREATE FUNCTION f() RETURNS void AS $$ BEGIN; END; $$ LANGUAGE plpgsql; SELECT 1"; + expect(splitSqlStatements(sql)).toEqual([ + "CREATE FUNCTION f() RETURNS void AS $$ BEGIN; END; $$ LANGUAGE plpgsql", + "SELECT 1", + ]); + }); + + it("ignores semicolons in tagged dollar quotes", () => { + expect(splitSqlStatements("SELECT $tag$a;b$tag$; SELECT 2")).toEqual([ + "SELECT $tag$a;b$tag$", + "SELECT 2", + ]); + }); + + it("ignores semicolons inside backtick identifiers (mysql)", () => { + expect(splitSqlStatements("SELECT `a;b`; SELECT 2")).toEqual(["SELECT `a;b`", "SELECT 2"]); + }); + + it("drops trailing empty statement and whitespace-only chunks", () => { + expect(splitSqlStatements("SELECT 1; ; ")).toEqual(["SELECT 1"]); + }); + + it("returns empty array for blank input", () => { + expect(splitSqlStatements(" ")).toEqual([]); + }); +}); diff --git a/apps/web/src/lib/sql-client-pool.ts b/apps/web/src/lib/sql-client-pool.ts new file mode 100644 index 00000000..7264d866 --- /dev/null +++ b/apps/web/src/lib/sql-client-pool.ts @@ -0,0 +1,143 @@ +import { Pool as PgPool } from "pg"; +import mysql from "mysql2/promise"; + +export type SqlDbType = "postgresql" | "mysql" | "mariadb"; + +export interface SqlConnParams { + type: SqlDbType; + host: string; + port: number; + database: string; + username: string; + password: string; + ssl: boolean; +} + +export interface SqlPoolHandle { + key: string; + type: SqlDbType; + pg?: PgPool; + mysql?: mysql.Pool; +} + +interface PooledSql { + key: string; + type: SqlDbType; + pg?: PgPool; + mysql?: mysql.Pool; + lastUsed: number; + refCount: number; +} + +const IDLE_TIMEOUT_MS = 300000; // 5 minutes +const STATEMENT_TIMEOUT_MS = 30000; +const MAX_DRIVER_POOL = 5; + +export function poolKey(p: SqlConnParams): string { + return JSON.stringify([p.type, p.host, p.port, p.database, p.username, p.password, p.ssl]); +} + +// ponytail: idle-evict, not a hard-capped LRU — distinct connections grow unbounded +// until the 60s idle sweep reclaims them (same as nosql-client-pool). Add max-entries +// eviction if a user cycles through many connections faster than the 5-min idle timeout. +class SqlPoolManager { + private static instance: SqlPoolManager; + private pool = new Map(); + private cleanupInterval: NodeJS.Timeout | null = null; + + private constructor() { + this.cleanupInterval = setInterval(() => this.cleanup(), 60000); + } + + static getInstance(): SqlPoolManager { + if (!SqlPoolManager.instance) SqlPoolManager.instance = new SqlPoolManager(); + return SqlPoolManager.instance; + } + + async get(p: SqlConnParams): Promise { + // ponytail: no in-flight `connecting` guard (unlike nosql-client-pool) because + // pg.Pool / mysql.createPool are synchronous & lazy — check-then-set below is + // atomic with no await between. If construction ever awaits (eager warm-up), add + // a connecting-map guard or the first set() becomes an unreleasable orphan. + const key = poolKey(p); + const existing = this.pool.get(key); + if (existing) { + existing.lastUsed = Date.now(); + existing.refCount++; + return { key, type: existing.type, pg: existing.pg, mysql: existing.mysql }; + } + + const entry: PooledSql = { key, type: p.type, lastUsed: Date.now(), refCount: 1 }; + + if (p.type === "postgresql") { + entry.pg = new PgPool({ + host: p.host, + port: p.port || 5432, + database: p.database, + user: p.username, + password: p.password, + ssl: p.ssl ? { rejectUnauthorized: false } : false, + max: MAX_DRIVER_POOL, + idleTimeoutMillis: 30000, + connectionTimeoutMillis: 10000, + statement_timeout: STATEMENT_TIMEOUT_MS, + }); + } else { + entry.mysql = mysql.createPool({ + host: p.host, + port: p.port || 3306, + database: p.database, + user: p.username, + password: p.password, + ssl: p.ssl ? { rejectUnauthorized: false } : undefined, + connectionLimit: MAX_DRIVER_POOL, + connectTimeout: 10000, + // We split statements ourselves (lib/sql-split), so no multipleStatements. + }); + } + + this.pool.set(key, entry); + return { key, type: entry.type, pg: entry.pg, mysql: entry.mysql }; + } + + release(key: string): void { + const entry = this.pool.get(key); + if (entry) { + entry.refCount = Math.max(0, entry.refCount - 1); + entry.lastUsed = Date.now(); + } + } + + private cleanup(): void { + const now = Date.now(); + for (const [key, entry] of this.pool.entries()) { + if (entry.refCount === 0 && now - entry.lastUsed > IDLE_TIMEOUT_MS) { + this.closeEntry(entry); + this.pool.delete(key); + } + } + } + + private closeEntry(entry: PooledSql): void { + entry.pg?.end().catch(console.error); + entry.mysql?.end().catch(console.error); + } + + async closeAll(): Promise { + if (this.cleanupInterval) clearInterval(this.cleanupInterval); + for (const entry of this.pool.values()) this.closeEntry(entry); + this.pool.clear(); + } +} + +export async function getSqlPool(p: SqlConnParams): Promise { + return SqlPoolManager.getInstance().get(p); +} + +export function releaseSqlPool(key: string): void { + SqlPoolManager.getInstance().release(key); +} + +export async function closeAllSqlPools(): Promise { + await SqlPoolManager.getInstance().closeAll(); +} diff --git a/apps/web/src/lib/sql-split.ts b/apps/web/src/lib/sql-split.ts new file mode 100644 index 00000000..7db7b453 --- /dev/null +++ b/apps/web/src/lib/sql-split.ts @@ -0,0 +1,96 @@ +/** + * Split a SQL script into individual statements, splitting only on top-level + * semicolons. Semicolons inside string literals, quoted identifiers, comments, + * and Postgres dollar-quoted bodies are ignored. + */ +export function splitSqlStatements(sql: string): string[] { + const statements: string[] = []; + let current = ""; + let i = 0; + const n = sql.length; + + const push = () => { + const trimmed = current.trim(); + if (trimmed) statements.push(trimmed); + current = ""; + }; + + while (i < n) { + const ch = sql[i]; + const next = sql[i + 1]; + + // Line comment: -- ... until newline + if (ch === "-" && next === "-") { + while (i < n && sql[i] !== "\n") current += sql[i++]; + continue; + } + + // Block comment: /* ... */ + if (ch === "/" && next === "*") { + current += ch; + current += next; + i += 2; + while (i < n && !(sql[i] === "*" && sql[i + 1] === "/")) current += sql[i++]; + if (i < n) { + current += sql[i]; // * + current += sql[i + 1]; // / + i += 2; + } + continue; + } + + // Single- or double-quoted string, or backtick identifier + if (ch === "'" || ch === '"' || ch === "`") { + const quote = ch; + current += ch; + i++; + while (i < n) { + current += sql[i]; + // Escaped quote by doubling (e.g. '') stays inside the string + if (sql[i] === quote && sql[i + 1] === quote) { + current += sql[i + 1]; + i += 2; + continue; + } + if (sql[i] === quote) { + i++; + break; + } + i++; + } + continue; + } + + // Dollar-quoted body: $tag$ ... $tag$ (tag may be empty: $$) + if (ch === "$") { + const tagMatch = /^\$([A-Za-z0-9_]*)\$/.exec(sql.slice(i)); + if (tagMatch) { + const tag = tagMatch[0]; // includes both $ delimiters + current += tag; + i += tag.length; + const end = sql.indexOf(tag, i); + if (end === -1) { + current += sql.slice(i); + i = n; + } else { + current += sql.slice(i, end + tag.length); + i = end + tag.length; + } + continue; + } + } + + // Top-level statement terminator + if (ch === ";") { + push(); + i++; + continue; + } + + current += ch; + i++; + } + + push(); + return statements; +}