|
6 | 6 | type TaskRunTemplate, |
7 | 7 | PrismaClientOrTransaction, |
8 | 8 | } from "@trigger.dev/database"; |
| 9 | +import { inferSchema } from "@jsonhero/schema-infer"; |
9 | 10 | import parse from "parse-duration"; |
10 | 11 | import { type PrismaClient } from "~/db.server"; |
11 | 12 | import { RunsRepository } from "~/services/runsRepository/runsRepository.server"; |
@@ -34,6 +35,8 @@ type Task = { |
34 | 35 | taskIdentifier: string; |
35 | 36 | filePath: string; |
36 | 37 | friendlyId: string; |
| 38 | + payloadSchema?: unknown; |
| 39 | + inferredPayloadSchema?: unknown; |
37 | 40 | }; |
38 | 41 |
|
39 | 42 | type Queue = { |
@@ -244,11 +247,30 @@ export class TestTaskPresenter { |
244 | 247 | }, |
245 | 248 | }); |
246 | 249 |
|
| 250 | + // Infer schema from existing run payloads when no explicit schema is defined |
| 251 | + let inferredPayloadSchema: unknown | undefined; |
| 252 | + if (!task.payloadSchema && latestRuns.length > 0 && task.triggerSource === "STANDARD") { |
| 253 | + let inference: ReturnType<typeof inferSchema> | undefined; |
| 254 | + for (const run of latestRuns) { |
| 255 | + try { |
| 256 | + const parsed = await parsePacket({ data: run.payload, dataType: run.payloadType }); |
| 257 | + inference = inferSchema(parsed, inference); |
| 258 | + } catch { |
| 259 | + // Skip malformed runs — inference is best-effort |
| 260 | + } |
| 261 | + } |
| 262 | + if (inference) { |
| 263 | + inferredPayloadSchema = inference.toJSONSchema(); |
| 264 | + } |
| 265 | + } |
| 266 | + |
247 | 267 | const taskWithEnvironment = { |
248 | 268 | id: task.id, |
249 | 269 | taskIdentifier: task.slug, |
250 | 270 | filePath: task.filePath, |
251 | 271 | friendlyId: task.friendlyId, |
| 272 | + payloadSchema: task.payloadSchema ?? undefined, |
| 273 | + inferredPayloadSchema, |
252 | 274 | }; |
253 | 275 |
|
254 | 276 | switch (task.triggerSource) { |
|
0 commit comments