@speles7172/sql-client

Run queries against AWS-hosted databases from any application, over a registry of named connections.

npm install @speles7172/sql-client

Four engines behind one API:

| Engine | Reached by | Use for | |---|---|---| | postgres | direct socket | a database this process can open a connection to | | remote-pg | bridge Lambda, or direct | PostgreSQL in another VPC or region | | athena | Athena API | querying S3 data | | neo4j | bridge Lambda | graph queries |

Register, then query

Registration is where a client declares which database, which engine, and which version it built against — the version is validated, not guessed, so a wrong assumption fails at startup rather than on a query months later.

import { registerDatabase, createSqlClient } from '@speles7172/sql-client';

registerDatabase({
  name: 'main',
  engine: 'postgres',
  version: '17',
  secretArn: 'arn:aws:secretsmanager:us-east-1:123456789012:secret:main-db',
  region: 'us-east-1',
});

const db = createSqlClient();

const users = await db.rows('main', 'SELECT id, email FROM users WHERE active = $1', {
  params: [true],
});

rows() returns the rows, one() the first row or null, and query() the full result including fields and counts.

Databases with expiring credentials (RDS IAM)

IAM authentication signs a password that is valid for about 15 minutes, so it is not a value you can store. Pass a function instead, and say how long what it returns stays good:

import { Signer } from '@aws-sdk/rds-signer';

const signer = new Signer({ hostname, port: 5432, username: 'app', region: 'us-east-1' });

registerDatabase({
  name: 'main',
  engine: 'postgres',
  version: '17',
  host: hostname,
  user: 'app',
  database: 'app',
  password: () => signer.getAuthToken(),
  credentialTtlMs: 10 * 60_000,   // re-sign well before the 15-minute expiry
});

The client re-mints the token when it expires and rebuilds the pool. Both halves matter: a pool keeps the password it opened with, so a token that expires mid-life leaves a pool that looks healthy while every new connection fails to authenticate.

Signing is local — no network call — which is what lets a Lambda in an isolated private subnet reach the database with no NAT gateway and no VPC endpoint.

Registering from the environment

Deployments that already carry a DB_CONNECTIONS JSON variable can load it as-is. PostgreSQL's own spellings (dbname, username) are accepted alongside this package's (database, user).

import { loadFromEnv } from '@speles7172/sql-client';

loadFromEnv(); // reads DB_CONNECTIONS

{
  "main":      { "secretArn": "arn:...", "region": "us-east-1", "version": "17" },
  "reporting": { "engine": "remote-pg", "secretArn": "arn:...", "region": "us-west-2",
                 "bridgeFunctionName": "db-bridge" },
  "lake":      { "engine": "athena", "outputLocation": "s3://results/", "database": "analytics" }
}

Pagination

Supplying page or pageSize turns it on. The statement is wrapped, not appended to, so a query with its own LIMIT, ORDER BY or UNION keeps its meaning:

const page = await db.query('main', 'SELECT * FROM events ORDER BY at DESC', {
  page: 2,
  pageSize: 50,
});
// { command: 'SELECT', rows: [...], totalCount: 4213, page: 2, pageSize: 50 }

LIMIT/OFFSET are interpolated because PostgreSQL accepts no placeholder there — they are integers this package computes, never caller text. Multiple statements are rejected rather than wrapped.

Running statements a human typed

query() is for an application talking to its own database: it does exactly what it is told. explore() is for anything a person types into a UI — a SQL console, an admin tool — where the default has to be that nothing is written by accident.

const result = await db.explore('main', sql, { maxRows: 500 });

It runs the statement on one connection, inside a transaction that is:

• READ ONLY unless writeMode, so a stray UPDATE fails at the engine rather than appearing to work and being silently undone — the user sees an error instead of believing a write happened;
• rolled back unless writeMode, which covers what READ ONLY does not, such as a function with side effects;
• bounded by a transaction-local statement_timeout, so a runaway query cannot pin the connection;
• capped at maxRows (1,000 by default, 10,000 ceiling), with truncated: true on the result when rows were dropped.

To let a statement persist, ask for it explicitly:

