diff --git a/nexus-messaging/src/ondemandpattern/README.md b/nexus-messaging/src/ondemandpattern/README.md index 4e35d9e9f..6ca1ca219 100644 --- a/nexus-messaging/src/ondemandpattern/README.md +++ b/nexus-messaging/src/ondemandpattern/README.md @@ -7,11 +7,14 @@ instance to target. The caller workflow: -1. Starts two remote `GreetingWorkflow` instances via `runFromRemote` (backed by `WorkflowRunOperation`) -2. Queries supported languages from workflow one and the current language from workflow two -3. Changes the language on each (Spanish and Hindi) -4. Approves both workflows -5. Waits for each to complete and returns their results +1. Attaches approval context for user one via `attachApprovalContext`, before anything has started + that user's workflow +2. Starts or attaches to two remote `GreetingWorkflow` instances via `runFromRemote` (backed by `TemporalOperation`) +3. Attaches approval context for user two, whose workflow now already exists +4. Queries supported languages from workflow one and the current language from workflow two +5. Changes the language on each (Spanish and Hindi) +6. Approves both workflows +7. Waits for each to complete and returns their results ### Running @@ -22,6 +25,7 @@ Start a compatible Temporal dev server with Workflow Update callbacks enabled: --dynamic-config-value history.enableCHASMCallbacks=true \ --dynamic-config-value history.enableUpdateCallbacks=true \ --dynamic-config-value history.enableCHASMSignalBacklinks=true \ + --dynamic-config-value history.enableSignalWithStartFromWorkflow=true \ --namespace nexus-messaging-handler-namespace \ --namespace nexus-messaging-caller-namespace ``` @@ -65,8 +69,10 @@ npm run workflow.ondemandpattern Expected output: ``` + attached approval context for user: UserId_One started workflow one for user: UserId_One started workflow two for user: UserId_Two + attached approval context to running workflow for user: UserId_Two workflow one languages: chinese, english workflow one: set language to spanish, previous was: english workflow two current language: english diff --git a/nexus-messaging/src/ondemandpattern/api.ts b/nexus-messaging/src/ondemandpattern/api.ts index 5e8e74427..c67839b1b 100644 --- a/nexus-messaging/src/ondemandpattern/api.ts +++ b/nexus-messaging/src/ondemandpattern/api.ts @@ -7,7 +7,7 @@ export const CALLER_NAMESPACE = 'nexus-messaging-caller-namespace'; export const nexusRemoteGreetingService = nexus.service('NexusRemoteGreetingService', { /** - * Starts a new GreetingWorkflow with the given workflowId (async WorkflowRunOperation). + * Starts a new GreetingWorkflow with the given workflowId (async TemporalOperation). */ runFromRemote: nexus.operation(), @@ -30,6 +30,12 @@ export const nexusRemoteGreetingService = nexus.service('NexusRemoteGreetingServ * Approves (completes) the given workflow via a signal. */ approve: nexus.operation(), + + /** + * Attaches supporting information for the eventual approval, either by messaging a running + * Workflow or by creating one (Signal-with-Start). + */ + attachApprovalContext: nexus.operation(), }); export type Language = 'arabic' | 'chinese' | 'english' | 'french' | 'hindi' | 'portuguese' | 'spanish'; @@ -62,3 +68,8 @@ export type SetLanguageOutput = Language; export interface ApproveInput { userId: string; } + +export interface AttachApprovalContextInput { + userId: string; + note: string; +} diff --git a/nexus-messaging/src/ondemandpattern/caller/workflows.ts b/nexus-messaging/src/ondemandpattern/caller/workflows.ts index 4b5e91942..d91e2db1b 100644 --- a/nexus-messaging/src/ondemandpattern/caller/workflows.ts +++ b/nexus-messaging/src/ondemandpattern/caller/workflows.ts @@ -12,7 +12,19 @@ export async function callerRemoteWorkflow(): Promise { const userIdOne = 'UserId_One'; const userIdTwo = 'UserId_Two'; - // Start both remote workflows concurrently + // Attach approval context before anything has started Workflow One. Because + // attachApprovalContext is backed by Signal-with-Start on the handler, this call creates the + // Workflow and delivers the note to it. + await nexusClient.executeOperation( + 'attachApprovalContext', + { userId: userIdOne, note: 'queued for localization review by the nightly batch' }, + { scheduleToCloseTimeout: '10s' }, + ); + log.push(`attached approval context for user: ${userIdOne}`); + + // Start both remote workflows concurrently. Workflow One is already running because of the call + // above; the handler sets the conflict policy to USE_EXISTING, so that start attaches the + // operation's completion callback to the running execution instead of failing. const [handleOne, handleTwo] = await Promise.all([ nexusClient.startOperation('runFromRemote', { userId: userIdOne }, { scheduleToCloseTimeout: '60s' }), nexusClient.startOperation('runFromRemote', { userId: userIdTwo }, { scheduleToCloseTimeout: '60s' }), @@ -21,6 +33,15 @@ export async function callerRemoteWorkflow(): Promise { log.push(`started workflow one for user: ${userIdOne}`); log.push(`started workflow two for user: ${userIdTwo}`); + // Workflow Two was just created by runFromRemote, so here Signal-with-Start skips the start and + // only delivers the Signal. + await nexusClient.executeOperation( + 'attachApprovalContext', + { userId: userIdTwo, note: 'translation approved by the localization team' }, + { scheduleToCloseTimeout: '10s' }, + ); + log.push(`attached approval context to running workflow for user: ${userIdTwo}`); + // Interact with workflow one: query languages, set language to spanish const languagesOne = await nexusClient.executeOperation( 'getLanguages', diff --git a/nexus-messaging/src/ondemandpattern/service/handler.ts b/nexus-messaging/src/ondemandpattern/service/handler.ts index aae3f67c5..067007afc 100644 --- a/nexus-messaging/src/ondemandpattern/service/handler.ts +++ b/nexus-messaging/src/ondemandpattern/service/handler.ts @@ -2,6 +2,7 @@ import * as nexus from 'nexus-rpc'; import * as temporalNexus from '@temporalio/nexus'; import { ApproveInput, + AttachApprovalContextInput, GetLanguageInput, GetLanguagesInput, nexusRemoteGreetingService, @@ -9,7 +10,14 @@ import { RunFromRemoteOutput, SetLanguageInput, } from '../api'; -import { approveSignal, getLanguageQuery, getLanguagesQuery, greetingWorkflow, setLanguageUpdate } from './workflows'; +import { + approveSignal, + attachApprovalContextSignal, + getLanguageQuery, + getLanguagesQuery, + greetingWorkflow, + setLanguageUpdate, +} from './workflows'; const WORKFLOW_ID_PREFIX = 'GreetingWorkflow_for_'; @@ -18,14 +26,17 @@ function getWorkflowId(userId: string): string { } export const nexusRemoteGreetingServiceHandler = nexus.serviceHandler(nexusRemoteGreetingService, { - runFromRemote: new temporalNexus.WorkflowRunOperationHandler( - async (ctx, input: RunFromRemoteInput) => { - return await temporalNexus.startWorkflow(ctx, greetingWorkflow, { + runFromRemote: new temporalNexus.TemporalOperationHandler({ + async start(_ctx, client, input: RunFromRemoteInput) { + return await client.startWorkflow(greetingWorkflow, { args: [], workflowId: getWorkflowId(input.userId), + // attachApprovalContext may have created the GreetingWorkflow already, so attach to the + // running execution instead of failing (the default behavior). + workflowIdConflictPolicy: 'USE_EXISTING', }); }, - ), + }), getLanguages: new temporalNexus.TemporalOperationHandler({ async start(_ctx, client, input: GetLanguagesInput) { @@ -57,4 +68,17 @@ export const nexusRemoteGreetingServiceHandler = nexus.serviceHandler(nexusRemot return temporalNexus.TemporalOperationResult.sync(undefined); }, }), + + // Signals the Workflow, starting it first if it is not already running. + attachApprovalContext: new temporalNexus.TemporalOperationHandler({ + async start(_ctx, client, input: AttachApprovalContextInput) { + await client.signalWithStartWorkflow(greetingWorkflow, { + args: [], + workflowId: getWorkflowId(input.userId), + signal: attachApprovalContextSignal, + signalArgs: [input], + }); + return temporalNexus.TemporalOperationResult.sync(undefined); + }, + }), }); diff --git a/nexus-messaging/src/ondemandpattern/service/workflows.ts b/nexus-messaging/src/ondemandpattern/service/workflows.ts index d09d1de99..8493bd1e2 100644 --- a/nexus-messaging/src/ondemandpattern/service/workflows.ts +++ b/nexus-messaging/src/ondemandpattern/service/workflows.ts @@ -1,5 +1,5 @@ import * as wf from '@temporalio/workflow'; -import { Language } from '../api'; +import { AttachApprovalContextInput, Language } from '../api'; import { createActivities } from './activities'; const { callGreetingService } = wf.proxyActivities>({ @@ -11,6 +11,7 @@ export const getLanguageQuery = wf.defineQuery('getLanguage'); export const setLanguageUpdate = wf.defineUpdate('setLanguage'); export const setLanguageUsingActivityUpdate = wf.defineUpdate('setLanguageUsingActivity'); export const approveSignal = wf.defineSignal<[]>('approve'); +export const attachApprovalContextSignal = wf.defineSignal<[AttachApprovalContextInput]>('attachApprovalContext'); const INITIAL_GREETINGS: Partial> = { chinese: '你好,世界', @@ -21,6 +22,7 @@ export async function greetingWorkflow(): Promise { let language: Language = 'english'; let greetings: Partial> = { ...INITIAL_GREETINGS }; let approved = false; + let approvalContext: string | undefined; wf.setHandler(getLanguagesQuery, () => Object.keys(greetings) as Language[]); @@ -56,9 +58,15 @@ export async function greetingWorkflow(): Promise { }); wf.setHandler(approveSignal, () => { + wf.log.info('approve signal received', { approvalContext }); approved = true; }); + wf.setHandler(attachApprovalContextSignal, (input: AttachApprovalContextInput) => { + wf.log.info('attachApprovalContext signal received', { userId: input.userId, note: input.note }); + approvalContext = input.note; + }); + await wf.condition(() => approved && wf.allHandlersFinished()); return greetings[language] ?? `Hello from ${language}`;