# PostgreSQL -- Advanced Examples

> Prepared statements, SSL/TLS configuration, migration patterns, and testing with transaction rollback. Reference from [SKILL.md](../SKILL.md).

**Related examples:**

- [core.md](core.md) -- Pool setup, parameterized queries, error handling
- [transactions.md](transactions.md) -- Transactions, savepoints, retry logic
- [streaming.md](streaming.md) -- Cursors, pg-query-stream, LISTEN/NOTIFY

---

## Prepared Statements

Named queries cache execution plans on the PostgreSQL server per connection. Useful for complex queries executed repeatedly.

```typescript
import type pg from "pg";

interface ProductRow {
  id: number;
  name: string;
  price: string; // numeric returns as string
  category: string;
}

const PRODUCT_SEARCH_QUERY = {
  name: "search-products",
  text: `
    SELECT id, name, price, category
    FROM products
    WHERE category = $1
      AND price BETWEEN $2 AND $3
    ORDER BY price ASC
    LIMIT $4
  `,
};

const DEFAULT_PAGE_SIZE = 50;

async function searchProducts(
  pool: pg.Pool,
  category: string,
  minPrice: number,
  maxPrice: number,
  limit: number = DEFAULT_PAGE_SIZE,
): Promise<ProductRow[]> {
  const result = await pool.query<ProductRow>({
    ...PRODUCT_SEARCH_QUERY,
    values: [category, minPrice, maxPrice, limit],
  });

  return result.rows;
}

export { searchProducts };
```

**Why good:** Named query (`name: "search-products"`) tells PostgreSQL to cache the execution plan, subsequent calls skip parsing and planning. The query object is defined once as a constant.

**When to use:** Complex queries with multiple JOINs where planning time is significant. For simple queries, the overhead of plan caching is not worth it.

**Gotcha:** Prepared statement plans are cached per connection. If a pool rotates connections (due to `maxUses` or `maxLifetimeSeconds`), plans are lost on the new connection. This is usually fine -- the first execution on a new connection parses, subsequent ones are cached.

**Gotcha:** If you run DDL (ALTER TABLE, CREATE INDEX) that changes the table structure, cached plans may become invalid. PostgreSQL automatically invalidates them, but be aware this causes a one-time re-plan.

---

## SSL/TLS Connections

```typescript
import pg from "pg";
import { readFileSync } from "node:fs";

// Cloud-managed databases (AWS RDS, GCP Cloud SQL, etc.)
// Most cloud providers use trusted CAs -- just enable SSL
function createCloudPool(): pg.Pool {
  return new pg.Pool({
    connectionString: process.env.DATABASE_URL,
    ssl: {
      rejectUnauthorized: true, // Verify server certificate
    },
  });
}

// Self-signed certificates (on-premise, custom CA)
function createMtlsPool(): pg.Pool {
  return new pg.Pool({
    connectionString: process.env.DATABASE_URL,
    ssl: {
      ca: readFileSync(process.env.PG_CA_CERT!),
      key: readFileSync(process.env.PG_CLIENT_KEY!),
      cert: readFileSync(process.env.PG_CLIENT_CERT!),
      rejectUnauthorized: true,
    },
  });
}

export { createCloudPool, createMtlsPool };
```

**Why good:** `rejectUnauthorized: true` validates the server certificate (prevents MITM), mTLS for mutual authentication, cert paths from environment variables

**Gotcha:** If the connection string contains any SSL parameters (`sslmode`, `sslcert`, `sslkey`, `sslrootcert`), the entire `ssl` config object is replaced and any additional options are lost. Use one or the other.

**Gotcha:** `ssl: true` relies on Node.js TLS defaults (which reject unauthorized certs), but does not explicitly set `rejectUnauthorized`. Always use the object form (`ssl: { rejectUnauthorized: true }`) to be explicit about intent.

---

## Dynamic Password Callback

Useful for AWS IAM database authentication or secrets managers that rotate credentials.

```typescript
import pg from "pg";

function createPoolWithDynamicPassword(
  getPassword: () => Promise<string>,
): pg.Pool {
  return new pg.Pool({
    host: process.env.PG_HOST,
    port: parseInt(process.env.PG_PORT ?? "5432", 10),
    database: process.env.PG_DATABASE,
    user: process.env.PG_USER,
    password: getPassword, // Called on each new connection
    ssl: { rejectUnauthorized: true },
  });
}

export { createPoolWithDynamicPassword };
```

