Kysely

Your application already writes through Kysely. When a request inserts a row and enqueues a follow-up job, those two writes must commit together — a job for a row that rolled back is a bug, and a committed row with no job is a silent gap. @stablemates/workhorse-kysely closes that gap: it converts Kysely databases and transactions into the core Queryable protocol, so Queue.enqueue runs inside the same transaction as your Kysely writes.

Workhorse keeps its schema lifecycle outside Kysely migrations. Kysely owns your tables; Workhorse owns its versioned SQL functions.

Create the adapter

Install the integration beside Kysely and node-postgres.

pnpm add @stablemates/workhorse-kysely kysely pg

Pass the database to createKyselyAdapter. Kysely's PostgresDialect already holds a node-postgres pool, so if the same process runs workers, pass that pool as notificationPool — workers then reserve a connection for LISTEN/NOTIFY and pick up new jobs immediately.

import { createKyselyAdapter } from "@stablemates/workhorse-kysely";
import { Pool } from "@stablemates/workhorse";
import { Kysely, PostgresDialect } from "kysely";

interface AppDatabase {
  account: { id: string; email: string };
}

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const database = new Kysely<AppDatabase>({ dialect: new PostgresDialect({ pool }) });
const workhorse = createKyselyAdapter(database, {
  notificationPool: pool,
  close: () => database.destroy(),
});

The returned WorkhorseAdapter exposes database, queue, admin, forTransaction, adminForTransaction, createWorker, and close — one stable surface for request handlers, operator tools, and worker processes.

Enqueue inside a transaction

This is the pattern the adapter exists for. Call forTransaction with the transaction passed to the callback. The queue executes its compiled SQL through that transaction, so the job follows the surrounding commit or rollback.

await database.transaction().execute(async (transaction) => {
  const account = await transaction
    .insertInto("account")
    .values({ email })
    .returning("id")
    .executeTakeFirstOrThrow();

  await workhorse.forTransaction(transaction).enqueue("account.created", {
    accountId: account.id,
  });
});

If the callback throws, Kysely rolls back both writes. Nothing outside the transaction ever sees a half-committed pair. The adapter's default queue remains outside that transaction — use it for enqueues that stand alone.

Keep resource ownership explicit

The adapter does not destroy a caller-owned database unless close is configured. Add that hook only when the adapter owns the database, because adapter.close may run during process shutdown or failed startup. Without notificationPool, workers use bounded polling and keep the same durable behavior — jobs still run, just on the polling cadence.

Database execution failures become KyselyQueryError, which retains the statement, original cause, and PostgreSQL error code. Typed Workhorse conflicts — idempotency, checkpoint, and wait errors — remain their core error classes.

Install the Workhorse schema with the core deployment tools rather than a Kysely migration — the schema is a versioned protocol, not application tables.

Next

• Enqueue and transactions — understand the transaction guarantee
• Workers — configure polling and handlers
• Installation — install the schema outside Kysely migrations

Exact adapter boundary and transaction ownership: architecture reference.