postgres: fix tests (#394)

* postgres: use non-deprecated drizzle signatures

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* postgres: store metadata in the hook

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* core: do not rely on module cache for world config. instead, use a global and a symbol.

this makes sure that streamers can use in-memory event emitters and that it won't be compiled away into the different flow.js and step.js files.

this was figured out when i was adding a hooks tests to world-testing.

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* add postgres world to all workbench packages

we try to run them with the postgres world but it's not installed

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* Replace jsonb with cbor because zero byte does not work in jsonb :(

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* fix error handling: attempts start at 0 now, and not 1 like when we released. so initial attempt in postgres should reflect that.

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* drain stuff

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* fallback metadata to metadataJson

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* Make code more readable

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* apply Vade fix

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

---------

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

Author: Schniz

packages/core/src/define-hook.ts

@@ -3,6 +3,43 @@ import type { Hook as HookEntity } from '@workflow/world';
 import type { Hook, HookOptions } from './create-hook.js';
 import { resumeHook } from './runtime/resume-hook.js';
 
+/**
+ * A typed hook interface for type-safe hook creation and resumption.
+ */
+export interface TypedHook<TInput, TOutput> {
+  /**
+   * Creates a new hook with the defined output type.
+   *
+   * Note: This method is not available in runtime bundles. Use it from workflow contexts only.
+   *
+   * @param options - Optional hook configuration
+   * @returns A Hook that resolves to the defined output type
+   */
+  create(options?: HookOptions): Hook<TOutput>;
+  /**
+   * Resumes a hook by sending a payload with the defined input type.
+   * This is a type-safe wrapper around the `resumeHook` runtime function.
+   *
+   * @param token - The unique token identifying the hook
+   * @param payload - The payload to send; if a `schema` is configured it is validated/transformed before resuming
+   * @returns Promise resolving to the hook entity
+   * @throws Error if the hook is not found or if there's an error during the process
+   */
+  resume(token: string, payload: TInput): Promise<HookEntity>;
+}
+
+export namespace TypedHook {
+  /**
+   * Extracts the input type from a {@link TypedHook}
+   */
+  export type Input<T extends TypedHook<any, any>> = T extends TypedHook<
+    infer I,
+    any
+  >
+    ? I
+    : never;
+}
+
 /**
  * Defines a typed hook for type-safe hook creation and resumption.
  *
@@ -38,31 +75,13 @@ export function defineHook<TInput, TOutput = TInput>({
   schema,
 }: {
   schema?: StandardSchemaV1<TInput, TOutput>;
-} = {}) {
+} = {}): TypedHook<TInput, TOutput> {
   return {
-    /**
-     * Creates a new hook with the defined output type.
-     *
-     * Note: This method is not available in runtime bundles. Use it from workflow contexts only.
-     *
-     * @param _options - Optional hook configuration
-     * @returns A Hook that resolves to the defined output type
-     */
     create(_options?: HookOptions): Hook<TOutput> {
       throw new Error(
         '`defineHook().create()` can only be called inside a workflow function.'
       );
     },
-
-    /**
-     * Resumes a hook by sending a payload with the defined input type.
-     * This is a type-safe wrapper around the `resumeHook` runtime function.
-     *
-     * @param token - The unique token identifying the hook
-     * @param payload - The payload to send; if a `schema` is configured it is validated/transformed before resuming
-     * @returns Promise resolving to the hook entity
-     * @throws Error if the hook is not found or if there's an error during the process
-     */
     async resume(token: string, payload: TInput): Promise<HookEntity> {
       if (!schema?.['~standard']) {
         return await resumeHook(token, payload);

packages/core/src/index.ts

@@ -24,7 +24,8 @@ export {
   type Webhook,
   type WebhookOptions,
 } from './create-hook.js';
-export { defineHook } from './define-hook.js';
+export { defineHook, type TypedHook } from './define-hook.js';
+export { sleep } from './sleep.js';
 export {
   getStepMetadata,
   type StepMetadata,
@@ -33,7 +34,6 @@ export {
   getWorkflowMetadata,
   type WorkflowMetadata,
 } from './step/get-workflow-metadata.js';
-export { sleep } from './sleep.js';
 export {
   getWritable,
   type WorkflowWritableStreamOptions,

packages/core/src/runtime/world.ts

@@ -6,8 +6,13 @@ import { createVercelWorld } from '@workflow/world-vercel';
 
 const require = createRequire(join(process.cwd(), 'index.js'));
 
-let worldCache: World | undefined;
-let stubbedWorldCache: World | undefined;
+const WorldCache = Symbol.for('@workflow/world//cache');
+const StubbedWorldCache = Symbol.for('@workflow/world//stubbedCache');
+
+const globalSymbols: typeof globalThis & {
+  [WorldCache]?: World;
+  [StubbedWorldCache]?: World;
+} = globalThis;
 
 function defaultWorld(): 'vercel' | 'embedded' {
   if (process.env.VERCEL_DEPLOYMENT_ID) {
@@ -67,29 +72,29 @@ export const createWorld = (): World => {
  * be able to re-combine getWorld and getWorldHandlers into one singleton.
  */
 export const getWorldHandlers = (): Pick<World, 'createQueueHandler'> => {
-  if (stubbedWorldCache) {
-    return stubbedWorldCache;
+  if (globalSymbols[StubbedWorldCache]) {
+    return globalSymbols[StubbedWorldCache];
   }
   const _world = createWorld();
-  stubbedWorldCache = _world;
+  globalSymbols[StubbedWorldCache] = _world;
   return {
     createQueueHandler: _world.createQueueHandler,
   };
 };
 
 export const getWorld = (): World => {
-  if (worldCache) {
-    return worldCache;
+  if (globalSymbols[WorldCache]) {
+    return globalSymbols[WorldCache];
   }
-  worldCache = createWorld();
-  return worldCache;
+  globalSymbols[WorldCache] = createWorld();
+  return globalSymbols[WorldCache];
 };
 
 /**
  * Reset the cached world instance. This should be called when environment
  * variables change and you need to reinitialize the world with new config.
  */
 export const setWorld = (world: World | undefined): void => {
-  worldCache = world;
-  stubbedWorldCache = world;
+  globalSymbols[WorldCache] = world;
+  globalSymbols[StubbedWorldCache] = world;
 };

packages/world-postgres/package.json

@@ -49,6 +49,7 @@
     "@workflow/errors": "workspace:*",
     "@workflow/world": "workspace:*",
     "@workflow/world-local": "workspace:*",
+    "cbor-x": "1.6.0",
     "dotenv": "16.4.5",
     "drizzle-orm": "0.44.7",
     "pg-boss": "11.0.7",

packages/world-postgres/src/drizzle/cbor.ts

@@ -0,0 +1,22 @@
+import { decode, encode } from 'cbor-x';
+import { customType } from 'drizzle-orm/pg-core';
+
+export function Cbor<T>() {
+  return customType<{ data: T; driverData: Buffer }>({
+    dataType: () => 'bytea',
+    fromDriver: (value) => decode(value),
+    toDriver: (value) => encode(value),
+  });
+}
+
+/**
+ * Adds a `{key}Json` property to the given type V, representing a key that was
+ * migrated to CBOR and can contain a previous JSONB representation.
+ *
+ * We migrated from JSONB to CBOR, but to avoid breaking changes in the codebase,
+ * we keep both representations in the database, and therefore we need to extend
+ * the types accordingly.
+ */
+export type Cborized<V extends object, K extends keyof V> = V & {
+  [key in `${Extract<K, string>}Json`]: unknown;
+};

packages/world-postgres/src/drizzle/migrations/0001_tricky_sersi.sql

@@ -0,0 +1,11 @@
+ALTER TABLE "workflow"."workflow_runs" ALTER COLUMN "input" DROP NOT NULL;--> statement-breakpoint
+ALTER TABLE "workflow"."workflow_steps" ALTER COLUMN "input" DROP NOT NULL;--> statement-breakpoint
+ALTER TABLE "workflow"."workflow_events" ADD COLUMN "payload_cbor" "bytea";--> statement-breakpoint
+ALTER TABLE "workflow"."workflow_hooks" ADD COLUMN "metadata_cbor" "bytea";--> statement-breakpoint
+ALTER TABLE "workflow"."workflow_runs" ADD COLUMN "output_cbor" "bytea";--> statement-breakpoint
+ALTER TABLE "workflow"."workflow_runs" ADD COLUMN "execution_context_cbor" "bytea";--> statement-breakpoint
+ALTER TABLE "workflow"."workflow_runs" ADD COLUMN "input_cbor" "bytea";--> statement-breakpoint
+ALTER TABLE "workflow"."workflow_steps" ADD COLUMN "input_cbor" "bytea";--> statement-breakpoint
+ALTER TABLE "workflow"."workflow_steps" ADD COLUMN "output_cbor" "bytea";--> statement-breakpoint
+ALTER TABLE "workflow"."workflow_runs" DROP COLUMN "error_code";--> statement-breakpoint
+ALTER TABLE "workflow"."workflow_steps" DROP COLUMN "error_code";