**Why good:** Password callback is invoked each time a new connection is established, so rotated credentials are picked up automatically without restarting the pool.

---

## Migration Pattern (Manual SQL Files)

A simple, dependency-free migration approach using sequential SQL files.

```typescript
import type pg from "pg";
import { readFileSync, readdirSync } from "node:fs";
import { join } from "node:path";

const MIGRATIONS_TABLE = "schema_migrations";

async function ensureMigrationsTable(pool: pg.Pool): Promise<void> {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS ${MIGRATIONS_TABLE} (
      id SERIAL PRIMARY KEY,
      name TEXT NOT NULL UNIQUE,
      applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
    )
  `);
}

async function getAppliedMigrations(pool: pg.Pool): Promise<Set<string>> {
  const result = await pool.query<{ name: string }>(
    `SELECT name FROM ${MIGRATIONS_TABLE} ORDER BY id`,
  );
  return new Set(result.rows.map((r) => r.name));
}

async function runMigrations(
  pool: pg.Pool,
  migrationsDir: string,
): Promise<string[]> {
  await ensureMigrationsTable(pool);
  const applied = await getAppliedMigrations(pool);

  const files = readdirSync(migrationsDir)
    .filter((f) => f.endsWith(".sql"))
    .sort(); // Alphabetical order: 001_create_users.sql, 002_add_email.sql

  const newMigrations: string[] = [];

  for (const file of files) {
    if (applied.has(file)) continue;

    const sql = readFileSync(join(migrationsDir, file), "utf-8");
    const client = await pool.connect();
    try {
      await client.query("BEGIN");
      await client.query(sql);
      await client.query(`INSERT INTO ${MIGRATIONS_TABLE} (name) VALUES ($1)`, [
        file,
      ]);
      await client.query("COMMIT");
      newMigrations.push(file);
    } catch (err) {
      await client.query("ROLLBACK");
      throw new Error(
        `Migration ${file} failed: ${err instanceof Error ? err.message : String(err)}`,
      );
    } finally {
      client.release();
    }
  }

  return newMigrations;
}

export { runMigrations };
```

**Why good:** Each migration runs in its own transaction (atomic per file), tracks applied migrations in database, alphabetical ordering ensures deterministic execution, no external dependencies

**When NOT to use:** If you already use a migration tool from your ORM or framework, use that instead. This pattern is for projects that use raw `pg` without higher-level tools.

---

## Testing with Transaction Rollback

Wrap each test in a transaction that rolls back after the test. This provides test isolation without database cleanup scripts.

```typescript
import type pg from "pg";

async function withTestTransaction<T>(
  pool: pg.Pool,
  testFn: (client: pg.PoolClient) => Promise<T>,
): Promise<T> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const result = await testFn(client);
    return result;
  } finally {
    await client.query("ROLLBACK");
    client.release();
  }
}

// Usage in tests (framework-agnostic)
// describe("UserRepository", () => {
//   it("creates a user", async () => {
//     await withTestTransaction(pool, async (client) => {
//       await client.query(
//         "INSERT INTO users (name, email) VALUES ($1, $2)",
//         ["Test User", "[email protected]"],
//       );
//
//       const { rows } = await client.query(
//         "SELECT * FROM users WHERE email = $1",
//         ["[email protected]"],
//       );
//
//       expect(rows).toHaveLength(1);
//       expect(rows[0].name).toBe("Test User");
//       // ROLLBACK happens in finally -- database is clean for next test
//     });
//   });
// });
```

**Why good:** Every test starts with a clean database state, no teardown needed, tests are isolated, fast because rollback is cheaper than truncate

**Gotcha:** The test function receives a `PoolClient`, not the pool itself. Code under test must accept a client/pool interface so you can inject the test client. If your code calls `pool.query()` internally, it will use a different connection and won't see the test transaction's uncommitted data.

---

## Connection Pool per Test Suite

For integration tests that need `pool.query()` to work (not just a dedicated client), create an isolated pool per test suite with a test database.

```typescript
import pg from "pg";

const POOL_MAX_TEST = 5;
const IDLE_TIMEOUT_TEST_MS = 1_000;
const CONNECTION_TIMEOUT_TEST_MS = 3_000;

