From b0d44ee4d6abd6e92c845c96e83d9383e33f0bb3 Mon Sep 17 00:00:00 2001 From: Rickylabs Date: Thu, 2 Jul 2026 22:38:09 +0200 Subject: [PATCH] fix(plugin-workers): derive trigger id from {id} path, fail loudly on missing id (triggerJob + triggerTask) (#279) Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_012wKHquACkXnWPDgJYhhFjN --- packages/contracts/src/domain/errors.ts | 43 +++++++++- packages/contracts/src/public/mod.ts | 8 +- packages/contracts/tests/errors_test.ts | 39 ++++++++- .../contracts/v1/workers.contract-schemas.ts | 14 ++- .../contracts/v1/workers.contract-types.ts | 6 +- .../integration/workers-trigger-rpc_test.ts | 85 +++++++++++++++++++ plugins/workers/services/src/routers/jobs.ts | 23 +++-- plugins/workers/services/src/routers/tasks.ts | 23 +++-- .../services/src/trigger-path-id_test.ts | 79 +++++++++++++++++ 9 files changed, 302 insertions(+), 18 deletions(-) create mode 100644 packages/sdk/tests/integration/workers-trigger-rpc_test.ts create mode 100644 plugins/workers/services/src/trigger-path-id_test.ts diff --git a/packages/contracts/src/domain/errors.ts b/packages/contracts/src/domain/errors.ts index 20b21f1a2..4ef02261a 100644 --- a/packages/contracts/src/domain/errors.ts +++ b/packages/contracts/src/domain/errors.ts @@ -1,4 +1,4 @@ -import type { NotFoundError } from './schemas.ts'; +import type { NotFoundError, ValidationError } from './schemas.ts'; const DEFAULT_RESOURCE_TYPE = 'resource'; const VERSION_SEGMENT_PATTERN = /^v\d+$/i; @@ -54,3 +54,44 @@ export function notFound(options: NotFoundOptions): never { function capitalize(value: string): string { return value.charAt(0).toUpperCase() + value.slice(1); } + +/** Options for throwing a contract `VALIDATION_ERROR` oRPC error. */ +export type ValidationFailedOptions = Readonly<{ + /** oRPC errors object from a handler context. */ + errors: unknown; + /** Human-readable summary of what failed validation. */ + message: string; + /** Field-level validation messages keyed by field name. */ + fieldErrors?: Record; + /** Form-level (non-field) validation messages. */ + formErrors?: string[]; +}>; + +type ValidationErrorFactory = (options: { + message: string; + data: ValidationError; +}) => unknown; + +type ValidationErrorContainer = Readonly<{ + VALIDATION_ERROR: ValidationErrorFactory; +}>; + +/** + * Throws the contract `VALIDATION_ERROR` oRPC error. + * + * Mirrors {@link notFound}: it accepts the handler's `errors` object as + * `unknown` and narrows it to the shared error vocabulary via the single + * sanctioned centralized-contract boundary cast, so callers fail loudly with a + * typed 422 envelope instead of proceeding with invalid input. + */ +export function validationFailed(options: ValidationFailedOptions): never { + const constructors = options.errors as ValidationErrorContainer; + + throw constructors.VALIDATION_ERROR({ + message: options.message, + data: { + formErrors: options.formErrors ?? [], + fieldErrors: options.fieldErrors ?? {}, + }, + }); +} diff --git a/packages/contracts/src/public/mod.ts b/packages/contracts/src/public/mod.ts index d41b23919..7eef9ca96 100644 --- a/packages/contracts/src/public/mod.ts +++ b/packages/contracts/src/public/mod.ts @@ -28,7 +28,13 @@ export { DEFAULT_PAGINATION_LIMIT_MAX, DEFAULT_PAGINATION_OFFSET, } from '../domain/constants.ts'; -export { getResourceType, notFound, type NotFoundOptions } from '../domain/errors.ts'; +export { + getResourceType, + notFound, + type NotFoundOptions, + validationFailed, + type ValidationFailedOptions, +} from '../domain/errors.ts'; export type { ErrorResult, OkResult, Result } from '../domain/result.ts'; export type { ContractDefaultableSchema, diff --git a/packages/contracts/tests/errors_test.ts b/packages/contracts/tests/errors_test.ts index 9b608c896..aeb608494 100644 --- a/packages/contracts/tests/errors_test.ts +++ b/packages/contracts/tests/errors_test.ts @@ -1,4 +1,4 @@ -import { getResourceType } from '../src/domain/errors.ts'; +import { getResourceType, validationFailed } from '../src/domain/errors.ts'; function assertEquals(actual: string, expected: string): void { if (actual !== expected) { @@ -17,3 +17,40 @@ Deno.test('getResourceType skips version segments and singularizes resources', ( assertEquals(getResourceType({ path: ['v2', 'sagas', 'getById'] }), 'saga'); assertEquals(getResourceType({ path: ['jobs'] }), 'job'); }); + +Deno.test('validationFailed throws a VALIDATION_ERROR envelope via the errors object', () => { + const calls: Array<{ message: string; data: unknown }> = []; + const errors = { + VALIDATION_ERROR(options: { message: string; data: unknown }): Error { + calls.push(options); + return new Error(options.message); + }, + }; + + let thrown: unknown; + try { + validationFailed({ + errors, + message: 'Job id is required in the {id} path segment.', + fieldErrors: { id: ['Job id is required in the {id} path segment.'] }, + }); + } catch (error) { + thrown = error; + } + + if (!(thrown instanceof Error)) { + throw new Error('validationFailed must throw the constructed oRPC error'); + } + assertEquals(thrown.message, 'Job id is required in the {id} path segment.'); + if (calls.length !== 1) { + throw new Error(`Expected 1 VALIDATION_ERROR call, received ${calls.length}`); + } + assertEquals(calls[0].message, 'Job id is required in the {id} path segment.'); + assertEquals( + JSON.stringify(calls[0].data), + JSON.stringify({ + formErrors: [], + fieldErrors: { id: ['Job id is required in the {id} path segment.'] }, + }), + ); +}); diff --git a/packages/plugin-workers-core/src/contracts/v1/workers.contract-schemas.ts b/packages/plugin-workers-core/src/contracts/v1/workers.contract-schemas.ts index 36e20b7ae..e0dec546d 100644 --- a/packages/plugin-workers-core/src/contracts/v1/workers.contract-schemas.ts +++ b/packages/plugin-workers-core/src/contracts/v1/workers.contract-schemas.ts @@ -205,7 +205,7 @@ type JobTriggerInputShape = 'delay' | 'payload' | 'priority' | 'traceparent' | 'tracestate' > & { - id: z.ZodString; + id: z.ZodOptional; correlationId: z.ZodOptional; }; @@ -217,7 +217,11 @@ const JobTriggerInputShape: JobTriggerInputShape = { traceparent: true, tracestate: true, }).shape, - id: z.string(), + // The `{id}` path segment is the single source of truth for the target job: + // oRPC merges the OpenAPI path param into `input.id`, so the body value is only + // an optional fallback for RPC transports. The handler fails loudly when no id + // resolves rather than persisting an `undefined` KV key. + id: z.string().optional().describe('Job id (resolved from the {id} path segment)'), correlationId: z.string().optional().describe('Correlation ID for tracing'), }; @@ -232,7 +236,11 @@ export const JobTriggerInputSchema: ContractSchema = JobTriggerInputZodSchema as unknown as ContractSchema; export const TaskTriggerInputZodSchema: z.ZodType = z.object({ - id: z.string(), + // The `{id}` path segment is the single source of truth for the target task: + // oRPC merges the OpenAPI path param into `input.id`, so the body value is only + // an optional fallback for RPC transports. The handler fails loudly when no id + // resolves rather than persisting an `undefined` KV key. + id: z.string().optional().describe('Task id (resolved from the {id} path segment)'), payload: z.record(z.string(), z.unknown()).optional().describe('Task payload'), priority: z.number().int().min(0).max(100).default(50).optional(), delay: z.number().int().nonnegative().optional().describe('Delay in ms'), diff --git a/packages/plugin-workers-core/src/contracts/v1/workers.contract-types.ts b/packages/plugin-workers-core/src/contracts/v1/workers.contract-types.ts index 3909413bb..dc3e82756 100644 --- a/packages/plugin-workers-core/src/contracts/v1/workers.contract-types.ts +++ b/packages/plugin-workers-core/src/contracts/v1/workers.contract-types.ts @@ -31,7 +31,8 @@ export type SSEEvent = Readonly>; /** Input accepted by the trigger-job procedure. */ export type JobTriggerInput = Readonly<{ - id: string; + /** Job id, resolved from the `{id}` path segment; optional in the body. */ + id?: string; payload?: Record; priority?: number; delay?: number; @@ -45,7 +46,8 @@ export type JobTriggerOutput = Readonly<{ jobId: string; triggered: boolean }>; /** Input accepted by the trigger-task procedure. */ export type TaskTriggerInput = Readonly<{ - id: string; + /** Task id, resolved from the `{id}` path segment; optional in the body. */ + id?: string; payload?: Record; priority?: number; delay?: number; diff --git a/packages/sdk/tests/integration/workers-trigger-rpc_test.ts b/packages/sdk/tests/integration/workers-trigger-rpc_test.ts new file mode 100644 index 000000000..43c33dc7e --- /dev/null +++ b/packages/sdk/tests/integration/workers-trigger-rpc_test.ts @@ -0,0 +1,85 @@ +import { assertEquals, assertRejects } from '@std/assert'; +import { ORPCError } from '@orpc/contract'; +import { os } from '@orpc/server'; +import { createService } from '../../../service/mod.ts'; +import { createServiceClient } from '../../src/client/service-client.ts'; +import { createServerServiceEnvKey } from '../../src/discovery/service-url.ts'; + +// Gap 1 regression (issue #279): the eis-chat consumer saw a live 404 on +// `GET /api/rpc/v1/workers/triggerJob`, i.e. the type-safe RPC endpoint appeared +// unmounted. This proves that a `createServiceClient({ serviceName, routerName })` +// call over the RPC transport reaches a plugin-workers-style `triggerJob` route +// (route is mounted and routed), and that a request that resolves no task id +// fails loudly with a `VALIDATION_ERROR` envelope (Gap 2 behavior over RPC) +// rather than 404-ing or silently succeeding with an `undefined` id. + +const SERVICE_NAME = 'workers'; +const ROUTER_NAME = 'workers'; +const RPC_PATH = `/api/rpc/v1/${ROUTER_NAME}`; + +interface TriggerJobInput { + readonly id?: string; +} + +// A plugin-workers-style router: `triggerJob` mirrors the real contract route +// (`POST /jobs/{id}/trigger`) with an optional body id, failing loudly when no +// id resolves — exactly the shape the #279 fix gives the real handler. +function createWorkersStyleRouter() { + return { + triggerJob: os.route({ method: 'POST', path: '/jobs/{id}/trigger' }).handler( + ({ input }: { input: unknown }) => { + const id = (input as TriggerJobInput).id; + if (!id) { + throw new ORPCError('VALIDATION_ERROR', { + status: 422, + message: 'Job id is required in the {id} path segment.', + data: { + formErrors: [], + fieldErrors: { id: ['Job id is required in the {id} path segment.'] }, + }, + }); + } + return { jobId: id, triggered: true }; + }, + ), + }; +} + +function clientOrigin(hostname: string, port: number): string { + const host = hostname === '0.0.0.0' ? '127.0.0.1' : hostname; + return `http://${host}:${port}`; +} + +Deno.test('createServiceClient RPC path reaches a plugin-workers triggerJob route', async () => { + const router = createWorkersStyleRouter(); + const running = await createService(router, { name: SERVICE_NAME }) + .withRPC({ rpcPath: RPC_PATH }) + .serve({ port: 0 }); + const envKey = createServerServiceEnvKey(SERVICE_NAME); + const previous = Deno.env.get(envKey); + Deno.env.set(envKey, clientOrigin(running.addr.hostname, running.addr.port)); + + try { + const client = createServiceClient({ + contract: router, + serviceName: SERVICE_NAME, + routerName: ROUTER_NAME, + }); + + // Route is mounted and routed over RPC: a body-carried id round-trips. + const ok = await client.triggerJob({ id: 'job-1' }); + assertEquals(ok, { jobId: 'job-1', triggered: true }); + + // No id resolves -> loud validation failure, not a 404 and not a silent + // success with an undefined id. + const error = await assertRejects(() => client.triggerJob({}), ORPCError); + assertEquals((error as ORPCError).code, 'VALIDATION_ERROR'); + } finally { + if (previous === undefined) { + Deno.env.delete(envKey); + } else { + Deno.env.set(envKey, previous); + } + await running.stop(); + } +}); diff --git a/plugins/workers/services/src/routers/jobs.ts b/plugins/workers/services/src/routers/jobs.ts index ce05b20c8..52782f416 100644 --- a/plugins/workers/services/src/routers/jobs.ts +++ b/plugins/workers/services/src/routers/jobs.ts @@ -1,5 +1,5 @@ import { DEFAULT_TOPIC, type JobMessage } from '@netscript/plugin-workers-core/runtime'; -import { notFound } from '@netscript/contracts'; +import { notFound, validationFailed } from '@netscript/contracts'; import { getJobQueue, getWorkersRuntime, router, type WorkersHandlers } from './router-context.ts'; export const jobHandlers: WorkersHandlers< @@ -85,18 +85,31 @@ export const jobHandlers: WorkersHandlers< }), triggerJob: router.triggerJob.handler(async ({ input, errors, path, context }) => { + // The `{id}` path segment is the single source of truth for the target job. + // oRPC merges the OpenAPI path param into `input.id`; the RPC transport carries + // it in the body. If neither resolves an id, fail loudly with a validation + // error rather than persisting an `undefined` KV key. + const jobId = input.id; + if (!jobId) { + validationFailed({ + errors, + message: 'Job id is required in the {id} path segment.', + fieldErrors: { id: ['Job id is required in the {id} path segment.'] }, + }); + } + const { jobRegistry: registry } = getWorkersRuntime(context); - const job = await registry.get(input.id); + const job = await registry.get(jobId); if (!job) { - notFound({ errors, path, resourceId: input.id }); + notFound({ errors, path, resourceId: jobId }); } const traceHeaders = (context as { traceHeaders?: { traceparent?: string; tracestate?: string } })?.traceHeaders; const jobMessage: JobMessage = { - jobId: input.id, + jobId, topic: job.topic ?? DEFAULT_TOPIC, triggeredBy: 'api', triggeredAt: new Date().toISOString(), @@ -119,6 +132,6 @@ export const jobHandlers: WorkersHandlers< : undefined, }); - return { jobId: input.id, triggered: true }; + return { jobId, triggered: true }; }), }; diff --git a/plugins/workers/services/src/routers/tasks.ts b/plugins/workers/services/src/routers/tasks.ts index dbeaec7ac..5347d4c4a 100644 --- a/plugins/workers/services/src/routers/tasks.ts +++ b/plugins/workers/services/src/routers/tasks.ts @@ -3,7 +3,7 @@ import { type ExecutionRecord, type TaskMessage, } from '@netscript/plugin-workers-core/runtime'; -import { notFound } from '@netscript/contracts'; +import { notFound, validationFailed } from '@netscript/contracts'; import { getTaskQueue, getWorkersRuntime, router, type WorkersHandlers } from './router-context.ts'; export const taskHandlers: WorkersHandlers< @@ -43,15 +43,28 @@ export const taskHandlers: WorkersHandlers< }), triggerTask: router.triggerTask.handler(async ({ input, errors, path, context }) => { + // The `{id}` path segment is the single source of truth for the target task. + // oRPC merges the OpenAPI path param into `input.id`; the RPC transport carries + // it in the body. If neither resolves an id, fail loudly with a validation + // error rather than persisting an `undefined` KV key. + const taskId = input.id; + if (!taskId) { + validationFailed({ + errors, + message: 'Task id is required in the {id} path segment.', + fieldErrors: { id: ['Task id is required in the {id} path segment.'] }, + }); + } + const { taskRegistry: registry } = getWorkersRuntime(context); - const task = await registry.get(input.id); + const task = await registry.get(taskId); if (!task) { - notFound({ errors, path, resourceId: input.id }); + notFound({ errors, path, resourceId: taskId }); } const taskMessage: TaskMessage = { - taskId: input.id, + taskId, topic: task.topic ?? DEFAULT_TOPIC, triggeredBy: 'api', triggeredAt: new Date().toISOString(), @@ -67,7 +80,7 @@ export const taskHandlers: WorkersHandlers< priority: input.priority, }); - return { taskId: input.id, triggered: true }; + return { taskId, triggered: true }; }), listTaskExecutions: router.listTaskExecutions.handler(async ({ input, context }) => { diff --git a/plugins/workers/services/src/trigger-path-id_test.ts b/plugins/workers/services/src/trigger-path-id_test.ts new file mode 100644 index 000000000..5c9a16cc8 --- /dev/null +++ b/plugins/workers/services/src/trigger-path-id_test.ts @@ -0,0 +1,79 @@ +import { assert, assertEquals } from 'jsr:@std/assert@^1'; +import { createPluginService } from '@netscript/plugin/service'; +import { router } from './router.ts'; +import type { WorkersServiceRuntime } from './routers/router-context.ts'; + +// Gap 2 regression (issue #279): `triggerJob` / `triggerTask` must take the +// target id from the `{id}` path segment, not a required body field. Previously +// the body `id` was required and read directly, so a consumer that omitted it +// persisted a message with `id=undefined` ("Workers KV key contains unsupported +// part: undefined"). These exercise the REAL workers contract + handlers through +// the OpenAPI (REST) mount: an empty request body must still resolve the id from +// the path and drive the lookup, proving the path is the single source of truth +// and that the registry is never queried with an undefined id. + +// Minimal stand-in runtime: only the relevant registry `get` is reached before +// the handler returns NOT_FOUND, so the rest of the runtime is intentionally +// absent. Test-only cast; the real runtime is wired by the host. +function buildApp(registryKey: 'jobRegistry' | 'taskRegistry', lookedUp: string[]) { + const stubRuntime = { + [registryKey]: { + get(id: string): Promise { + lookedUp.push(id); + return Promise.resolve(undefined); + }, + }, + } as unknown as WorkersServiceRuntime; + + return createPluginService(router, { + name: 'workers', + version: '1.0.0', + openApi: { + title: 'Workers API', + description: 'Workers service for job management and execution', + }, + context: () => ({ workers: stubRuntime }), + }).build(); +} + +function assertPathIdResolved(lookedUp: string[], expected: string, status: number): void { + // The path id flowed through to the registry lookup (source of truth), and the + // handler failed with NOT_FOUND (unknown resource) rather than persisting undefined. + assert( + lookedUp.includes(expected), + `expected registry lookup for the path id, got: ${JSON.stringify(lookedUp)}`, + ); + assert( + !lookedUp.includes('undefined') && !lookedUp.some((id) => id === undefined), + 'registry must never be queried with an undefined id', + ); + assertEquals(status, 404); +} + +Deno.test('triggerJob resolves the target job id from the {id} path, not the body', async () => { + const lookedUp: string[] = []; + const app = buildApp('jobRegistry', lookedUp); + + // Empty body: the id must come from the `{id}` path segment. + const response = await app.request('/api/v1/workers/jobs/job-from-path/trigger', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({}), + }); + + assertPathIdResolved(lookedUp, 'job-from-path', response.status); +}); + +Deno.test('triggerTask resolves the target task id from the {id} path, not the body', async () => { + const lookedUp: string[] = []; + const app = buildApp('taskRegistry', lookedUp); + + // Empty body: the id must come from the `{id}` path segment. + const response = await app.request('/api/v1/workers/tasks/task-from-path/trigger', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({}), + }); + + assertPathIdResolved(lookedUp, 'task-from-path', response.status); +});