diff --git a/CHANGELOG.md b/CHANGELOG.md index ef86f77a8..10c460b8e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,11 +25,18 @@ to docs, or any other relevant information. - **Experimental**: `TemporalOperationHandler` can now use Standalone Activities as asynchronous Nexus Operation backing executions through `TemporalNexusClient.startActivity` and `typedActivity`. +- **Experimental**: Adding a link from a Nexus Query operation to the destination activity. ### Changed - Nexus is now generally available (GA) for calling Nexus Operations from Workflows and handling Workflow-backed Operations with `WorkflowRunOperationHandler`. +- A `common.v1.Link.Workflow` now serializes to the Workflow path + `temporal:///namespaces/{ns}/workflows/{wid}/{rid}` with the optional `reason` as a query param, + rather than reusing the workflow event path with a `/history` suffix and dropping `reason`. The + previous form was indistinguishable from a workflow event link except by its type, and did not match + the other SDKs. Inbound Workflow links are now parsed as well, and a link with a trailing path + segment is rejected. - `@temporalio/ai-sdk` now requires `ai@>=7.0.59` as a peer dependency, up from `7.0.0`, since earlier releases threw a `TypeError` on import in runtimes without a global `fetch`. diff --git a/packages/client/src/internal.ts b/packages/client/src/internal.ts index 52489e025..dd7d9bffb 100644 --- a/packages/client/src/internal.ts +++ b/packages/client/src/internal.ts @@ -4,7 +4,7 @@ import type { WorkflowOptions, WorkflowUpdateOptions } from './workflow-options' import type { ActivityOptions } from './activity-client'; import type { WorkflowHandle, WorkflowUpdateHandle } from './workflow-client'; import type { WorkflowUpdateStage } from './workflow-update-stage'; -import type { WorkflowSignalInput } from './interceptors'; +import type { WorkflowQueryInput, WorkflowSignalInput } from './interceptors'; /** * A symbol used to attach extra, SDK-internal options to the `WorkflowClient.start()` call. @@ -134,6 +134,39 @@ export interface InternalWorkflowSignalOptions { */ export type InternalWorkflowSignalInput = WorkflowSignalInput & InternalWorkflowSignalOptions; +/** + * A symbol used to attach Nexus-specific options to a `WorkflowHandle.query()` call, so the client + * can capture the response link returned on the QueryWorkflowResponse. + * + * Unlike the signal variant there are no request links: a Query writes nothing to history, so + * `QueryWorkflowRequest` has no `links` field and there is no event to link from. + * + * @internal + * @hidden + */ +export const InternalWorkflowQueryOptionsSymbol = Symbol.for('__temporal_internal_client_workflow_query_options'); +export interface InternalWorkflowQueryOptions { + [InternalWorkflowQueryOptionsSymbol]?: { + /** + * Response link copied by the client from the QueryWorkflowResponse. Points at the Workflow + * execution that processed the Query rather than at an event, since a Query writes none. Only + * populated by servers that support Query response links; left unset otherwise. + */ + responseLink?: temporal.api.common.v1.ILink; + }; +} + +/** + * The SDK-internal variant of {@link WorkflowQueryInput} that carries the + * {@link InternalWorkflowQueryOptionsSymbol} payload used by the Temporal Nexus helpers to capture + * the response link. Kept off the public `WorkflowQueryInput` so the symbol does not leak onto the + * interceptor surface. + * + * @internal + * @hidden + */ +export type InternalWorkflowQueryInput = WorkflowQueryInput & InternalWorkflowQueryOptions; + /** * The SDK-internal surface of a `WorkflowHandle` used by the Temporal Nexus helpers to send a Signal * while forwarding request links and capturing the response link the server returns. @@ -147,7 +180,8 @@ export type InternalWorkflowSignalInput = WorkflowSignalInput & InternalWorkflow * @hidden */ export type InternalWorkflowHandle = WorkflowHandle & - InternalWorkflowSignalOptions & { + InternalWorkflowSignalOptions & + InternalWorkflowQueryOptions & { /** * A single, non-overloaded view of {@link WorkflowHandle.startUpdate} used by the Temporal Nexus * helpers to start an Update-backed operation. The public `startUpdate` is overloaded to diff --git a/packages/client/src/workflow-client.ts b/packages/client/src/workflow-client.ts index 6fdb4cf3c..e6cef1a62 100644 --- a/packages/client/src/workflow-client.ts +++ b/packages/client/src/workflow-client.ts @@ -113,8 +113,14 @@ import type { BaseClientOptions, LoadedWithDefaults, WithDefaults } from './base import { BaseClient, defaultBaseClientOptions } from './base-client'; import { mapAsyncIterable } from './iterators-utils'; import { WorkflowUpdateStage, encodeWorkflowUpdateStage } from './workflow-update-stage'; -import type { InternalWorkflowHandle, InternalWorkflowSignalInput, InternalWorkflowStartOptions } from './internal'; +import type { + InternalWorkflowHandle, + InternalWorkflowQueryInput, + InternalWorkflowSignalInput, + InternalWorkflowStartOptions, +} from './internal'; import { + InternalWorkflowQueryOptionsSymbol, InternalWorkflowSignalOptionsSymbol, InternalWorkflowStartOptionsSymbol, type InternalWorkflowUpdateOptions, @@ -1010,6 +1016,7 @@ export class WorkflowClient extends BaseClient { protected async _queryWorkflowHandler(input: WorkflowQueryInput): Promise { const dataConverter = this.dataConverter; const context = this.workflowSerializationContext(input.workflowExecution.workflowId!); + const internalOptions = (input as InternalWorkflowQueryInput)[InternalWorkflowQueryOptionsSymbol]; const req: temporal.api.workflowservice.v1.IQueryWorkflowRequest = { queryRejectCondition: input.queryRejectCondition, namespace: this.options.namespace, @@ -1047,6 +1054,12 @@ export class WorkflowClient extends BaseClient { this.rethrowGrpcError(err, 'Failed to query Workflow', input.workflowExecution); } await visit(response, walkQueryWorkflowResponse, extstoreInboundOptions(externalStorage)); + if (internalOptions != null) { + // A Query writes nothing to history, so the server returns a link to the Workflow execution + // that processed it rather than to an event. Captured before the rejection check below so a + // rejected Query still records its link. Older servers leave it unset. + internalOptions.responseLink = response.link ?? undefined; + } if (response.queryRejected) { if (response.queryRejected.status === undefined || response.queryRejected.status === null) { throw new TypeError('Received queryRejected from server with no status'); @@ -1765,13 +1778,18 @@ export class WorkflowClient extends BaseClient { async query(def: QueryDefinition | string, ...args: Args): Promise { const next = this.client._queryWorkflowHandler.bind(this.client); const fn = composeInterceptors(interceptors, 'query', next); - return fn({ + const input: InternalWorkflowQueryInput = { workflowExecution: { workflowId, runId }, queryRejectCondition: encodeQueryRejectCondition(this.client.options.queryRejectCondition), queryType: typeof def === 'string' ? def : def.name, args, headers: {}, - }) as Promise; + // Forward any SDK-internal query options (e.g. the Nexus response-link slot) that were + // attached to this handle, and let the query handler write the response link back onto the + // same payload. + [InternalWorkflowQueryOptionsSymbol]: (this as InternalWorkflowHandle)[InternalWorkflowQueryOptionsSymbol], + }; + return fn(input) as Promise; }, }; } diff --git a/packages/nexus/src/__tests__/helpers.ts b/packages/nexus/src/__tests__/helpers.ts new file mode 100644 index 000000000..035b2451e --- /dev/null +++ b/packages/nexus/src/__tests__/helpers.ts @@ -0,0 +1,34 @@ +import type * as nexus from 'nexus-rpc'; +import type { HandlerContext } from '../context'; + +/** + * Builds a minimal {@link nexus.StartOperationContext} for driving a start handler directly, without + * a live worker. + */ +export function makeStartContext(overrides: Partial = {}): nexus.StartOperationContext { + return { + service: 'service', + operation: 'operation', + headers: {}, + abortSignal: new AbortController().signal, + requestId: 'request-id', + inboundLinks: [], + outboundLinks: [], + ...overrides, + }; +} + +/** + * Builds the handler context the worker would otherwise install, backed by the given stand-in client. + * Only the fields the Workflow helpers read are populated. + */ +export function makeHandlerContext(client: HandlerContext['client']): HandlerContext { + return { + log: { trace() {}, debug() {}, info() {}, warn() {}, error() {} } as unknown as HandlerContext['log'], + metrics: {} as HandlerContext['metrics'], + client, + namespace: 'ns', + taskQueue: 'tq', + endpoint: 'endpoint', + }; +} diff --git a/packages/nexus/src/__tests__/test-nexus-link-converter.ts b/packages/nexus/src/__tests__/test-nexus-link-converter.ts index 791078987..d6661ca65 100644 --- a/packages/nexus/src/__tests__/test-nexus-link-converter.ts +++ b/packages/nexus/src/__tests__/test-nexus-link-converter.ts @@ -5,6 +5,7 @@ import { convertActivityLinkToNexusLink, convertNexusLinkToTemporalLink, convertNexusLinkToWorkflowEventLink, + convertNexusLinkToWorkflowLink, convertNexusOperationLinkToNexusLink, convertTemporalLinkToNexusLink, convertWorkflowEventLinkToNexusLink, @@ -140,14 +141,26 @@ test('convertActivityLinkToNexusLink escapes URL path components', (t) => { t.deepEqual(roundTrip, { activity }); }); -test('convertWorkflowLinkToNexusLink produces a history URL with the Workflow link type', (t) => { +test('convertWorkflowLinkToNexusLink produces a workflow URL with the Workflow link type', (t) => { + // A Workflow link addresses the execution itself, so there is no '/history' suffix. That suffix + // belongs to the workflow event form, and its absence is what distinguishes the two paths. const nexusLink = convertWorkflowLinkToNexusLink({ namespace: 'ns', workflowId: 'wid', runId: 'rid', }); t.is(nexusLink.type, WORKFLOW_TYPE); - t.is(nexusLink.url.toString(), 'temporal:///namespaces/ns/workflows/wid/rid/history'); + t.is(nexusLink.url.toString(), 'temporal:///namespaces/ns/workflows/wid/rid'); +}); + +test('convertWorkflowLinkToNexusLink carries reason as a query param', (t) => { + const nexusLink = convertWorkflowLinkToNexusLink({ + namespace: 'ns', + workflowId: 'wid', + runId: 'rid', + reason: 'rejected update', + }); + t.is(nexusLink.url.toString(), 'temporal:///namespaces/ns/workflows/wid/rid?reason=rejected+update'); }); test('convertWorkflowLinkToNexusLink escapes URL path components', (t) => { @@ -156,7 +169,7 @@ test('convertWorkflowLinkToNexusLink escapes URL path components', (t) => { workflowId: 'work id', runId: 'run/id', }); - t.is(nexusLink.url.toString(), 'temporal:///namespaces/name%2Fspace/workflows/work%20id/run%2Fid/history'); + t.is(nexusLink.url.toString(), 'temporal:///namespaces/name%2Fspace/workflows/work%20id/run%2Fid'); }); test('convertWorkflowLinkToNexusLink throws on missing required fields', (t) => { @@ -166,8 +179,8 @@ test('convertWorkflowLinkToNexusLink throws on missing required fields', (t) => t.throws(() => convertWorkflowLinkToNexusLink({ namespace: 'ns', workflowId: '', runId: 'rid' }), { instanceOf: TypeError, }); - // An empty run ID would produce `.../workflows/wid//history`, whose double slash does not resolve - // to a valid UI page, so the converter rejects it rather than emit a malformed URL. + // An empty run ID would address no particular run, so the converter rejects it and lets the + // caller drop the link rather than attach one that resolves nowhere useful. t.throws(() => convertWorkflowLinkToNexusLink({ namespace: 'ns', workflowId: 'wid', runId: '' }), { instanceOf: TypeError, }); @@ -381,3 +394,108 @@ test('throws on unknown eventType in requestIdRef', (t) => { }; t.throws(() => convertNexusLinkToWorkflowEventLink(fakeLink), { message: /Unknown eventType parameter/ }); }); + +test('convertNexusLinkToWorkflowLink parses a workflow URL', (t) => { + const workflowLink = convertNexusLinkToWorkflowLink({ + url: new URL('temporal:///namespaces/ns/workflows/wid/rid'), + type: WORKFLOW_TYPE, + }); + t.deepEqual(workflowLink, { namespace: 'ns', workflowId: 'wid', runId: 'rid' }); +}); + +test('convertNexusLinkToWorkflowLink parses reason', (t) => { + const workflowLink = convertNexusLinkToWorkflowLink({ + url: new URL('temporal:///namespaces/ns/workflows/wid/rid?reason=rejected+update'), + type: WORKFLOW_TYPE, + }); + t.is(workflowLink.reason, 'rejected update'); +}); + +test('convertNexusLinkToWorkflowLink finds reason by key, not position', (t) => { + const workflowLink = convertNexusLinkToWorkflowLink({ + url: new URL('temporal:///namespaces/ns/workflows/wid/rid?foo=bar&reason=Query+processed'), + type: WORKFLOW_TYPE, + }); + t.is(workflowLink.reason, 'Query processed'); +}); + +test('convertNexusLinkToWorkflowLink leaves reason unset when absent', (t) => { + // A key that merely starts with 'reason' must not be treated as 'reason'. + const workflowLink = convertNexusLinkToWorkflowLink({ + url: new URL('temporal:///namespaces/ns/workflows/wid/rid?reasonx=nope'), + type: WORKFLOW_TYPE, + }); + t.is(workflowLink.reason, undefined); +}); + +test('convertNexusLinkToWorkflowLink rejects a trailing path segment', (t) => { + // The workflow event form addresses an event inside the Workflow, so it must not be accepted as a + // Workflow link even when the type says otherwise. + t.throws( + () => + convertNexusLinkToWorkflowLink({ + url: new URL('temporal:///namespaces/ns/workflows/wid/rid/history'), + type: WORKFLOW_TYPE, + }), + { instanceOf: TypeError } + ); +}); + +test('convertNexusLinkToWorkflowLink rejects a missing run ID', (t) => { + t.throws( + () => + convertNexusLinkToWorkflowLink({ + url: new URL('temporal:///namespaces/ns/workflows/wid'), + type: WORKFLOW_TYPE, + }), + { instanceOf: TypeError } + ); +}); + +test('convertNexusLinkToWorkflowEventLink rejects a suffixless workflow path', (t) => { + // The inverse of the trailing-segment case: a Workflow link must not be readable as a workflow + // event. + t.throws( + () => + convertNexusLinkToWorkflowEventLink({ + url: new URL('temporal:///namespaces/ns/workflows/wid/rid'), + type: WORKFLOW_EVENT_TYPE, + }), + { instanceOf: TypeError } + ); +}); + +test('convertNexusLinkToTemporalLink dispatches the Workflow link type', (t) => { + const temporalLink = convertNexusLinkToTemporalLink({ + url: new URL('temporal:///namespaces/ns/workflows/wid/rid?reason=Query+processed'), + type: WORKFLOW_TYPE, + }); + t.deepEqual(temporalLink, { + workflow: { namespace: 'ns', workflowId: 'wid', runId: 'rid', reason: 'Query processed' }, + }); +}); + +test('Workflow link round trips through both converters', (t) => { + // Reserved characters in every field at once: path segments are percent escaped and the reason is a + // query value, so a reason containing '=' and '&' must not be split as query syntax. + const workflowLink = { + namespace: 'ns/with/slash', + workflowId: 'wf id with space', + runId: 'rid', + reason: 'reason with = and &', + }; + t.deepEqual(convertNexusLinkToWorkflowLink(convertWorkflowLinkToNexusLink(workflowLink)), workflowLink); +}); + +test('Workflow link reason round trips a literal plus', (t) => { + // URLSearchParams form encodes, writing a space as '+' and a literal '+' as '%2B', and its reader + // reverses that. Percent decoding alone would turn this reason into 'a b'. + const nexusLink = convertWorkflowLinkToNexusLink({ + namespace: 'ns', + workflowId: 'wid', + runId: 'rid', + reason: 'a+b', + }); + t.is(nexusLink.url.search, '?reason=a%2Bb'); + t.is(convertNexusLinkToWorkflowLink(nexusLink).reason, 'a+b'); +}); diff --git a/packages/nexus/src/__tests__/test-nexus-query-workflow.ts b/packages/nexus/src/__tests__/test-nexus-query-workflow.ts new file mode 100644 index 000000000..74f231a01 --- /dev/null +++ b/packages/nexus/src/__tests__/test-nexus-query-workflow.ts @@ -0,0 +1,144 @@ +import test from 'ava'; +import type * as nexus from 'nexus-rpc'; +import { temporal } from '@temporalio/proto'; +import { InternalWorkflowQueryOptionsSymbol, type InternalWorkflowHandle } from '@temporalio/client/lib/internal'; +import { asyncLocalStorage, type HandlerContext } from '../context'; +import { TemporalOperationHandler, TemporalOperationResult } from '../workflow-helpers'; +import { makeHandlerContext, makeStartContext } from './helpers'; + +const WORKFLOW_TYPE = (temporal.api.common.v1.Link.Workflow as any).fullName.slice(1); + +function workflowLink(workflowId: string, runId: string, reason: string): temporal.api.common.v1.ILink { + return { workflow: { namespace: 'ns', workflowId, runId, reason } }; +} + +/** + * Stands in for the client the worker would put on the handler context. `query` behaves like the real + * client handler: it writes the server's response link (when there is one) onto the SDK-internal + * options payload the caller attached to the handle, then resolves or rejects. + */ +function makeFakeClient( + responses: Array<{ link?: temporal.api.common.v1.ILink; result?: unknown; reject?: Error }> +): HandlerContext['client'] { + let next = 0; + return { + workflow: { + getHandle(workflowId: string, runId?: string) { + const handle = { + workflowId, + runId, + async query(_def: unknown, ..._args: unknown[]): Promise { + const response = responses[next++]; + if (response == null) { + throw new Error('fake client: more queries than canned responses'); + } + const internalOptions = (handle as unknown as InternalWorkflowHandle)[InternalWorkflowQueryOptionsSymbol]; + if (internalOptions != null) { + internalOptions.responseLink = response.link; + } + if (response.reject != null) { + throw response.reject; + } + return response.result; + }, + }; + return handle; + }, + }, + } as unknown as HandlerContext['client']; +} + +/** + * Runs a start handler that queries the given Workflow, inside a handler context backed by the fake + * client, and returns the operation context so its `outboundLinks` can be asserted. + */ +async function runQueryOperation( + client: HandlerContext['client'], + queries = 1 +): Promise<{ ctx: nexus.StartOperationContext; result: unknown; error?: Error }> { + const ctx = makeStartContext(); + const handler = new TemporalOperationHandler({ + async start(_startCtx, temporalClient) { + const handle = temporalClient.getWorkflowHandle('wid', 'rid'); + let last: unknown; + for (let i = 0; i < queries; i++) { + last = await handle.query('getCount'); + } + return TemporalOperationResult.sync(last); + }, + }); + + try { + const result = await asyncLocalStorage.run(makeHandlerContext(client), () => handler.start(ctx, undefined)); + return { ctx, result: (result as unknown as { value: unknown }).value }; + } catch (error) { + return { ctx, result: undefined, error: error as Error }; + } +} + +test('a Query response link is attached to the operation outbound links', async (t) => { + // A Query writes nothing to history, so the server answers with a Workflow link naming the + // execution that processed it. That link has to reach the operation so the caller's NexusOperation + // event points back at the queried Workflow. + const link = workflowLink('wid', 'rid', 'Query processed'); + const { ctx, result, error } = await runQueryOperation(makeFakeClient([{ link, result: 2 }])); + + t.is(error, undefined); + // Capturing the link must not disturb the Query's own result. + t.is(result, 2); + t.is(ctx.outboundLinks.length, 1); + t.is(ctx.outboundLinks[0]!.type, WORKFLOW_TYPE); + t.is(ctx.outboundLinks[0]!.url.toString(), 'temporal:///namespaces/ns/workflows/wid/rid?reason=Query+processed'); +}); + +test('an older server that returns no Query link attaches nothing', async (t) => { + const { ctx, result, error } = await runQueryOperation(makeFakeClient([{ result: 2 }])); + + t.is(error, undefined); + t.is(result, 2); + t.deepEqual(ctx.outboundLinks, []); +}); + +test('two Queries attach both response links in call order', async (t) => { + const first = workflowLink('callee-a', 'run-a', 'Query processed'); + const second = workflowLink('callee-b', 'run-b', 'Query processed'); + const { ctx, error } = await runQueryOperation( + makeFakeClient([ + { link: first, result: 1 }, + { link: second, result: 2 }, + ]), + 2 + ); + + t.is(error, undefined); + t.deepEqual( + ctx.outboundLinks.map((l) => l.url.toString()), + [ + 'temporal:///namespaces/ns/workflows/callee-a/run-a?reason=Query+processed', + 'temporal:///namespaces/ns/workflows/callee-b/run-b?reason=Query+processed', + ] + ); +}); + +test('a failed Query still attaches its response link', async (t) => { + // The server returns a link alongside a rejection or failure, and the client records it before + // throwing. Pins that the link is attached in a `finally` rather than only on the success path. + const link = workflowLink('wid', 'rid', 'Query processed'); + const { ctx, error } = await runQueryOperation(makeFakeClient([{ link, reject: new Error('query rejected') }])); + + t.truthy(error); + t.is(ctx.outboundLinks.length, 1); + t.is(ctx.outboundLinks[0]!.url.toString(), 'temporal:///namespaces/ns/workflows/wid/rid?reason=Query+processed'); +}); + +test('an unconvertible response link is dropped rather than failing the operation', async (t) => { + // A link variant the converter does not handle is logged and skipped; links are not essential to + // the operation succeeding. + const { ctx, result, error } = await runQueryOperation( + makeFakeClient([{ link: { batchJob: { jobId: 'batch' } }, result: 2 }]) + ); + + t.is(error, undefined); + t.is(result, 2); + t.deepEqual(ctx.outboundLinks, []); +}); diff --git a/packages/nexus/src/__tests__/test-nexus-signal-with-start-handle.ts b/packages/nexus/src/__tests__/test-nexus-signal-with-start-handle.ts new file mode 100644 index 000000000..42dcc6627 --- /dev/null +++ b/packages/nexus/src/__tests__/test-nexus-signal-with-start-handle.ts @@ -0,0 +1,89 @@ +import test from 'ava'; +import * as nexus from 'nexus-rpc'; +import { asyncLocalStorage, type HandlerContext } from '../context'; +import { signalWithStartWorkflow } from '../workflow-helpers'; +import { makeHandlerContext, makeStartContext } from './helpers'; + +/** + * Stands in for the client the worker would put on the handler context, recording the calls the + * returned handle makes so they can be asserted. + */ +function makeFakeClient(calls: string[]): HandlerContext['client'] { + return { + workflow: { + async signalWithStart(_wf: unknown, _options: unknown) { + calls.push('signalWithStart'); + return { workflowId: 'wid', signaledRunId: 'rid' }; + }, + getHandle(workflowId: string, runId?: string) { + return { + workflowId, + runId, + async signal(def: unknown) { + calls.push(`signal:${String(def)}`); + }, + async query(def: unknown) { + calls.push(`query:${String(def)}`); + return 'query-result'; + }, + }; + }, + }, + } as unknown as HandlerContext['client']; +} + +/** + * Runs `body` against the handle `signalWithStartWorkflow` returns, inside the handler context. The + * handle's methods read that context themselves, so they have to be called within it, exactly as they + * would be from inside an operation handler. + */ +async function withSignalWithStartHandle( + body: (handle: Awaited>, calls: string[]) => Promise, + ctxOverrides: Partial = {} +): Promise { + const calls: string[] = []; + const ctx = makeStartContext(ctxOverrides); + return await asyncLocalStorage.run(makeHandlerContext(makeFakeClient(calls)), async () => { + const handle = await signalWithStartWorkflow(ctx, 'SomeWorkflow', { + workflowId: 'wid', + signal: 'someSignal', + signalArgs: [], + } as any); + return await body(handle, calls); + }); +} + +test('the handle returned by signalWithStartWorkflow can signal', async (t) => { + // The handle used to be a bare object literal cast to WorkflowHandle, so the `signal` its type + // advertised was undefined at runtime. + await withSignalWithStartHandle(async (handle, calls) => { + t.is(handle.workflowId, 'wid'); + t.is(handle.runId, 'rid'); + + await handle.signal('anotherSignal'); + t.deepEqual(calls, ['signalWithStart', 'signal:anotherSignal']); + }); +}); + +test('the handle returned by signalWithStartWorkflow can query', async (t) => { + await withSignalWithStartHandle(async (handle, calls) => { + t.is(await handle.query('someQuery'), 'query-result'); + t.deepEqual(calls, ['signalWithStart', 'query:someQuery']); + }); +}); + +test('the handle returned by signalWithStartWorkflow rejects update with a handler error', async (t) => { + // The Workflow run this handle refers to already backs the operation, so starting an Update-backed + // operation from it is a caller error. It must surface as a BAD_REQUEST handler error rather than a + // TypeError from calling a method the cast only pretended to provide. A callback URL is supplied so + // the earlier callback-URL guard passes and the reservation is what rejects. + await withSignalWithStartHandle( + async (handle) => { + const err = await t.throwsAsync(() => (handle as any).update('someUpdate')); + t.true(err instanceof nexus.HandlerError); + t.is((err as nexus.HandlerError).type, 'BAD_REQUEST'); + t.regex(err?.message ?? '', /already backs it/); + }, + { callbackUrl: 'http://localhost/callback' } + ); +}); diff --git a/packages/nexus/src/__tests__/test-nexus-update-workflow.ts b/packages/nexus/src/__tests__/test-nexus-update-workflow.ts index 2f9bdad2b..2aeb3d5f7 100644 --- a/packages/nexus/src/__tests__/test-nexus-update-workflow.ts +++ b/packages/nexus/src/__tests__/test-nexus-update-workflow.ts @@ -1,25 +1,7 @@ import test from 'ava'; import * as nexus from 'nexus-rpc'; import { TemporalOperationHandler } from '../workflow-helpers'; - -/** - * Builds a minimal {@link nexus.StartOperationContext} for driving a {@link TemporalOperationHandler} - * start handler directly. The {@link WorkflowHandle.update} input-validation guards run before any - * Temporal Client or handler context is touched, so these paths can be exercised without a live - * worker. - */ -function makeStartContext(overrides: Partial = {}): nexus.StartOperationContext { - return { - service: 'service', - operation: 'operation', - headers: {}, - abortSignal: new AbortController().signal, - requestId: 'request-id', - inboundLinks: [], - outboundLinks: [], - ...overrides, - }; -} +import { makeStartContext } from './helpers'; test('update without a callback URL fails with a BAD_REQUEST handler error', async (t) => { const handler = new TemporalOperationHandler({ diff --git a/packages/nexus/src/link-converter.ts b/packages/nexus/src/link-converter.ts index f8c56c11e..319fcac06 100644 --- a/packages/nexus/src/link-converter.ts +++ b/packages/nexus/src/link-converter.ts @@ -15,6 +15,7 @@ const LINK_EVENT_ID_PARAM = 'eventID'; const LINK_EVENT_TYPE_PARAM = 'eventType'; const LINK_REQUEST_ID_PARAM = 'requestID'; const LINK_REFERENCE_TYPE_KEY = 'referenceType'; +const LINK_REASON_PARAM = 'reason'; const EVENT_REFERENCE_TYPE = 'EventReference'; const REQUEST_ID_REFERENCE_TYPE = 'RequestIdReference'; @@ -60,6 +61,11 @@ export function convertNexusLinkToTemporalLink(link: NexusLink): TemporalLink { nexusOperation: convertNexusLinkToNexusOperationLink(link), }; + case WORKFLOW_TYPE: + return { + workflow: convertNexusLinkToWorkflowLink(link), + }; + case ACTIVITY_TYPE: return { activity: convertNexusLinkToActivityLink(link), @@ -95,16 +101,17 @@ export function convertWorkflowEventLinkToNexusLink(we: WorkflowEventLink): Nexu /** * Converts a plain Workflow link (as opposed to a {@link WorkflowEventLink}) to a Nexus link. * - * Used to point at a Workflow when there is no specific history event to reference, e.g. when an - * UpdateWorkflow-backed Nexus operation fails validation and no Update Accepted event exists. The - * resulting URL is the workflow history URL without a reference query; the link is distinguished - * from a workflow event link by its {@link NexusLink.type}. + * A Workflow link addresses a Workflow execution as a whole rather than one event within it, so the + * URL carries no event path suffix and no reference query params. It is used when there is no history + * event to point at, e.g. a Query, or an UpdateWorkflow-backed Nexus operation that fails validation + * and so has no Update Accepted event. The absence of the `/history` suffix is what distinguishes + * this path from a workflow event link's. * - * `runId` is required: an empty run segment produces `.../workflows///history`, whose - * double slash does not resolve to a valid Temporal UI workflow page. The server populates the - * resolved run ID even on the plain Workflow link it returns for a validation failure, so a missing - * run ID is unexpected; throwing (rather than coalescing to '') lets callers drop the link instead - * of attaching a malformed URL. + * The optional `reason` explaining why the link exists is carried as a query param. + * + * `runId` is required: the server populates the resolved run ID even on the plain Workflow link it + * returns, so a missing one is unexpected; throwing (rather than coalescing to '') lets callers drop + * the link instead of attaching one that addresses no particular run. */ export function convertWorkflowLinkToNexusLink(wl: WorkflowLink): NexusLink { if (!wl.namespace || !wl.workflowId || !wl.runId) { @@ -115,15 +122,64 @@ export function convertWorkflowLinkToNexusLink(wl: WorkflowLink): NexusLink { const url = new URL( `temporal:///namespaces/${encodeURIComponent(wl.namespace)}/workflows/${encodeURIComponent( wl.workflowId - )}/${encodeURIComponent(wl.runId)}/history` + )}/${encodeURIComponent(wl.runId)}` ); + if (wl.reason) { + const params = new URLSearchParams(); + params.set(LINK_REASON_PARAM, wl.reason); + url.search = params.toString(); + } + return { url, type: WORKFLOW_TYPE, }; } +/** + * Converts a Nexus link back to a plain Workflow link. + * + * The run ID ends a Workflow link, so anything trailing is rejected. In particular this rejects the + * workflow event form, which ends in `history` and is otherwise identical. + */ +export function convertNexusLinkToWorkflowLink(link: NexusLink): WorkflowLink { + // /namespaces/:namespace/workflows/:workflowId/:runId + const [namespace, workflowId, runId] = parseTemporalLinkPath(link, 'workflows'); + + if (!namespace || !workflowId || !runId) { + throw new TypeError('Missing required fields: namespace, workflowId, or runId'); + } + + const workflowLink: WorkflowLink = { namespace, workflowId, runId }; + const reason = link.url.searchParams.get(LINK_REASON_PARAM); + if (reason != null) { + workflowLink.reason = reason; + } + return workflowLink; +} + +/** + * Validates a Temporal link path of the shape `/namespaces/:namespace/:collection/:id/:runId[/:tail]` + * and returns its three decoded variable segments. + * + * Passing no `tail` means nothing may follow the run ID, which is what separates a plain Workflow + * link from a workflow event link since both live under `workflows`. + */ +function parseTemporalLinkPath(link: NexusLink, collection: string, tail?: string): [string, string, string] { + const parts = link.url.pathname.split('/'); + const expectedLength = tail == null ? 6 : 7; + if ( + parts.length !== expectedLength || + parts[1] !== 'namespaces' || + parts[3] !== collection || + (tail != null && parts[6] !== tail) + ) { + throw new TypeError(`Invalid URL path: ${link.url}`); + } + return [decodeURIComponent(parts[2]!), decodeURIComponent(parts[4]!), decodeURIComponent(parts[5]!)]; +} + export function convertNexusOperationLinkToNexusLink(opLink: NexusOperationLink): NexusLink { if (!opLink.namespace || !opLink.operationId || !opLink.runId) { throw new TypeError('Missing required fields: namespace, operationId, or runId'); @@ -160,13 +216,7 @@ export function convertActivityLinkToNexusLink(activityLink: ActivityLink): Nexu export function convertNexusLinkToWorkflowEventLink(link: NexusLink): WorkflowEventLink { // /namespaces/:namespace/workflows/:workflowId/:runId/history - const parts = link.url.pathname.split('/'); - if (parts.length !== 7 || parts[1] !== 'namespaces' || parts[3] !== 'workflows' || parts[6] !== 'history') { - throw new TypeError(`Invalid URL path: ${link.url}`); - } - const namespace = decodeURIComponent(parts[2]!); - const workflowId = decodeURIComponent(parts[4]!); - const runId = decodeURIComponent(parts[5]!); + const [namespace, workflowId, runId] = parseTemporalLinkPath(link, 'workflows', 'history'); const query = link.url.searchParams; const refType = query.get(LINK_REFERENCE_TYPE_KEY); @@ -192,13 +242,7 @@ export function convertNexusLinkToWorkflowEventLink(link: NexusLink): WorkflowEv function convertNexusLinkToNexusOperationLink(link: NexusLink): NexusOperationLink { // /namespaces/:namespace/nexus-operations/:operationId/:runId/details - const parts = link.url.pathname.split('/'); - if (parts.length !== 7 || parts[1] !== 'namespaces' || parts[3] !== 'nexus-operations' || parts[6] !== 'details') { - throw new TypeError(`Invalid URL path: ${link.url}`); - } - const namespace = decodeURIComponent(parts[2]!); - const operationId = decodeURIComponent(parts[4]!); - const runId = decodeURIComponent(parts[5]!); + const [namespace, operationId, runId] = parseTemporalLinkPath(link, 'nexus-operations', 'details'); if (!namespace || !operationId || !runId) { throw new TypeError('Missing required fields: namespace, operationId, or runId'); @@ -213,13 +257,7 @@ function convertNexusLinkToNexusOperationLink(link: NexusLink): NexusOperationLi function convertNexusLinkToActivityLink(link: NexusLink): ActivityLink { // /namespaces/:namespace/activities/:activityId/:runId/details - const parts = link.url.pathname.split('/'); - if (parts.length !== 7 || parts[1] !== 'namespaces' || parts[3] !== 'activities' || parts[6] !== 'details') { - throw new TypeError(`Invalid URL path: ${link.url}`); - } - const namespace = decodeURIComponent(parts[2]!); - const activityId = decodeURIComponent(parts[4]!); - const runId = decodeURIComponent(parts[5]!); + const [namespace, activityId, runId] = parseTemporalLinkPath(link, 'activities', 'details'); if (!namespace || !activityId || !runId) { throw new TypeError('Missing required fields: namespace, activityId, or runId'); diff --git a/packages/nexus/src/workflow-helpers.ts b/packages/nexus/src/workflow-helpers.ts index e944a2dd1..9f4eb680f 100644 --- a/packages/nexus/src/workflow-helpers.ts +++ b/packages/nexus/src/workflow-helpers.ts @@ -3,6 +3,7 @@ import type { Workflow, WorkflowResultType, WithWorkflowArgs, + QueryDefinition, SignalDefinition, UpdateDefinition, } from '@temporalio/common'; @@ -22,12 +23,14 @@ import { type temporal } from '@temporalio/proto'; import type { InternalActivityStartOptions, InternalWorkflowHandle, + InternalWorkflowQueryOptions, InternalWorkflowSignalOptions, InternalWorkflowStartOptions, InternalWorkflowUpdateOptions, } from '@temporalio/client/lib/internal'; import { InternalActivityStartOptionsSymbol, + InternalWorkflowQueryOptionsSymbol, InternalWorkflowSignalOptionsSymbol, InternalWorkflowStartOptionsSymbol, InternalWorkflowUpdateOptionsSymbol, @@ -73,6 +76,16 @@ export interface WorkflowHandle { ...args: Args ): Promise; + /** + * Queries the Workflow as part of this Nexus Operation. + * + * A Query resolves immediately and writes nothing to history, so this backs a synchronous + * operation: there is no operation token and nothing to cancel. The link the server returns for the + * Workflow that processed the Query is attached to the operation's outbound links, so the caller's + * NexusOperation history event points back at that Workflow. + */ + query(def: QueryDefinition | string, ...args: Args): Promise; + /** * Virtual type brand to maintain a distinction between {@link WorkflowHandle} provided by the * {@link startWorkflow} helper (which will have attached links, request ID, completion URL, etc) @@ -288,6 +301,26 @@ function createWorkflowHandle( } }, + async query(def: QueryDefinition | string, ...args: Args): Promise { + const { client } = getHandlerContext(); + + // Query through a regular WorkflowHandle for the same reason as signal above. There are no + // request links to forward, since a Query writes no event to link from; the payload exists + // only so the query handler can write the server's response link back onto it. + const handle = client.workflow.getHandle(this.workflowId, this.runId) as InternalWorkflowHandle; + const internalOptions: InternalWorkflowQueryOptions[typeof InternalWorkflowQueryOptionsSymbol] = {}; + handle[InternalWorkflowQueryOptionsSymbol] = internalOptions; + try { + return await handle.query(def, ...args); + } finally { + // In the `finally` so a rejected or failed Query still contributes its link: the server + // returns one alongside the rejection, and the client records it before throwing. + if (internalOptions.responseLink != null) { + pushResponseLink(ctx, internalOptions.responseLink); + } + } + }, + // Single permissive implementation signature; the no-arg vs with-args overload pair that callers // type against is declared on `UpdatableWorkflowHandle.update` update( @@ -334,10 +367,9 @@ export async function signalWithStartWorkflow>; + // As in `startWorkflow`, the Workflow run this handle refers to is the operation's async backing + // operation, so the handle's `update()` must not be able to start another one. + return createWorkflowHandle(ctx, handle.workflowId, handle.signaledRunId, alreadyBackedReservation); } /**