function createTestPool(): pg.Pool {
  const pool = new pg.Pool({
    connectionString: process.env.TEST_DATABASE_URL,
    max: POOL_MAX_TEST,
    idleTimeoutMillis: IDLE_TIMEOUT_TEST_MS,
    connectionTimeoutMillis: CONNECTION_TIMEOUT_TEST_MS,
    allowExitOnIdle: true, // Let test runner exit cleanly
  });

  pool.on("error", (err) => {
    console.error("Test pool error:", err.message);
  });

  return pool;
}

// Truncate all tables between test suites
async function cleanDatabase(pool: pg.Pool): Promise<void> {
  await pool.query(`
    DO $$ DECLARE
      r RECORD;
    BEGIN
      FOR r IN (
        SELECT tablename FROM pg_tables
        WHERE schemaname = 'public'
        AND tablename != 'schema_migrations'
      ) LOOP
        EXECUTE 'TRUNCATE TABLE ' || quote_ident(r.tablename) || ' CASCADE';
      END LOOP;
    END $$
  `);
}

export { createTestPool, cleanDatabase };
```

**Why good:** `allowExitOnIdle: true` prevents test runner from hanging, low pool size for tests, truncate cascades handle foreign keys, skips migration table

---

_Full skill documentation: [SKILL.md](../SKILL.md) | Quick reference: [reference.md](../reference.md)_

# PostgreSQL -- Core Pattern Examples

> Pool setup, parameterized queries, typed results, and error handling. Reference from [SKILL.md](../SKILL.md).

**Related examples:**

- [transactions.md](transactions.md) -- BEGIN/COMMIT/ROLLBACK, savepoints, retry logic
- [streaming.md](streaming.md) -- Cursors, pg-query-stream, LISTEN/NOTIFY
- [advanced.md](advanced.md) -- SSL/TLS, prepared statements, testing patterns

---

## Pool Setup with Full Error Handling

```typescript
import pg from "pg";

const POOL_MAX_CLIENTS = 20;
const IDLE_TIMEOUT_MS = 30_000;
const CONNECTION_TIMEOUT_MS = 5_000;
const MAX_LIFETIME_SECONDS = 1_800;

function createPool(): pg.Pool {
  const connectionString = process.env.DATABASE_URL;
  if (!connectionString) {
    throw new Error("DATABASE_URL environment variable is required");
  }

  const pool = new pg.Pool({
    connectionString,
    max: POOL_MAX_CLIENTS,
    idleTimeoutMillis: IDLE_TIMEOUT_MS,
    connectionTimeoutMillis: CONNECTION_TIMEOUT_MS,
    maxLifetimeSeconds: MAX_LIFETIME_SECONDS,
  });

  // REQUIRED: idle client errors crash the process if unhandled
  pool.on("error", (err) => {
    console.error("Unexpected idle client error:", err.message);
  });

  return pool;
}

export { createPool };
```

**Why good:** Environment variable validation, named constants for all config values, error handler prevents process crash, `connectionTimeoutMillis` prevents infinite waits

```typescript
// ❌ Bad Example - Hardcoded, no error handling
import pg from "pg";

const pool = new pg.Pool({
  host: "localhost",
  database: "mydb",
  user: "root",
  password: "secret123",
});

// No pool.on("error") -- idle client errors crash process
// Hardcoded credentials -- will be committed to version control
// Default connectionTimeoutMillis: 0 -- infinite wait on pool exhaustion
```

**Why bad:** Hardcoded credentials leak in version control, missing error handler crashes process, default timeout means pool exhaustion hangs forever

---

## Parameterized Queries

```typescript
import type pg from "pg";

interface UserRow {
  id: number;
  name: string;
  email: string;
  created_at: Date;
}

// Simple SELECT with parameters
async function getUserById(
  pool: pg.Pool,
  userId: number,
): Promise<UserRow | undefined> {
  const result = await pool.query<UserRow>(
    "SELECT id, name, email, created_at FROM users WHERE id = $1",
    [userId],
  );
  return result.rows[0];
}

// INSERT with RETURNING
async function createUser(
  pool: pg.Pool,
  name: string,
  email: string,
): Promise<UserRow> {
  const result = await pool.query<UserRow>(
    "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id, name, email, created_at",
    [name, email],
  );
  return result.rows[0];
}

