diff --git a/packages/cli-v3/src/build/bundle.ts b/packages/cli-v3/src/build/bundle.ts index 8a01c8422..3dad16e92 100644 --- a/packages/cli-v3/src/build/bundle.ts +++ b/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) | 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"), }; } diff --git a/packages/cli-v3/src/build/packageModules.ts b/packages/cli-v3/src/build/packageModules.ts index 3580e5248..26596ffae 100644 --- a/packages/cli-v3/src/build/packageModules.ts +++ b/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")); } diff --git a/packages/cli-v3/src/dev/backgroundWorker.ts b/packages/cli-v3/src/dev/backgroundWorker.ts index c61a3911c..dec9f50f6 100644 --- a/packages/cli-v3/src/dev/backgroundWorker.ts +++ b/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 = new Evt(); private _onClose: Evt = new Evt(); public deprecated: boolean = false; public manifest: WorkerManifest | undefined; public serverWorker: ServerBackgroundWorker | undefined; - public stderr: Array = []; _taskRunProcesses: Map = new Map(); private _taskRunProcessesBeingKilled: Set = 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((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, diff --git a/packages/cli-v3/src/dev/devOutput.ts b/packages/cli-v3/src/dev/devOutput.ts index fd69b9b03..ebab3f566 100644 --- a/packages/cli-v3/src/dev/devOutput.ts +++ b/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( diff --git a/packages/cli-v3/src/dev/devSession.ts b/packages/cli-v3/src/dev/devSession.ts index e297e704f..e353168f0 100644 --- a/packages/cli-v3/src/dev/devSession.ts +++ b/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: {}, diff --git a/packages/cli-v3/src/dev/workerRuntime.ts b/packages/cli-v3/src/dev/workerRuntime.ts index ab2b896b0..e86a5fab3 100644 --- a/packages/cli-v3/src/dev/workerRuntime.ts +++ b/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; diff --git a/packages/cli-v3/src/entryPoints/dev.ts b/packages/cli-v3/src/entryPoints/dev.ts index e8b5a29e4..2cbce9b9f 100644 --- a/packages/cli-v3/src/entryPoints/dev.ts +++ b/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, diff --git a/packages/cli-v3/src/entryPoints/indexer.ts b/packages/cli-v3/src/entryPoints/indexer.ts new file mode 100644 index 000000000..f4c1a9117 --- /dev/null +++ b/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; +}); diff --git a/packages/core/src/v3/schemas/build.ts b/packages/core/src/v3/schemas/build.ts index 3f9d97816..dc8caba92 100644 --- a/packages/core/src/v3/schemas/build.ts +++ b/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(), diff --git a/packages/core/src/v3/schemas/messages.ts b/packages/core/src/v3/schemas/messages.ts index af0007b50..cfb103281 100644 --- a/packages/core/src/v3/schemas/messages.ts +++ b/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({ diff --git a/packages/core/src/v3/schemas/schemas.ts b/packages/core/src/v3/schemas/schemas.ts index e4d37a919..4ae7588d6 100644 --- a/packages/core/src/v3/schemas/schemas.ts +++ b/packages/core/src/v3/schemas/schemas.ts @@ -170,6 +170,7 @@ export type TaskFile = z.infer; const taskFileMetadata = { filePath: z.string(), exportName: z.string(), + entryPoint: z.string(), }; export const TaskFileMetadata = z.object(taskFileMetadata); diff --git a/packages/core/src/v3/zodMessageHandler.ts b/packages/core/src/v3/zodMessageHandler.ts index 7f0ac0c61..c380529fb 100644 --- a/packages/core/src/v3/zodMessageHandler.ts +++ b/packages/core/src/v3/zodMessageHandler.ts @@ -202,6 +202,34 @@ export class ZodMessageHandler } } +export function parseMessageFromCatalog( + message: unknown, + schema: TMessageCatalog +): MessageFromCatalog { + 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 = (message: { type: keyof TMessageCatalog; payload: z.infer; @@ -276,6 +304,27 @@ export class ZodMessageSender { } } +export async function sendMessageInCatalog( + catalog: TMessageCatalog, + type: keyof TMessageCatalog, + payload: z.input, + sender: ZodMessageSenderCallback +) { + 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 = { [K in keyof TCatalog]: (message: z.infer) => void; }; diff --git a/references/v3-catalog/src/trigger/simple.ts b/references/v3-catalog/src/trigger/simple.ts index 5e563acd8..118e33847 100644 --- a/references/v3-catalog/src/trigger/simple.ts +++ b/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 }); }