productionize

Author: jaredLunde

.sqlx/query-5c4b0ca90761c24ad202cf91affecae645162448622ff5b19df624e791b85b04.json

@@ -59,8 +59,29 @@ run = "npm test"
 dir = "sdk/ts/events"
 depends = ["generate:types", "build:rs"]
 
+[tasks."test:ts"]
+depends = [
+  "test:integration:queue",
+  "test:integration:events",
+  "test:types:queue",
+  "test:types:events",
+]
+
 [tasks."test:integration:ts"]
-depends = ["test:integration:queue", "test:integration:events"]
+depends = ["test:ts"]
+
+[tasks."test:types:queue"]
+run = "npx vitest --config vitest.typecheck.config.ts --run"
+dir = "sdk/ts/queue"
+depends = ["generate:types"]
+
+[tasks."test:types:events"]
+run = "npx vitest --config vitest.typecheck.config.ts --run"
+dir = "sdk/ts/events"
+depends = ["generate:types"]
+
+[tasks."test:types:ts"]
+depends = ["test:types:queue", "test:types:events"]
 
 [tasks."generate:openapi"]
 run = "cargo run -- generate-openapi"

mise.toml

@@ -29,7 +29,7 @@ async function waitForHealthy(url: string, timeoutMs = 30_000): Promise<void> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
     try {
-      const res = await fetch(`${url}/healthz`);
+      const res = await fetch(`${url}/livez`);
       if (res.ok) return;
     } catch {
       // server not up yet
@@ -50,18 +50,18 @@ export async function setup(): Promise<void> {
   const sql = postgres(databaseUrl, { max: 1 });
   const schemaPath = resolve(
     __dirname,
-    "../../../../../beyond-queue-extension/sql/schema.sql",
+    "../../../../beyond-queue-extension/sql/schema.sql",
   );
   const hotPathsPath = resolve(
     __dirname,
-    "../../../../../tests/fixtures/hot_paths.sql",
+    "../../../../tests/fixtures/hot_paths.sql",
   );
   await sql.unsafe(readFileSync(schemaPath, "utf8"));
   await sql.unsafe(readFileSync(hotPathsPath, "utf8"));
   await sql.end();
 
   const binaryPath = process.env["BEYOND_QUEUE_BINARY"]
-    ?? resolve(__dirname, "../../../../../target/debug/beyond-queue");
+    ?? resolve(__dirname, "../../../../target/debug/beyond-queue");
 
   serverProcess = spawn(binaryPath, ["serve"], {
     env: {

sdk/ts/events/__tests__/global-setup.ts

@@ -0,0 +1,67 @@
+import { describe, expectTypeOf, it } from "vitest";
+import {
+  createEventClient,
+  type EventClient,
+  type EventSchemaClient,
+} from "../src/index.js";
+
+const schema = {
+  "user.*": { parse: (v: unknown) => v as { userId: string } },
+  "order.placed": {
+    parse: (v: unknown) => v as { orderId: string; amount: number },
+  },
+};
+
+const ev = createEventClient({ schema, url: "http://localhost" });
+const plain = createEventClient({ url: "http://localhost" });
+
+describe("createEventClient overloads", () => {
+  it("returns EventSchemaClient when schema is provided", () => {
+    expectTypeOf(ev).toMatchTypeOf<EventSchemaClient<typeof schema>>();
+  });
+
+  it("returns EventClient when no schema is provided", () => {
+    expectTypeOf(plain).toMatchTypeOf<EventClient>();
+  });
+});
+
+describe("publish — payload type", () => {
+  it("infers payload from exact key match", () => {
+    expectTypeOf(ev.publish<"order.placed">).parameter(1).toMatchTypeOf<{
+      orderId: string;
+      amount: number;
+    }>();
+  });
+
+  it("infers payload via glob match (user.created → user.*)", () => {
+    expectTypeOf(ev.publish<"user.created">).parameter(1).toMatchTypeOf<{
+      userId: string;
+    }>();
+  });
+
+  it("infers payload via glob match (user.deleted → user.*)", () => {
+    expectTypeOf(ev.publish<"user.deleted">).parameter(1).toMatchTypeOf<{
+      userId: string;
+    }>();
+  });
+
+  it("falls back to JsonValue for unmatched routing key", () => {
+    expectTypeOf(ev.publish<"payment.failed">).parameter(1).toMatchTypeOf<
+      string | number | boolean | null | object
+    >();
+  });
+
+  it("plain client accepts JsonValue for any routing key", () => {
+    expectTypeOf(plain.publish).parameter(1).toMatchTypeOf<
+      string | number | boolean | null | object
+    >();
+  });
+});
+
+describe("subscriptions passthrough", () => {
+  it("schema client retains subscriptions interface unchanged", () => {
+    expectTypeOf(ev.subscriptions).toMatchTypeOf<
+      EventClient["subscriptions"]
+    >();
+  });
+});

sdk/ts/events/__tests__/schema.test-d.ts

@@ -78,6 +78,30 @@ export interface EventClientOptions {
   onResponse?: (event: EventResponseEvent) => void;
 }
 
+// ── Schema types ──────────────────────────────────────────────────────────────
+
+/** A value parser. Compatible with Zod schemas and any object with `.parse`. */
+export type Schema<T> = { parse: (value: unknown) => T };
+
+/** Maps routing key patterns (including globs) to their payload schemas. */
+export type EventSchemaMap = Record<string, Schema<unknown>>;
+
+type GlobMatch<K extends string, P extends string> = P extends
+  `${infer Pre}*${infer Suf}` ? K extends `${Pre}${string}${Suf}` ? true : false
+  : K extends P ? true
+  : false;
+
+type MatchedPattern<K extends string, Map extends EventSchemaMap> = {
+  [P in keyof Map & string]: GlobMatch<K, P> extends true ? P : never;
+}[keyof Map & string];
+
+export type EventPayloadType<K extends string, Map extends EventSchemaMap> =
+  [MatchedPattern<K, Map>] extends [never] ? JsonValue
+    : Map[MatchedPattern<K, Map> & keyof Map] extends Schema<infer T> ? T
+    : JsonValue;
+
+// ── Client interfaces ─────────────────────────────────────────────────────────
+
 export interface EventClient {
   /**
    * Publish a message to a routing key. All subscriptions whose pattern
@@ -110,6 +134,21 @@ export interface EventClient {
   close(): Promise<void>;
 }
 
+/**
+ * An event client with schema-aware payload types. Returned by `createEventClient`
+ * when a `schema` map is provided. Routing keys matching a schema pattern get typed
+ * payloads; unmatched keys fall back to `JsonValue`.
+ */
+export interface EventSchemaClient<Map extends EventSchemaMap>
+  extends Omit<EventClient, "publish">
+{
+  publish<K extends string>(
+    routingKey: K,
+    payload: EventPayloadType<K, Map>,
+    opts?: PublishOptions,
+  ): EventResult<PublishResult>;
+}
+
 // ── Internal helpers ──────────────────────────────────────────────────────────
 
 function toEventError(error: unknown, response: Response): EventError {
@@ -184,22 +223,29 @@ function buildFetch(
 
 // ── Factory ───────────────────────────────────────────────────────────────────
 
+/** Creates a schema-aware event client. Publish payload types are inferred from the schema map. */
+export function createEventClient<Map extends EventSchemaMap>(
+  opts: EventClientOptions & { schema: Map },
+): EventSchemaClient<Map>;
 /** Creates an event client backed by the beyond-queue HTTP API. */
-export function createEventClient(opts: EventClientOptions = {}): EventClient {
-  const url = opts.url ?? env["BEYOND_EVENTS_URL"];
+export function createEventClient(opts?: EventClientOptions): EventClient;
+export function createEventClient(
+  opts?: EventClientOptions & { schema?: EventSchemaMap },
+): EventClient {
+  const url = opts?.url ?? env["BEYOND_EVENTS_URL"];
   if (!url) {
     throw new Error(
       "BEYOND_EVENTS_URL is required (pass `url` or set the BEYOND_EVENTS_URL env var)",
     );
   }
   const base = url.replace(/\/+$/, "");
-  const token = opts.token ?? env["BEYOND_EVENTS_TOKEN"];
-  const { onRequest, onResponse } = opts;
+  const token = opts?.token ?? env["BEYOND_EVENTS_TOKEN"];
+  const { onRequest, onResponse } = opts ?? {};
 
   const client = createFetchClient<paths>({
     baseUrl: base,
     headers: { Authorization: `Bearer ${token ?? "anon"}` },
-    fetch: buildFetch(opts.fetch, opts.retries ?? 2, opts.timeout),
+    fetch: buildFetch(opts?.fetch, opts?.retries ?? 2, opts?.timeout),
   });
 
   function cmd<A extends unknown[], R>(

sdk/ts/events/src/client.ts

@@ -18,11 +18,15 @@ export { createEventClient } from "./client.js";
 export type {
   EventClient,
   EventClientOptions,
+  EventPayloadType,
   EventResult,
+  EventSchemaClient,
+  EventSchemaMap,
   EventTarget,
   JsonValue,
   PublishOptions,
   PublishResult,
+  Schema,
   Subscription,
 } from "./client.js";
 export { EventError } from "./errors.js";

sdk/ts/events/src/index.ts

@@ -0,0 +1,11 @@
+import { defineConfig } from "vitest/config";
+
+export default defineConfig({
+  test: {
+    environment: "node",
+    include: ["__tests__/**/*.test-d.ts"],
+    typecheck: {
+      include: ["__tests__/**/*.test-d.ts"],
+    },
+  },
+});

sdk/ts/events/vitest.typecheck.config.ts

@@ -29,7 +29,7 @@ async function waitForHealthy(url: string, timeoutMs = 30_000): Promise<void> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
     try {
-      const res = await fetch(`${url}/healthz`);
+      const res = await fetch(`${url}/livez`);
       if (res.ok) return;
     } catch {
       // server not up yet
@@ -50,18 +50,18 @@ export async function setup(): Promise<void> {
   const sql = postgres(databaseUrl, { max: 1 });
   const schemaPath = resolve(
     __dirname,
-    "../../../../../beyond-queue-extension/sql/schema.sql",
+    "../../../../beyond-queue-extension/sql/schema.sql",
   );
   const hotPathsPath = resolve(
     __dirname,
-    "../../../../../tests/fixtures/hot_paths.sql",
+    "../../../../tests/fixtures/hot_paths.sql",
   );
   await sql.unsafe(readFileSync(schemaPath, "utf8"));
   await sql.unsafe(readFileSync(hotPathsPath, "utf8"));
   await sql.end();
 
   const binaryPath = process.env["BEYOND_QUEUE_BINARY"]
-    ?? resolve(__dirname, "../../../../../target/debug/beyond-queue");
+    ?? resolve(__dirname, "../../../../target/debug/beyond-queue");
 
   serverProcess = spawn(binaryPath, ["serve"], {
     env: {