// UPDATE with rowCount check
async function updateUserEmail(
  pool: pg.Pool,
  userId: number,
  newEmail: string,
): Promise<boolean> {
  const result = await pool.query(
    "UPDATE users SET email = $1, updated_at = NOW() WHERE id = $2",
    [newEmail, userId],
  );
  return (result.rowCount ?? 0) > 0;
}

// DELETE
async function deleteUser(pool: pg.Pool, userId: number): Promise<boolean> {
  const result = await pool.query("DELETE FROM users WHERE id = $1", [userId]);
  return (result.rowCount ?? 0) > 0;
}

export { getUserById, createUser, updateUserEmail, deleteUser };
```

**Why good:** Generic `<UserRow>` types result rows, `$1`/`$2` prevents injection, `RETURNING` avoids a second query, `rowCount` check for update/delete success

---

## Array Parameters

PostgreSQL array parameters have a specific gotcha: you cannot use `IN ($1)` with a JavaScript array. Use `= ANY($1)` instead.

```typescript
// ✅ Good Example - Array parameter with ANY
async function getUsersByIds(
  pool: pg.Pool,
  userIds: number[],
): Promise<UserRow[]> {
  const result = await pool.query<UserRow>(
    "SELECT id, name, email FROM users WHERE id = ANY($1)",
    [userIds], // pg auto-converts JS array to PostgreSQL array
  );
  return result.rows;
}
```

**Why good:** `= ANY($1)` works with pg's automatic array conversion; no need to generate `$1, $2, $3, ...` placeholders dynamically

```typescript
// ❌ Bad Example - IN with single parameter
const result = await pool.query("SELECT * FROM users WHERE id IN ($1)", [
  [1, 2, 3],
]);
// Error: invalid input syntax for type integer: "{1,2,3}"
// pg converts [1,2,3] to the string "{1,2,3}" which IN cannot parse
```

**Why bad:** `IN ($1)` expects a single scalar value, not an array; pg's array conversion creates a PostgreSQL array literal that IN cannot parse

---

## Error Handling with PostgreSQL Error Codes

```typescript
import type pg from "pg";

// Named constants for PostgreSQL SQLSTATE error codes
const PG_UNIQUE_VIOLATION = "23505";
const PG_FOREIGN_KEY_VIOLATION = "23503";
const PG_NOT_NULL_VIOLATION = "23502";
const PG_CHECK_VIOLATION = "23514";
const PG_DEADLOCK_DETECTED = "40P01";
const PG_SERIALIZATION_FAILURE = "40001";

// Type guard for PostgreSQL errors
interface PgError extends Error {
  code: string;
  constraint?: string;
  detail?: string;
  table?: string;
  column?: string;
  schema?: string;
}

function isPgError(err: unknown): err is PgError {
  return err instanceof Error && "code" in err;
}

// Usage: map PostgreSQL errors to application errors
async function createUser(
  pool: pg.Pool,
  email: string,
  name: string,
): Promise<UserRow> {
  try {
    const result = await pool.query<UserRow>(
      "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *",
      [email, name],
    );
    return result.rows[0];
  } catch (err) {
    if (!isPgError(err)) throw err;

    switch (err.code) {
      case PG_UNIQUE_VIOLATION:
        throw new ConflictError(
          `Duplicate value for constraint: ${err.constraint}`,
        );
      case PG_FOREIGN_KEY_VIOLATION:
        throw new NotFoundError(
          `Referenced entity does not exist: ${err.detail}`,
        );
      case PG_NOT_NULL_VIOLATION:
        throw new ValidationError(`Missing required field: ${err.column}`);
      case PG_CHECK_VIOLATION:
        throw new ValidationError(`Validation failed: ${err.constraint}`);
      default:
        throw err;
    }
  }
}

export {
  isPgError,
  PG_UNIQUE_VIOLATION,
  PG_FOREIGN_KEY_VIOLATION,
  PG_NOT_NULL_VIOLATION,
  PG_CHECK_VIOLATION,
  PG_DEADLOCK_DETECTED,
  PG_SERIALIZATION_FAILURE,
};
```

**Why good:** Named constants replace magic strings, type guard enables safe property access, switch on code maps to domain-specific errors, re-throws unknown errors

---

## Graceful Pool Shutdown

```typescript
import type pg from "pg";