await db.explore('main', sql, { writeMode: true });

One statement per call. A batch is refused, because PostgreSQL's simple query protocol would run all of it — so COMMIT; DELETE FROM users would end the sandbox's transaction and then delete outside it, while the rollback at the end quietly did nothing. A semicolon inside a string literal or a comment is still fine; the check is a scanner, not a search for ;.

This is a second line of defence, not the first. The real guardrail is the database grant: a role with SELECT/INSERT/UPDATE/DELETE and no DDL cannot be talked into DROP TABLE by any client, whereas everything above is enforced by code that could have a bug in it. Give the console its own role.

Athena has no transactions, so explore() refuses there rather than implying a safety net that does not exist — restrict the query role instead.

Importing a CSV

importRows() loads parsed rows into a table. Unlike explore(), it commits.

const result = await db.importRows('main', {
  table: 'subscribers',
  mode: 'insert',
  errorMode: 'skip',
  matchColumns: [],
  fields: ['First', 'Last', 'Email'],   // the CSV headers
  rows: [['Ada', 'Lovelace', '[email protected]']],
  mappings: [
    { column: 'id', csvField: null, expression: 'gen_random_uuid()::text' },
    { column: 'name', csvField: 'First', expression: "{value} || ' ' || {Last}" },
    { column: 'email', csvField: 'Email', expression: null },
  ],
});
// → { total: 1, applied: 1, errorCount: 0, errors: [], rolledBack: false, … }

Mappings. Each target column takes a CSV column, a SQL expression, or both:

| csvField | expression | Result | | --- | --- | --- | | 'Email' | null | The cell, verbatim | | null | 'now()' | A constant — no cell is read | | 'Age' | '{value}::int' | The cell, transformed | | 'First' | "{value} \|\| ' ' \|\| {Last}" | Several columns combined |

{value} is the selected csvField's cell; {Header} is any other column by header. Both compile to bind parameters — file content never reaches the SQL text, so a header crafted to close a quote and append its own statement is inert. Braces inside string literals are left alone, so '{new,pending}' stays an array literal.

The two contradictory pairings are rejected rather than guessed at: {value} with no csvField has nothing to bind, and a csvField whose expression never reads it would silently drop the cell. A CSV with two columns of the same name is rejected for the same reason — resolving it to one of them would import that column's data and discard the other's without saying so.

Errors. Every row runs inside its own savepoint, so one bad row cannot poison the rest of the file:

• errorMode: 'skip' — failures are recorded, everything else commits.
• errorMode: 'rollback' — every row still runs, so you get the complete error list, but nothing persists and applied is 0.

In update mode, matchColumns locate the row and the remaining mappings are what gets set. A row that matches nothing is reported as an error, not counted as a success.

Limits. 5,000 rows per import (MAX_IMPORT_ROWS) — the file is one transaction, which also keeps the payload under API Gateway's 10 MB. At most 50 errors are returned (IMPORT_ERROR_CAP); errorCount is the true total.

Mapping expressions are free SQL, at the same trust level as explore(). They come from the operator, not the file — but the grant on the connecting role is still what bounds them.

Postgres only. Other engines throw UNSUPPORTED: without savepoints, one bad row would abandon the whole file, which is a different feature.

Auditing every statement

const db = createSqlClient({
  onQuery: (event) => auditLog.write(event),
});

Called after every statement, successful or not, with the database, engine, SQL, parameters, duration, row count and any error — plus writeMode for explore().

A hook rather than a table, because "audit" means a different thing in every application. Failures inside the hook are swallowed: a logging sink that is down must not turn a working query into a failed one. If auditing has to be mandatory, enforce that where you call the client.

In Lambda

Lambda freezes the container the moment a handler returns, so pool idle timers never fire. Without cleanup, every warm instance holds its connection for the container's life and concurrent instances exhaust the database's connection limit.

import { createSqlClient, withSqlCleanup } from '@speles7172/sql-client';

const db = createSqlClient();

