@venturekit/data API

Functions

RDS Configuration

Function	Description
createRdsConfig(options)	Create RDS configuration
buildRdsConfig(config)	Build RDS infrastructure config

Query Utilities

Function	Signature	Description
query	(sql: string, params?: unknown[], resultsMapper?: (rows) => T) => Promise<T>	Execute a SQL query with automatic dot-notation mapping
getPool	() => Pool	Get the PostgreSQL connection pool
mapResults	(rows: Record<string, unknown>[]) => Record<string, unknown>[]	Map flat rows with dot-notation keys into nested objects
mapRow	(row: Record<string, unknown>) => Record<string, unknown>	Map a single flat row into a nested object

Result Mapping (dot-notation)

All queries automatically run through mapResults() which converts dot-notation column aliases into nested objects:

SELECT p.id, p.title, u.id AS "author.id", u.name AS "author.name"
FROM posts p JOIN users u ON u.id = p.author_id

Returns:

[{ "id": "1", "title": "Hello", "author": { "id": "5", "name": "Alice" } }]

Custom Result Mappers

The optional third argument to query() transforms the mapped rows into any shape:

// Extract a single column
const titles = await query<string[]>(
  'SELECT title FROM posts',
  [],
  (rows) => rows.map(r => r.title as string),
);

// Return a single row or null
const post = await query<Post | null>(
  'SELECT * FROM posts WHERE id = $1',
  [id],
  (rows) => (rows[0] as Post) ?? null,
);

Error Handling

• Throws the underlying pg driver error on connection/query failures.
• If resultsMapper throws, that error propagates unchanged.
• An empty result set is not an error — it returns [] (or whatever your mapper produces for an empty array).

Transactions

Function	Signature	Description
beginTransaction	() => Promise<Transaction>	Start a new transaction
withTransaction	(fn: (tx: Transaction) => Promise<T>) => Promise<T>	Execute within auto-commit/rollback transaction
buildTransaction	(client: PoolClient) => Transaction	Build a Transaction from a pool client (used internally by beginTransaction and the handler’s transactional option)

Migration Utilities

Function	Description
createMigrationConfig(options)	Create migration configuration
getFlywayEnv(config)	Get Flyway-compatible environment variables

Tenant context (row-level security)

Move tenant isolation into Postgres: wrap a request in runWithTenant() and every query issued inside — pooled or transactional — runs with the app.tenant_id GUC set, so RLS policies filter the rows application SQL forgot to. An unset GUC yields zero rows (fail-closed).

ALTER TABLE posts ENABLE ROW LEVEL SECURITY;
ALTER TABLE posts FORCE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON posts
  USING (tenant_id::text = ANY (string_to_array(NULLIF(current_setting('app.tenant_ids', true), ''), ',')));

Export	Description
runWithTenant(tenantId, fn)	Run fn with app.tenant_id and app.tenant_ids set to one tenant
runWithTenantScope(tenantIds, fn)	Hierarchies (parent org over subsidiaries): app.tenant_ids holds every readable tenant, the first is also the acting app.tenant_id
runWithoutTenant(fn)	Explicitly leave a scope (platform jobs, the sweep that lists tenants)
getCurrentTenantId() / getCurrentTenantScope() / isTenancyEngaged()	Inspect the active scope
TENANT_GUC / TENANT_SCOPE_GUC	'app.tenant_id' / 'app.tenant_ids'
tenantRole(env?) / TENANT_ROLE_ENV	DB_TENANT_ROLE — when it names a NOLOGIN role that owns nothing (e.g. app_user), every tenant-scoped statement runs under SET ROLE to it, so RLS applies even when the connection is the table owner or postgres under vk dev

RLS is skipped for table owners (without FORCE) and for BYPASSRLS roles — run the application as a role that owns nothing and keep migrations on a privileged one. @venturekit-pro/tenancy’s createTenantRlsMiddleware wires runWithTenant around each request.

Subpath exports

Four Postgres-backed building blocks, each shipped with its own migration under the package’s vk.migrations directory (vk_data_001_idempotency.sql … vk_data_004_file_object.sql, auto-applied by vk migrate). Every helper accepts an optional querier / tableName so it can join an open transaction.

