@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.

{my_uuid} — who is running the import. The one token whose value is not in the file: it binds spec.actorId, so an imported row can record who imported it.

await db.importRows('main', {
  // …
  // From the session. Never from the request body — see below.
  actorId: session.userId,
  mappings: [
    { column: 'created_by', csvField: null, expression: '{my_uuid}' },
    // A `text` column takes the cast, exactly as `{value}::int` does.
    { column: 'imported_by', csvField: null, expression: '{my_uuid}::text' },
  ],
});

It is matched in any casing ({MY_UUID} works), and it binds ::uuid because inside an expression a parameter has no column to infer its type from.

actorId is the endpoint's to set, from the session. A browser-supplied "who am I" is a claim rather than a fact, and a created_by column anybody can fill in with anybody's name is worse than no column at all — the same rule @speles7172/controls-client applies to an option source's scope.

Three things are refused rather than resolved quietly: an expression using the token when nothing supplied an actorId (a null created_by reported as a clean import), an actorId that is not a uuid (fifty thousand identical invalid input syntax for type uuid errors that read as a bad file), and a CSV column named my_uuid alongside an expression that uses the token — the token wins the name, so shadowing would write the operator's own id over the file's.

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.

What a failure tells you. The mapping expressions are the operator's own SQL, so an import that misbehaves is usually a formula to correct — and syntax error at or near "FROM" is not correctable without the statement it is about. So:

• result.statement is the compiled statement every row ran, with each cell a $n. It carries no file content, and is what a row error's position refers to. onQuery logs it in place of a summary of the spec.

• Each ImportRowError carries what the database said past its first line — code (SQLSTATE), detail (which is where PostgreSQL puts the offending value on a unique violation), hint, column, constraint, position.

• A statement-level failure on the first row throws ImportStatementError rather than reporting the same mistake once per row. Those are the SQLSTATEs settled when the statement is planned — syntax, unknown column, unknown table, unknown function, a cast that does not exist — so they fail identically for every row, and grinding through fifty thousand of them costs minutes and buries the cause. Its .statement and .detail are on the error, and its message carries the statement with a caret under the offending character.

  It is an allowlist rather than "class 42", deliberately: 42501 is what a row-level security WITH CHECK policy raises, and that depends on the row being inserted. A row the policy refuses is skipped like any other bad row.

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 failure carries errorDetail as well as error: the SQLSTATE, the position into sql, the detail line, the hint, the column and the constraint. error is one line, and a log holding only it cannot say where — which is the whole of what a syntax error needs to become a fix. QueryError puts the same thing in its message, laid out with the statement and a caret:

Query against "main" failed: syntax error at or near "FROM" [42601]
  hint: perhaps you meant to write a comma
  statement (line 1, character 41):
    INSERT INTO "people" ("name") VALUES ($1::text FROM)
                                            ^

For an import, sql is the compiled statement rather than a summary of the spec, and rowErrors/rowErrorCount report the rows it dropped even when the import as a whole succeeded — a skip run that discarded nine thousand rows resolves rather than throws, and without those the only record of them is the browser that started the run. Use formatDatabaseError(detail, sql) to lay any of it out yourself.

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, and say where. A driver error alone says "syntax error at or near"; QueryError carries the database name, the statement, and the rest of what the server reported — SQLSTATE, position, detail, hint, column, constraint — on .detail, and in its message with a caret under the offending character. 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.