async function gracefulShutdown(pool: pg.Pool): Promise<void> {
  console.log("Shutting down database pool...");
  await pool.end(); // Waits for checked-out clients to be released, then closes all
  console.log("Database pool closed");
}

// Wire into process signals
process.on("SIGTERM", async () => {
  await gracefulShutdown(pool);
  process.exit(0);
});

process.on("SIGINT", async () => {
  await gracefulShutdown(pool);
  process.exit(0);
});
```

**Why good:** `pool.end()` drains the pool cleanly, signal handlers prevent abrupt disconnection

**Gotcha:** `pool.end()` waits for ALL checked-out clients to be released. If any client was never released (leaked), `pool.end()` will hang forever. This is a symptom of missing `finally` blocks in `pool.connect()` usage.

---

_Full skill documentation: [SKILL.md](../SKILL.md) | Quick reference: [reference.md](../reference.md)_

# PostgreSQL -- Streaming & Real-Time Examples

> Cursors, pg-query-stream for large result sets, and LISTEN/NOTIFY for real-time notifications. Reference from [SKILL.md](../SKILL.md).

**Related examples:**

- [core.md](core.md) -- Pool setup, parameterized queries, error handling
- [transactions.md](transactions.md) -- Transactions, savepoints, retry logic
- [advanced.md](advanced.md) -- Prepared statements, SSL, testing patterns

---

## Cursor-Based Batch Processing (pg-cursor)

Use `pg-cursor` when you need to process rows in batches without loading the entire result set into memory.

```typescript
import type pg from "pg";
import Cursor from "pg-cursor";

const BATCH_SIZE = 500;

async function processLargeTable(
  pool: pg.Pool,
  status: string,
  processor: (rows: OrderRow[]) => Promise<void>,
): Promise<number> {
  const client = await pool.connect();
  let totalProcessed = 0;

  try {
    const cursor = client.query(
      new Cursor(
        "SELECT id, user_id, total, status FROM orders WHERE status = $1 ORDER BY id",
        [status],
      ),
    );

    let rows = await cursor.read(BATCH_SIZE);
    while (rows.length > 0) {
      await processor(rows);
      totalProcessed += rows.length;
      rows = await cursor.read(BATCH_SIZE);
    }

    await cursor.close();
    return totalProcessed;
  } finally {
    client.release();
  }
}

interface OrderRow {
  id: number;
  user_id: number;
  total: string; // numeric comes back as string
  status: string;
}

export { processLargeTable };
```

**Why good:** Fixed memory footprint regardless of result set size, parameterized cursor query, proper client release, cursor closed explicitly

**When to use:** Processing 1,000-100,000 rows where you need to control batch size and do async work per batch.

---

## Streaming with pg-query-stream

Use `pg-query-stream` when you need a Node.js Readable stream -- ideal for piping to file writes, HTTP responses, or transform streams.

```typescript
import type pg from "pg";
import QueryStream from "pg-query-stream";
import { pipeline } from "node:stream/promises";
import { createWriteStream } from "node:fs";
import { Transform } from "node:stream";

const STREAM_BATCH_SIZE = 100;

async function exportOrdersToCsv(
  pool: pg.Pool,
  outputPath: string,
): Promise<void> {
  const client = await pool.connect();

  try {
    const queryStream = new QueryStream(
      "SELECT id, user_id, total, created_at FROM orders ORDER BY id",
      [],
      { batchSize: STREAM_BATCH_SIZE },
    );

    const dbStream = client.query(queryStream);

    const csvTransform = new Transform({
      objectMode: true,
      transform(row, _encoding, callback) {
        const line = `${row.id},${row.user_id},${row.total},${row.created_at.toISOString()}\n`;
        callback(null, line);
      },
    });

    const fileStream = createWriteStream(outputPath);

    // Write CSV header
    fileStream.write("id,user_id,total,created_at\n");

    await pipeline(dbStream, csvTransform, fileStream);
  } finally {
    client.release();
  }
}