@venturekit/data/idempotency

Export	Description
createPostgresIdempotencyStore({ querier?, tableName? })	An IdempotencyStore for @venturekit/runtime/patterns’ idempotencyMiddleware over vk_idempotency_record — the Postgres alternative to DynamoDB
purgeExpiredIdempotencyRecords({ querier?, tableName? })	Postgres has no TTL — run this from a cron
DEFAULT_IDEMPOTENCY_TABLE	'vk_idempotency_record'

@venturekit/data/outbox

The store half of the transactional outbox:

Export	Description
appendToOutbox(querier, event, { tableName? })	Insert a DomainEvent in the same transaction as the state change
createPostgresOutboxStore({ querier?, tableName? })	The OutboxStore (claim with FOR UPDATE SKIP LOCKED, markPublished, recordFailures) for relayOnce()
discardOutboxEvent(querier, id, …)	Drop a quarantined row an operator has dealt with
DEFAULT_OUTBOX_TABLE	'vk_outbox'

@venturekit/data/jobs

Durable, queryable background jobs — for work the product talks about (“which imports are still geocoding for this tenant”), where an SQS queue has no query surface. The claim is FOR UPDATE SKIP LOCKED, so several workers can drain the same table.

import { enqueueJob, runJobsOnce } from '@venturekit/data/jobs';

// enqueue in the transaction that made the work necessary
await withTransaction(async (tx) => {
  const batch = await stageImport(tx, rows);
  await enqueueJob(tx.query, { kind: 'geocode-import', tenantId, payload: { batch: batch.id } });
});

// src/crons/jobs.ts — drain
await runJobsOnce({ 'geocode-import': async (job) => geocode(job.payload) }, { limit: 20 });

Export	Description
enqueueJob(querier, { kind, payload?, tenantId?, runAfter?, maxAttempts? })	Insert a queued job (Job<TPayload>)
claimJobs({ limit?, kinds?, workerId? })	Claim due jobs → running
completeJob(id) / failJob(id, error) / releaseJob(id, reason)	Finish, fail (retry until maxAttempts, then failed), or put back
reclaimStuckJobs({ staleAfterSeconds? })	Return jobs whose worker died to queued
listJobs({ kind?, status?, limit? })	The query surface — JobStatus is 'queued' | 'running' | 'succeeded' | 'failed' | 'cancelled'
runJobsOnce(handlers, options?)	Claim + dispatch to Record<kind, JobHandler> → { claimed, succeeded, failed }
DEFAULT_JOB_TABLE	'vk_job'

@venturekit/data/files

Metadata for uploaded objects — the record, where @venturekit/storage is the bytes. Deliberately uncoupled: sweepArchivedFiles takes object deletion as a port.

Export	Description
recordFileObject(querier, { purpose, bucket, objectKey, contentType, sizeBytes, tenantId?, checksum?, originalName?, uploadedBy? })	Register an upload (FileObject)
getFileObject(id) / listFileObjects({ … }) / findFileObjectByChecksum(checksum)	Look-ups — “every document for this tenant, newest first”, dedupe identical bytes
archiveFileObject(id)	Soft-delete; the bytes stay until swept
sweepArchivedFiles({ deleteObjects, retainForSeconds?, limit? })	Delete archived objects through your deleteObjects(keys) port (e.g. storage.deleteMany) and drop the rows
DEFAULT_FILE_TABLE	'vk_file_object'

@venturekit/data/flags

Feature flags and tenant settings over vk_feature_flags, vk_feature_flag_overrides and vk_tenant_settings (vk_data_005). Every function takes a Querier (query or tx.query). Flag keys match ^[a-z0-9][a-z0-9._-]{0,127}$.

Evaluation order, first match wins: user pin → tenant pin → global enabled → rollout (rolloutPercent > 0: the subject’s stable bucket) → off. The bucket hashes ${key}:${subject} (the user id when given, else the tenant id), so a subject sees the same answer on every request and across containers, and raising the percentage from 10 to 20 keeps the first 10 % on. Unknown keys evaluate to false (or your fallback), never an error.