export const handler = withSqlCleanup(db, async (event) => {
  const user = await db.one('main', 'SELECT * FROM users WHERE id = $1', {
    params: [event.userId],
  });
  return { statusCode: 200, body: JSON.stringify(user) };
});

The client detects Lambda and sizes pools to a single connection there. Disposal runs whether the handler returned or threw, and never masks its outcome.

Behaviour worth knowing

Retries are narrow on purpose. "The connection failed" hides two different situations, and only one is safe to re-send:

| Failure | Example | Retried? | |---|---|---| | Connection never obtained | timeout exceeded when trying to connect, ECONNREFUSED | Always — the server provably never saw the statement | | Connection died mid-flight | ECONNRESET, EPIPE, socket hang up | Only if you pass replayable: true |

The second case is genuinely ambiguous: the server may have received, executed and committed the statement before the socket dropped, and the client cannot tell. Re-sending an INSERT there would apply it twice with nothing downstream able to detect it.

You have to declare it, because it cannot be read off the statement. These both look like reads and both write:

SELECT record_access(user_id) FROM sessions            -- volatile function
WITH moved AS (INSERT INTO archive … RETURNING *)      -- data-modifying CTE
  SELECT * FROM moved

For a genuinely pure read:

await db.rows('main', 'SELECT id FROM users', { replayable: true });

DATE stays a string. node-postgres turns a DATE into a JS Date at local midnight, so 2026-02-10 read from a US timezone reports as Feb 9. Dates without times are calendar values and are returned as YYYY-MM-DD.

Registration fields beat secret fields. That is what lets one RDS secret serve several registrations — point at a different database on the same cluster and override only that.

Errors name the database. A driver error alone says "syntax error at or near"; QueryError carries the database name and the statement. Unknown database names throw UnknownDatabaseError, which lists what is registered.

Credentials and TLS

No credentials are stored by this package. Secrets are read from AWS Secrets Manager via the ambient credential chain and cached per database; call resolver.invalidate() after a rotation. A registration may instead carry host/user/password directly, which is intended for local development.

TLS defaults to encrypted but not server-authenticated. Amazon RDS presents a certificate signed by a private Amazon CA that is not in Node's trust store, so verifying by default would break every RDS connection on first use — and the reflexive fix people reach for is disabling TLS altogether.

On any network where interception is a concern, verify. Supplying a ca turns verification on:

import { readFileSync } from 'node:fs';

registerDatabase({
  name: 'main',
  engine: 'postgres',
  secretArn: 'arn:…',
  // https://truststore.pki.rds.amazonaws.com/global/global-bundle.pem
  ssl: { ca: readFileSync('global-bundle.pem', 'utf8') },
});

| ssl | Behaviour | |---|---| | true (default) | Encrypted, server not authenticated | | false | No TLS | | { ca } | Encrypted and verified against that CA | | { rejectUnauthorized: false } | Same as true, stated explicitly |

Athena

Athena is a job API, not a connection: the client starts an execution, polls until it settles, then reads results. Two consequences worth knowing:

Paging is cursor-based. Athena has no OFFSET; later pages are reachable only by following NextToken in order, so requesting page 5 costs five calls. Deep paging over Athena is a bad idea, not merely a slow one.

Page size is capped at 999, one below Athena's own limit: Athena repeats the column names as the first row of page one and counts it against MaxResults, so page one needs room to ask for that extra row — otherwise it would return one row fewer than every other page.

Without pagination, every page is collected up to maxRows (default 10,000). Reaching that ceiling sets truncated: true on the result rather than quietly returning a prefix.

API

| | | |---|---| | registerDatabase(reg) · registerDatabases([...]) | add to the shared registry | | loadFromEnv({ variable }) | load DB_CONNECTIONS-style JSON | | new DatabaseRegistry(opts) | an isolated registry | | createSqlClient(opts) | a client bound to a registry | | client.query(db, sql, opts) | full result | | client.rows(db, sql, opts) · client.one(...) | rows, or the first row | | client.databases() | what this client can reach | | client.dispose() | close every pool | | withSqlCleanup(client, handler) | Lambda wrapper |

Engines, the pagination helpers and the AWS adapters are exported too, for callers building their own wiring.