export { exportOrdersToCsv };
```

**Why good:** Constant memory usage for arbitrarily large tables, `pipeline` handles backpressure and error propagation, client released in finally

**When to use:** Exporting 100,000+ rows to files, HTTP streaming responses, or ETL pipelines. For smaller result sets or when you need per-batch async processing, use pg-cursor instead.

---

## LISTEN/NOTIFY -- Subscribing to PostgreSQL Notifications

LISTEN/NOTIFY is PostgreSQL's built-in pub/sub mechanism. A client executes `LISTEN channel` and receives notifications sent via `NOTIFY channel` or `pg_notify(channel, payload)`.

```typescript
import type pg from "pg";

const ORDER_CHANNEL = "order_events";

interface OrderEvent {
  orderId: number;
  action: "created" | "updated" | "canceled";
  timestamp: string;
}

async function subscribeToOrderEvents(
  pool: pg.Pool,
  handler: (event: OrderEvent) => void,
): Promise<{ unsubscribe: () => Promise<void> }> {
  // LISTEN requires a dedicated client -- it stays checked out
  const client = await pool.connect();

  client.on("notification", (msg) => {
    if (msg.channel === ORDER_CHANNEL && msg.payload) {
      try {
        const event = JSON.parse(msg.payload) as OrderEvent;
        handler(event);
      } catch (err) {
        console.error("Failed to parse notification payload:", err);
      }
    }
  });

  await client.query(`LISTEN ${ORDER_CHANNEL}`);

  return {
    unsubscribe: async () => {
      await client.query(`UNLISTEN ${ORDER_CHANNEL}`);
      client.release();
    },
  };
}

export { subscribeToOrderEvents };
export type { OrderEvent };
```

**Why good:** Dedicated client for listener, JSON payload parsing with error handling, unsubscribe releases client back to pool, typed event interface

---

## LISTEN/NOTIFY -- Publishing Notifications

Use `pg_notify()` with parameterized queries for safe payload publishing. This can use `pool.query()` since it is a single command.

```typescript
import type pg from "pg";

const ORDER_CHANNEL = "order_events";

async function publishOrderEvent(
  pool: pg.Pool,
  orderId: number,
  action: "created" | "updated" | "canceled",
): Promise<void> {
  const payload = JSON.stringify({
    orderId,
    action,
    timestamp: new Date().toISOString(),
  });

  // pg_notify() with parameterized channel and payload
  await pool.query("SELECT pg_notify($1, $2)", [ORDER_CHANNEL, payload]);
}

export { publishOrderEvent };
```

**Why good:** `pg_notify($1, $2)` is parameterized (safe from injection), `pool.query()` is fine for publishing (no need for dedicated client), JSON payload is structured

```typescript
// ❌ Bad Example - String interpolation in NOTIFY
await pool.query(`NOTIFY ${channel}, '${payload}'`);
// SQL injection if channel or payload contain quotes
// Also: NOTIFY payload is limited to 8000 bytes and must be a string literal
```

**Why bad:** String interpolation allows injection, `NOTIFY` with literal payload has an 8000-byte limit and requires manual escaping

---

## LISTEN/NOTIFY -- Database Trigger Integration

A common pattern is to fire NOTIFY from a PostgreSQL trigger so that application code is notified of data changes automatically.

```sql
-- PostgreSQL trigger function
CREATE OR REPLACE FUNCTION notify_order_change()
RETURNS TRIGGER AS $$
BEGIN
  PERFORM pg_notify(
    'order_events',
    json_build_object(
      'orderId', NEW.id,
      'action', TG_OP,
      'timestamp', NOW()
    )::text
  );
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- Attach trigger to table
CREATE TRIGGER order_change_trigger
  AFTER INSERT OR UPDATE ON orders
  FOR EACH ROW
  EXECUTE FUNCTION notify_order_change();
```

**Why good:** Notifications are automatic -- no application code needed for publishing. Trigger fires within the transaction, so notification is only sent on COMMIT.

**Gotcha:** If the transaction rolls back, the notification is NOT sent. This is usually desired behavior -- you don't want to notify about changes that didn't persist.

**Gotcha:** LISTEN is NOT transactional -- `BEGIN; LISTEN channel; ROLLBACK;` still leaves the listener active. Only UNLISTEN or disconnecting removes it.

---

_Full skill documentation: [SKILL.md](../SKILL.md) | Quick reference: [reference.md](../reference.md)_

# PostgreSQL -- Transaction Examples

> BEGIN/COMMIT/ROLLBACK, savepoints, retry logic for deadlocks, and advisory locks. Reference from [SKILL.md](../SKILL.md).

**Related examples:**

- [core.md](core.md) -- Pool setup, parameterized queries, error handling
- [streaming.md](streaming.md) -- Cursors, pg-query-stream, LISTEN/NOTIFY
- [advanced.md](advanced.md) -- Prepared statements, testing with transaction rollback

---

## Basic Transaction

```typescript
import type pg from "pg";