Export	Description
createFlags({ querier, ttlMs? })	Flags — isEnabled(key, { tenantId?, userId? }, fallback?), evaluateAll(subject) (Record<key, boolean> to hydrate once per request), snapshot(), invalidate(). Keeps an in-memory snapshot of both tables refreshed lazily when older than ttlMs (default 30 000; 0 re-reads every call), so a flag check costs no query
evaluateFlag(snapshot, key, subject?, fallback?) / rolloutBucket(flagKey, subject)	The pure evaluation and the 0–99 bucket (SHA-256 based), exported for tests and client-side consistency
listFlags(q) / upsertFlag(q, { key, enabled?, rolloutPercent?, description?, metadata? }) / deleteFlag(q, key)	Admin: upsert on key; omitted fields keep their value; deleting a flag cascades to its pins
setFlagOverride(q, { key, scope: 'tenant' | 'user', scopeId, enabled }) / clearFlagOverride(q, { key, scope, scopeId }) / listFlagOverrides(q, { key?, scope?, scopeId? })	Pins
getTenantSettings(q, tenantId, { validate?, table? })	The tenant’s JSONB document, {} when none — validate (typically schema.parse) normalises / rejects
setTenantSettings(q, tenantId, patch, { validate?, table? })	Shallow-merge patch (a null value deletes the key) atomically in SQL (`

@venturekit/data/search

The SQL-building half of “search my tables”, pure and unit-testable. Every fragment takes the $n parameter index you will bind, so it composes with the rest of your statement; identifiers are validated before interpolation.

Export	Description
escapeLike(term) / containsPattern(term) / prefixPattern(term) / LIKE_ESCAPE	%term% / term% with %, _ and \ escaped — a member typing 100% matches the literal text. WHERE name ILIKE $1
trigramIndexSql(table, column)	Migration line for the pg_trgm GIN index that makes ILIKE '%term%' fast (needs CREATE EXTENSION IF NOT EXISTS pg_trgm once)
toWebSearchQuery(input, { maxLength? })	Normalise user input for websearch_to_tsquery — strips control characters, collapses whitespace, caps at 200 chars; never throws
ftsColumnSql(table, { sources, column?, config? }) / ftsIndexSql(table, column?)	Migration lines for a STORED generated, weighted tsvector column (sources: [{ column: 'title', weight: 'A' }, …], default column search_vector, config simple — no stemming, safe for mixed-language content) and its GIN index
ftsWhereSql(paramIndex, { column?, config? }) / ftsRankSql(…) / ftsHeadlineSql(textColumn, paramIndex, { maxWords?, minWords?, config? })	The query-side fragments: predicate, ts_rank_cd for ORDER BY … DESC, and a <mark>-wrapped snippet
encodeCursor({ value, id }) / decodeCursor(token) / keysetWhereSql(sortColumn, paramIndex, { idColumn?, direction? })	Keyset pagination: an opaque URL-safe cursor over (sortValue, id) and the row-value comparison (created_at, id) < ($n, $n+1) for the next page

import { containsPattern, ftsWhereSql, ftsRankSql, keysetWhereSql, decodeCursor, encodeCursor } from '@venturekit/data/search';

const q = toWebSearchQuery(input.q);
const cursor = decodeCursor(input.cursor);
const rows = await query(
  `SELECT id, title, created_at FROM posts
    WHERE ${ftsWhereSql(1)} ${cursor ? `AND ${keysetWhereSql('created_at', 2)}` : ''}
    ORDER BY ${ftsRankSql(1)} DESC, created_at DESC, id DESC LIMIT 20`,
  cursor ? [q, cursor.value, cursor.id] : [q],
);
const next = rows.length === 20 ? encodeCursor({ value: rows[19].createdAt.toISOString(), id: rows[19].id }) : null;

Types

Type	Description
RdsOutputs	RDS deployment outputs (endpoint, port, secret ARN)
RdsInfraConfig	RDS infrastructure configuration
Transaction	Transaction object with query(), commit(), rollback()
ResultsMapper	Function type for mapping query results

Constants

Constant	Description
DEFAULT_RDS_CONFIG	Default RDS configuration
DEFAULT_MIGRATION_CONFIG	Default migration configuration