diff --git a/bun.lock b/bun.lock index 6c5d256d..e4c20ded 100644 --- a/bun.lock +++ b/bun.lock @@ -292,7 +292,7 @@ }, "packages/cli": { "name": "@wrnexus/cli", - "version": "0.8.53", + "version": "0.8.54", "bin": { "wrnexus": "src/index.ts", }, @@ -349,7 +349,7 @@ }, "packages/db": { "name": "@wrnexus/db", - "version": "0.8.18", + "version": "0.8.19", "devDependencies": { "@types/bun": "^1.3.14", "typescript": "^6.0.3", @@ -357,7 +357,7 @@ }, "packages/dev-server": { "name": "@wrnexus/dev-server", - "version": "0.8.47", + "version": "0.8.48", "dependencies": { "@wrnexus/authz": "workspace:*", "@wrnexus/cache": "workspace:*", @@ -371,6 +371,7 @@ "@wrnexus/plugin": "workspace:*", "@wrnexus/pubsub": "workspace:*", "@wrnexus/pwa": "workspace:*", + "@wrnexus/queue": "workspace:*", "@wrnexus/router": "workspace:*", "@wrnexus/rpc": "workspace:*", "@wrnexus/security": "workspace:*", @@ -546,9 +547,10 @@ }, "packages/queue": { "name": "@wrnexus/queue", - "version": "0.8.11", + "version": "0.8.12", "dependencies": { "@wrnexus/core": "workspace:*", + "@wrnexus/db": "workspace:*", "@wrnexus/rpc": "workspace:*", }, }, @@ -633,7 +635,7 @@ }, "packages/styles": { "name": "@wrnexus/styles", - "version": "0.8.17", + "version": "0.8.18", "dependencies": { "@wrnexus/core": "workspace:*", "@wrnexus/plugin": "workspace:*", @@ -646,7 +648,7 @@ }, "packages/test": { "name": "@wrnexus/test", - "version": "0.8.10", + "version": "0.8.11", }, "packages/tracking": { "name": "@wrnexus/tracking", diff --git a/packages/cli/package.json b/packages/cli/package.json index 75d8eece..d4a621df 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/cli", - "version": "0.8.53", + "version": "0.8.54", "type": "module", "main": "src/index.ts", "exports": { diff --git a/packages/cli/src/build.ts b/packages/cli/src/build.ts index 5521c4f3..c9f704b0 100644 --- a/packages/cli/src/build.ts +++ b/packages/cli/src/build.ts @@ -825,6 +825,7 @@ await createProductionServer( i18n: ${i18n ? JSON.stringify(i18n) : "undefined"}, db: ${config.db ? JSON.stringify(config.db) : "undefined"}, databases: ${config.databases ? JSON.stringify(config.databases) : "undefined"}, + queue: ${config.queue ? JSON.stringify(config.queue) : "undefined"}, storage: ${config.storage ? JSON.stringify(config.storage) : "undefined"}, ${hasDefaultMigrations ? `migrationsDir: join(import.meta.dir, "migrations"),` : ""} ${ diff --git a/packages/db/README.md b/packages/db/README.md index 7fb4f5e6..aa23e0fd 100644 --- a/packages/db/README.md +++ b/packages/db/README.md @@ -297,3 +297,16 @@ const users = await usersRepo.all({ Repository SQL identifiers are validated and values remain parameterized. Placeholder generation is dialect-aware: PostgreSQL uses `$1`, `$2`, and SQLite/MySQL use `?`. List limits are bounded. `retryTransaction()` retries recognized serialization, deadlock, and database-lock errors by default. Supply `shouldRetry` for application-specific retryable errors; ordinary validation or business errors are not retried automatically. + +## Reusable seed helpers + +```ts +import { addSeedData, removeSeedData, runSeedQuery, getDb } from "@wrnexus/db"; + +await addSeedData(getDb, [{ code: "free", credits: 100 }], "plans", { conflict: "ignore" }); +await removeSeedData(getDb(), { code: "legacy" }, "plans"); +await runSeedQuery(getDb(), "UPDATE plans SET credits = ? WHERE code = ?", [200, "free"]); +``` + +Table and column identifiers are validated, values are always parameterized, and empty bulk +deletes are refused unless `{ all: true }` is explicit. diff --git a/packages/db/package.json b/packages/db/package.json index df244c45..68c98c40 100644 --- a/packages/db/package.json +++ b/packages/db/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/db", - "version": "0.8.18", + "version": "0.8.19", "private": true, "type": "module", "main": "./src/index.ts", diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index c2ad991a..5a8a6302 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -61,3 +61,5 @@ export { createRepository, } from "./helpers.ts"; export type { Repository } from "./helpers.ts"; +export { addSeedData, removeSeedData, runSeedQuery } from "./seed.ts"; +export type { AddSeedOptions, SeedDatabase, SeedRow } from "./seed.ts"; diff --git a/packages/db/src/seed.ts b/packages/db/src/seed.ts new file mode 100644 index 00000000..21142ec4 --- /dev/null +++ b/packages/db/src/seed.ts @@ -0,0 +1,83 @@ +import type { Db, ExecResult } from "./driver.ts"; + +export type SeedRow = Record; +export type SeedDatabase = Db | (() => Db); + +export interface AddSeedOptions { + conflict?: "error" | "ignore" | "replace"; +} + +function identifier(value: string, label: string): string { + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(value)) { + throw new TypeError(`${label} must be a safe SQL identifier`); + } + return value; +} + +function database(value: SeedDatabase): Db { + return typeof value === "function" ? value() : value; +} + +/** Insert one or many object rows with parameterized values. */ +export async function addSeedData( + source: SeedDatabase, + data: SeedRow | readonly SeedRow[], + table: string, + options: AddSeedOptions = {}, +): Promise { + const db = database(source); + const target = identifier(table, "seed table"); + const rows = Array.isArray(data) ? data : [data]; + const results: ExecResult[] = []; + const clause = + options.conflict === "ignore" + ? " OR IGNORE" + : options.conflict === "replace" + ? " OR REPLACE" + : ""; + for (const row of rows) { + const entries = Object.entries(row); + if (!entries.length) throw new TypeError("seed row cannot be empty"); + const columns = entries.map(([column]) => identifier(column, "seed column")); + results.push( + await db.exec( + `INSERT${clause} INTO ${target} (${columns.join(", ")}) VALUES (${columns.map(() => "?").join(", ")})`, + entries.map(([, value]) => value), + ), + ); + } + return results; +} + +/** Delete matching seed rows. An empty match is refused unless `all` is explicit. */ +export async function removeSeedData( + source: SeedDatabase, + match: SeedRow, + table: string, + options: { all?: boolean } = {}, +): Promise { + const db = database(source); + const target = identifier(table, "seed table"); + const entries = Object.entries(match); + if (!entries.length && !options.all) { + throw new TypeError("removeSeedData requires match fields or { all: true }"); + } + const where = entries.length + ? ` WHERE ${entries.map(([column]) => `${identifier(column, "seed column")} = ?`).join(" AND ")}` + : ""; + return db.exec( + `DELETE FROM ${target}${where}`, + entries.map(([, value]) => value), + ); +} + +/** Execute an application-owned parameterized seed statement. */ +export function runSeedQuery( + source: SeedDatabase, + query: string, + data: readonly unknown[] = [], +): Promise { + const db = database(source); + if (!query.trim()) throw new TypeError("seed query cannot be empty"); + return db.exec(query, [...data]); +} diff --git a/packages/db/test/seed.test.ts b/packages/db/test/seed.test.ts new file mode 100644 index 00000000..6e09e709 --- /dev/null +++ b/packages/db/test/seed.test.ts @@ -0,0 +1,37 @@ +import { expect, test } from "bun:test"; +import type { Db } from "../src/driver.ts"; +import { addSeedData, removeSeedData, runSeedQuery } from "../src/seed.ts"; + +function fakeDb() { + const calls: Array<{ sql: string; parameters: unknown[] }> = []; + const db = { + exec: async (sql: string, parameters: unknown[] = []) => { + calls.push({ sql, parameters }); + return { changes: 1 }; + }, + } as unknown as Db; + return { db, calls }; +} + +test("seed helpers parameterize object rows and removals", async () => { + const { db, calls } = fakeDb(); + await addSeedData(db, [{ code: "free", credits: 10 }, { code: "pro", credits: 20 }], "plans", { + conflict: "ignore", + }); + await removeSeedData(db, { code: "free" }, "plans"); + expect(calls[0]).toEqual({ + sql: "INSERT OR IGNORE INTO plans (code, credits) VALUES (?, ?)", + parameters: ["free", 10], + }); + expect(calls[2]).toEqual({ sql: "DELETE FROM plans WHERE code = ?", parameters: ["free"] }); +}); + +test("custom seed queries stay parameterized and unsafe identifiers are refused", async () => { + const { db, calls } = fakeDb(); + await runSeedQuery(db, "UPDATE plans SET credits = ? WHERE code = ?", [50, "free"]); + expect(calls[0]?.parameters).toEqual([50, "free"]); + await expect(addSeedData(db, { id: 1 }, "plans; DROP TABLE plans")).rejects.toThrow( + "safe SQL identifier", + ); + await expect(removeSeedData(db, {}, "plans")).rejects.toThrow("match fields"); +}); diff --git a/packages/dev-server/package.json b/packages/dev-server/package.json index ea3f2d0f..64d8b3e6 100644 --- a/packages/dev-server/package.json +++ b/packages/dev-server/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/dev-server", - "version": "0.8.47", + "version": "0.8.48", "type": "module", "main": "src/index.ts", "exports": { @@ -24,6 +24,7 @@ "@wrnexus/pubsub": "workspace:*", "@wrnexus/uploader": "workspace:*", "@wrnexus/plugin": "workspace:*", + "@wrnexus/queue": "workspace:*", "@wrnexus/store": "workspace:*", "@wrnexus/security": "workspace:*", "@wrnexus/observability": "workspace:*", diff --git a/packages/dev-server/src/index.ts b/packages/dev-server/src/index.ts index aa4a91fe..66c73de2 100644 --- a/packages/dev-server/src/index.ts +++ b/packages/dev-server/src/index.ts @@ -33,6 +33,7 @@ import { getDbPerformanceSnapshot, } from "@wrnexus/db"; import { connectFromConfig } from "@wrnexus/db/connect"; +import { configureQueueStorage, type QueueStorageConfig } from "@wrnexus/queue"; import { configureStorage, type StorageConfig } from "@wrnexus/uploader"; import { setAuthzCatalog, type AuthzCatalog, type AuthzModule } from "@wrnexus/authz"; import { loadAppAuthzCatalog } from "./authz-boot.ts"; @@ -102,6 +103,8 @@ export interface ServeOptions { /** Inject the live-reload client (defaults to true in development). */ hmr?: boolean; appConfig?: Record; + /** Queue persistence; durable SQLite is used when omitted. */ + queue?: QueueStorageConfig; /** Resolved absolute path to the global CSS entry, or null. */ styleEntry?: string | null; /** Custom styles config (e.g. a Tailwind/PostCSS processor). */ @@ -220,6 +223,7 @@ function resolveDevToolbarConfig( export async function startServer(opts: ServeOptions): Promise { const appDir = resolve(opts.appDir); const appRoot = dirname(appDir); + configureQueueStorage({ ...opts.queue, appRoot }); const mode: Mode = opts.mode ?? "development"; const importConfig = (opts.appConfig?.imports ?? {}) as { mode?: "legacy" | "compatible" | "explicit"; diff --git a/packages/dev-server/src/prod.ts b/packages/dev-server/src/prod.ts index 0aefde9c..7c22b0f3 100644 --- a/packages/dev-server/src/prod.ts +++ b/packages/dev-server/src/prod.ts @@ -40,6 +40,7 @@ import { VALIDATE_RUNTIME } from "@wrnexus/validation"; import { I18N_RUNTIME, type ResolvedI18n } from "@wrnexus/i18n"; import { setDb, registerLazyDb, getDb, hasDb, migrate } from "@wrnexus/db"; import { connectFromConfig } from "@wrnexus/db/connect"; +import { configureQueueStorage, type QueueStorageConfig } from "@wrnexus/queue"; import { hasAuthzCatalog, mergeCatalogs, setAuthzCatalog, type AuthzModule } from "@wrnexus/authz"; import { configureStorage, @@ -138,6 +139,8 @@ export interface ProdOptions { db?: { driver: string; url: string }; /** Named databases, reached with `getDb("")`. */ databases?: Record; + /** Queue persistence; durable SQLite is used when omitted. */ + queue?: QueueStorageConfig; /** * Absolute path to the default db's migrations bundled into the build * (`dist/migrations`). When set, they are applied on startup — like dev. @@ -499,6 +502,7 @@ export function createProductionHandlers( for (const [name, cfg] of Object.entries(opts.databases ?? {})) { registerLazyDb(name, () => connectFromConfig(cfg)); } + configureQueueStorage({ ...opts.queue, appRoot: process.cwd() }); // File-upload storage. Relative local dirs resolve against the deployment cwd // (NOT dist/, which is rebuilt) so uploads persist across deploys. diff --git a/packages/dev-server/src/serve-entry.ts b/packages/dev-server/src/serve-entry.ts index 43f0a3fd..a6b0249d 100644 --- a/packages/dev-server/src/serve-entry.ts +++ b/packages/dev-server/src/serve-entry.ts @@ -58,6 +58,7 @@ const server = await startServer({ i18n: config.i18n, db: config.db, databases: config.databases, + queue: config.queue, realtime: config.realtime, storage: config.storage, mobile: config.mobile, diff --git a/packages/queue/README.md b/packages/queue/README.md index fc3cf0d7..110bf367 100644 --- a/packages/queue/README.md +++ b/packages/queue/README.md @@ -6,11 +6,11 @@ Part of the **WrNexus** framework — an SSR-first, Bun-native full-stack web fr ## Overview -`@wrnexus/queue` is a server-side in-process job queue. You register named +`@wrnexus/queue` is a server-side durable job queue. You register named workers, enqueue jobs (optionally delayed or recurring), and let the queue poll and run them on a timer — with per-job retry limits and doubling backoff between -attempts. The default store lives in memory; the design allows a pluggable driver -to back it with Redis/SQL for durability across restarts. Reach for it when you +attempts. `defineQueue` persists to `.wrnexus/queue.sqlite` by default, while +low-level `createQueue` remains an intentionally in-memory primitive. Reach for it when you need to defer work (emails, webhooks, cleanup) off the request path without a heavyweight external broker. Tests can drive it deterministically via `drain()`. @@ -100,6 +100,20 @@ interface Job { ## Usage +Application queues belong in `app/queues/*.ts` and default-export `defineQueue(...)`. +Configure persistence once in `wrnexus.config.ts`: + +```ts +export default { + queue: { storage: "database", databaseName: "default" }, + // Or omit queue entirely for .wrnexus/queue.sqlite. +}; +``` + +Use `storage: "memory"` only for disposable work. A selected database must exist; +WrNexus fails startup/first use with a clear error instead of silently falling back. +Each job may declare `success(data, job)` and `failed(data, error, job)` lifecycle hooks. + Register workers, enqueue jobs, then start the poller: ```ts diff --git a/packages/queue/package.json b/packages/queue/package.json index d0d10767..50beabbf 100644 --- a/packages/queue/package.json +++ b/packages/queue/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/queue", - "version": "0.8.11", + "version": "0.8.12", "private": true, "type": "module", "main": "src/index.ts", @@ -10,6 +10,7 @@ }, "dependencies": { "@wrnexus/core": "workspace:*", + "@wrnexus/db": "workspace:*", "@wrnexus/rpc": "workspace:*" } } diff --git a/packages/queue/src/configured.ts b/packages/queue/src/configured.ts new file mode 100644 index 00000000..5f059f8b --- /dev/null +++ b/packages/queue/src/configured.ts @@ -0,0 +1,89 @@ +import { getDb, hasDb, registerDb, type Db } from "@wrnexus/db"; +import { connectFromConfig } from "@wrnexus/db/connect"; +import { memoryQueueStore, type QueueStore } from "./durable.ts"; +import { installSqliteQueueSchema, sqliteQueueStore } from "./sqlite.ts"; + +export type QueueStorage = "sqlite" | "database" | "memory"; + +export interface QueueStorageConfig { + /** Durable SQLite is the default. Use memory only for disposable/test queues. */ + storage?: QueueStorage; + /** Named `databases` connection, or `default` for the app's `db`. */ + databaseName?: string; + table?: string; + /** Used internally to resolve the default `.wrnexus/queue.sqlite` path. */ + appRoot?: string; +} + +let configured: QueueStorageConfig = { storage: "sqlite" }; + +export function configureQueueStorage(config: QueueStorageConfig = {}): void { + configured = { storage: "sqlite", ...config }; +} + +export function queueStorageConfig(): Readonly { + return { ...configured }; +} + +function lazyStore(resolve: () => Promise): QueueStore { + let pending: Promise | undefined; + const ready = () => (pending ??= resolve()); + return { + async put(job) { return (await ready()).put(job); }, + async get(id) { return (await ready()).get(id); }, + async remove(id) { return (await ready()).remove(id); }, + async due(now, limit) { return (await ready()).due(now, limit); }, + async list(name) { return (await ready()).list(name); }, + async findByIdempotencyKey(name, key) { + return (await ready()).findByIdempotencyKey?.(name, key) ?? null; + }, + async size() { return (await ready()).size?.() ?? (await (await ready()).list()).length; }, + async claim(id, worker, leaseUntil, now) { + return (await ready()).claim?.(id, worker, leaseUntil, now) ?? true; + }, + async release(id, worker) { await (await ready()).release?.(id, worker); }, + async archive(record) { await (await ready()).archive?.(record); }, + async history(id) { return (await ready()).history?.(id) ?? []; }, + }; +} + +async function resolveConfiguredStore(): Promise { + const config = configured; + if (config.storage === "memory") return memoryQueueStore(); + + const databaseName = config.databaseName ?? "default"; + let db: Db; + if (config.storage === "database") { + if (!hasDb(databaseName)) { + throw new Error( + `WRN-QUEUE-DATABASE: database '${databaseName}' is not configured. ` + + "Add it to db/databases or choose storage: 'sqlite'.", + ); + } + db = getDb(databaseName); + } else { + if (hasDb("__wrnexus_queue")) db = getDb("__wrnexus_queue"); + else { + db = connectFromConfig( + { driver: "sqlite", url: "file:./.wrnexus/queue.sqlite" }, + config.appRoot ?? process.cwd(), + ); + registerDb("__wrnexus_queue", db); + } + } + + if (db.driver.dialect !== "sqlite") { + throw new Error( + `WRN-QUEUE-DATABASE: configured queue storage currently requires SQLite; ` + + `database '${databaseName}' uses ${db.driver.dialect}. Pass a custom queue store for that driver.`, + ); + } + const table = config.table ?? "wrnexus_jobs"; + await installSqliteQueueSchema(db, table); + return sqliteQueueStore(db, table); +} + +/** Lazy global store used by defineQueue; runtime config is read on first operation. */ +export function configuredQueueStore(): QueueStore { + return lazyStore(resolveConfiguredStore); +} diff --git a/packages/queue/src/defined.ts b/packages/queue/src/defined.ts index 7d658740..811f1e15 100644 --- a/packages/queue/src/defined.ts +++ b/packages/queue/src/defined.ts @@ -5,11 +5,13 @@ import { type DurableQueueOptions, type QueueHealth, } from "./durable.ts"; +import { configuredQueueStore } from "./configured.ts"; export type JobStatus = "queued" | "completed" | "failed" | "cancelled" | "missing"; export interface DefinedJobOptions extends Omit { run: (data: T, context: JobContext & { job: Job }) => void | Promise; + success?: (data: T, job: Job) => void | Promise; failed?: (data: T, error: unknown, job: Job) => void | Promise; idempotency?: (data: T) => string | undefined; validate?: (data: unknown) => data is T; @@ -65,6 +67,7 @@ export function defineQueue( definition.queue ?? createDurableQueue({ ...definition.options, + store: definition.options?.store ?? configuredQueueStore(), async onDeadLetter(job, error) { await definition.options?.onDeadLetter?.(job, error); const prefix = `${name}:`; @@ -82,6 +85,7 @@ export function defineQueue( throw new TypeError(`WRN-QUEUE-PAYLOAD: invalid payload for '${jobName}'`); } await jobDefinition.run(job.data, { ...context, job }); + await jobDefinition.success?.(job.data, job); }; queue.process(jobName, handler); diff --git a/packages/queue/src/index.ts b/packages/queue/src/index.ts index af96a612..74ce3564 100644 --- a/packages/queue/src/index.ts +++ b/packages/queue/src/index.ts @@ -317,6 +317,8 @@ export type { } from "./defined.ts"; export { installSqliteQueueSchema, sqliteQueueSchema, sqliteQueueStore } from "./sqlite.ts"; export type { SqliteQueueClient } from "./sqlite.ts"; +export { configureQueueStorage, configuredQueueStore, queueStorageConfig } from "./configured.ts"; +export type { QueueStorage, QueueStorageConfig } from "./configured.ts"; export { redisQueueStore, postgresQueueStore, POSTGRES_QUEUE_SCHEMA } from "./stores.ts"; export type { RedisQueueClient, SqlQueueClient } from "./stores.ts"; export { diff --git a/packages/queue/test/defined.test.ts b/packages/queue/test/defined.test.ts index 13179a9d..b6e99663 100644 --- a/packages/queue/test/defined.test.ts +++ b/packages/queue/test/defined.test.ts @@ -3,6 +3,7 @@ import { createDurableQueue, defineQueue, memoryQueueStore } from "../src/index. test("defineQueue creates typed producers and registers workers", async () => { const seen: number[] = []; + const completed: number[] = []; const email = defineQueue({ name: "email", options: { store: memoryQueueStore() }, @@ -15,6 +16,7 @@ test("defineQueue creates typed producers and registers workers", async () => { data !== null && Number.isInteger((data as { messageId?: unknown }).messageId), run: async ({ messageId }: { messageId: number }) => void seen.push(messageId), + success: async ({ messageId }: { messageId: number }) => void completed.push(messageId), }, }, }); @@ -25,6 +27,7 @@ test("defineQueue creates typed producers and registers workers", async () => { expect(await email.send.status(first.id)).toBe("queued"); expect(await email.drain()).toBe(1); expect(seen).toEqual([7]); + expect(completed).toEqual([7]); expect(await email.send.status(first.id)).toBe("completed"); }); diff --git a/packages/styles/package.json b/packages/styles/package.json index 6045a8f9..1d6865c8 100644 --- a/packages/styles/package.json +++ b/packages/styles/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/styles", - "version": "0.8.17", + "version": "0.8.18", "type": "module", "main": "src/index.ts", "exports": { diff --git a/packages/styles/src/config.ts b/packages/styles/src/config.ts index 1306d0be..3fcc0fbd 100644 --- a/packages/styles/src/config.ts +++ b/packages/styles/src/config.ts @@ -300,6 +300,12 @@ export interface AppConfig { }; /** Default database connection (driver + url); reached with `getDb()`. */ db?: { driver: "sqlite" | "postgres" | "mysql" | "mongo"; url: string }; + /** Background-job persistence. Defaults to durable `.wrnexus/queue.sqlite`. */ + queue?: { + storage?: "sqlite" | "database" | "memory"; + databaseName?: string; + table?: string; + }; /** * File-upload storage. Declare named stores (local dir or S3-compatible), * upload with `handleUpload`/`upload` from `@wrnexus/uploader`, and serve diff --git a/packages/test/package.json b/packages/test/package.json index 9233a06c..eb7ef94b 100644 --- a/packages/test/package.json +++ b/packages/test/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/test", - "version": "0.8.10", + "version": "0.8.11", "private": true, "type": "module", "main": "src/index.ts", diff --git a/packages/test/src/index.ts b/packages/test/src/index.ts index 407bb754..3d3c1c5b 100644 --- a/packages/test/src/index.ts +++ b/packages/test/src/index.ts @@ -171,6 +171,7 @@ export async function createHarness( i18n: config.i18n, db: config.db, databases: config.databases, + queue: config.queue, }); return {