interface TransferResult {
  fromBalance: number;
  toBalance: number;
}

async function transferFunds(
  pool: pg.Pool,
  fromAccountId: number,
  toAccountId: number,
  amount: number,
): Promise<TransferResult> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");

    // Lock rows in consistent order to prevent deadlocks
    const { rows } = await client.query<{ id: number; balance: number }>(
      `SELECT id, balance FROM accounts
       WHERE id = ANY($1)
       ORDER BY id FOR UPDATE`,
      [[fromAccountId, toAccountId]],
    );

    const fromAccount = rows.find((r) => r.id === fromAccountId);
    const toAccount = rows.find((r) => r.id === toAccountId);

    if (!fromAccount || !toAccount) {
      throw new Error("Account not found");
    }
    if (fromAccount.balance < amount) {
      throw new Error("Insufficient balance");
    }

    await client.query(
      "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
      [amount, fromAccountId],
    );
    await client.query(
      "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
      [amount, toAccountId],
    );

    await client.query("COMMIT");

    return {
      fromBalance: fromAccount.balance - amount,
      toBalance: toAccount.balance + amount,
    };
  } catch (err) {
    await client.query("ROLLBACK");
    throw err;
  } finally {
    client.release();
  }
}

export { transferFunds };
```

**Why good:** `FOR UPDATE` locks rows to prevent concurrent modification, consistent `ORDER BY id` prevents deadlocks, balance check after locking prevents race conditions, ROLLBACK on any error, release in finally

---

## Transaction Helper (Reusable)

Extract the try/BEGIN/COMMIT/catch/ROLLBACK/finally/release boilerplate into a reusable helper.

```typescript
import type pg from "pg";

async function withTransaction<T>(
  pool: pg.Pool,
  callback: (client: pg.PoolClient) => Promise<T>,
): Promise<T> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const result = await callback(client);
    await client.query("COMMIT");
    return result;
  } catch (err) {
    await client.query("ROLLBACK");
    throw err;
  } finally {
    client.release();
  }
}

export { withTransaction };
```

Usage:

```typescript
const order = await withTransaction(pool, async (client) => {
  const {
    rows: [newOrder],
  } = await client.query<OrderRow>(
    "INSERT INTO orders (user_id, total) VALUES ($1, $2) RETURNING *",
    [userId, total],
  );

  for (const item of items) {
    await client.query(
      "INSERT INTO order_items (order_id, product_id, quantity) VALUES ($1, $2, $3)",
      [newOrder.id, item.productId, item.quantity],
    );
  }

  return newOrder;
});
```

**Why good:** Eliminates boilerplate, impossible to forget ROLLBACK or release, type-safe return value

---

## Savepoints (Nested Transaction Behavior)

PostgreSQL does not support nested transactions, but savepoints provide similar functionality within a transaction.

```typescript
import type pg from "pg";

async function createOrderWithOptionalDiscount(
  pool: pg.Pool,
  userId: number,
  items: Array<{ productId: number; quantity: number }>,
  discountCode: string | null,
): Promise<void> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");

    // Core order creation
    const {
      rows: [order],
    } = await client.query<{ id: number }>(
      "INSERT INTO orders (user_id) VALUES ($1) RETURNING id",
      [userId],
    );

    for (const item of items) {
      await client.query(
        "INSERT INTO order_items (order_id, product_id, quantity) VALUES ($1, $2, $3)",
        [order.id, item.productId, item.quantity],
      );
    }

    // Optional: apply discount -- if it fails, order still succeeds
    if (discountCode) {
      await client.query("SAVEPOINT apply_discount");
      try {
        await client.query(
          "UPDATE discount_codes SET used = true WHERE code = $1 AND used = false",
          [discountCode],
        );
        await client.query(
          "UPDATE orders SET discount_code = $1 WHERE id = $2",
          [discountCode, order.id],
        );
      } catch {
        // Discount failed -- rollback to savepoint, continue with order
        await client.query("ROLLBACK TO SAVEPOINT apply_discount");
      }
    }

    await client.query("COMMIT");
  } catch (err) {
    await client.query("ROLLBACK");
    throw err;
  } finally {
    client.release();
  }
}

