diff --git a/bun.lock b/bun.lock index e4c20ded..0af423be 100644 --- a/bun.lock +++ b/bun.lock @@ -292,7 +292,7 @@ }, "packages/cli": { "name": "@wrnexus/cli", - "version": "0.8.54", + "version": "0.8.55", "bin": { "wrnexus": "src/index.ts", }, @@ -342,14 +342,14 @@ }, "packages/csr": { "name": "@wrnexus/csr", - "version": "0.8.28", + "version": "0.8.29", "dependencies": { "@wrnexus/core": "workspace:*", }, }, "packages/db": { "name": "@wrnexus/db", - "version": "0.8.19", + "version": "0.8.20", "devDependencies": { "@types/bun": "^1.3.14", "typescript": "^6.0.3", @@ -393,9 +393,10 @@ }, "packages/encryption": { "name": "@wrnexus/encryption", - "version": "0.8.9", + "version": "0.8.10", "dependencies": { "@wrnexus/core": "workspace:*", + "@wrnexus/db": "workspace:*", }, "devDependencies": { "@types/bun": "^1.3.14", @@ -411,9 +412,11 @@ }, "packages/helpers": { "name": "@wrnexus/helpers", - "version": "0.8.8", + "version": "0.8.9", "dependencies": { + "@wrnexus/authz": "workspace:*", "@wrnexus/core": "workspace:*", + "@wrnexus/db": "workspace:*", }, }, "packages/i18n": { @@ -482,6 +485,13 @@ "vscode-html-languageservice": "^5.6.2", }, }, + "packages/mail": { + "name": "@wrnexus/mail", + "version": "0.8.1", + "dependencies": { + "@wrnexus/queue": "workspace:*", + }, + }, "packages/mcp": { "name": "@wrnexus/mcp", "version": "0.8.8", @@ -547,7 +557,7 @@ }, "packages/queue": { "name": "@wrnexus/queue", - "version": "0.8.12", + "version": "0.8.13", "dependencies": { "@wrnexus/core": "workspace:*", "@wrnexus/db": "workspace:*", @@ -575,7 +585,7 @@ }, "packages/reactive": { "name": "@wrnexus/reactive", - "version": "0.8.8", + "version": "0.8.9", }, "packages/realtime": { "name": "@wrnexus/realtime", @@ -635,7 +645,7 @@ }, "packages/styles": { "name": "@wrnexus/styles", - "version": "0.8.18", + "version": "0.8.19", "dependencies": { "@wrnexus/core": "workspace:*", "@wrnexus/plugin": "workspace:*", @@ -1036,6 +1046,8 @@ "@wrnexus/language-server": ["@wrnexus/language-server@workspace:packages/language-server"], + "@wrnexus/mail": ["@wrnexus/mail@workspace:packages/mail"], + "@wrnexus/managed-captcha-service": ["@wrnexus/managed-captcha-service@workspace:services/managed-captcha"], "@wrnexus/mcp": ["@wrnexus/mcp@workspace:packages/mcp"], diff --git a/docs/application-productivity.md b/docs/application-productivity.md new file mode 100644 index 00000000..43d02351 --- /dev/null +++ b/docs/application-productivity.md @@ -0,0 +1,43 @@ +# Application productivity kit + +WrNexus applications should declare infrastructure once and keep only domain decisions locally. + +## Included foundations + +1. HTTP routes: `defineApiRoute`, `authorized`, `requireUser`, `requirePermission`, `requireParam`, `requireJson`, and `json`. +2. CRUD resources: `defineResource` generates owned list/create/get/update/delete handlers. +3. Seed data: `addSeedData`, `upsertSeedData`, `seedIfMissing`, `defineSeed`, `removeSeedData`, and `runSeedQuery`. +4. Transactional delivery: `withOutbox`, `drainOutbox`, and `installOutboxSchema`. +5. Job state transitions: `defineJobStateMachine`. +6. SQLite/PostgreSQL/MySQL queue tables: configured storage plus `databaseQueueStore`. +7. Queue operations: success/failed hooks, dashboard handler and queue CLI. +8. Queue testing: `testQueue` with deterministic `runNext`/`runAll`. +9. Workflows and compensation: durable workflows and `defineSaga`. +10. Mail: `defineMail`, `defineMailTemplate`, sandbox allowlists and `queuedMail`. +11. Encrypted settings: `defineSecretResource`. +12. Ledgers: `defineLedger` with atomic debit and audit entries. +13. Forms: `useForm` for values, errors, dirty/submitting state, validation and submission. +14. Typed HTTP calls: `createApiClient`; generated OpenAPI/SDK clients remain available through `wrnexus api generate`. +15. Full-stack testing: `createTestApp`, database rollback/factories, and queue test controls. +16. Package migrations: queue/auth/authz packages contribute migrations through the plugin lifecycle. +17. Auth/OAuth: auth and authz plugins own routes, middleware, roles and provider lifecycle. +18. Code generation: `wrnexus generate resource `. +19. Incremental features: `wrnexus add queue|mail|seed|testing|resource`. +20. Starter presets: `wrnexus create app --preset=saas|email` and `--features=...`. + +## Recommended new-project flow + +```sh +wrnexus create acme --preset=saas +cd acme && bun install +wrnexus generate resource projects +wrnexus add mail +wrnexus db migrate +wrnexus db seed +wrnexus dev +``` + +Use generated resources for ordinary owned CRUD. Use the outbox for business operations that +change data and enqueue work together. Use sagas only when a workflow crosses systems that cannot +share a transaction. Application code should retain product-specific policy; framework code owns +storage, validation, lifecycle, retries, diagnostics and scaffolding. diff --git a/packages/cli/package.json b/packages/cli/package.json index d4a621df..36a1ace5 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/cli", - "version": "0.8.54", + "version": "0.8.55", "type": "module", "main": "src/index.ts", "exports": { diff --git a/packages/cli/src/add.ts b/packages/cli/src/add.ts new file mode 100644 index 00000000..e0bf6efa --- /dev/null +++ b/packages/cli/src/add.ts @@ -0,0 +1,35 @@ +import { existsSync, mkdirSync, writeFileSync } from "node:fs"; +import { dirname, join, resolve } from "node:path"; +import { runGenerateResource } from "./generate.ts"; + +const FEATURES: Record> = { + queue: { + "app/queues/example.ts": `import { defineQueue } from "@wrnexus/queue";\n\nexport default defineQueue({\n name: "example",\n jobs: {\n run: {\n run: async (data: { id: string }) => { console.log("job", data.id); },\n success: async (data) => { console.log("completed", data.id); },\n failed: async (data, error) => { console.error("failed", data.id, error); },\n },\n },\n});\n`, + }, + mail: { + "app/lib/mail.ts": `import { defineMail } from "@wrnexus/mail";\n\nexport const mail = defineMail({\n from: process.env.MAIL_FROM,\n driver: { send: async () => { throw new Error("Configure a mail driver"); } },\n sandbox: { enabled: process.env.NODE_ENV !== "production", allowlist: [] },\n});\n`, + }, + seed: { + "app/db/seed.ts": `import { addSeedData, defineSeed } from "@wrnexus/db";\n\nexport default defineSeed(async (db) => {\n await addSeedData(db, { name: "Example" }, "examples", { conflict: "ignore" });\n});\n`, + }, + testing: { + "test/app.test.ts": `import { expect, test } from "bun:test";\nimport { createTestApp } from "@wrnexus/test";\n\ntest("app boots", async () => {\n const app = await createTestApp(import.meta.dir + "/..");\n expect((await app.fetch("/healthz")).ok).toBe(true);\n app.close();\n});\n`, + }, +}; + +export function runAdd(root: string, feature: string | undefined, name?: string) { + if (feature === "resource") return runGenerateResource(root, name); + const files = feature ? FEATURES[feature] : undefined; + if (!files) + throw new Error(`Use wrnexus add ${Object.keys(FEATURES).join(" | ")} | resource `); + for (const [relative, content] of Object.entries(files)) { + const target = join(resolve(root), relative); + if (existsSync(target)) { + console.log(`• ${relative} exists — skipped`); + continue; + } + mkdirSync(dirname(target), { recursive: true }); + writeFileSync(target, content, "utf8"); + console.log(`✓ Added ${relative}`); + } +} diff --git a/packages/cli/src/create.ts b/packages/cli/src/create.ts index 2637926b..9100205c 100644 --- a/packages/cli/src/create.ts +++ b/packages/cli/src/create.ts @@ -7,6 +7,7 @@ import { existsSync, mkdirSync, writeFileSync } from "node:fs"; import { dirname, join, resolve } from "node:path"; +import { runAdd } from "./add.ts"; import { AI_GUIDE, CLAUDE_MD } from "./ai-guide.ts"; import { currentCliVersion } from "./update-notifier.ts"; @@ -672,13 +673,26 @@ export function scaffoldApp(root: string, appName: string): void { } } -export function createApp(name: string): void { +export function createApp(name: string, options: readonly string[] = []): void { if (!name) { console.error("Usage: wrnexus create "); process.exit(1); } - scaffoldApp(resolve(process.cwd(), name), name); + const root = resolve(process.cwd(), name); + scaffoldApp(root, name); + const preset = options.find((value) => value.startsWith("--preset="))?.split("=")[1]; + const requested = + options + .find((value) => value.startsWith("--features=")) + ?.split("=")[1] + ?.split(",") ?? []; + const features = new Set([ + ...(preset === "saas" ? ["queue", "mail", "seed", "testing"] : []), + ...(preset === "email" ? ["queue", "mail", "testing"] : []), + ...requested, + ]); + for (const feature of features) runAdd(root, feature); console.log(`✓ Created ${name}`); console.log(`\nNext steps:`); diff --git a/packages/cli/src/generate.ts b/packages/cli/src/generate.ts index d95d4a03..dec92c84 100644 --- a/packages/cli/src/generate.ts +++ b/packages/cli/src/generate.ts @@ -30,6 +30,46 @@ export interface GeneratedFile { content: string; } +export function scaffoldResource(name: string): GeneratedFile[] { + const clean = name.replace(/[^A-Za-z0-9_-]/g, "").toLowerCase(); + if (!clean) throw new TypeError("resource name is required"); + const singular = clean.replace(/s$/, ""); + const pascal = toPascalCase(singular); + return [ + { + path: `schemas/${singular}.ts`, + content: `import { v } from "@wrnexus/validation";\n\nexport default v.object({\n name: v.string().min(1, "Required"),\n});\n`, + }, + { + path: `resources/${clean}.ts`, + content: `import { getDb } from "@wrnexus/db";\nimport { defineResource } from "@wrnexus/helpers";\n\nexport default defineResource({\n name: "${clean}",\n db: getDb,\n table: "${clean}",\n owner: "owner_id",\n fields: ["name"],\n permissions: { read: "${singular}:read", write: "${singular}:write" },\n});\n`, + }, + { + path: `api/${clean}.ts`, + content: `import resource from "../resources/${clean}.ts";\nexport const GET = resource.list;\nexport const POST = resource.create;\n`, + }, + { + path: `api/${clean}/[id].ts`, + content: `import resource from "../../resources/${clean}.ts";\nexport const GET = resource.get;\nexport const PUT = resource.update;\nexport const DELETE = resource.remove;\n`, + }, + { + path: `pages/${clean}.wrn`, + content: `page ${pascal}List {\n view {\n

${pascal}

\n

Generated resource page. Connect it to /api/${clean}.

\n }\n}\n`, + }, + ]; +} + +export function runGenerateResource(appRoot: string, name: string | undefined): void { + if (!name) throw new TypeError("Usage: wrnexus generate resource "); + for (const file of scaffoldResource(name)) { + const target = join(resolve(appRoot), "app", file.path); + if (existsSync(target)) continue; + mkdirSync(dirname(target), { recursive: true }); + writeFileSync(target, file.content, "utf8"); + console.log(`✓ Generated app/${file.path}`); + } +} + function toPascalCase(name: string): string { return name .split(/[^A-Za-z0-9]+/) diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts index 64842f28..4cf1f3ac 100644 --- a/packages/cli/src/index.ts +++ b/packages/cli/src/index.ts @@ -167,7 +167,7 @@ async function main(): Promise { break; } case "create": - createApp(rest[0] ?? ""); + createApp(rest.find((value) => !value.startsWith("--")) ?? "", rest); break; case "workspace": { const { addWorkspaceApp, createWorkspace } = await import("./workspace.ts"); @@ -192,6 +192,11 @@ async function main(): Promise { } case "generate": case "g": { + if (rest[0] === "resource") { + const { runGenerateResource } = await import("./generate.ts"); + runGenerateResource(".", rest[1]); + break; + } if (rest[0] === "routes") { const { regenerateRoutes } = await import("./routes.ts"); const n = regenerateRoutes(join(resolve(rest[1] ?? "."), "app")); @@ -307,6 +312,21 @@ async function main(): Promise { if (!result.ok) process.exitCode = 1; break; } + case "queue": { + const values = rest.filter((value) => !value.startsWith("--")); + const { runQueueCommand } = await import("./queue-command.ts"); + await runQueueCommand(values[2] ?? ".", values[0] ?? "status", [ + values[1] ?? "", + ...rest.filter((value) => value.startsWith("--")), + ]); + break; + } + case "add": { + const values = rest.filter((value) => !value.startsWith("--")); + const { runAdd } = await import("./add.ts"); + runAdd(values[2] ?? ".", values[0], values[1]); + break; + } case "security": { const values = rest.filter((value) => !value.startsWith("--")); const { runSecurityCommand } = await import("./security-command.ts"); diff --git a/packages/cli/src/queue-command.ts b/packages/cli/src/queue-command.ts new file mode 100644 index 00000000..760e8249 --- /dev/null +++ b/packages/cli/src/queue-command.ts @@ -0,0 +1,86 @@ +import { resolve } from "node:path"; +import { loadAppConfig } from "@wrnexus/styles"; +import { connectFromConfig } from "@wrnexus/db/connect"; +import { installSqliteQueueSchema } from "@wrnexus/queue"; + +export async function runQueueCommand(root: string, command: string, args: readonly string[]) { + const appRoot = resolve(root); + const config = await loadAppConfig(appRoot, process.env.WRNEXUS_PROFILE); + const selected = config.queue?.databaseName ?? "default"; + const dbConfig = selected === "default" ? config.db : config.databases?.[selected]; + const effective = + config.queue?.storage === "database" + ? dbConfig + : { driver: "sqlite" as const, url: "file:./.wrnexus/queue.sqlite" }; + if (!effective || effective.driver === "mongo") + throw new Error(`Queue database '${selected}' is not configured as SQL.`); + const db = connectFromConfig(effective, appRoot); + if (db.driver.dialect !== "sqlite") + throw new Error("Queue CLI currently requires SQLite configured storage."); + const table = config.queue?.table ?? "wrnexus_jobs"; + await installSqliteQueueSchema(db, table); + try { + const id = args.find((value) => !value.startsWith("--")); + if (command === "list") { + console.table( + await db.all( + `SELECT id,name,run_at,priority,lease_owner,lease_until FROM ${table} ORDER BY run_at`, + ), + ); + return; + } + if (command === "status") { + const pending = await db.one<{ total: number }>(`SELECT COUNT(*) AS total FROM ${table}`); + const history = await db.one<{ total: number }>( + `SELECT COUNT(*) AS total FROM ${table}_history`, + ); + console.log( + JSON.stringify( + { pending: Number(pending?.total ?? 0), history: Number(history?.total ?? 0) }, + null, + 2, + ), + ); + return; + } + if (!id && command !== "clear") throw new Error(`wrnexus queue ${command} requires a job id`); + if (command === "inspect") { + console.log( + JSON.stringify(await db.one(`SELECT * FROM ${table} WHERE id = ?`, [id]), null, 2), + ); + return; + } + if (command === "cancel") { + await db.exec(`DELETE FROM ${table} WHERE id = ?`, [id]); + return; + } + if (command === "retry") { + const row = await db.one<{ payload: string }>( + `SELECT payload FROM ${table}_history WHERE id = ? AND state = 'failed'`, + [id], + ); + if (!row) throw new Error(`Failed job '${id}' was not found.`); + const job = JSON.parse(row.payload) as { + name: string; + runAt: number; + priority: number; + idempotencyKey?: string; + }; + job.runAt = Date.now(); + await db.exec( + `INSERT INTO ${table} (id,name,payload,run_at,priority,idempotency_key) VALUES (?,?,?,?,?,?)`, + [id, job.name, JSON.stringify(job), job.runAt, job.priority, job.idempotencyKey ?? null], + ); + return; + } + if (command === "clear" && args.includes("--completed")) { + await db.exec(`DELETE FROM ${table}_history WHERE state = 'completed'`); + return; + } + throw new Error( + "Use queue list | status | inspect | retry | cancel | clear --completed", + ); + } finally { + await db.close(); + } +} diff --git a/packages/cli/test/generate.test.ts b/packages/cli/test/generate.test.ts index 3f1ac94c..1e6033fc 100644 --- a/packages/cli/test/generate.test.ts +++ b/packages/cli/test/generate.test.ts @@ -1,5 +1,5 @@ import { test, expect } from "bun:test"; -import { scaffold } from "../src/generate.ts"; +import { scaffold, scaffoldResource } from "../src/generate.ts"; test("scaffold page: kebab file, PascalCase page name", () => { const f = scaffold("page", "about-us"); @@ -32,3 +32,15 @@ test("scaffold strips extensions from the given name", () => { expect(scaffold("page", "home.wrn").path).toBe("pages/home.wrn"); expect(scaffold("api", "ping.ts").path).toBe("api/ping.ts"); }); + +test("resource scaffold emits schema, handlers, routes and a page", () => { + const files = scaffoldResource("templates"); + expect(files.map((file) => file.path)).toEqual([ + "schemas/template.ts", + "resources/templates.ts", + "api/templates.ts", + "api/templates/[id].ts", + "pages/templates.wrn", + ]); + expect(files[1]?.content).toContain("defineResource"); +}); diff --git a/packages/csr/package.json b/packages/csr/package.json index cd4e9af4..769ac305 100644 --- a/packages/csr/package.json +++ b/packages/csr/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/csr", - "version": "0.8.28", + "version": "0.8.29", "type": "module", "main": "src/index.ts", "exports": { diff --git a/packages/csr/src/api-client.ts b/packages/csr/src/api-client.ts new file mode 100644 index 00000000..2a10dfc3 --- /dev/null +++ b/packages/csr/src/api-client.ts @@ -0,0 +1,41 @@ +export interface ApiRequest { + method?: string; + params?: Record; + query?: Record; + body?: Input; +} + +export interface ApiClientOptions { + baseUrl?: string; + fetch?: (input: RequestInfo | URL, init?: RequestInit) => Promise; +} + +/** Runtime used by generated API clients; throws a typed response error on non-2xx results. */ +export function createApiClient(options: ApiClientOptions = {}) { + const fetcher = options.fetch ?? globalThis.fetch; + return async function call( + template: string, + request: ApiRequest = {}, + ): Promise { + let path = template; + for (const [name, value] of Object.entries(request.params ?? {})) { + path = path.replace(`[${name}]`, encodeURIComponent(String(value))); + } + const url = new URL(path, options.baseUrl ?? globalThis.location?.origin ?? "http://localhost"); + for (const [name, value] of Object.entries(request.query ?? {})) + if (value !== undefined) url.searchParams.set(name, String(value)); + const response = await fetcher(url, { + method: request.method ?? (request.body === undefined ? "GET" : "POST"), + credentials: "same-origin", + headers: request.body === undefined ? undefined : { "content-type": "application/json" }, + body: request.body === undefined ? undefined : JSON.stringify(request.body), + }); + const data = response.status === 204 ? undefined : await response.json(); + if (!response.ok) + throw Object.assign( + new Error((data as { error?: string })?.error ?? `HTTP ${response.status}`), + { status: response.status, data }, + ); + return data as Output; + }; +} diff --git a/packages/csr/src/index.ts b/packages/csr/src/index.ts index 4b2ea357..7d0e0438 100644 --- a/packages/csr/src/index.ts +++ b/packages/csr/src/index.ts @@ -16,6 +16,8 @@ export { REACTIVE_RUNTIME } from "./reactive-runtime.ts"; export { NAV_RUNTIME } from "./nav-runtime.ts"; export { REALTIME_RUNTIME } from "./realtime-runtime.ts"; export { ACTION_RUNTIME } from "./action-runtime.ts"; +export { createApiClient } from "./api-client.ts"; +export type { ApiClientOptions, ApiRequest } from "./api-client.ts"; const CONTROLLER_SECTIONS = ["PRIMARY", "UI", "PIN"] as const; diff --git a/packages/csr/test/api-client.test.ts b/packages/csr/test/api-client.test.ts new file mode 100644 index 00000000..94604114 --- /dev/null +++ b/packages/csr/test/api-client.test.ts @@ -0,0 +1,23 @@ +import { expect, test } from "bun:test"; +import { createApiClient } from "../src/index.ts"; + +test("typed API client expands params, query and JSON input", async () => { + let request: Request | undefined; + const call = createApiClient({ + baseUrl: "https://app.test", + fetch: async (input, init) => { + request = new Request(input, init); + return Response.json({ ok: true }); + }, + }); + expect( + await call<{ ok: boolean }, { name: string }>("/api/users/[id]", { + method: "PUT", + params: { id: 7 }, + query: { audit: true }, + body: { name: "Ada" }, + }), + ).toEqual({ ok: true }); + expect(request?.url).toBe("https://app.test/api/users/7?audit=true"); + expect(await request?.json()).toEqual({ name: "Ada" }); +}); diff --git a/packages/db/package.json b/packages/db/package.json index 68c98c40..cec0b53e 100644 --- a/packages/db/package.json +++ b/packages/db/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/db", - "version": "0.8.19", + "version": "0.8.20", "private": true, "type": "module", "main": "./src/index.ts", diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 5a8a6302..98dc1303 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -61,5 +61,22 @@ 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"; +export { + addSeedData, + defineSeed, + removeSeedData, + runSeedQuery, + seedIfMissing, + seedUsers, + upsertSeedData, +} from "./seed.ts"; +export type { + AddSeedOptions, + SeedDatabase, + SeedRow, + SeedStep, + SeedUserAccount, + UpsertSeedOptions, +} from "./seed.ts"; +export { defineLedger } from "./ledger.ts"; +export type { LedgerOptions } from "./ledger.ts"; diff --git a/packages/db/src/ledger.ts b/packages/db/src/ledger.ts new file mode 100644 index 00000000..b23bd2d3 --- /dev/null +++ b/packages/db/src/ledger.ts @@ -0,0 +1,76 @@ +import type { Db } from "./driver.ts"; + +const SAFE = /^[A-Za-z_][A-Za-z0-9_]*$/; +const ident = (value: string) => { + if (!SAFE.test(value)) throw new TypeError(`Unsafe ledger identifier: ${value}`); + return value; +}; + +export interface LedgerOptions { + db: () => Db; + table: string; + subjectColumn?: string; + amountColumn?: string; + kindColumn?: string; + reasonColumn?: string; + referenceColumn?: string; +} + +export function defineLedger(options: LedgerOptions) { + const table = ident(options.table); + const subject = ident(options.subjectColumn ?? "subject_id"); + const amount = ident(options.amountColumn ?? "amount"); + const kind = ident(options.kindColumn ?? "kind"); + const reason = ident(options.reasonColumn ?? "reason"); + const reference = ident(options.referenceColumn ?? "ref"); + const positive = (value: number) => { + if (!Number.isSafeInteger(value) || value < 1) + throw new RangeError("ledger amount must be a positive safe integer"); + }; + const write = async ( + subjectId: string, + delta: number, + entryKind: string, + entryReason: string, + ref = "", + ) => { + if (!subjectId.trim()) throw new TypeError("ledger subject cannot be empty"); + await options + .db() + .exec( + `INSERT INTO ${table} (${subject},${amount},${kind},${reason},${reference}) VALUES (?,?,?,?,?)`, + [subjectId, delta, entryKind, entryReason, ref], + ); + }; + const balance = async (subjectId: string, db = options.db()) => { + const row = await db.one<{ balance: number | string | null }>( + `SELECT COALESCE(SUM(${amount}),0) AS balance FROM ${table} WHERE ${subject} = ?`, + [subjectId], + ); + return Number(row?.balance ?? 0); + }; + return { + balance, + async credit(subjectId: string, value: number, entryReason: string, ref = "") { + positive(value); + await write(subjectId, value, "credit", entryReason, ref); + }, + async debit(subjectId: string, value: number, entryReason: string, ref = "") { + positive(value); + return options.db().tx(async (db) => { + if ((await balance(subjectId, db)) < value) return false; + await db.exec( + `INSERT INTO ${table} (${subject},${amount},${kind},${reason},${reference}) VALUES (?,?,?,?,?)`, + [subjectId, -value, "debit", entryReason, ref], + ); + return true; + }); + }, + refund: (subjectId: string, value: number, entryReason: string, ref = "") => { + positive(value); + return write(subjectId, value, "refund", entryReason, ref); + }, + entries: (subjectId: string) => + options.db().all(`SELECT * FROM ${table} WHERE ${subject} = ? ORDER BY id DESC`, [subjectId]), + }; +} diff --git a/packages/db/src/seed.ts b/packages/db/src/seed.ts index 21142ec4..c9dc2912 100644 --- a/packages/db/src/seed.ts +++ b/packages/db/src/seed.ts @@ -7,6 +7,11 @@ export interface AddSeedOptions { conflict?: "error" | "ignore" | "replace"; } +export interface UpsertSeedOptions { + key: string | readonly string[]; + update?: readonly string[]; +} + 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`); @@ -81,3 +86,98 @@ export function runSeedQuery( if (!query.trim()) throw new TypeError("seed query cannot be empty"); return db.exec(query, [...data]); } + +/** Portable object upsert for SQLite, PostgreSQL and MySQL. */ +export async function upsertSeedData( + source: SeedDatabase, + data: SeedRow | readonly SeedRow[], + table: string, + options: UpsertSeedOptions, +): Promise { + const db = database(source); + const target = identifier(table, "seed table"); + const keys = (Array.isArray(options.key) ? options.key : [options.key]).map((key) => + identifier(key, "seed key"), + ); + if (!keys.length) throw new TypeError("upsertSeedData requires at least one key"); + const results: ExecResult[] = []; + for (const row of Array.isArray(data) ? data : [data]) { + 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")); + for (const key of keys) + if (!columns.includes(key)) throw new TypeError(`seed row is missing key '${key}'`); + const updates = (options.update ?? columns.filter((column) => !keys.includes(column))).map( + (column) => identifier(column, "seed update column"), + ); + const prefix = db.driver.dialect === "mysql" ? "INSERT INTO" : "INSERT INTO"; + const suffix = + db.driver.dialect === "mysql" + ? ` ON DUPLICATE KEY UPDATE ${updates.map((column) => `${column} = VALUES(${column})`).join(", ")}` + : ` ON CONFLICT (${keys.join(", ")}) DO ${ + updates.length + ? `UPDATE SET ${updates.map((column) => `${column} = excluded.${column}`).join(", ")}` + : "NOTHING" + }`; + results.push( + await db.exec( + `${prefix} ${target} (${columns.join(", ")}) VALUES (${columns.map(() => "?").join(", ")})${suffix}`, + entries.map(([, value]) => value), + ), + ); + } + return results; +} + +export async function seedIfMissing( + source: SeedDatabase, + table: string, + match: SeedRow, + data: SeedRow, +): Promise { + const db = database(source); + const target = identifier(table, "seed table"); + const where = Object.entries(match); + if (!where.length) throw new TypeError("seedIfMissing requires match fields"); + const existing = await db.one( + `SELECT 1 AS found FROM ${target} WHERE ${where + .map(([column]) => `${identifier(column, "seed column")} = ?`) + .join(" AND ")} LIMIT 1`, + where.map(([, value]) => value), + ); + if (existing) return false; + await addSeedData(db, data, target); + return true; +} + +export type SeedStep = (db: Db) => void | Promise; + +export function defineSeed(...steps: readonly SeedStep[]): SeedStep { + return async (db) => + db.tx(async (transaction) => { + for (const step of steps) await step(transaction); + }); +} + +export interface SeedUserAccount { + identifier: string; + password?: string; + [key: string]: unknown; +} + +export async function seedUsers( + accounts: readonly T[], + options: { + find(identifier: string): Promise; + create(account: T): Promise; + configure?(user: U, account: T): void | Promise; + }, +): Promise { + const users: U[] = []; + for (const account of accounts) { + const user = (await options.find(account.identifier)) ?? (await options.create(account)); + await options.configure?.(user, account); + users.push(user); + } + return users; +} diff --git a/packages/db/test/ledger-seed.test.ts b/packages/db/test/ledger-seed.test.ts new file mode 100644 index 00000000..a0bf503f --- /dev/null +++ b/packages/db/test/ledger-seed.test.ts @@ -0,0 +1,25 @@ +import { expect, test } from "bun:test"; +import { createDb, defineLedger, upsertSeedData } from "../src/index.ts"; +import { sqlite } from "../src/adapters/sqlite.ts"; + +test("portable seed upserts and ledgers remove application boilerplate", async () => { + const db = createDb(sqlite(":memory:")); + await db.exec("CREATE TABLE plans (code TEXT PRIMARY KEY, credits INTEGER NOT NULL)"); + await upsertSeedData(db, { code: "free", credits: 10 }, "plans", { key: "code" }); + await upsertSeedData(db, { code: "free", credits: 20 }, "plans", { key: "code" }); + expect((await db.one<{ credits: number }>("SELECT credits FROM plans"))?.credits).toBe(20); + await db.exec( + "CREATE TABLE ledger (id INTEGER PRIMARY KEY, user_id TEXT, delta INTEGER, kind TEXT, reason TEXT, ref TEXT)", + ); + const ledger = defineLedger({ + db: () => db, + table: "ledger", + subjectColumn: "user_id", + amountColumn: "delta", + }); + await ledger.credit("u1", 10, "grant"); + expect(await ledger.debit("u1", 4, "use")).toBe(true); + expect(await ledger.balance("u1")).toBe(6); + expect(await ledger.debit("u1", 7, "too much")).toBe(false); + await db.close(); +}); diff --git a/packages/encryption/package.json b/packages/encryption/package.json index d04f783c..2967668d 100644 --- a/packages/encryption/package.json +++ b/packages/encryption/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/encryption", - "version": "0.8.9", + "version": "0.8.10", "private": true, "type": "module", "main": "./src/index.ts", @@ -23,6 +23,7 @@ "typescript": "^6.0.3" }, "dependencies": { - "@wrnexus/core": "workspace:*" + "@wrnexus/core": "workspace:*", + "@wrnexus/db": "workspace:*" } } diff --git a/packages/encryption/src/index.ts b/packages/encryption/src/index.ts index 467b57f4..2ec0c9d2 100644 --- a/packages/encryption/src/index.ts +++ b/packages/encryption/src/index.ts @@ -160,3 +160,5 @@ export type { DecryptedHttpBody, ReplayStore, } from "./http.ts"; +export { defineSecretResource } from "./resource.ts"; +export type { SecretResourceOptions } from "./resource.ts"; diff --git a/packages/encryption/src/resource.ts b/packages/encryption/src/resource.ts new file mode 100644 index 00000000..2ce6807b --- /dev/null +++ b/packages/encryption/src/resource.ts @@ -0,0 +1,53 @@ +import type { Db } from "@wrnexus/db"; +import { decrypt, encrypt } from "./index.ts"; + +export interface SecretResourceOptions> { + db: () => Db; + table: string; + key: string | (() => string); + encrypted: readonly (keyof T & string)[]; + hidden?: readonly (keyof T & string)[]; + id?: string; +} + +const SAFE = /^[A-Za-z_][A-Za-z0-9_]*$/; + +/** Encrypt selected database fields and guarantee hidden fields never leave `safe()`. */ +export function defineSecretResource>( + options: SecretResourceOptions, +) { + const checked = (value: string) => { + if (!SAFE.test(value)) throw new TypeError(`Unsafe secret resource identifier: ${value}`); + return value; + }; + const table = checked(options.table); + const id = checked(options.id ?? "id"); + const encrypted = options.encrypted.map(checked); + const hidden = new Set((options.hidden ?? options.encrypted).map(checked)); + const key = () => (typeof options.key === "function" ? options.key() : options.key); + return { + async seal(input: T): Promise { + const output = { ...input }; + const writable = output as Record; + for (const field of encrypted) + if (writable[field] !== undefined) + writable[field] = await encrypt(String(writable[field]), key()); + return output; + }, + async open(input: T): Promise { + const output = { ...input }; + const writable = output as Record; + for (const field of encrypted) + if (writable[field]) writable[field] = await decrypt(String(writable[field]), key()); + return output; + }, + safe(input: T): Partial { + return Object.fromEntries( + Object.entries(input).filter(([field]) => !hidden.has(field)), + ) as Partial; + }, + async find(value: string | number): Promise { + return options.db().one(`SELECT * FROM ${table} WHERE ${id} = ?`, [value]); + }, + }; +} diff --git a/packages/helpers/package.json b/packages/helpers/package.json index 203daded..cc165504 100644 --- a/packages/helpers/package.json +++ b/packages/helpers/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/helpers", - "version": "0.8.8", + "version": "0.8.9", "private": true, "type": "module", "description": "Safe convenience helpers for WrNexus request contexts and common application flows.", @@ -9,6 +9,8 @@ ".": "./src/index.ts" }, "dependencies": { + "@wrnexus/authz": "workspace:*", + "@wrnexus/db": "workspace:*", "@wrnexus/core": "workspace:*" } } diff --git a/packages/helpers/src/http.ts b/packages/helpers/src/http.ts new file mode 100644 index 00000000..e2d93ae6 --- /dev/null +++ b/packages/helpers/src/http.ts @@ -0,0 +1,90 @@ +import type { Context } from "@wrnexus/core"; +import { can } from "@wrnexus/authz"; + +export class HttpError extends Error { + constructor( + public readonly status: number, + message: string, + public readonly details?: unknown, + ) { + super(message); + this.name = "HttpError"; + } +} + +export function subjectId(ctx: Pick): string { + return String((ctx.user as { id?: unknown } | undefined)?.id ?? "").trim(); +} + +export function requireUser(ctx: Pick): { id: string; user: unknown } { + const id = subjectId(ctx); + if (!id) throw new HttpError(401, "Unauthorized"); + return { id, user: ctx.user }; +} + +export function requireParam( + ctx: Pick, + name: string, + options: { integer?: boolean; positive?: boolean } = {}, +): string | number { + const raw = String(ctx.params[name] ?? "").trim(); + if (!raw) throw new HttpError(400, `Route parameter '${name}' is required.`); + if (!options.integer) return raw; + const value = Number(raw); + if (!Number.isInteger(value) || (options.positive && value < 1)) { + throw new HttpError(400, `Route parameter '${name}' must be a positive integer.`); + } + return value; +} + +export async function requireJson>( + ctx: Pick, +): Promise { + try { + return (await ctx.req.json()) as T; + } catch { + throw new HttpError(400, "Request body must be valid JSON."); + } +} + +export async function requirePermission( + ctx: Context, + permission: string, + resource?: unknown, +): Promise { + if (!(await can(ctx, permission, resource))) throw new HttpError(403, "Forbidden"); +} + +export const json = { + ok: (value: T) => Response.json(value), + created: (value: T) => Response.json(value, { status: 201 }), + noContent: () => new Response(null, { status: 204 }), + error: (status: number, message: string, details?: unknown) => + Response.json({ error: message, ...(details === undefined ? {} : { details }) }, { status }), +}; + +export type ApiHandler = (ctx: Context) => Response | Promise; + +/** Convert thrown HttpError values into the framework's standard JSON error shape. */ +export function defineApiRoute(handler: ApiHandler): ApiHandler { + return async (ctx) => { + try { + return await handler(ctx); + } catch (error) { + if (error instanceof HttpError) return json.error(error.status, error.message, error.details); + throw error; + } + }; +} + +export function authorized( + permission: string, + handler: ApiHandler, + resource?: (ctx: Context) => unknown | Promise, +): ApiHandler { + return defineApiRoute(async (ctx) => { + requireUser(ctx); + await requirePermission(ctx, permission, await resource?.(ctx)); + return handler(ctx); + }); +} diff --git a/packages/helpers/src/index.ts b/packages/helpers/src/index.ts index 0035e501..3c01ca9d 100644 --- a/packages/helpers/src/index.ts +++ b/packages/helpers/src/index.ts @@ -157,3 +157,17 @@ export { once, } from "./resilience.ts"; export type { RetryOptions } from "./resilience.ts"; +export { + HttpError, + authorized, + defineApiRoute, + json, + requireJson, + requireParam, + requirePermission, + requireUser, + subjectId, +} from "./http.ts"; +export type { ApiHandler } from "./http.ts"; +export { defineResource } from "./resource.ts"; +export type { ResourceDefinition, ResourceHandlers } from "./resource.ts"; diff --git a/packages/helpers/src/resource.ts b/packages/helpers/src/resource.ts new file mode 100644 index 00000000..8dc2ade9 --- /dev/null +++ b/packages/helpers/src/resource.ts @@ -0,0 +1,118 @@ +import type { Context } from "@wrnexus/core"; +import type { Db } from "@wrnexus/db"; +import { can } from "@wrnexus/authz"; +import { HttpError, defineApiRoute, json, requireJson, requireParam, requireUser } from "./http.ts"; + +const SAFE = /^[A-Za-z_][A-Za-z0-9_]*$/; +const ident = (value: string) => { + if (!SAFE.test(value)) throw new TypeError(`Unsafe resource identifier: ${value}`); + return value; +}; + +export interface ResourceDefinition> { + name: string; + db: () => Db; + table: string; + id?: string; + owner?: string; + fields: readonly (keyof T & string)[]; + permissions?: { read?: string; readAny?: string; write?: string; writeAny?: string }; + validate?: (input: unknown, mode: "create" | "update") => T | Promise; + serialize?: (row: T) => unknown; +} + +export interface ResourceHandlers { + list(ctx: Context): Response | Promise; + create(ctx: Context): Response | Promise; + get(ctx: Context): Response | Promise; + update(ctx: Context): Response | Promise; + remove(ctx: Context): Response | Promise; +} + +/** Build conventional owned CRUD handlers from one checked resource declaration. */ +export function defineResource>( + definition: ResourceDefinition, +): ResourceHandlers { + const table = ident(definition.table); + const id = ident(definition.id ?? "id"); + const owner = definition.owner ? ident(definition.owner) : undefined; + const fields = definition.fields.map(ident); + const shape = (row: T) => definition.serialize?.(row) ?? row; + const authorize = async (ctx: Context, action: "read" | "write", row?: T) => { + const subject = requireUser(ctx).id; + const permission = definition.permissions?.[action]; + const any = definition.permissions?.[`${action}Any`]; + if (any && (await can(ctx, any))) return subject; + if ( + permission && + !(await can(ctx, permission, owner ? { ownerId: row?.[owner] ?? subject } : row)) + ) { + throw new HttpError(403, "Forbidden"); + } + if (owner && row && String(row[owner]) !== subject) throw new HttpError(403, "Forbidden"); + return subject; + }; + const find = async (ctx: Context) => { + const value = requireParam(ctx, id, { integer: true, positive: true }); + const row = await definition.db().one(`SELECT * FROM ${table} WHERE ${id} = ?`, [value]); + if (!row) throw new HttpError(404, `${definition.name} not found`); + return row; + }; + return { + list: defineApiRoute(async (ctx) => { + const subject = await authorize(ctx, "read"); + const seeAll = + definition.permissions?.readAny && (await can(ctx, definition.permissions.readAny)); + const rows = await definition + .db() + .all( + `SELECT * FROM ${table}${owner && !seeAll ? ` WHERE ${owner} = ?` : ""} ORDER BY ${id} DESC`, + owner && !seeAll ? [subject] : [], + ); + return json.ok({ [definition.name]: rows.map(shape) }); + }), + create: defineApiRoute(async (ctx) => { + const subject = await authorize(ctx, "write"); + const input = definition.validate + ? await definition.validate(await requireJson(ctx), "create") + : await requireJson(ctx); + const values = Object.fromEntries( + fields.filter((key) => key in input).map((key) => [key, input[key]]), + ); + if (owner) values[owner] = subject; + const columns = Object.keys(values).map(ident); + const result = await definition.db().exec( + `INSERT INTO ${table} (${columns.join(", ")}) VALUES (${columns.map(() => "?").join(", ")})`, + columns.map((key) => values[key]), + ); + return json.created({ ok: true, id: result.lastInsertId }); + }), + get: defineApiRoute(async (ctx) => { + const row = await find(ctx); + await authorize(ctx, "read", row); + return json.ok({ [definition.name.replace(/s$/, "")]: shape(row) }); + }), + update: defineApiRoute(async (ctx) => { + const row = await find(ctx); + await authorize(ctx, "write", row); + const input = definition.validate + ? await definition.validate(await requireJson(ctx), "update") + : await requireJson(ctx); + const keys = fields.filter((key) => key in input); + if (!keys.length) throw new HttpError(400, "No writable fields supplied."); + await definition + .db() + .exec(`UPDATE ${table} SET ${keys.map((key) => `${key} = ?`).join(", ")} WHERE ${id} = ?`, [ + ...keys.map((key) => input[key]), + row[id], + ]); + return json.ok({ ok: true }); + }), + remove: defineApiRoute(async (ctx) => { + const row = await find(ctx); + await authorize(ctx, "write", row); + await definition.db().exec(`DELETE FROM ${table} WHERE ${id} = ?`, [row[id]]); + return json.ok({ ok: true }); + }), + }; +} diff --git a/packages/mail/README.md b/packages/mail/README.md new file mode 100644 index 00000000..bcb8b122 --- /dev/null +++ b/packages/mail/README.md @@ -0,0 +1,5 @@ +# @wrnexus/mail + +Driver-neutral transactional mail with templates, sandbox allowlists, connection tests and +optional durable queue producers. Applications supply an SMTP/API driver, while policy and +message rendering stay portable. diff --git a/packages/mail/package.json b/packages/mail/package.json new file mode 100644 index 00000000..d0135f62 --- /dev/null +++ b/packages/mail/package.json @@ -0,0 +1,13 @@ +{ + "name": "@wrnexus/mail", + "version": "0.8.1", + "private": true, + "type": "module", + "main": "src/index.ts", + "exports": { + ".": "./src/index.ts" + }, + "dependencies": { + "@wrnexus/queue": "workspace:*" + } +} diff --git a/packages/mail/src/index.ts b/packages/mail/src/index.ts new file mode 100644 index 00000000..4e528038 --- /dev/null +++ b/packages/mail/src/index.ts @@ -0,0 +1,88 @@ +import type { DefinedJob } from "@wrnexus/queue"; + +export interface MailMessage { + from?: string; + to: string | readonly string[]; + subject: string; + html?: string; + text?: string; + headers?: Record; +} + +export interface MailDriver { + send(message: MailMessage): Promise; + test?(): Promise<{ ok: boolean; error?: string }>; +} + +export interface MailTemplate> { + subject(data: T): string; + html?(data: T): string; + text?(data: T): string; +} + +export interface MailOptions { + driver: MailDriver; + from?: string; + sandbox?: { enabled?: boolean; allowlist?: readonly string[] }; +} + +export function defineMail(options: MailOptions) { + const allowed = new Set( + (options.sandbox?.allowlist ?? []).map((value) => value.trim().toLowerCase()), + ); + const assertAllowed = (recipients: readonly string[]) => { + if (!options.sandbox?.enabled) return; + const denied = recipients.find((recipient) => !allowed.has(recipient.trim().toLowerCase())); + if (denied) throw new Error(`WRN-MAIL-SANDBOX: recipient '${denied}' is not allow-listed`); + }; + return { + async send(message: MailMessage) { + const recipients = typeof message.to === "string" ? [message.to] : [...message.to]; + assertAllowed(recipients); + return options.driver.send({ ...message, from: message.from ?? options.from }); + }, + async render(template: MailTemplate, data: T, to: string | readonly string[]) { + return { + from: options.from, + to, + subject: template.subject(data), + html: template.html?.(data), + text: template.text?.(data), + } satisfies MailMessage; + }, + test: () => options.driver.test?.() ?? Promise.resolve({ ok: true }), + }; +} + +export function defineMailTemplate(template: MailTemplate): MailTemplate { + return template; +} + +export function interpolateTemplate(source: string, data: Record): string { + return source.replace(/\{\{\s*([A-Za-z_][A-Za-z0-9_.]*)\s*\}\}/g, (_match, path: string) => { + let value: unknown = data; + for (const part of path.split(".")) + value = + value && typeof value === "object" ? (value as Record)[part] : undefined; + return value == null ? "" : String(value); + }); +} + +export function storedMailTemplate(template: { + subject: string; + html?: string; + text?: string; +}): MailTemplate> { + return { + subject: (data) => interpolateTemplate(template.subject, data), + html: + template.html === undefined ? undefined : (data) => interpolateTemplate(template.html!, data), + text: + template.text === undefined ? undefined : (data) => interpolateTemplate(template.text!, data), + }; +} + +/** Queue a fully rendered message without coupling the mail package to one queue backend. */ +export function queuedMail(job: Pick, "add">) { + return { send: (message: T) => job.add(message) }; +} diff --git a/packages/mail/test/mail.test.ts b/packages/mail/test/mail.test.ts new file mode 100644 index 00000000..768e0a0f --- /dev/null +++ b/packages/mail/test/mail.test.ts @@ -0,0 +1,18 @@ +import { expect, test } from "bun:test"; +import { defineMail, defineMailTemplate } from "../src/index.ts"; + +test("mail templates render and sandbox blocks accidental recipients", async () => { + const sent: unknown[] = []; + const mail = defineMail({ + from: "hello@example.com", + sandbox: { enabled: true, allowlist: ["dev@example.com"] }, + driver: { send: async (message) => void sent.push(message) }, + }); + const welcome = defineMailTemplate<{ name: string }>({ + subject: ({ name }) => `Hello ${name}`, + html: ({ name }) => `${name}`, + }); + await mail.send(await mail.render(welcome, { name: "Ada" }, "dev@example.com")); + expect(sent).toHaveLength(1); + await expect(mail.send({ to: "real@example.com", subject: "no" })).rejects.toThrow("SANDBOX"); +}); diff --git a/packages/queue/package.json b/packages/queue/package.json index 50beabbf..27363b1a 100644 --- a/packages/queue/package.json +++ b/packages/queue/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/queue", - "version": "0.8.12", + "version": "0.8.13", "private": true, "type": "module", "main": "src/index.ts", diff --git a/packages/queue/src/admin.ts b/packages/queue/src/admin.ts new file mode 100644 index 00000000..a9a2cd61 --- /dev/null +++ b/packages/queue/src/admin.ts @@ -0,0 +1,37 @@ +import type { DurableQueue } from "./durable.ts"; +import { queueDashboardSnapshot, renderQueueDashboard } from "./scheduler.ts"; + +export interface QueueAdminOptions { + queues: Record; + authorize?: (request: Request) => boolean | Promise; +} + +/** Framework-neutral protected queue dashboard/API handler, mountable at any route. */ +export function createQueueAdminHandler(options: QueueAdminOptions) { + return async (request: Request): Promise => { + if (options.authorize && !(await options.authorize(request))) { + return Response.json({ error: "Forbidden" }, { status: 403 }); + } + const url = new URL(request.url); + const name = url.searchParams.get("queue") ?? Object.keys(options.queues)[0]; + const queue = name ? options.queues[name] : undefined; + if (!queue) return Response.json({ error: "Queue not found" }, { status: 404 }); + const id = url.searchParams.get("id"); + if (request.method === "POST" && id) { + const action = url.searchParams.get("action"); + const changed = + action === "retry" + ? await queue.retry(id) + : action === "cancel" + ? await queue.cancel(id) + : false; + return Response.json({ ok: changed }); + } + const snapshot = await queueDashboardSnapshot(queue); + if (url.searchParams.get("format") === "json") + return Response.json({ queue: name, ...snapshot }); + return new Response(renderQueueDashboard(snapshot), { + headers: { "content-type": "text/html; charset=utf-8" }, + }); + }; +} diff --git a/packages/queue/src/configured.ts b/packages/queue/src/configured.ts index 5f059f8b..f442b28e 100644 --- a/packages/queue/src/configured.ts +++ b/packages/queue/src/configured.ts @@ -2,6 +2,7 @@ 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"; +import { databaseQueueStore, installDatabaseQueueSchema } from "./database.ts"; export type QueueStorage = "sqlite" | "database" | "memory"; @@ -29,21 +30,39 @@ 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 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 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 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) ?? []; + }, }; } @@ -72,15 +91,13 @@ async function resolveConfiguredStore(): Promise { } } - 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); + if (db.driver.dialect === "sqlite") { + await installSqliteQueueSchema(db, table); + return sqliteQueueStore(db, table); + } + await installDatabaseQueueSchema(db, table); + return databaseQueueStore(db, table); } /** Lazy global store used by defineQueue; runtime config is read on first operation. */ diff --git a/packages/queue/src/database.ts b/packages/queue/src/database.ts new file mode 100644 index 00000000..eb61d94e --- /dev/null +++ b/packages/queue/src/database.ts @@ -0,0 +1,158 @@ +import type { Db } from "@wrnexus/db"; +import type { Job } from "./index.ts"; +import type { QueueJobRecord, QueueStore } from "./durable.ts"; + +const SAFE = /^[A-Za-z_][A-Za-z0-9_]*$/; +const check = (value: string) => { + if (!SAFE.test(value)) throw new TypeError(`Invalid queue table: ${value}`); + return value; +}; + +export async function installDatabaseQueueSchema(db: Db, table = "wrnexus_jobs") { + const target = check(table); + await db.exec(`CREATE TABLE IF NOT EXISTS ${target} ( + id VARCHAR(191) PRIMARY KEY, name VARCHAR(191) NOT NULL, payload TEXT NOT NULL, + run_at BIGINT NOT NULL, priority INTEGER NOT NULL DEFAULT 0, + idempotency_key VARCHAR(191), lease_owner VARCHAR(191), lease_until BIGINT + )`); + await db.exec(`CREATE TABLE IF NOT EXISTS ${target}_history ( + id VARCHAR(191) PRIMARY KEY, name VARCHAR(191) NOT NULL, state VARCHAR(32) NOT NULL, + payload TEXT NOT NULL, error TEXT, finished_at BIGINT NOT NULL + )`); + for (const statement of [ + `CREATE INDEX ${target}_due ON ${target} (run_at, priority)`, + `CREATE UNIQUE INDEX ${target}_idempotency ON ${target} (name, idempotency_key)`, + ]) { + try { + await db.exec(statement); + } catch (error) { + if (!/exist|duplicate|already/i.test(String(error))) throw error; + } + } +} + +export function databaseQueueStore(db: Db, table = "wrnexus_jobs"): QueueStore { + const target = check(table); + type Row = { payload: string }; + const decode = (row: Row) => JSON.parse(row.payload) as Job; + return { + async put(job) { + if (db.driver.dialect === "mysql") { + await db.exec( + `INSERT INTO ${target} (id,name,payload,run_at,priority,idempotency_key,lease_owner,lease_until) + VALUES (?,?,?,?,?,?,NULL,NULL) ON DUPLICATE KEY UPDATE name=VALUES(name),payload=VALUES(payload),run_at=VALUES(run_at),priority=VALUES(priority),lease_owner=NULL,lease_until=NULL`, + [ + job.id, + job.name, + JSON.stringify(job), + job.runAt, + job.priority, + job.idempotencyKey ?? null, + ], + ); + } else { + await db.exec( + `INSERT INTO ${target} (id,name,payload,run_at,priority,idempotency_key,lease_owner,lease_until) + VALUES (?,?,?,?,?,?,NULL,NULL) ON CONFLICT(id) DO UPDATE SET name=excluded.name,payload=excluded.payload,run_at=excluded.run_at,priority=excluded.priority,lease_owner=NULL,lease_until=NULL`, + [ + job.id, + job.name, + JSON.stringify(job), + job.runAt, + job.priority, + job.idempotencyKey ?? null, + ], + ); + } + }, + async get(id) { + const row = await db.one(`SELECT payload FROM ${target} WHERE id = ?`, [id]); + return row ? decode(row) : null; + }, + async remove(id) { + await db.exec(`DELETE FROM ${target} WHERE id = ?`, [id]); + }, + async due(now, limit) { + return ( + await db.all( + `SELECT payload FROM ${target} WHERE run_at <= ? AND (lease_until IS NULL OR lease_until < ?) ORDER BY priority DESC,run_at LIMIT ?`, + [now, now, limit], + ) + ).map(decode); + }, + async list(name) { + return ( + await db.all( + `SELECT payload FROM ${target}${name ? " WHERE name = ?" : ""} ORDER BY run_at`, + name ? [name] : [], + ) + ).map(decode); + }, + async findByIdempotencyKey(name, key) { + const row = await db.one( + `SELECT payload FROM ${target} WHERE name = ? AND idempotency_key = ?`, + [name, key], + ); + return row ? decode(row) : null; + }, + async size() { + return Number( + (await db.one<{ total: number | string }>(`SELECT COUNT(*) AS total FROM ${target}`)) + ?.total ?? 0, + ); + }, + async claim(id, worker, leaseUntil, now = Date.now()) { + return ( + ( + await db.exec( + `UPDATE ${target} SET lease_owner=?,lease_until=? WHERE id=? AND (lease_until IS NULL OR lease_until < ?)`, + [worker, leaseUntil, id, now], + ) + ).changes > 0 + ); + }, + async release(id, worker) { + await db.exec( + `UPDATE ${target} SET lease_owner=NULL,lease_until=NULL WHERE id=? AND lease_owner=?`, + [id, worker], + ); + }, + async archive(record) { + const values = [ + record.job.id, + record.job.name, + record.state, + JSON.stringify(record.job), + record.error ?? null, + record.finishedAt, + ]; + if (db.driver.dialect === "mysql") + await db.exec( + `INSERT INTO ${target}_history (id,name,state,payload,error,finished_at) VALUES (?,?,?,?,?,?) ON DUPLICATE KEY UPDATE state=VALUES(state),payload=VALUES(payload),error=VALUES(error),finished_at=VALUES(finished_at)`, + values, + ); + else + await db.exec( + `INSERT INTO ${target}_history (id,name,state,payload,error,finished_at) VALUES (?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET state=excluded.state,payload=excluded.payload,error=excluded.error,finished_at=excluded.finished_at`, + values, + ); + }, + async history(id) { + const rows = await db.all<{ + state: QueueJobRecord["state"]; + payload: string; + error: string | null; + finished_at: number; + }>( + `SELECT state,payload,error,finished_at FROM ${target}_history${id ? " WHERE id = ?" : ""} ORDER BY finished_at DESC`, + id ? [id] : [], + ); + return rows.map((row) => ({ + job: JSON.parse(row.payload), + state: row.state, + error: row.error ?? undefined, + finishedAt: Number(row.finished_at), + })); + }, + }; +} diff --git a/packages/queue/src/index.ts b/packages/queue/src/index.ts index 74ce3564..c0179ee0 100644 --- a/packages/queue/src/index.ts +++ b/packages/queue/src/index.ts @@ -317,6 +317,7 @@ export type { } from "./defined.ts"; export { installSqliteQueueSchema, sqliteQueueSchema, sqliteQueueStore } from "./sqlite.ts"; export type { SqliteQueueClient } from "./sqlite.ts"; +export { databaseQueueStore, installDatabaseQueueSchema } from "./database.ts"; export { configureQueueStorage, configuredQueueStore, queueStorageConfig } from "./configured.ts"; export type { QueueStorage, QueueStorageConfig } from "./configured.ts"; export { redisQueueStore, postgresQueueStore, POSTGRES_QUEUE_SCHEMA } from "./stores.ts"; @@ -340,4 +341,15 @@ export type { } from "./workflow.ts"; export { subjectQueue } from "./subject.ts"; export type { SubjectJob, SubjectQueue } from "./subject.ts"; +export { + defineJobStateMachine, + defineSaga, + drainOutbox, + installOutboxSchema, + testQueue, + withOutbox, +} from "./patterns.ts"; +export { createQueueAdminHandler } from "./admin.ts"; +export type { QueueAdminOptions } from "./admin.ts"; +export type { JobStateMachineOptions, OutboxEntry, OutboxWriter, SagaStep } from "./patterns.ts"; import { createExecutionContext, type ExecutionContext } from "@wrnexus/core"; diff --git a/packages/queue/src/patterns.ts b/packages/queue/src/patterns.ts new file mode 100644 index 00000000..a1da550a --- /dev/null +++ b/packages/queue/src/patterns.ts @@ -0,0 +1,171 @@ +import type { Db } from "@wrnexus/db"; +import type { AddOptions, Job } from "./index.ts"; +import type { DefinedJob } from "./defined.ts"; +import type { DurableQueue } from "./durable.ts"; + +const SAFE = /^[A-Za-z_][A-Za-z0-9_]*$/; +const ident = (value: string) => { + if (!SAFE.test(value)) throw new TypeError(`Unsafe queue identifier: ${value}`); + return value; +}; + +export interface OutboxEntry { + id: number; + queue: string; + payload: T; + options?: AddOptions; +} + +export interface OutboxWriter { + enqueue(job: Pick, "name">, data: T, options?: AddOptions): Promise; +} + +export async function installOutboxSchema(db: Db, table = "wrnexus_outbox"): Promise { + const target = ident(table); + const id = + db.driver.dialect === "postgres" + ? "BIGSERIAL PRIMARY KEY" + : "INTEGER PRIMARY KEY AUTOINCREMENT"; + await db.exec(`CREATE TABLE IF NOT EXISTS ${target} ( + id ${id}, queue_name TEXT NOT NULL, payload TEXT NOT NULL, + options TEXT, created_at BIGINT NOT NULL, dispatched_at BIGINT + )`); + await db.exec(`CREATE INDEX IF NOT EXISTS ${target}_pending ON ${target} (dispatched_at, id)`); +} + +/** Commit business writes and durable enqueue intents in the same database transaction. */ +export async function withOutbox( + db: Db, + operation: (context: { db: Db; enqueue: OutboxWriter["enqueue"] }) => Promise, + table = "wrnexus_outbox", +): Promise { + await installOutboxSchema(db, table); + const target = ident(table); + return db.tx((transaction) => + operation({ + db: transaction, + enqueue: async (job, data, options) => { + await transaction.exec( + `INSERT INTO ${target} (queue_name,payload,options,created_at,dispatched_at) VALUES (?,?,?,?,NULL)`, + [job.name, JSON.stringify(data), options ? JSON.stringify(options) : null, Date.now()], + ); + }, + }), + ); +} + +/** Dispatch committed outbox rows. A crash before marking simply retries idempotently. */ +export async function drainOutbox( + db: Db, + resolve: ( + queueName: string, + ) => { add(data: unknown, options?: AddOptions): Promise } | undefined, + options: { table?: string; limit?: number } = {}, +): Promise { + const target = ident(options.table ?? "wrnexus_outbox"); + await installOutboxSchema(db, target); + const rows = await db.all<{ + id: number; + queue_name: string; + payload: string; + options: string | null; + }>( + `SELECT id,queue_name,payload,options FROM ${target} WHERE dispatched_at IS NULL ORDER BY id LIMIT ?`, + [options.limit ?? 100], + ); + let dispatched = 0; + for (const row of rows) { + const producer = resolve(row.queue_name); + if (!producer) continue; + const addOptions = row.options ? (JSON.parse(row.options) as AddOptions) : {}; + await producer.add(JSON.parse(row.payload), { + idempotencyKey: `outbox:${row.id}`, + ...addOptions, + }); + await db.exec(`UPDATE ${target} SET dispatched_at = ? WHERE id = ? AND dispatched_at IS NULL`, [ + Date.now(), + row.id, + ]); + dispatched++; + } + return dispatched; +} + +export interface JobStateMachineOptions { + db: () => Db; + table: string; + idColumn?: string; + stateColumn?: string; + states: { queued: string; running: string; success: string; failed: string }; +} + +export function defineJobStateMachine(options: JobStateMachineOptions) { + const table = ident(options.table); + const id = ident(options.idColumn ?? "id"); + const state = ident(options.stateColumn ?? "status"); + const transition = async (recordId: string | number, from: string, to: string) => + ( + await options + .db() + .exec(`UPDATE ${table} SET ${state} = ? WHERE ${id} = ? AND ${state} = ?`, [ + to, + recordId, + from, + ]) + ).changes > 0; + return { + queue: (recordId: string | number) => + transition(recordId, options.states.failed, options.states.queued), + claim: (recordId: string | number) => + transition(recordId, options.states.queued, options.states.running), + complete: (recordId: string | number) => + transition(recordId, options.states.running, options.states.success), + fail: (recordId: string | number) => + transition(recordId, options.states.running, options.states.failed), + transition, + }; +} + +export function testQueue(queue: DurableQueue) { + return { + add: queue.add.bind(queue), + runNext: () => queue.drain(), + async runAll(limit = 1_000) { + let total = 0; + while (total < limit) { + const count = await queue.drain(); + if (!count) break; + total += count; + } + return total; + }, + get: queue.get.bind(queue), + status: queue.status.bind(queue), + history: queue.history.bind(queue), + shutdown: queue.shutdown.bind(queue), + }; +} + +export interface SagaStep { + name: string; + run(value: T): void | Promise; + compensate?(value: T, cause: unknown): void | Promise; +} + +export function defineSaga(definition: { name: string; steps: readonly SagaStep[] }) { + return { + name: definition.name, + async run(value: T): Promise { + const completed: SagaStep[] = []; + try { + for (const step of definition.steps) { + await step.run(value); + completed.push(step); + } + } catch (cause) { + for (const step of completed.reverse()) await step.compensate?.(value, cause); + throw cause; + } + }, + }; +} diff --git a/packages/queue/test/patterns.test.ts b/packages/queue/test/patterns.test.ts new file mode 100644 index 00000000..f4e98f44 --- /dev/null +++ b/packages/queue/test/patterns.test.ts @@ -0,0 +1,50 @@ +import { expect, test } from "bun:test"; +import { createDb } from "@wrnexus/db"; +import { sqlite } from "@wrnexus/db/sqlite"; +import { + createDurableQueue, + defineSaga, + drainOutbox, + memoryQueueStore, + testQueue, + withOutbox, +} from "../src/index.ts"; + +test("outbox commits intents and deterministic queue helpers dispatch them", async () => { + const db = createDb(sqlite(":memory:")); + const queue = createDurableQueue({ store: memoryQueueStore() }); + const seen: number[] = []; + queue.process("email:send", async (job) => void seen.push((job.data as { id: number }).id)); + await withOutbox(db, async ({ enqueue }) => enqueue({ name: "email:send" } as never, { id: 7 })); + expect( + await drainOutbox(db, (name) => + name === "email:send" + ? { add: (data, options) => queue.add(name, data, options) } + : undefined, + ), + ).toBe(1); + expect(await testQueue(queue).runAll()).toBe(1); + expect(seen).toEqual([7]); +}); + +test("sagas compensate completed steps in reverse order", async () => { + const calls: string[] = []; + const saga = defineSaga({ + name: "send", + steps: [ + { + name: "reserve", + run: () => void calls.push("reserve"), + compensate: () => void calls.push("refund"), + }, + { + name: "queue", + run: () => { + throw new Error("down"); + }, + }, + ], + }); + await expect(saga.run({})).rejects.toThrow("down"); + expect(calls).toEqual(["reserve", "refund"]); +}); diff --git a/packages/reactive/package.json b/packages/reactive/package.json index a18aba7f..e427a7e7 100644 --- a/packages/reactive/package.json +++ b/packages/reactive/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/reactive", - "version": "0.8.8", + "version": "0.8.9", "type": "module", "main": "src/index.ts", "exports": { diff --git a/packages/reactive/src/form.ts b/packages/reactive/src/form.ts new file mode 100644 index 00000000..c23dfe4a --- /dev/null +++ b/packages/reactive/src/form.ts @@ -0,0 +1,75 @@ +import { computed, signal, type ReadonlySignal, type Signal } from "./signal.ts"; + +export interface FormState> { + values: Signal; + errors: Signal>>; + submitting: Signal; + dirty: ReadonlySignal; + field( + name: K, + ): { + value: T[K]; + onInput(payload: { value?: unknown; checked?: boolean }): void; + }; + set(name: K, value: T[K]): void; + reset(next?: T): void; + validate(): Promise; + submit(operation: (values: T) => R | Promise): Promise; +} + +export function useForm>( + initial: T, + validator?: ( + values: T, + ) => Partial> | Promise>>, +): FormState { + let baseline = structuredClone(initial); + const values = signal(structuredClone(initial)); + const errors = signal>>({}); + const submitting = signal(false); + const dirty = computed(() => JSON.stringify(values.get()) !== JSON.stringify(baseline)); + const set = (name: K, value: T[K]) => { + values.set({ ...values.get(), [name]: value }); + if (errors.get()[name]) errors.set({ ...errors.get(), [name]: undefined }); + }; + const validate = async () => { + errors.set((await validator?.(values.get())) ?? {}); + return Object.values(errors.get()).every((value) => !value); + }; + return { + values, + errors, + submitting, + dirty, + field(name) { + return { + value: values.get()[name], + onInput(payload) { + const current = values.get()[name]; + set( + name, + (typeof current === "boolean" + ? Boolean(payload.checked) + : payload.value) as T[typeof name], + ); + }, + }; + }, + set, + reset(next = initial) { + baseline = structuredClone(next); + values.set(structuredClone(next)); + errors.set({}); + }, + validate, + async submit(operation) { + if (!(await validate())) throw new Error("WRN-FORM-INVALID"); + submitting.set(true); + try { + return await operation(values.get()); + } finally { + submitting.set(false); + } + }, + }; +} diff --git a/packages/reactive/src/index.ts b/packages/reactive/src/index.ts index b345832c..c6ac7608 100644 --- a/packages/reactive/src/index.ts +++ b/packages/reactive/src/index.ts @@ -17,6 +17,8 @@ export { createTimeline, urlSignal, } from "./advanced.ts"; +export { useForm } from "./form.ts"; +export type { FormState } from "./form.ts"; export type { HistorySignal, ReactiveContext, diff --git a/packages/reactive/test/form.test.ts b/packages/reactive/test/form.test.ts new file mode 100644 index 00000000..61c0eb6d --- /dev/null +++ b/packages/reactive/test/form.test.ts @@ -0,0 +1,14 @@ +import { expect, test } from "bun:test"; +import { useForm } from "../src/index.ts"; + +test("form state binds fields, validates, submits and resets", async () => { + const form = useForm({ name: "", active: false }, (value) => ({ + name: value.name ? undefined : "Required", + })); + form.field("name").onInput({ value: "Ada" }); + form.field("active").onInput({ checked: true }); + expect(form.dirty.get()).toBe(true); + expect(await form.submit(async (value) => value.name)).toBe("Ada"); + form.reset(); + expect(form.values.get()).toEqual({ name: "", active: false }); +}); diff --git a/packages/styles/package.json b/packages/styles/package.json index 1d6865c8..3ef9a82a 100644 --- a/packages/styles/package.json +++ b/packages/styles/package.json @@ -1,6 +1,6 @@ { "name": "@wrnexus/styles", - "version": "0.8.18", + "version": "0.8.19", "type": "module", "main": "src/index.ts", "exports": { diff --git a/packages/styles/src/config.ts b/packages/styles/src/config.ts index 3fcc0fbd..999392c9 100644 --- a/packages/styles/src/config.ts +++ b/packages/styles/src/config.ts @@ -305,6 +305,13 @@ export interface AppConfig { storage?: "sqlite" | "database" | "memory"; databaseName?: string; table?: string; + dashboard?: { enabled?: boolean; path?: string; permission?: string }; + }; + /** Driver-neutral mail defaults consumed by @wrnexus/mail/application drivers. */ + mail?: { + from?: string; + queue?: string | boolean; + sandbox?: { enabled?: boolean; allowlist?: string[] }; }; /** * File-upload storage. Declare named stores (local dir or S3-compatible),