Move indexing to it’s own entry point, simplify code

packages/cli-v3/src/build/bundle.ts

@@ -10,6 +10,7 @@ import {
   devEntryPoints,
   isDeployEntryPoint,
   isDevEntryPoint,
+  isIndexerEntryPoint,
   isLoaderEntryPoint,
   shims,
 } from "./packageModules.js";
@@ -35,6 +36,7 @@ export type BundleResult = {
   configPath: string;
   loaderEntryPoint: string | undefined;
   workerEntryPoint: string | undefined;
+  indexerEntryPoint: string | undefined;
   stop: (() => Promise<void>) | undefined;
 };
 
@@ -144,6 +146,7 @@ export async function getBundleResultFromBuild(
   let configPath: string | undefined;
   let loaderEntryPoint: string | undefined;
   let workerEntryPoint: string | undefined;
+  let indexerEntryPoint: string | undefined;
 
   for (const [outputPath, outputMeta] of Object.entries(result.metafile.outputs)) {
     if (outputPath.endsWith(".mjs")) {
@@ -159,6 +162,8 @@ export async function getBundleResultFromBuild(
         loaderEntryPoint = $outputPath;
       } else if (isEntryPointForTarget(outputMeta.entryPoint, target)) {
         workerEntryPoint = $outputPath;
+      } else if (isIndexerEntryPoint(outputMeta.entryPoint)) {
+        indexerEntryPoint = $outputPath;
       } else {
         if (
           !outputMeta.entryPoint.startsWith("..") &&
@@ -182,6 +187,7 @@ export async function getBundleResultFromBuild(
     configPath: configPath,
     loaderEntryPoint,
     workerEntryPoint,
+    indexerEntryPoint,
     contentHash: hasher.digest("hex"),
   };
 }

packages/cli-v3/src/build/packageModules.ts

@@ -1,11 +1,12 @@
-import { join, relative } from "node:path";
+import { join } from "node:path";
 import { sourceDir } from "../sourceDir.js";
 
 export const devEntryPoint = join(sourceDir, "entryPoints", "dev.js");
 export const deployEntryPoint = join(sourceDir, "entryPoints", "deploy.js");
 export const telemetryEntryPoint = join(sourceDir, "entryPoints", "loader.js");
+export const indexerEntryPoint = join(sourceDir, "entryPoints", "indexer.js");
 
-export const devEntryPoints = [devEntryPoint, telemetryEntryPoint];
+export const devEntryPoints = [devEntryPoint, indexerEntryPoint, telemetryEntryPoint];
 
 export const deployEntryPoints = [devEntryPoint, deployEntryPoint, telemetryEntryPoint];
 
@@ -17,6 +18,10 @@ export function isDevEntryPoint(entryPoint: string) {
   return entryPoint.includes(join("dist", "esm", "entryPoints", "dev.js"));
 }
 
+export function isIndexerEntryPoint(entryPoint: string) {
+  return entryPoint.includes(join("dist", "esm", "entryPoints", "indexer.js"));
+}
+
 export function isDeployEntryPoint(entryPoint: string) {
   return entryPoint.includes(join("dist", "esm", "entryPoints", "deploy.js"));
 }

packages/cli-v3/src/dev/backgroundWorker.ts

@@ -13,9 +13,14 @@ import {
   childToWorkerMessages,
   correctErrorStackTrace,
   formatDurationMilliseconds,
+  indexerToWorkerMessages,
   workerToChildMessages,
 } from "@trigger.dev/core/v3";
-import { ZodMessageHandler, ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
+import {
+  parseMessageFromCatalog,
+  ZodMessageHandler,
+  ZodMessageSender,
+} from "@trigger.dev/core/v3/zodMessageHandler";
 import { Evt } from "evt";
 import { ChildProcess, fork } from "node:child_process";
 import {
@@ -41,6 +46,8 @@ import {
   UnexpectedExitError,
   getFriendlyErrorMessage,
 } from "./errors.js";
+import { writeJSONFile } from "../utilities/fileSystem.js";
+import { join } from "node:path";
 
 export type CurrentWorkers = BackgroundWorkerCoordinator["currentWorkers"];
 export class BackgroundWorkerCoordinator {
@@ -208,18 +215,12 @@ export type BackgroundWorkerOptions = {
 };
 
 export class BackgroundWorker {
-  private _initialized: boolean = false;
-  private _handler = new ZodMessageHandler({
-    schema: childToWorkerMessages,
-  });
-
   public onTaskRunHeartbeat: Evt<string> = new Evt();
   private _onClose: Evt<void> = new Evt();
 
   public deprecated: boolean = false;
   public manifest: WorkerManifest | undefined;
   public serverWorker: ServerBackgroundWorker | undefined;
-  public stderr: Array<string> = [];
 
   _taskRunProcesses: Map<string, TaskRunProcess> = new Map();
   private _taskRunProcessesBeingKilled: Set<number> = new Set();
@@ -258,19 +259,27 @@ export class BackgroundWorker {
   }
 
   async initialize() {
-    if (this._initialized) {
+    if (this.manifest) {
       throw new Error("Worker already initialized");
     }
 
     let resolved = false;
 
+    const buildManifestPath = join(this.build.outputPath, "build.json");
+
+    // Write the build manifest to this.build.outputPath/build.json
+    await writeJSONFile(buildManifestPath, this.build, true);
+
     logger.debug("Initializing worker", { build: this.build, params: this.params });
 
     this.manifest = await new Promise<WorkerManifest>((resolve, reject) => {
-      const child = fork(this.build.workerEntryPoint, {
+      const child = fork(this.build.indexerEntryPoint, {
         stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"],
         cwd: this.params.cwd,
-        env: this.params.env,
+        env: {
+          ...this.params.env,
+          TRIGGER_BUILD_MANIFEST_PATH: buildManifestPath,
+        },
       });
 
       // Set a timeout to kill the child process if it doesn't respond
@@ -285,35 +294,30 @@ export class BackgroundWorker {
       }, 20_000);
 
       child.on("message", async (msg: any) => {
-        const message = this._handler.parseMessage(msg);
+        const message = parseMessageFromCatalog(msg, indexerToWorkerMessages);
 
-        if (!message.success) {
-          clearTimeout(timeout);
-          resolved = true;
-          reject(new Error(`Failed to parse message: ${message.error}`));
-          child.kill();
-          return;
-        }
-
-        if (message.data.type === "INDEX_COMPLETE" && !resolved) {
-          clearTimeout(timeout);
-          resolved = true;
-          resolve(message.data.payload.manifest);
-          child.kill();
-        } else if (message.data.type === "UNCAUGHT_EXCEPTION") {
-          clearTimeout(timeout);
-          resolved = true;
-          reject(
-            new UncaughtExceptionError(message.data.payload.error, message.data.payload.origin)
-          );
-          child.kill();
-        } else if (message.data.type === "TASKS_FAILED_TO_PARSE") {
-          clearTimeout(timeout);
-          resolved = true;
-          reject(
-            new TaskMetadataParseError(message.data.payload.zodIssues, message.data.payload.tasks)
-          );
-          child.kill();
+        switch (message.type) {
+          case "INDEX_COMPLETE": {
+            clearTimeout(timeout);
+            resolved = true;
+            resolve(message.payload.manifest);
+            child.kill();
+            break;
+          }
+          case "TASKS_FAILED_TO_PARSE": {
+            clearTimeout(timeout);
+            resolved = true;
+            reject(new TaskMetadataParseError(message.payload.zodIssues, message.payload.tasks));
+            child.kill();
+            break;
+          }
+          case "UNCAUGHT_EXCEPTION": {
+            clearTimeout(timeout);
+            resolved = true;
+            reject(new UncaughtExceptionError(message.payload.error, message.payload.origin));
+            child.kill();
+            break;
+          }
         }
       });
 
@@ -326,35 +330,20 @@ export class BackgroundWorker {
       });
 
       child.stdout?.on("data", (data) => {
-        logger.debug(data.toString());
+        logger.debug(`indexer: ${data.toString()}`);
       });
 
       child.stderr?.on("data", (data) => {
-        logger.debug(data.toString());
-      });
-
-      const sender = new ZodMessageSender({
-        schema: workerToChildMessages,
-        sender: async (message) => {
-          if (child.connected && !resolved) {
-            child.send(message);
-          }
-        },
-      });
-
-      sender.send("INDEX", { build: this.build }).catch((err) => {
-        if (err instanceof Error) {
-          clearTimeout(timeout);
-          resolved = true;
-          reject(err);
-          child.kill();
-        }
+        logger.debug(`indexer: ${data.toString()}`);
       });
     });
 
-    logger.debug("Worker initialized", { manifest: this.manifest });
+    const indexManifestPath = join(this.build.outputPath, "index.json");
 
-    this._initialized = true;
+    // Write the build manifest to this.build.outputPath/worker.json
+    await writeJSONFile(indexManifestPath, this.manifest, true);
+
+    logger.debug("Worker initialized", { index: indexManifestPath });
   }
 
   // We need to notify all the task run processes that a task run has completed,

packages/cli-v3/src/dev/devOutput.ts

@@ -39,7 +39,7 @@ export function startDevOutput(options: DevOutputOptions) {
     const testLink = chalkLink(cliLink("Test tasks", testUrl));
     const runsLink = chalkLink(cliLink("View runs", runsUrl));
 
-    const workerStarted = chalkGrey("Background worker started");
+    const workerStarted = chalkGrey("Background worker ready");
     const workerVersion = chalkWorker(worker.serverWorker!.version);
 
     logger.log(

packages/cli-v3/src/dev/devSession.ts

@@ -16,15 +16,14 @@ import {
   resolvePluginsForContext,
 } from "../build/extensions.js";
 import { createExternalsBuildExtension } from "../build/externals.js";
-import { devEntryPoint, telemetryEntryPoint } from "../build/packageModules.js";
+import { copyManifestToDir } from "../build/manifests.js";
+import { devEntryPoint, indexerEntryPoint, telemetryEntryPoint } from "../build/packageModules.js";
 import { type DevCommandOptions } from "../commands/dev.js";
+import { eventBus } from "../utilities/eventBus.js";
 import { logger } from "../utilities/logger.js";
 import { EphemeralDirectory, getTmpDir } from "../utilities/tempDirectories.js";
-import { copyManifestToDir } from "../build/manifests.js";
-import { startWorkerRuntime } from "./workerRuntime.js";
-import { chalkGrey } from "../utilities/cliOutput.js";
-import { eventBus } from "../utilities/eventBus.js";
 import { startDevOutput } from "./devOutput.js";
+import { startWorkerRuntime } from "./workerRuntime.js";
 
 export type DevSessionOptions = {
   name: string | undefined;
@@ -183,6 +182,7 @@ async function createBuildManifestFromBundle(
     outputPath: destination,
     workerEntryPoint: bundle.workerEntryPoint ?? devEntryPoint,
     loaderEntryPoint: bundle.loaderEntryPoint ?? telemetryEntryPoint,
+    indexerEntryPoint: bundle.indexerEntryPoint ?? indexerEntryPoint,
     configPath: bundle.configPath,
     deploy: {
       env: {},

packages/cli-v3/src/dev/workerRuntime.ts

@@ -9,22 +9,22 @@ import {
   WorkerManifest,
 } from "@trigger.dev/core/v3";
 import { ResolvedConfig } from "@trigger.dev/core/v3/build";
-import { ClientRequestArgs } from "node:http";
-import { WebSocket } from "partysocket";
-import { ClientOptions, WebSocket as wsWebSocket } from "ws";
-import { CliApiClient } from "../apiClient.js";
-import { DevCommandOptions } from "../commands/dev.js";
-import { chalkError, chalkGrey, chalkTask } from "../utilities/cliOutput.js";
-import { logger } from "../utilities/logger.js";
-import { BackgroundWorker, BackgroundWorkerCoordinator } from "./backgroundWorker.js";
 import {
   MessagePayloadFromSchema,
   ZodMessageHandler,
   ZodMessageSender,
 } from "@trigger.dev/core/v3/zodMessageHandler";
+import { ClientRequestArgs } from "node:http";
+import { WebSocket } from "partysocket";
+import { ClientOptions, WebSocket as wsWebSocket } from "ws";
+import { CliApiClient } from "../apiClient.js";
+import { DevCommandOptions } from "../commands/dev.js";
+import { chalkError, chalkTask } from "../utilities/cliOutput.js";
 import { resolveDotEnvVars } from "../utilities/dotEnv.js";
-import { VERSION } from "../version.js";
 import { eventBus } from "../utilities/eventBus.js";
+import { logger } from "../utilities/logger.js";
+import { VERSION } from "../version.js";
+import { BackgroundWorker, BackgroundWorkerCoordinator } from "./backgroundWorker.js";
 
 export interface WorkerRuntime {
   shutdown(): Promise<void>;

packages/cli-v3/src/entryPoints/dev.ts

@@ -1,6 +1,7 @@
 import type { Tracer } from "@opentelemetry/api";
 import type { Logger } from "@opentelemetry/api-logs";
 import {
+  BuildManifest,
   childToWorkerMessages,
   clock,
   type HandleErrorFunction,
@@ -12,6 +13,7 @@ import {
   TaskRunExecution,
   TriggerConfig,
   TriggerTracer,
+  WorkerManifest,
   workerToChildMessages,
 } from "@trigger.dev/core/v3";
 import { DevRuntimeManager } from "@trigger.dev/core/v3/dev";
@@ -28,11 +30,8 @@ import {
   TracingSDK,
   usage,
 } from "@trigger.dev/core/v3/workers";
-import {
-  ZodMessageHandler,
-  ZodMessageSender,
-  ZodSchemaParsedError,
-} from "@trigger.dev/core/v3/zodMessageHandler";
+import { ZodMessageHandler, ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
+import { readFile } from "node:fs/promises";
 import sourceMapSupport from "source-map-support";
 import { VERSION } from "../version.js";
 
@@ -71,8 +70,25 @@ async function importConfig(
   };
 }
 
-async function bootstrap(configPath: string) {
-  const { config, handleError } = await importConfig(configPath);
+async function loadBuildManifest() {
+  const manifestContents = await readFile(process.env.TRIGGER_BUILD_MANIFEST_PATH!, "utf-8");
+  const raw = JSON.parse(manifestContents);
+
+  return BuildManifest.parse(raw);
+}
+
+async function loadWorkerManifest() {
+  const manifestContents = await readFile(process.env.TRIGGER_WORKER_MANIFEST_PATH!, "utf-8");
+  const raw = JSON.parse(manifestContents);
+
+  return WorkerManifest.parse(raw);
+}
+
+async function bootstrap() {
+  const buildManifest = await loadBuildManifest();
+  const workerManifest = await loadWorkerManifest();
+
+  const { config, handleError } = await importConfig(process.env.TRIGGER_BUILD_MANIFEST_PATH!);
 
   const tracingSDK = new TracingSDK({
     url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
@@ -100,7 +116,14 @@ async function bootstrap(configPath: string) {
 
   logger.setGlobalTaskLogger(otelTaskLogger);
 
-  return { tracer, tracingSDK, consoleInterceptor, config, handleErrorFn: handleError };
+  return {
+    tracer,
+    tracingSDK,
+    consoleInterceptor,
+    config,
+    handleErrorFn: handleError,
+    workerManifest,
+  };
 }
 
 async function registerTaskFileMetadata(files: Array<{ entry: string; out: string }>) {
@@ -135,23 +158,6 @@ let _tracingSDK: TracingSDK | undefined;
 const handler = new ZodMessageHandler({
   schema: workerToChildMessages,
   messages: {
-    INDEX: async ({ build }) => {
-      await bootstrap(build.configPath);
-      await registerTaskFileMetadata(build.files);
-
-      const tasks = taskCatalog.listTaskManifests();
-
-      await sender.send("INDEX_COMPLETE", { manifest: { tasks } }).catch((err) => {
-        if (err instanceof ZodSchemaParsedError) {
-          sender.send("TASKS_FAILED_TO_PARSE", {
-            zodIssues: err.error.issues,
-            tasks,
-          });
-        } else {
-          console.error("Failed to send TASKS_READY message", err);
-        }
-      });
-    },
     EXECUTE_TASK_RUN: async ({ execution, traceContext, metadata }) => {
       if (_isRunning) {
         console.error("Worker is already running a task");
@@ -199,9 +205,8 @@ const handler = new ZodMessageHandler({
         return;
       }
 
-      const { tracer, tracingSDK, consoleInterceptor, config, handleErrorFn } = await bootstrap(
-        "./trigger.config.js"
-      );
+      const { tracer, tracingSDK, consoleInterceptor, config, handleErrorFn, workerManifest } =
+        await bootstrap();
 
       const executor = new TaskExecutor(task, {
         tracer,

packages/cli-v3/src/entryPoints/indexer.ts

@@ -0,0 +1,144 @@
+import {
+  BuildManifest,
+  type HandleErrorFunction,
+  indexerToWorkerMessages,
+  taskCatalog,
+  TriggerConfig,
+} from "@trigger.dev/core/v3";
+import {
+  StandardTaskCatalog,
+  TracingDiagnosticLogLevel,
+  TracingSDK,
+} from "@trigger.dev/core/v3/workers";
+import { sendMessageInCatalog, ZodSchemaParsedError } from "@trigger.dev/core/v3/zodMessageHandler";
+import { readFile } from "node:fs/promises";
+import sourceMapSupport from "source-map-support";
+
+sourceMapSupport.install({
+  handleUncaughtExceptions: false,
+  environment: "node",
+  hookRequire: false,
+});
+
+process.on("uncaughtException", function (error, origin) {
+  if (error instanceof Error) {
+    process.send &&
+      process.send({
+        type: "UNCAUGHT_EXCEPTION",
+        payload: {
+          error: { name: error.name, message: error.message, stack: error.stack },
+          origin,
+        },
+        version: "v1",
+      });
+  } else {
+    process.send &&
+      process.send({
+        type: "UNCAUGHT_EXCEPTION",
+        payload: {
+          error: {
+            name: "Error",
+            message: typeof error === "string" ? error : JSON.stringify(error),
+          },
+          origin,
+        },
+        version: "v1",
+      });
+  }
+});
+
+taskCatalog.setGlobalTaskCatalog(new StandardTaskCatalog());
+
+async function importConfig(
+  configPath: string
+): Promise<{ config: TriggerConfig; handleError?: HandleErrorFunction }> {
+  const configModule = await import(configPath);
+
+  const config = configModule?.default ?? configModule?.config;
+
+  return {
+    config,
+    handleError: configModule?.handleError,
+  };
+}
+
+async function loadBuildManifest() {
+  const manifestContents = await readFile(process.env.TRIGGER_BUILD_MANIFEST_PATH!, "utf-8");
+  const raw = JSON.parse(manifestContents);
+
+  return BuildManifest.parse(raw);
+}
+
+async function bootstrap() {
+  const buildManifest = await loadBuildManifest();
+
+  const { config } = await importConfig(buildManifest.configPath);
+
+  // This needs to run or the PrismaInstrumentation will throw an error
+  const tracingSDK = new TracingSDK({
+    url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
+    instrumentations: config.instrumentations ?? [],
+    diagLogLevel: (process.env.OTEL_LOG_LEVEL as TracingDiagnosticLogLevel) ?? "none",
+    forceFlushTimeoutMillis: 30_000,
+  });
+
+  for (const file of buildManifest.files) {
+    const module = await import(file.out);
+
+    for (const exportName of Object.keys(module)) {
+      const task = module[exportName];
+
+      if (!task) {
+        continue;
+      }
+
+      if (task[Symbol.for("trigger.dev/task")]) {
+        if (taskCatalog.taskExists(task.id)) {
+          taskCatalog.registerTaskFileMetadata(task.id, {
+            exportName,
+            filePath: file.entry,
+            entryPoint: file.out,
+          });
+        }
+      }
+    }
+  }
+
+  return {
+    tracingSDK,
+    config,
+    buildManifest,
+  };
+}
+
+await bootstrap();
+
+const tasks = taskCatalog.listTaskManifests();
+
+await sendMessageInCatalog(
+  indexerToWorkerMessages,
+  "INDEX_COMPLETE",
+  {
+    manifest: {
+      tasks,
+    },
+  },
+  async (msg) => {
+    process.send?.(msg);
+  }
+).catch((err) => {
+  if (err instanceof ZodSchemaParsedError) {
+    return sendMessageInCatalog(
+      indexerToWorkerMessages,
+      "TASKS_FAILED_TO_PARSE",
+      { zodIssues: err.error.issues, tasks },
+      async (msg) => {
+        process.send?.(msg);
+      }
+    );
+  } else {
+    console.error("Failed to send TASKS_READY message", err);
+  }
+
+  return;
+});

packages/core/src/v3/schemas/build.ts

@@ -24,6 +24,7 @@ export const BuildManifest = z.object({
   config: ConfigManifest,
   files: z.array(TaskFile),
   outputPath: z.string(),
+  indexerEntryPoint: z.string(),
   workerEntryPoint: z.string(),
   loaderEntryPoint: z.string().optional(),
   configPath: z.string(),

packages/core/src/v3/schemas/messages.ts

@@ -109,9 +109,7 @@ export const clientWebsocketMessages = {
 };
 
 export const workerToChildMessages = {
-  INDEX: z.object({
-    build: BuildManifest,
-  }),
+  INDEX: z.object({}),
   EXECUTE_TASK_RUN: z.object({
     version: z.literal("v1").default("v1"),
     execution: TaskRunExecution,
@@ -160,11 +158,6 @@ export const childToWorkerMessages = {
     execution: TaskRunExecution,
     result: TaskRunExecutionResult,
   }),
-  INDEX_COMPLETE: z.object({
-    version: z.literal("v1").default("v1"),
-    manifest: WorkerManifest,
-  }),
-  TASKS_FAILED_TO_PARSE: TaskMetadataFailedToParseData,
   TASK_HEARTBEAT: z.object({
     version: z.literal("v1").default("v1"),
     id: z.string(),
@@ -190,6 +183,15 @@ export const childToWorkerMessages = {
   UNCAUGHT_EXCEPTION: UncaughtExceptionMessage,
 };
 
+export const indexerToWorkerMessages = {
+  INDEX_COMPLETE: z.object({
+    version: z.literal("v1").default("v1"),
+    manifest: WorkerManifest,
+  }),
+  TASKS_FAILED_TO_PARSE: TaskMetadataFailedToParseData,
+  UNCAUGHT_EXCEPTION: UncaughtExceptionMessage,
+};
+
 export const ProdChildToWorkerMessages = {
   TASK_RUN_COMPLETED: {
     message: z.object({

packages/core/src/v3/schemas/schemas.ts

@@ -170,6 +170,7 @@ export type TaskFile = z.infer<typeof TaskFile>;
 const taskFileMetadata = {
   filePath: z.string(),
   exportName: z.string(),
+  entryPoint: z.string(),
 };
 
 export const TaskFileMetadata = z.object(taskFileMetadata);

packages/core/src/v3/zodMessageHandler.ts

@@ -202,6 +202,34 @@ export class ZodMessageHandler<TMessageCatalog extends ZodMessageCatalogSchema>
   }
 }
 
+export function parseMessageFromCatalog<TMessageCatalog extends ZodMessageCatalogSchema>(
+  message: unknown,
+  schema: TMessageCatalog
+): MessageFromCatalog<TMessageCatalog> {
+  const parsedMessage = ZodMessageSchema.safeParse(message);
+
+  if (!parsedMessage.success) {
+    throw new Error(`Failed to parse message: ${JSON.stringify(parsedMessage.error)}`);
+  }
+
+  const messageSchema = schema[parsedMessage.data.type];
+
+  if (!messageSchema) {
+    throw new Error(`Unknown message type: ${parsedMessage.data.type}`);
+  }
+
+  const parsedPayload = messageSchema.safeParse(parsedMessage.data.payload);
+
+  if (!parsedPayload.success) {
+    throw new Error(`Failed to parse message payload: ${JSON.stringify(parsedPayload.error)}`);
+  }
+
+  return {
+    type: parsedMessage.data.type,
+    payload: parsedPayload.data,
+  };
+}
+
 type ZodMessageSenderCallback<TMessageCatalog extends ZodMessageCatalogSchema> = (message: {
   type: keyof TMessageCatalog;
   payload: z.infer<TMessageCatalog[keyof TMessageCatalog]>;
@@ -276,6 +304,27 @@ export class ZodMessageSender<TMessageCatalog extends ZodMessageCatalogSchema> {
   }
 }
 
+export async function sendMessageInCatalog<TMessageCatalog extends ZodMessageCatalogSchema>(
+  catalog: TMessageCatalog,
+  type: keyof TMessageCatalog,
+  payload: z.input<TMessageCatalog[keyof TMessageCatalog]>,
+  sender: ZodMessageSenderCallback<TMessageCatalog>
+) {
+  const schema = catalog[type];
+
+  if (!schema) {
+    throw new Error(`Unknown message type: ${type as string}`);
+  }
+
+  const parsedPayload = schema.safeParse(payload);
+
+  if (!parsedPayload.success) {
+    throw new ZodSchemaParsedError(parsedPayload.error, payload);
+  }
+
+  await sender({ type, payload, version: "v1" });
+}
+
 export type MessageCatalogToSocketIoEvents<TCatalog extends ZodMessageCatalogSchema> = {
   [K in keyof TCatalog]: (message: z.infer<TCatalog[K]>) => void;
 };

references/v3-catalog/src/trigger/simple.ts

@@ -21,7 +21,7 @@ export const anyPayloadTask = task({
     });
 
     if (result.ok) {
-      logger.info("Result from fetch-post-task 211111s", { output: result.output });
+      logger.info("Result from fetch-post-task 211111sss", { output: result.output });
     } else {
       logger.error("Error from fetch-post-task", { error: result.error });
     }