export { createOrderWithOptionalDiscount };
```

**Why good:** Savepoint isolates optional logic -- discount failure doesn't kill the order. `ROLLBACK TO SAVEPOINT` undoes only the savepoint's changes. The outer transaction can still COMMIT.

---

## Retry Logic for Deadlocks and Serialization Failures

Deadlocks (`40P01`) and serialization failures (`40001`) are safe to retry. Wrap transactional code in a retry loop.

```typescript
import type pg from "pg";

const PG_DEADLOCK_DETECTED = "40P01";
const PG_SERIALIZATION_FAILURE = "40001";
const MAX_RETRIES = 3;
const BASE_DELAY_MS = 50;

const RETRYABLE_CODES = new Set([
  PG_DEADLOCK_DETECTED,
  PG_SERIALIZATION_FAILURE,
]);

interface PgError extends Error {
  code: string;
}

function isPgError(err: unknown): err is PgError {
  return err instanceof Error && "code" in err;
}

async function withRetry<T>(
  pool: pg.Pool,
  operation: (client: pg.PoolClient) => Promise<T>,
): Promise<T> {
  for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
    const client = await pool.connect();
    try {
      await client.query("BEGIN");
      const result = await operation(client);
      await client.query("COMMIT");
      return result;
    } catch (err) {
      await client.query("ROLLBACK");

      const isRetryable = isPgError(err) && RETRYABLE_CODES.has(err.code);
      if (!isRetryable || attempt === MAX_RETRIES) {
        throw err;
      }

      // Exponential backoff with jitter
      const delay =
        BASE_DELAY_MS * Math.pow(2, attempt) + Math.random() * BASE_DELAY_MS;
      await new Promise((resolve) => setTimeout(resolve, delay));
    } finally {
      client.release();
    }
  }

  // Unreachable, but TypeScript needs it
  throw new Error("Retry loop exited unexpectedly");
}

export { withRetry };
```

**Why good:** Only retries errors that are safe to retry (deadlocks, serialization failures), exponential backoff with jitter prevents thundering herd, bounded retries prevent infinite loops, fresh client per attempt

---

## Advisory Locks

PostgreSQL advisory locks are application-level locks that don't lock any table rows. Use them to coordinate distributed operations.

```typescript
import type pg from "pg";

// Advisory locks use a bigint key. Use a consistent hash for string-based lock names.
const LOCK_REPORT_GENERATION = 1001;
const LOCK_CACHE_REBUILD = 1002;

async function withAdvisoryLock<T>(
  pool: pg.Pool,
  lockId: number,
  operation: (client: pg.PoolClient) => Promise<T>,
): Promise<T | null> {
  const client = await pool.connect();
  try {
    // pg_try_advisory_lock returns true if lock acquired, false if already held
    const { rows } = await client.query<{ pg_try_advisory_lock: boolean }>(
      "SELECT pg_try_advisory_lock($1)",
      [lockId],
    );

    if (!rows[0].pg_try_advisory_lock) {
      return null; // Another process holds the lock
    }

    try {
      return await operation(client);
    } finally {
      await client.query("SELECT pg_advisory_unlock($1)", [lockId]);
    }
  } finally {
    client.release();
  }
}

// Usage
const result = await withAdvisoryLock(
  pool,
  LOCK_REPORT_GENERATION,
  async (client) => {
    // Only one process runs this at a time across all app instances
    return generateExpensiveReport(client);
  },
);

if (result === null) {
  console.log("Report generation already in progress");
}

export { withAdvisoryLock, LOCK_REPORT_GENERATION, LOCK_CACHE_REBUILD };
```

**Why good:** `pg_try_advisory_lock` is non-blocking (returns false immediately if held), lock is released in a nested finally, named constants for lock IDs prevent collisions

**Gotcha:** Advisory locks are bound to the **session** (connection), not the transaction. If the client disconnects, the lock is automatically released. If you use `pg_advisory_xact_lock`, the lock is released when the transaction ends instead.

---

_Full skill documentation: [SKILL.md](../SKILL.md) | Quick reference: [reference.md](../reference.md)_