diff --git a/exercise-1302-async-nexus/exercise-1302-README.md b/exercise-1302-async-nexus/exercise-1302-README.md new file mode 100644 index 0000000..d0a6fbc --- /dev/null +++ b/exercise-1302-async-nexus/exercise-1302-README.md @@ -0,0 +1,691 @@ +# Exercise 1302 — Async Nexus: One New Pattern, Infinite New Possibilities + +> **Bridge from Exercise 1301:** You know sync Nexus — handlers that run inline and return immediately. One new pattern changes everything: `fromWorkflowMethod()`. The compliance check now starts a real Temporal workflow. You hold the handle. Temporal does the waiting. + +--- + +## Scenario + +**The Deep Investigation** — The Compliance team upgraded their pipeline from a 2-second AI call to a 3-phase forensic investigation: + +``` +Phase 1: OFAC Screening — check sanctioned entities and countries +Phase 2: Pattern Analysis — detect structuring, layering, unusual routing +Phase 3: Final Review — senior analyst sign-off + risk scoring +``` + +Total time: **15 to 60 seconds** per transaction. The Payments team's sync Nexus calls now timeout. Transactions are being lost. The on-call engineer just opened a ticket: + +> *"Compliance is breaking us again. TXN-BETA timed out at 5 seconds but the investigation took 35. By the time compliance finished, our workflow was already dead."* + +Your job: upgrade the integration from sync to async. The Compliance handler will **start** a `ComplianceInvestigationWorkflow` instead of running inline. The Payments workflow will hold a `NexusOperationHandle` and await it durably. + +--- + +## Quickstart Docs By Temporal + +🚀 [Get started in a few mins](https://docs.temporal.io/quickstarts?utm_campaign=awareness-nikolay-advolodkin&utm_medium=code&utm_source=github) + +--- + +## Working in This Exercise + +``` +exercise-1302-async-nexus/ +├── exercise/ ← Work here. All TODO files are inside. +├── solution/ ← Reference implementation. Peek only after genuinely trying. +└── ui/ ← Interactive diagrams. Open in browser directly. + ├── async-nexus-flow.html ← Primary diagram: toggle sync/async, kill-worker demo + └── sync-vs-async.html ← Side-by-side comparison +``` + +**All checkpoint commands run from `exercise/`:** +```bash +cd exercise-1302-async-nexus/exercise +``` + +--- + +## Checkpoint 0 — See the Problem First + +Before writing a single line, run the baseline to **feel** the problem: + +**Terminal 1:** +```bash +temporal server start-dev +``` + +**Terminal 2:** +```bash +cd exercise-1302-async-nexus/exercise +mvn compile exec:java +``` + +**You will see:** +``` +╔══════════════════════════════════════════════════════════╗ +║ CHECKPOINT 0: The Sync Timeout Problem ║ +║ Payments team calling slow Compliance pipeline ║ +╚══════════════════════════════════════════════════════════╝ + + Calling compliance team for TXN-ALPHA... + Timeout: 5 seconds + Compliance pipeline takes: ~30 seconds + +[SlowComplianceAgent] Starting compliance check for TXN-ALPHA +[SlowComplianceAgent] Phase 1/3: OFAC screening (this takes a while...) + +╔══════════════════════════════════════════════════════════╗ +║ ERROR: java.util.concurrent.TimeoutException ║ +║ Transaction TXN-ALPHA: LOST ║ +║ Retries: 0 ║ +║ Audit trail: none ║ +║ On-call engineer: paged ║ +╚══════════════════════════════════════════════════════════╝ +``` + +**What this means:** A sync call to a slow pipeline is a ticking time bomb. The compliance investigation ran perfectly — it just took longer than the caller expected. Without async Nexus, the payment is permanently lost. + +> **Open the interactive diagram to see this visually:** +> `ui/async-nexus-flow.html` → click "Sync Mode (1301) — breaks" + +--- + +## Sync vs Async: The Mental Model + +### The Phone Call vs the Ticket System + +**Sync Nexus (Exercise 1301):** +> You call Compliance. They answer and you wait on hold for 2 seconds. Fine. +> But now it takes 35 seconds. You're still holding. Your boss calls. You miss it. You're fired. + +**Async Nexus (Exercise 1302):** +> You open a ticket: "Please investigate TXN-BETA." You get a ticket number. +> You go do other things. When Compliance closes the ticket, you get a notification. +> You resume where you left off — durably, automatically. + +The ticket number is the `NexusOperationHandle`. Temporal holds it. You retrieve the result when it's ready. + +### Timeline Comparison + +``` +SYNC (breaks with slow compliance) +──────────────────────────────────────────────────────────────────────── + +Payments │▓▓▓ validate │░░░░░░░░░░░ WAITING (blocked) ░░░░░│💥 TIMEOUT│ + │ │ │ │ +Compliance│ │▓▓▓ Phase 1 ▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓│ + 0s 2s 5s (crash) 30s + + +ASYNC (works — TXN-BETA at 35s, TXN-GAMMA at 60s, no problem) +──────────────────────────────────────────────────────────────────────── + +Payments │▓▓ validate │▓ start+handle │░ suspended (durable, no thread) ░│▓ execute │ + │ │ │ │ │ +Compliance│ │▓▓ Phase 1 ▓▓▓▓▓▓▓▓ Phase 2 ▓▓▓▓▓▓▓▓▓ Phase 3 ▓▓▓│ + 0s 2s 5s 35s +``` + +### The API Difference + +| | Exercise 1301 (sync) | Exercise 1302 (async) | +|---|---|---| +| **Handler** | `WorkflowClientOperationHandlers.sync(...)` | `WorkflowClientOperationHandlers.fromWorkflowMethod(...)` | +| **Handler starts?** | Runs inline | Starts a Temporal workflow | +| **Caller API** | `result = complianceService.checkCompliance(req)` | `handle = Workflow.startNexusOperation(...)` | +| **Await** | Inline (blocks) | `handle.getResult().get()` (durable suspend) | +| **Survives timeout?** | No | Yes | +| **Survives worker crash?** | No | Yes — resumes from last completed activity | +| **Visible in Temporal UI?** | Handler only | Workflow + 3 activity completions | + +--- + +## Quick Check — Before You Code + +> **3 questions before you start. No peeking. Think it through.** + +**Q1:** The compliance worker crashes mid-investigation (Phase 2 of 3). The payment workflow is holding a `NexusOperationHandle`. What happens? + +- A) The payment workflow crashes too — it was waiting +- B) The investigation restarts from Phase 1 on the next worker +- C) Temporal retries delivery; the investigation resumes from the last completed activity (Phase 3) +- D) The handle expires and the payment fails + +**Q2:** What is the difference between `complianceService.investigate(req)` and `Workflow.startNexusOperation(complianceService::investigate, req)`? + +- A) No difference — both return InvestigationResult +- B) The first is a sync call (blocks). The second starts async and gives back a handle +- C) The second runs the investigation inline but in a separate thread +- D) The first uses HTTP; the second uses gRPC + +**Q3:** In `fromWorkflowMethod(...)`, why does the lambda end with `::investigate`? + +- A) It's a method reference — tells Temporal which workflow method to invoke when starting `ComplianceInvestigationWorkflow` +- B) It filters responses to only "investigate" type operations +- C) It's the handler method name — must match the @OperationImpl method +- D) It specifies the activity to run inside the workflow + +*(Answers: C, B, A)* + +--- + +## Architecture: What You Are Building + +``` +┌─────────────────────────────────┐ Nexus ┌──────────────────────────────────────┐ +│ PAYMENTS TEAM │ Endpoint │ COMPLIANCE TEAM │ +│ task-queue: payments-process │◄──────────►│ task-queue: compliance-investigation │ +│ │ │ │ +│ PaymentProcessingWorkflowImpl │ │ ComplianceNexusServiceImpl │ +│ 1. validatePayment() │ ─── ► ─── │ fromWorkflowMethod() │ +│ 2. startNexusOperation() │ │ starts ▼ │ +│ 3. [suspended] │ │ ComplianceInvestigationWorkflowImpl │ +│ 4. handle.getResult().get() │ ◄── ─ ─── │ activity.investigate(request) │ +│ 5. executePayment() │ │ Phase 1: OFAC screening │ +│ │ │ Phase 2: Pattern analysis │ +│ PaymentsWorkerApp │ │ Phase 3: Final review │ +│ PaymentActivityImpl │ │ ComplianceWorkerApp │ +│ │ │ ComplianceInvestigationActivityImpl │ +└─────────────────────────────────┘ └──────────────────────────────────────┘ +``` + +### What Is Given vs What You Implement + +| File | Status | Est. Time | +|---|---|---| +| `PaymentActivityImpl.java` | **TODO File 1** — warm-up | 5 min | +| `ComplianceInvestigationActivityImpl.java` | **TODO File 2** — same pattern, new domain | 10 min | +| `ComplianceInvestigationWorkflowImpl.java` | **TODO File 3** — workflow orchestration | 10 min | +| `ComplianceNexusServiceImpl.java` | **TODO File 4** — THE key new concept | 20 min | +| `PaymentProcessingWorkflowImpl.java` | **TODO File 5** — using NexusOperationHandle | 20 min | +| `ComplianceWorkerApp.java` | **TODO File 6** — CRAWL extended | 15 min | +| `PaymentsWorkerApp.java` | **TODO File 7** — CRAWL familiar | 10 min | +| All interfaces, domain POJOs, business logic | **GIVEN** — do not modify | — | + +--- + +## Step 1 — `PaymentActivityImpl.java` (TODO File 1) + +**Purpose:** Warm-up. Thin wrapper that delegates to `PaymentGateway`. Identical pattern to previous exercises. + +Open `exercise/src/main/java/payments/temporal/activity/PaymentActivityImpl.java` and implement the 4 TODOs: + +1. Add `private final PaymentGateway gateway;` field +2. Add constructor that accepts `PaymentGateway` +3. `validatePayment()` → `return gateway.validatePayment(request);` +4. `executePayment()` → `return gateway.executePayment(request);` + +--- + +## Step 2 — `ComplianceInvestigationActivityImpl.java` (TODO File 2) + +**Purpose:** Same thin-wrapper pattern, new domain. Confirms you understand the pattern before tackling Nexus. + +Open `exercise/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivityImpl.java`: + +1. Add `private final ComplianceInvestigator investigator;` field +2. Add constructor that accepts `ComplianceInvestigator` +3. `investigate()` → `return investigator.runFullInvestigation(request);` + +### Checkpoint 1 — After File 2 + +```bash +mvn compile exec:java@activity-test +``` + +**Expected output:** +``` +[INV-ALPHA] Phase 1/3: OFAC screening — checking sanctioned entities... +[INV-ALPHA] Phase 1/3: OFAC screening complete. +[INV-ALPHA] Phase 2/3: Pattern analysis — detecting structuring and layering... +[INV-ALPHA] Phase 2/3: Pattern analysis complete. +[INV-ALPHA] Phase 3/3: Final review — senior analyst sign-off... +[INV-ALPHA] Phase 3/3: Final review complete. +Checkpoint 1 PASSED: ComplianceInvestigationActivityImpl delegates correctly. +``` + +**If you see `FAILED`:** Check that you're calling `investigator.runFullInvestigation(request)` and not returning `null`. + +--- + +## Step 3 — `ComplianceInvestigationWorkflowImpl.java` (TODO File 3) + +**Purpose:** Wire the investigation activity inside a workflow. The interface `ComplianceInvestigationWorkflow` is given; you implement the body. + +Open `exercise/src/main/java/compliance/temporal/ComplianceInvestigationWorkflowImpl.java`: + +**TODO 1 — Define `ACTIVITY_OPTIONS`:** +```java +private static final ActivityOptions ACTIVITY_OPTIONS = ActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofMinutes(2)) + .setHeartbeatTimeout(Duration.ofSeconds(10)) + .setRetryOptions(RetryOptions.newBuilder() + .setMaximumAttempts(3) + .setInitialInterval(Duration.ofSeconds(1)) + .setBackoffCoefficient(2.0) + .build()) + .build(); +``` + +> **Why heartbeat timeout?** `ComplianceInvestigator` calls `Activity.getExecutionContext().heartbeat()` during each phase. If the activity stops heartbeating (worker crash), Temporal detects it within 10 seconds and reschedules. This is what makes the Heist Test (Checkpoint 4) work correctly. + +**TODO 2 — Create activity stub:** +```java +private final ComplianceInvestigationActivity activity = + Workflow.newActivityStub(ComplianceInvestigationActivity.class, ACTIVITY_OPTIONS); +``` + +**TODO 3 — Implement `investigate()`:** +```java +return activity.investigate(request); +``` + +--- + +## Step 4 — `ComplianceNexusServiceImpl.java` (TODO File 4) — The Core Lesson + +**This is the key file.** The entire exercise builds to this moment. + +Open `exercise/src/main/java/compliance/temporal/ComplianceNexusServiceImpl.java`. + +### What You Did in 1301 (sync) + +```java +// Handler runs inline — works for fast operations, breaks for slow ones +return WorkflowClientOperationHandlers.sync( + (context, details, client, input) -> + complianceAgent.checkCompliance(input) +); +``` + +### What You Implement in 1302 (async) + +```java +// Handler STARTS a workflow — works for any duration +return WorkflowClientOperationHandlers.fromWorkflowMethod( + (context, details, client, input) -> + client.newWorkflowStub( + ComplianceInvestigationWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId("investigation-" + input.getTransactionId()) + .build() + )::investigate +); +``` + +**Mental model for `fromWorkflowMethod()`:** +> You are not running the workflow. You are registering **how to start it**. When a Nexus request arrives, Temporal runs your lambda to get a workflow stub + method reference, starts the workflow, and returns an operation token to the caller. The caller doesn't know (or care) that a workflow is running — they just have a handle. + +**The `::investigate` method reference — why it matters:** +This tells Temporal which `@WorkflowMethod` to invoke on `ComplianceInvestigationWorkflow`. The method name must exactly match the interface declaration. A mismatch causes a runtime error when the workflow is started. + +**Implement the 3 TODOs:** + +1. Add `@ServiceImpl(service = ComplianceNexusService.class)` to the class +2. Add `@OperationImpl` to the `investigate()` method + *(method name must be exactly `investigate` — Temporal matches by name)* +3. Return `WorkflowClientOperationHandlers.fromWorkflowMethod(...)` with: + - Workflow stub for `ComplianceInvestigationWorkflow.class` + - `workflowId = "investigation-" + input.getTransactionId()` + - Method reference: `stub::investigate` + +--- + +## Step 5 — `PaymentProcessingWorkflowImpl.java` (TODO File 5) — The New API + +**Purpose:** Use `NexusOperationHandle` on the calling side. The conceptual partner to File 4. + +Open `exercise/src/main/java/payments/temporal/PaymentProcessingWorkflowImpl.java`. + +The `ACTIVITY_OPTIONS` are already given as a starting point. You implement the stubs and the 5 orchestration steps. + +### TODO 1b — Create paymentActivity stub + +```java +private final PaymentActivity paymentActivity = + Workflow.newActivityStub(PaymentActivity.class, ACTIVITY_OPTIONS); +``` + +### TODO 1c — Create complianceService Nexus stub + +```java +private final ComplianceNexusService complianceService = Workflow.newNexusServiceStub( + ComplianceNexusService.class, + NexusServiceOptions.newBuilder() + .setOperationOptions(NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofMinutes(2)) + .build()) + .build()); +``` + +> **Where does `"compliance-endpoint"` come from?** +> Not here — the endpoint name is configured in `PaymentsWorkerApp` via `WorkflowImplementationOptions`. The workflow only knows the *service* (`ComplianceNexusService`). The *worker* knows the *endpoint*. This keeps the workflow portable. + +### TODO 2 — Validate payment + +```java +boolean valid = paymentActivity.validatePayment(request); +if (!valid) { + return new PaymentResult(false, request.getTransactionId(), + "REJECTED", null, null, null, "Validation failed"); +} +``` + +### TODO 3 — Build request and start Nexus operation + +```java +InvestigationRequest investigationReq = new InvestigationRequest( + request.getTransactionId(), request.getAmount(), + request.getSenderCountry(), request.getReceiverCountry(), + request.getDescription()); + +NexusOperationHandle handle = + Workflow.startNexusOperation(complianceService::investigate, investigationReq); +``` + +> **`startNexusOperation()` returns immediately.** No blocking. Temporal schedules the Nexus call and returns a handle. The workflow can continue doing other work — or, in this case, immediately await the result. + +### TODO 4 — Await result + +```java +InvestigationResult investigation = handle.getResult().get(); +if (investigation.isBlocked()) { + return new PaymentResult(false, request.getTransactionId(), + "BLOCKED_COMPLIANCE", investigation.getRiskLevel(), + investigation.getSummary(), null, null); +} +``` + +> **`handle.getResult().get()` is a durable await.** The workflow suspends here. No thread is blocked. If the Payments worker crashes, Temporal replays the workflow from history — re-attaches to the in-progress investigation — and waits again. The investigation is not restarted. + +### TODO 5 — Execute payment + +```java +String confirmation = paymentActivity.executePayment(request); +return new PaymentResult(true, request.getTransactionId(), + "COMPLETED", investigation.getRiskLevel(), + investigation.getSummary(), confirmation, null); +``` + +Wrap all steps in `try/catch` and return a `"FAILED"` result on unexpected error. + +--- + +## Step 6 — `ComplianceWorkerApp.java` (TODO File 6) + +**Purpose:** Register the full async compliance infrastructure. Unlike 1301's compliance worker (handler only), this worker also registers the investigation workflow and activity. + +Open `exercise/src/main/java/compliance/temporal/ComplianceWorkerApp.java`. + +Follow the CRAWL pattern (the comments in the file guide you): + +``` +C — Connect: WorkflowServiceStubs + WorkflowClient +R — Register workflow: worker.registerWorkflowImplementationTypes(ComplianceInvestigationWorkflowImpl.class) +A — Register activity: worker.registerActivitiesImplementations(new ComplianceInvestigationActivityImpl(new ComplianceInvestigator())) +W — Wire Nexus: worker.registerNexusServiceImplementation(new ComplianceNexusServiceImpl()) +L — Launch: factory.start() + startup banner +``` + +**Task queue: `"compliance-investigation"`** — must match `--target-task-queue` in the endpoint creation command below. + +### Checkpoint 2 — After Files 4 + 6 + +Create the Nexus endpoint (run once, then it persists): + +```bash +temporal operator nexus endpoint create \ + --name compliance-endpoint \ + --target-namespace default \ + --target-task-queue compliance-investigation +``` + +Start the workers: + +```bash +# Terminal 3: +mvn compile exec:java@compliance-worker + +# Terminal 4: +mvn compile exec:java@payments-worker +``` + +**Expected output (Terminal 3):** +``` +╔══════════════════════════════════════════════════════════╗ +║ Compliance Investigation Worker — ONLINE ║ +║ Task queue: compliance-investigation ║ +║ Waiting for payment investigation requests... ║ +╚══════════════════════════════════════════════════════════╝ +``` + +**Open Temporal UI** (`http://localhost:8233`) → **Nexus** tab → `compliance-endpoint` should appear with status `HEALTHY`. + +If the endpoint shows as unhealthy: verify the task queue in `ComplianceWorkerApp.TASK_QUEUE` matches `--target-task-queue` exactly. + +--- + +## Step 7 — `PaymentsWorkerApp.java` (TODO File 7) + +**Purpose:** Wire the payments side. Same CRAWL pattern as Exercise 1301. + +Open `exercise/src/main/java/payments/temporal/PaymentsWorkerApp.java`. + +Follow the CRAWL pattern: + +``` +C — Connect +R — Register PaymentProcessingWorkflowImpl with Nexus endpoint mapping: + WorkflowImplementationOptions.newBuilder() + .setNexusServiceOptions(Collections.singletonMap( + "ComplianceNexusService", + NexusServiceOptions.newBuilder() + .setEndpoint("compliance-endpoint") + .build())) + .build() +A — Register: new PaymentActivityImpl(new PaymentGateway()) +L — Launch + banner +``` + +> **The mapping key `"ComplianceNexusService"`** is the class name (no package prefix). Temporal uses this to find the correct NexusServiceOptions when the workflow calls the service. + +### Checkpoint 3 — Full End-to-End + +Start the payments worker (if not already running), then in a new terminal: + +```bash +# Terminal 5: +mvn compile exec:java@starter +``` + +**Expected output:** +``` +╔══════════════════════════════════════════════════════════╗ +║ PAYMENT STARTER — Exercise 1302: Async Nexus ║ +║ Starting 3 transactions (parallel investigations) ║ +╚══════════════════════════════════════════════════════════╝ + + Launching: payment-TXN-ALPHA ($3,500 -> Germany) + Launching: payment-TXN-BETA ($47,500 -> Cayman Islands) + Launching: payment-TXN-GAMMA ($180,000 -> Iran) + + All 3 workflows started. Waiting for results... + +╔══════════════════════════════════════════════════════════╗ +║ PAYMENT RESULTS ║ +╚══════════════════════════════════════════════════════════╝ + [✓] TXN-ALPHA | COMPLETED | Risk: LOW | Transaction approved... + [✓] TXN-BETA | COMPLETED | Risk: MEDIUM | Transaction approved with elevated... + [✗] TXN-GAMMA | BLOCKED_COMPLIANCE | Risk: CRITICAL | Transaction blocked: destination... +``` + +**In Temporal UI you should see 6 workflows:** +- `payment-TXN-ALPHA`, `payment-TXN-BETA`, `payment-TXN-GAMMA` (payment workflows) +- `investigation-TXN-ALPHA`, `investigation-TXN-BETA`, `investigation-TXN-GAMMA` (investigation workflows) + +Click any payment workflow → look for `NexusOperationScheduled` and `NexusOperationStarted` events in the event history. Click the linked workflow ID to jump to the investigation workflow. + +--- + +## Temporal UI Walkthrough + +### 1. Nexus Tab +Navigate to **Nexus** in the left sidebar. You should see `compliance-endpoint` with: +- Status: HEALTHY +- Target: `default/compliance-investigation` + +### 2. Investigation Workflow Event History +Click any `investigation-TXN-*` workflow. You'll see: +- `WorkflowExecutionStarted` — triggered by the Nexus handler +- `ActivityTaskScheduled` / `ActivityTaskStarted` / `ActivityTaskCompleted` for each phase +- `WorkflowExecutionCompleted` — carries the `InvestigationResult` + +### 3. Linked Workflows via Nexus Events +Click any `payment-TXN-*` workflow → scroll to the `NexusOperationScheduled` event → expand it. You'll find the operation token. Temporal uses this to link the two workflows across the Nexus boundary. + +### 4. TXN-GAMMA Blocked — Full Audit Trail +Even though `investigation-TXN-GAMMA` results in `blocked=true`, the investigation workflow runs all 3 phases and completes normally. The **decision** is in the data; the **execution** is always durable. This is the audit trail you couldn't get with a plain timeout. + +--- + +## The Heist Test — Checkpoint 4 + +> **Prove durability by killing the worker mid-investigation.** + +Run with timestamp-suffixed IDs (so you don't hit ALREADY_EXISTS): + +```bash +mvn compile exec:java@starter-rerun +``` + +Watch for Phase 2 of TXN-BETA: +``` +[INV-BETA] Phase 2/3: Pattern analysis — detecting structuring and layering... +``` + +**Press Ctrl+C** on the compliance worker (Terminal 3). Watch what happens: + +1. The payment starter does **not** crash — TXN-BETA is durable +2. In Temporal UI, `investigation-TXN-BETA` shows **Running** (not Failed) +3. `payment-TXN-BETA` shows **Running** (suspended at `handle.getResult().get()`) + +**Restart the compliance worker:** +```bash +# Terminal 3 (restart): +mvn compile exec:java@compliance-worker +``` + +**You will see:** +``` +[INV-BETA] Phase 3/3: Final review — senior analyst sign-off... +[INV-BETA] Phase 3/3: Final review complete. +``` + +Temporal **resumed from Phase 3** — it did not restart the investigation from Phase 1. Phase 1 and Phase 2 completion events were already written to the event log. The payment eventually completes. + +> **This is what "durable" means.** A crash mid-operation is a deployment detail, not a failure. + +> **Also try the diagram:** `ui/async-nexus-flow.html` → click "Start Animation" → click "Kill Compliance Worker" during Phase 2 → click "Restart Compliance Worker". + +--- + +## Sync vs Async Retrospective + +### Side by Side — The Handler + +**Exercise 1301 (sync):** +```java +@ServiceImpl(service = ComplianceNexusService.class) +public class ComplianceNexusServiceImpl { + private final ComplianceAgent agent; + + @OperationImpl + public OperationHandler checkCompliance() { + return WorkflowClientOperationHandlers.sync( + (context, details, client, input) -> + agent.checkCompliance(input) // ← runs inline, blocks caller + ); + } +} +``` + +**Exercise 1302 (async):** +```java +@ServiceImpl(service = ComplianceNexusService.class) +public class ComplianceNexusServiceImpl { + + @OperationImpl + public OperationHandler investigate() { + return WorkflowClientOperationHandlers.fromWorkflowMethod( + (context, details, client, input) -> + client.newWorkflowStub( + ComplianceInvestigationWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId("investigation-" + input.getTransactionId()) + .build() + )::investigate // ← starts workflow, returns token + ); + } +} +``` + +**What changed:** `sync()` → `fromWorkflowMethod()`. The caller (`PaymentProcessingWorkflowImpl`) changed from a direct call to `Workflow.startNexusOperation()` + `handle.getResult().get()`. Nothing else changed. The Nexus service *interface* is identical in structure. + +### When to Use Each + +| Use `sync()` when... | Use `fromWorkflowMethod()` when... | +|---|---| +| Operation completes in < 10 seconds | Operation takes minutes, hours, or has no SLA | +| No worker crash risk during operation | Worker restarts are normal (deploys, scaling) | +| Stateless, inline calculation | Multi-step process with checkpoints | +| No audit trail needed | Full event history required | + +--- + +## What's Next: Exercise 1300 + +Exercise 1300 builds on everything you've done here: + +| Feature | 1302 | 1300 | +|---|---|---| +| Async Nexus | ✓ | ✓ | +| Multiple NexusOperationHandles (parallel) | — | ✓ | +| Signals for human approval | — | ✓ | +| Two LLM agents (GPT + Claude) | — | ✓ | +| Next.js real-time dashboard | — | ✓ | +| 5 transactions in parallel | — | ✓ | + +--- + +## Troubleshooting + +### "Nexus endpoint not found" / ComplianceNexusService call fails +- Verify endpoint was created: `temporal operator nexus endpoint list` +- Check that `--target-task-queue compliance-investigation` matches `ComplianceWorkerApp.TASK_QUEUE` +- Check that `--name compliance-endpoint` matches the endpoint name in `PaymentsWorkerApp` + +### `OperationHandler` returns `null` for `investigate()` +- You forgot `@OperationImpl` on the method, or +- The method name in `ComplianceNexusServiceImpl` doesn't match the interface (`investigate`) + +### `WorkflowExecutionAlreadyStarted` +- Use `mvn compile exec:java@starter-rerun` for re-runs — it uses timestamp-suffixed IDs +- Or restart `temporal server start-dev` to clear state + +### Investigation never completes / workflow stuck +- Check that `ComplianceWorkerApp` registers **all three**: workflow, activity, Nexus handler +- Check heartbeat timeout: `setHeartbeatTimeout(Duration.ofSeconds(10))` in `ComplianceInvestigationWorkflowImpl` + +### `NullPointerException` in `PaymentProcessingWorkflowImpl` +- Verify activity stub field is initialized (not null) +- Verify Nexus stub field is initialized +- Check that `PaymentActivityImpl` returns real values, not `false`/`null` + +### Python version / wrong JDK +- Java 11+ required: `java -version` +- Maven: `mvn -version` + +--- + +> **You did it.** Sync Nexus was a phone call. Async Nexus is a durable ticket system — one that survives crashes, restarts, and long investigations. The callers don't block. The compliance team runs independently. And Temporal keeps the receipt. diff --git a/exercise-1302-async-nexus/exercise/pom.xml b/exercise-1302-async-nexus/exercise/pom.xml new file mode 100644 index 0000000..6986855 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/pom.xml @@ -0,0 +1,99 @@ + + + 4.0.0 + + io.temporal.warmups + exercise-1302-async-nexus-java + 1.0-SNAPSHOT + jar + + Exercise 1302 - Temporal Async Nexus (Java) + Upgrade from sync to async Nexus: one new concept, infinite new possibilities + + + 11 + 11 + UTF-8 + 1.27.0 + + + + + + io.temporal + temporal-sdk + ${temporal.version} + + + + + org.slf4j + slf4j-simple + 2.0.9 + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + 3.8.1 + + 11 + 11 + + + + + + org.codehaus.mojo + exec-maven-plugin + 3.0.0 + + + exercise.PaymentProcessingService + + + + + activity-test + + compliance.temporal.InvestigationActivityTest + + + + + compliance-worker + + compliance.temporal.ComplianceWorkerApp + + + + + payments-worker + + payments.temporal.PaymentsWorkerApp + + + + + starter + + payments.temporal.PaymentStarter + + + + + starter-rerun + + payments.temporal.PaymentStarterRerun + + + + + + + diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/ComplianceInvestigator.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/ComplianceInvestigator.java new file mode 100644 index 0000000..a1a1d34 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/ComplianceInvestigator.java @@ -0,0 +1,113 @@ +package compliance; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.temporal.activity.Activity; + +/** + * [GIVEN] The 3-phase forensic investigation logic. + * + * Phase 1: OFAC Screening — checks sanctioned entities and countries + * Phase 2: Pattern Analysis — detects structuring, layering, unusual routing + * Phase 3: Final Review — senior analyst sign-off with risk scoring + * + * Investigation durations per transaction: + * TXN-ALPHA ($3,500 US→Germany): ~15 seconds total + * TXN-BETA ($47,500 US→Caymans): ~35 seconds total + * TXN-GAMMA ($180,000 US→Iran): ~60 seconds total + * + * Students use this class as-is — focus is on wiring it into Temporal activities, + * not the investigation logic itself. + */ +public class ComplianceInvestigator { + + public InvestigationResult runFullInvestigation(InvestigationRequest request) { + String txn = request.getTransactionId(); + String prefix = "[INV-" + txn.replace("TXN-", "") + "]"; + + // Determine investigation profile based on transaction + int[] phaseDurations = getInvestigationProfile(request); + + try { + // Phase 1: OFAC Screening + System.out.println(prefix + " Phase 1/3: OFAC screening — checking sanctioned entities..."); + heartbeat("Phase 1/3: OFAC screening"); + sleep(phaseDurations[0]); + System.out.println(prefix + " Phase 1/3: OFAC screening complete."); + + // Phase 2: Pattern Analysis + System.out.println(prefix + " Phase 2/3: Pattern analysis — detecting structuring and layering..."); + heartbeat("Phase 2/3: Pattern analysis"); + sleep(phaseDurations[1]); + System.out.println(prefix + " Phase 2/3: Pattern analysis complete."); + + // Phase 3: Final Review + System.out.println(prefix + " Phase 3/3: Final review — senior analyst sign-off..."); + heartbeat("Phase 3/3: Final review"); + sleep(phaseDurations[2]); + System.out.println(prefix + " Phase 3/3: Final review complete."); + + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Investigation interrupted for " + txn, e); + } + + return buildResult(request); + } + + private int[] getInvestigationProfile(InvestigationRequest request) { + double amount = request.getAmount(); + String dest = request.getReceiverCountry(); + + // TXN-GAMMA: sanctioned country — full 60-second deep investigation + if (dest.equalsIgnoreCase("Iran") || dest.equalsIgnoreCase("North Korea") + || dest.equalsIgnoreCase("Cuba") || dest.equalsIgnoreCase("Syria")) { + return new int[]{15000, 20000, 25000}; // 60s total + } + // TXN-BETA: large international — 35-second investigation + if (amount >= 10000 && !request.getSenderCountry().equalsIgnoreCase(dest)) { + return new int[]{10000, 15000, 10000}; // 35s total + } + // TXN-ALPHA: routine — 15-second investigation + return new int[]{5000, 5000, 5000}; // 15s total + } + + private InvestigationResult buildResult(InvestigationRequest request) { + String dest = request.getReceiverCountry(); + double amount = request.getAmount(); + + // Sanctioned country — always blocked + if (dest.equalsIgnoreCase("Iran") || dest.equalsIgnoreCase("North Korea") + || dest.equalsIgnoreCase("Cuba") || dest.equalsIgnoreCase("Syria")) { + return new InvestigationResult(request.getTransactionId(), true, "CRITICAL", + "Transaction blocked: destination country is OFAC-sanctioned. " + + "All 3 investigation phases confirm: payment to " + dest + " violates regulatory requirements."); + } + + // Large international — medium risk, approved + if (amount >= 10000) { + return new InvestigationResult(request.getTransactionId(), false, "MEDIUM", + "Transaction approved with elevated monitoring. Amount $" + + String.format("%.0f", amount) + " to " + dest + + " triggers enhanced due diligence. No structuring patterns detected."); + } + + // Routine — low risk, approved + return new InvestigationResult(request.getTransactionId(), false, "LOW", + "Transaction approved. Routine transfer to " + dest + + ". All 3 investigation phases passed. No flags raised."); + } + + private void heartbeat(String phase) { + try { + // Heartbeat so Temporal knows the activity is still alive during long phases + Activity.getExecutionContext().heartbeat(phase); + } catch (Exception e) { + // Not running inside a Temporal activity (e.g., test runner) — ignore + } + } + + private void sleep(int millis) throws InterruptedException { + Thread.sleep(millis); + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/SlowComplianceAgent.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/SlowComplianceAgent.java new file mode 100644 index 0000000..2d74f07 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/SlowComplianceAgent.java @@ -0,0 +1,42 @@ +package compliance; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; + +/** + * [GIVEN] A mock compliance agent used ONLY in Checkpoint 0. + * + * This simulates what happened before the async Nexus upgrade: + * the compliance team's new 3-phase investigation pipeline takes 30+ seconds, + * but the Payments team's sync Nexus call has a 5-second timeout. + * + * Result: every transaction times out. The Payments team pages on-call. + * + * This class is NOT used in your actual implementation — it exists to + * demonstrate the problem that async Nexus solves. + * + * Run Checkpoint 0 to see the timeout: + * mvn compile exec:java + */ +public class SlowComplianceAgent { + + public InvestigationResult runFullInvestigation(InvestigationRequest request) { + System.out.println("[SlowComplianceAgent] Starting compliance pipeline for " + + request.getTransactionId()); + System.out.println("[SlowComplianceAgent] Phase 1/3: OFAC screening (this takes a while...)"); + + try { + Thread.sleep(10000); // Phase 1: 10 seconds + System.out.println("[SlowComplianceAgent] Phase 2/3: Pattern analysis..."); + Thread.sleep(10000); // Phase 2: 10 seconds + System.out.println("[SlowComplianceAgent] Phase 3/3: Final review..."); + Thread.sleep(10000); // Phase 3: 10 seconds + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + // We never get here in practice — the caller times out first + return new InvestigationResult(request.getTransactionId(), false, "LOW", + "Investigation complete (but caller already gave up)"); + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/domain/InvestigationRequest.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/domain/InvestigationRequest.java new file mode 100644 index 0000000..ca640eb --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/domain/InvestigationRequest.java @@ -0,0 +1,31 @@ +package compliance.domain; + +/** + * [GIVEN] Input to a compliance investigation. + * Sent by the Payments team via Nexus to the Compliance team. + */ +public class InvestigationRequest { + public String transactionId; + public double amount; + public String senderCountry; + public String receiverCountry; + public String description; + + public InvestigationRequest() {} + + public InvestigationRequest(String transactionId, double amount, + String senderCountry, String receiverCountry, + String description) { + this.transactionId = transactionId; + this.amount = amount; + this.senderCountry = senderCountry; + this.receiverCountry = receiverCountry; + this.description = description; + } + + public String getTransactionId() { return transactionId; } + public double getAmount() { return amount; } + public String getSenderCountry() { return senderCountry; } + public String getReceiverCountry() { return receiverCountry; } + public String getDescription() { return description; } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/domain/InvestigationResult.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/domain/InvestigationResult.java new file mode 100644 index 0000000..b6f5c2d --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/domain/InvestigationResult.java @@ -0,0 +1,38 @@ +package compliance.domain; + +/** + * [GIVEN] Result of a 3-phase compliance investigation. + * Returned by the Compliance team to the Payments team via async Nexus. + * + * Fields: + * blocked — true = block the payment, false = allow it + * riskLevel — "LOW", "MEDIUM", or "CRITICAL" + * summary — one-line summary of all 3 investigation phases + */ +public class InvestigationResult { + public String transactionId; + public boolean blocked; + public String riskLevel; // "LOW", "MEDIUM", "CRITICAL" + public String summary; + + public InvestigationResult() {} + + public InvestigationResult(String transactionId, boolean blocked, + String riskLevel, String summary) { + this.transactionId = transactionId; + this.blocked = blocked; + this.riskLevel = riskLevel; + this.summary = summary; + } + + public String getTransactionId() { return transactionId; } + public boolean isBlocked() { return blocked; } + public String getRiskLevel() { return riskLevel; } + public String getSummary() { return summary; } + + @Override + public String toString() { + return String.format("InvestigationResult{txn=%s, blocked=%s, risk=%s, summary='%s'}", + transactionId, blocked, riskLevel, summary); + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceInvestigationWorkflow.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceInvestigationWorkflow.java new file mode 100644 index 0000000..e616aa6 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceInvestigationWorkflow.java @@ -0,0 +1,26 @@ +package compliance.temporal; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +/** + * [GIVEN] Workflow interface for the compliance investigation workflow. + * Implemented by you in ComplianceInvestigationWorkflowImpl.java. + * + * This workflow runs on the Compliance team's worker. It orchestrates + * the 3-phase investigation via ComplianceInvestigationActivity. + * + * IMPORTANT: The method name here must exactly match what you use in + * ComplianceNexusServiceImpl when building the fromWorkflowMethod() handler. + * + * client.newWorkflowStub(ComplianceInvestigationWorkflow.class, ...)::investigate + * ^^^^^^^^^^^ + * must match this method name + */ +@WorkflowInterface +public interface ComplianceInvestigationWorkflow { + @WorkflowMethod + InvestigationResult investigate(InvestigationRequest request); +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceInvestigationWorkflowImpl.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceInvestigationWorkflowImpl.java new file mode 100644 index 0000000..37f7de2 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceInvestigationWorkflowImpl.java @@ -0,0 +1,58 @@ +package compliance.temporal; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import compliance.temporal.activity.ComplianceInvestigationActivity; +import io.temporal.activity.ActivityOptions; +import io.temporal.common.RetryOptions; +import io.temporal.workflow.Workflow; + +import java.time.Duration; + +/** + * ═══════════════════════════════════════════════════════════════════ + * TODO FILE 3 of 7: ComplianceInvestigationWorkflowImpl — Workflow Pattern (10 min) + * ═══════════════════════════════════════════════════════════════════ + * + * PURPOSE: Wire the 3-phase investigation activity inside a workflow. + * You're given the interface (ComplianceInvestigationWorkflow); you implement the body. + * + * WHAT TO IMPLEMENT: + * + * 1. Define ACTIVITY_OPTIONS as a static final ActivityOptions: + * - startToCloseTimeout: Duration.ofMinutes(2) + * (TXN-GAMMA investigation takes ~60s — give it room) + * - retryPolicy: maxAttempts(3), initialInterval(1s), backoffCoefficient(2) + * - heartbeatTimeout: Duration.ofSeconds(10) + * (ComplianceInvestigator calls heartbeat — set a timeout shorter than each phase) + * + * 2. Create a ComplianceInvestigationActivity stub: + * - Use Workflow.newActivityStub(ComplianceInvestigationActivity.class, ACTIVITY_OPTIONS) + * + * 3. Implement investigate(): + * - Call activity.investigate(request) + * - Return the result + * + * WHY HEARTBEAT TIMEOUT? + * ComplianceInvestigator calls Activity.getExecutionContext().heartbeat() in each phase. + * If the activity stops heartbeating (worker crash), Temporal detects it via heartbeatTimeout + * and can reschedule the activity on another worker. + * This is what makes "The Heist Test" (Checkpoint 4) work. + * + * LOGGING REMINDER: + * Use Workflow.getLogger(), never System.out.println() in workflow code. + * Workflows can replay — println fires on every replay. getLogger() does not. + */ +public class ComplianceInvestigationWorkflowImpl implements ComplianceInvestigationWorkflow { + + // TODO 1: Define ACTIVITY_OPTIONS + // Hint: startToCloseTimeout=2min, heartbeatTimeout=10s, retryOptions with maxAttempts=3 + + // TODO 2: Create activity stub using Workflow.newActivityStub(...) + + @Override + public InvestigationResult investigate(InvestigationRequest request) { + // TODO 3: call activity.investigate(request) and return the result + return null; + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceNexusServiceImpl.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceNexusServiceImpl.java new file mode 100644 index 0000000..0beaa38 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceNexusServiceImpl.java @@ -0,0 +1,83 @@ +package compliance.temporal; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.client.WorkflowOptions; +import io.temporal.nexus.WorkflowClientOperationHandlers; +import shared.nexus.ComplianceNexusService; + +/** + * ═══════════════════════════════════════════════════════════════════ + * TODO FILE 4 of 7: ComplianceNexusServiceImpl — Core New Concept (20 min) + * ═══════════════════════════════════════════════════════════════════ + * + * PURPOSE: THE KEY LESSON of this exercise. Upgrade from sync to async Nexus + * by starting a workflow instead of running inline. + * + * WHAT YOU DID IN 1301 (sync — runs inline): + * ┌─────────────────────────────────────────────────────────────────┐ + * │ return WorkflowClientOperationHandlers.sync( │ + * │ (context, details, client, input) -> │ + * │ complianceAgent.checkCompliance(input) │ + * │ ); │ + * └─────────────────────────────────────────────────────────────────┘ + * The handler runs, returns immediately. Caller waits synchronously. + * + * WHAT YOU IMPLEMENT IN 1302 (async — starts a workflow): + * ┌─────────────────────────────────────────────────────────────────┐ + * │ return WorkflowClientOperationHandlers.fromWorkflowMethod( │ + * │ (context, details, client, input) -> │ + * │ client.newWorkflowStub( │ + * │ ComplianceInvestigationWorkflow.class, │ + * │ WorkflowOptions.newBuilder() │ + * │ .setWorkflowId("investigation-" │ + * │ + input.getTransactionId()) │ + * │ .build() │ + * │ )::investigate │ + * │ ); │ + * └─────────────────────────────────────────────────────────────────┘ + * The handler starts a workflow and returns an operation token. + * The caller (PaymentProcessingWorkflow) holds a NexusOperationHandle + * and calls handle.getResult().get() when it's ready to collect the result. + * + * WHAT TO IMPLEMENT: + * + * 1. Add @ServiceImpl(service = ComplianceNexusService.class) to the class + * + * 2. Add @OperationImpl to the investigate() method + * NOTE: The method name must exactly match the interface — "investigate" + * Temporal matches handler methods to @Operation interface methods by name. + * + * 3. Return WorkflowClientOperationHandlers.fromWorkflowMethod(...) with: + * - Create a workflow stub: client.newWorkflowStub(ComplianceInvestigationWorkflow.class, options) + * - WorkflowId: "investigation-" + input.getTransactionId() + * - Method reference: stub::investigate + * + * MENTAL MODEL: + * fromWorkflowMethod() doesn't run your workflow. It registers how to START it. + * When Temporal receives a Nexus request, it: + * 1. Runs your lambda to get the workflow stub + method reference + * 2. Starts the workflow + * 3. Returns an operation token to the caller + * The caller polls for completion using the token (handled by Temporal automatically). + * + * CHECKPOINT 2 — After this file + ComplianceWorkerApp: + * Start both workers and check: Temporal UI → Nexus tab → compliance-endpoint = HEALTHY + */ +// TODO 1: Add @ServiceImpl annotation +public class ComplianceNexusServiceImpl { + + // TODO 2: Add @OperationImpl annotation below + // NOTE: method name must match the interface — "investigate" + public OperationHandler investigate() { + // TODO 3: Return WorkflowClientOperationHandlers.fromWorkflowMethod(...) + // Lambda: (context, details, client, input) -> ... + // Create workflow stub for ComplianceInvestigationWorkflow + // with workflowId = "investigation-" + input.getTransactionId() + // Return stub::investigate + return null; + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceWorkerApp.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceWorkerApp.java new file mode 100644 index 0000000..625e26a --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/ComplianceWorkerApp.java @@ -0,0 +1,97 @@ +package compliance.temporal; + +import compliance.ComplianceInvestigator; +import compliance.temporal.activity.ComplianceInvestigationActivityImpl; +import io.temporal.client.WorkflowClient; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; + +/** + * ═══════════════════════════════════════════════════════════════════ + * TODO FILE 6 of 7: ComplianceWorkerApp — CRAWL Extended (15 min) + * ═══════════════════════════════════════════════════════════════════ + * + * PURPOSE: Register the full async compliance infrastructure. + * Unlike 1301's compliance worker (Nexus handler only), this worker + * ALSO registers the investigation workflow and activity. + * + * WHY THE DIFFERENCE? + * In 1301: The compliance logic ran INLINE inside the handler. + * Only the Nexus handler needed registration. + * + * In 1302: The compliance logic runs as a REAL WORKFLOW. + * The handler just starts the workflow. + * The actual work (3 phases) happens in ComplianceInvestigationWorkflowImpl, + * which calls ComplianceInvestigationActivityImpl. + * So this worker must register ALL THREE. + * + * CRAWL PATTERN — Extended: + * + * C — Connect: + * WorkflowServiceStubs.newLocalServiceStubs() + * WorkflowClient.newInstance(service) + * + * R — Register workflow: ← NEW vs 1301 + * worker.registerWorkflowImplementationTypes(ComplianceInvestigationWorkflowImpl.class) + * + * A — Register activities: ← NEW vs 1301 + * worker.registerActivitiesImplementations( + * new ComplianceInvestigationActivityImpl(new ComplianceInvestigator())) + * + * W — Wire Nexus handler: + * worker.registerNexusServiceImplementation(new ComplianceNexusServiceImpl()) + * + * L — Launch: + * factory.start() + * + * TASK QUEUE: "compliance-investigation" + * This MUST match --target-task-queue when you created the Nexus endpoint: + * temporal operator nexus endpoint create \ + * --name compliance-endpoint \ + * --target-namespace default \ + * --target-task-queue compliance-investigation + * + * If you used a different queue for the Nexus endpoint, Nexus calls will never + * reach this worker. The string must be identical. + * + * STARTUP BANNER: + * After factory.start(), print a banner like: + * System.out.println("Compliance Investigation Worker started on compliance-investigation"); + * System.out.println("Waiting for Nexus requests from Payments team..."); + * + * WHAT TO IMPLEMENT: + * Follow the CRAWL steps above, in order. + */ +public class ComplianceWorkerApp { + + static final String TASK_QUEUE = "compliance-investigation"; + + public static void main(String[] args) { + // TODO: C — Connect to Temporal + // WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); + // WorkflowClient client = WorkflowClient.newInstance(service); + + // TODO: Create WorkerFactory and Worker on TASK_QUEUE ("compliance-investigation") + // WorkerFactory factory = WorkerFactory.newInstance(client); + // Worker worker = factory.newWorker(TASK_QUEUE); + + // TODO: R — Register ComplianceInvestigationWorkflowImpl + // worker.registerWorkflowImplementationTypes(ComplianceInvestigationWorkflowImpl.class); + + // TODO: A — Register ComplianceInvestigationActivityImpl + // Inject a new ComplianceInvestigator + // worker.registerActivitiesImplementations( + // new ComplianceInvestigationActivityImpl(/* new ComplianceInvestigator() */ )); + + // TODO: W — Register the Nexus handler + // worker.registerNexusServiceImplementation(new ComplianceNexusServiceImpl()); + + // TODO: L — Start the factory + // factory.start(); + + // TODO: Print startup banner + System.out.println("ComplianceWorkerApp: not implemented yet."); + System.out.println("Implement the TODO steps above."); + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/InvestigationActivityTest.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/InvestigationActivityTest.java new file mode 100644 index 0000000..a789ed8 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/InvestigationActivityTest.java @@ -0,0 +1,85 @@ +package compliance.temporal; + +import compliance.ComplianceInvestigator; +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import compliance.temporal.activity.ComplianceInvestigationActivityImpl; + +/** + * [GIVEN] Checkpoint 1 runner — tests your ComplianceInvestigationActivityImpl. + * + * This is NOT a JUnit test. It's a simple main class that calls your + * activity implementation directly (without Temporal), so you can verify + * your delegation code is correct before wiring it into the full system. + * + * Expected output: + * [INV-ALPHA] Phase 1/3: OFAC screening — checking sanctioned entities... + * [INV-ALPHA] Phase 1/3: OFAC screening complete. + * [INV-ALPHA] Phase 2/3: Pattern analysis — detecting structuring and layering... + * ... + * Checkpoint 1 PASSED: ComplianceInvestigationActivityImpl delegates correctly. + * + * Run with: mvn compile exec:java@activity-test + */ +public class InvestigationActivityTest { + + public static void main(String[] args) { + System.out.println("═══════════════════════════════════════════════════"); + System.out.println(" Checkpoint 1: Testing ComplianceInvestigationActivityImpl"); + System.out.println(" (Running investigation inline — no Temporal needed)"); + System.out.println("═══════════════════════════════════════════════════\n"); + + ComplianceInvestigator investigator = new ComplianceInvestigator(); + // Uses reflection to support both: stub (no-arg) and solution (with-arg) constructors + ComplianceInvestigationActivityImpl activity; + try { + activity = ComplianceInvestigationActivityImpl.class + .getConstructor(ComplianceInvestigator.class) + .newInstance(investigator); + } catch (NoSuchMethodException e) { + System.out.println(" *** Constructor TODO not yet implemented — using no-arg fallback ***"); + System.out.println(" *** Implement the constructor in ComplianceInvestigationActivityImpl ***\n"); + try { + activity = ComplianceInvestigationActivityImpl.class + .getConstructor() + .newInstance(); + } catch (Exception ex) { + System.out.println(" Cannot instantiate ComplianceInvestigationActivityImpl: " + ex.getMessage()); + System.exit(1); + return; + } + } catch (Exception e) { + System.out.println(" Error instantiating activity: " + e.getMessage()); + System.exit(1); + return; + } + + InvestigationRequest request = new InvestigationRequest( + "TXN-ALPHA", 3500.00, "US", "Germany", + "Equipment purchase from European supplier"); + + System.out.println("Running investigation for TXN-ALPHA (expect ~15 seconds)...\n"); + + long start = System.currentTimeMillis(); + InvestigationResult result = activity.investigate(request); + long elapsed = (System.currentTimeMillis() - start) / 1000; + + System.out.println("\n═══════════════════════════════════════════════════"); + System.out.println(" RESULT:"); + System.out.println(" Transaction: " + result.getTransactionId()); + System.out.println(" Blocked: " + result.isBlocked()); + System.out.println(" Risk Level: " + result.getRiskLevel()); + System.out.println(" Summary: " + result.getSummary()); + System.out.println(" Time: " + elapsed + " seconds"); + + if (result.getTransactionId() != null && !result.isBlocked() && "LOW".equals(result.getRiskLevel())) { + System.out.println("\n Checkpoint 1 PASSED: ComplianceInvestigationActivityImpl delegates correctly."); + } else if (result.getTransactionId() == null) { + System.out.println("\n Checkpoint 1 FAILED: result is null — did you implement the activity?"); + System.exit(1); + } else { + System.out.println("\n Checkpoint 1 PASSED: Activity delegates to ComplianceInvestigator."); + } + System.out.println("═══════════════════════════════════════════════════\n"); + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivity.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivity.java new file mode 100644 index 0000000..bab206b --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivity.java @@ -0,0 +1,19 @@ +package compliance.temporal.activity; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; + +/** + * [GIVEN] Activity interface for the 3-phase compliance investigation. + * Implemented by you in ComplianceInvestigationActivityImpl.java. + * + * This activity runs on the Compliance team's worker. It delegates + * to ComplianceInvestigator which performs the actual 3-phase logic. + */ +@ActivityInterface +public interface ComplianceInvestigationActivity { + @ActivityMethod + InvestigationResult investigate(InvestigationRequest request); +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivityImpl.java b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivityImpl.java new file mode 100644 index 0000000..a3e0d6d --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivityImpl.java @@ -0,0 +1,52 @@ +package compliance.temporal.activity; + +import compliance.ComplianceInvestigator; +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; + +/** + * ═══════════════════════════════════════════════════════════════════ + * TODO FILE 2 of 7: ComplianceInvestigationActivityImpl — Confidence Builder (10 min) + * ═══════════════════════════════════════════════════════════════════ + * + * PURPOSE: Apply the same thin-wrapper pattern to a new domain. + * After this file, you'll have confirmed the activity pattern works + * before tackling the more complex Nexus files. + * + * WHAT TO IMPLEMENT: + * + * 1. Add a private field: ComplianceInvestigator investigator + * + * 2. Add a constructor that accepts a ComplianceInvestigator and assigns it. + * + * 3. Implement investigate(): + * → return investigator.runFullInvestigation(request) + * + * CHECKPOINT 1 — After this file: + * Run: mvn compile exec:java@activity-test + * + * Expected output: + * [INV-ALPHA] Phase 1/3: OFAC screening — checking sanctioned entities... + * [INV-ALPHA] Phase 2/3: Pattern analysis — detecting structuring and layering... + * [INV-ALPHA] Phase 3/3: Final review — senior analyst sign-off... + * Checkpoint 1 PASSED: ComplianceInvestigationActivityImpl delegates correctly. + * + * If you see "Checkpoint 1 FAILED" — check that you're calling + * investigator.runFullInvestigation(request), not returning null. + */ +public class ComplianceInvestigationActivityImpl implements ComplianceInvestigationActivity { + + // TODO 1: Add private ComplianceInvestigator field + + // TODO 2: Add constructor that accepts ComplianceInvestigator + // (The no-arg constructor below is a temporary placeholder so the project compiles. + // Replace it with your parameterized constructor once you add the field.) + public ComplianceInvestigationActivityImpl() { + // Placeholder — replace with: public ComplianceInvestigationActivityImpl(ComplianceInvestigator investigator) + } + + @Override + public InvestigationResult investigate(InvestigationRequest request) { + return null; // TODO 3: delegate to investigator.runFullInvestigation(request) + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/exercise/PaymentProcessingService.java b/exercise-1302-async-nexus/exercise/src/main/java/exercise/PaymentProcessingService.java new file mode 100644 index 0000000..8b44aa5 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/exercise/PaymentProcessingService.java @@ -0,0 +1,79 @@ +package exercise; + +import compliance.SlowComplianceAgent; +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; + +import java.util.concurrent.*; + +/** + * [GIVEN] Checkpoint 0: The Problem — sync call to a slow compliance pipeline. + * + * The Compliance team upgraded their AI pipeline from a 2-second check + * to a 3-phase forensic investigation (OFAC → pattern analysis → final review). + * Total time: 30 seconds. But the Payments team's SLA is 5 seconds. + * + * This class demonstrates what happens without async Nexus: + * - Sync call with a 5-second timeout + * - SlowComplianceAgent takes 30 seconds + * - Every transaction times out + * - No retry. No audit trail. Transaction is lost. + * + * Run this first, BEFORE implementing anything: + * mvn compile exec:java + * + * You should see a timeout exception. That's the problem you're solving. + */ +public class PaymentProcessingService { + + private static final int COMPLIANCE_TIMEOUT_SECONDS = 5; + + public static void main(String[] args) { + System.out.println("╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ CHECKPOINT 0: The Sync Timeout Problem ║"); + System.out.println("║ Payments team calling slow Compliance pipeline ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝\n"); + + SlowComplianceAgent agent = new SlowComplianceAgent(); + InvestigationRequest request = new InvestigationRequest( + "TXN-ALPHA", 3500.00, "US", "Germany", + "Equipment purchase from European supplier"); + + System.out.println(" Calling compliance team for TXN-ALPHA..."); + System.out.println(" Timeout: " + COMPLIANCE_TIMEOUT_SECONDS + " seconds"); + System.out.println(" Compliance pipeline takes: ~30 seconds"); + System.out.println(); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + Future future = executor.submit( + () -> agent.runFullInvestigation(request)); + + try { + InvestigationResult result = future.get(COMPLIANCE_TIMEOUT_SECONDS, TimeUnit.SECONDS); + // We never reach here + System.out.println(" Result: " + result); + + } catch (TimeoutException e) { + future.cancel(true); + System.out.println(); + System.out.println("╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ ERROR: java.util.concurrent.TimeoutException ║"); + System.out.println("║ Compliance check exceeded " + COMPLIANCE_TIMEOUT_SECONDS + "-second SLA ║"); + System.out.println("║ ║"); + System.out.println("║ Transaction TXN-ALPHA: LOST ║"); + System.out.println("║ Retries: 0 ║"); + System.out.println("║ Audit trail: none ║"); + System.out.println("║ On-call engineer: paged ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝\n"); + System.out.println(" This is what async Nexus solves."); + System.out.println(" Instead of blocking for 30 seconds, Temporal holds the handle"); + System.out.println(" and resumes the Payments workflow when the investigation completes."); + System.out.println(" Read the README to see how."); + + } catch (Exception e) { + System.out.println(" Unexpected error: " + e.getMessage()); + } finally { + executor.shutdownNow(); + } + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/PaymentGateway.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/PaymentGateway.java new file mode 100644 index 0000000..6defe82 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/PaymentGateway.java @@ -0,0 +1,48 @@ +package payments; + +import payments.domain.PaymentRequest; + +/** + * [GIVEN] Simulated payment gateway for executing transactions. + * In production this would call Stripe, PayPal, SWIFT, etc. + * + * Students use this class as-is — focus is on Temporal integration, not payment logic. + */ +public class PaymentGateway { + + public boolean validatePayment(PaymentRequest request) { + if (request.getAmount() <= 0) { + System.out.println("[PaymentGateway] REJECTED: Invalid amount for " + request.getTransactionId()); + return false; + } + if (request.getSenderAccount() == null || request.getReceiverAccount() == null) { + System.out.println("[PaymentGateway] REJECTED: Missing account info for " + request.getTransactionId()); + return false; + } + System.out.println("[PaymentGateway] Validation passed for " + request.getTransactionId()); + return true; + } + + public String executePayment(PaymentRequest request) { + System.out.println("[PaymentGateway] Processing " + request.getTransactionId() + + " | $" + String.format("%.2f", request.getAmount()) + + " | " + request.getSenderCountry() + " -> " + request.getReceiverCountry()); + + // Simulate processing time + try { + Thread.sleep(300 + (long) (Math.random() * 300)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + // Simulate occasional gateway failures (10% chance) — Temporal retries automatically + if (Math.random() < 0.10) { + throw new RuntimeException("Payment gateway timeout for " + request.getTransactionId() + + " — connection to banking network failed"); + } + + String confirmationNumber = "CONF-" + request.getTransactionId() + "-" + System.currentTimeMillis(); + System.out.println("[PaymentGateway] Payment executed: " + confirmationNumber); + return confirmationNumber; + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/Shared.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/Shared.java new file mode 100644 index 0000000..51d03e0 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/Shared.java @@ -0,0 +1,5 @@ +package payments; + +public interface Shared { + String TASK_QUEUE = "payments-processing"; +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/domain/PaymentRequest.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/domain/PaymentRequest.java new file mode 100644 index 0000000..6bbcc4e --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/domain/PaymentRequest.java @@ -0,0 +1,39 @@ +package payments.domain; + +/** + * [GIVEN] A payment transaction to be processed. + */ +public class PaymentRequest { + public String transactionId; + public double amount; + public String currency; + public String senderCountry; + public String receiverCountry; + public String description; + public String senderAccount; + public String receiverAccount; + + public PaymentRequest() {} + + public PaymentRequest(String transactionId, double amount, String currency, + String senderCountry, String receiverCountry, + String description, String senderAccount, String receiverAccount) { + this.transactionId = transactionId; + this.amount = amount; + this.currency = currency; + this.senderCountry = senderCountry; + this.receiverCountry = receiverCountry; + this.description = description; + this.senderAccount = senderAccount; + this.receiverAccount = receiverAccount; + } + + public String getTransactionId() { return transactionId; } + public double getAmount() { return amount; } + public String getCurrency() { return currency; } + public String getSenderCountry() { return senderCountry; } + public String getReceiverCountry() { return receiverCountry; } + public String getDescription() { return description; } + public String getSenderAccount() { return senderAccount; } + public String getReceiverAccount() { return receiverAccount; } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/domain/PaymentResult.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/domain/PaymentResult.java new file mode 100644 index 0000000..99c1cb7 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/domain/PaymentResult.java @@ -0,0 +1,48 @@ +package payments.domain; + +/** + * [GIVEN] Result of a payment workflow execution. + * + * status values: + * "COMPLETED" — payment processed successfully + * "REJECTED" — failed payment validation + * "BLOCKED_COMPLIANCE" — compliance investigation returned blocked=true + * "FAILED" — unexpected error + */ +public class PaymentResult { + public boolean success; + public String transactionId; + public String status; + public String riskLevel; // from compliance investigation: "LOW", "MEDIUM", "CRITICAL" + public String summary; // from compliance investigation AI summary + public String confirmationNumber; // set when status = COMPLETED + public String error; + + public PaymentResult() {} + + public PaymentResult(boolean success, String transactionId, String status, + String riskLevel, String summary, + String confirmationNumber, String error) { + this.success = success; + this.transactionId = transactionId; + this.status = status; + this.riskLevel = riskLevel; + this.summary = summary; + this.confirmationNumber = confirmationNumber; + this.error = error; + } + + public boolean isSuccess() { return success; } + public String getTransactionId() { return transactionId; } + public String getStatus() { return status; } + public String getRiskLevel() { return riskLevel; } + public String getSummary() { return summary; } + public String getConfirmationNumber() { return confirmationNumber; } + public String getError() { return error; } + + @Override + public String toString() { + return String.format("PaymentResult{txn=%s, status=%s, risk=%s, confirm=%s, error=%s}", + transactionId, status, riskLevel, confirmationNumber, error); + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentProcessingWorkflow.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentProcessingWorkflow.java new file mode 100644 index 0000000..ea31733 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentProcessingWorkflow.java @@ -0,0 +1,16 @@ +package payments.temporal; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import payments.domain.PaymentRequest; +import payments.domain.PaymentResult; + +/** + * [GIVEN] Workflow interface for the payment processing workflow. + * Implemented by you in PaymentProcessingWorkflowImpl.java. + */ +@WorkflowInterface +public interface PaymentProcessingWorkflow { + @WorkflowMethod + PaymentResult processPayment(PaymentRequest request); +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentProcessingWorkflowImpl.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentProcessingWorkflowImpl.java new file mode 100644 index 0000000..a09c3fe --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentProcessingWorkflowImpl.java @@ -0,0 +1,130 @@ +package payments.temporal; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.temporal.activity.ActivityOptions; +import io.temporal.common.RetryOptions; +import io.temporal.workflow.NexusOperationHandle; +import io.temporal.workflow.NexusOperationOptions; +import io.temporal.workflow.NexusServiceOptions; +import io.temporal.workflow.Workflow; +import payments.domain.PaymentRequest; +import payments.domain.PaymentResult; +import payments.temporal.activity.PaymentActivity; +import shared.nexus.ComplianceNexusService; + +import java.time.Duration; + +/** + * ═══════════════════════════════════════════════════════════════════ + * TODO FILE 5 of 7: PaymentProcessingWorkflowImpl — New API (20 min) + * ═══════════════════════════════════════════════════════════════════ + * + * PURPOSE: Use NexusOperationHandle on the calling side — the conceptual + * partner to ComplianceNexusServiceImpl. Where File 4 starts the workflow, + * this file holds the handle and awaits the result. + * + * ─── THE KEY API CHANGE FROM 1301 ──────────────────────────────── + * + * In Exercise 1301 (sync): + * ComplianceResult result = complianceService.checkCompliance(compReq); + * // Blocks until compliance returns (fast in 1301, but breaks with slow pipelines) + * + * In Exercise 1302 (async): + * NexusOperationHandle handle = + * Workflow.startNexusOperation(complianceService::investigate, investigationReq); + * // Returns immediately — Temporal holds the handle; compliance runs independently + * InvestigationResult result = handle.getResult().get(); + * // Durable await — workflow suspends here until investigation completes + * + * TWO LINES instead of one. But the difference is enormous: + * - No thread is blocked waiting + * - If your worker crashes between startNexusOperation and getResult().get(), + * Temporal replays the workflow and re-attaches to the running investigation + * - You can start multiple handles in parallel (see Exercise 1300) + * + * ─── ORCHESTRATION STEPS ───────────────────────────────────────── + * + * TODO 1 — Define ACTIVITY_OPTIONS and stubs: + * - ACTIVITY_OPTIONS: startToCloseTimeout=30s, retryOptions maxAttempts=3 + * - paymentActivity stub: Workflow.newActivityStub(PaymentActivity.class, ...) + * - complianceService stub: Workflow.newNexusServiceStub(ComplianceNexusService.class, ...) + * with NexusServiceOptions → NexusOperationOptions → scheduleToCloseTimeout=2min + * + * TODO 2 — Validate payment: + * boolean valid = paymentActivity.validatePayment(request); + * if (!valid) return new PaymentResult(false, request.getTransactionId(), + * "REJECTED", null, null, null, "Validation failed"); + * + * TODO 3 — Build InvestigationRequest and start Nexus operation: + * InvestigationRequest investigationReq = new InvestigationRequest( + * request.getTransactionId(), request.getAmount(), + * request.getSenderCountry(), request.getReceiverCountry(), + * request.getDescription()); + * NexusOperationHandle handle = + * Workflow.startNexusOperation(complianceService::investigate, investigationReq); + * + * TODO 4 — Await the investigation result: + * InvestigationResult investigation = handle.getResult().get(); + * if (investigation.isBlocked()) { + * return new PaymentResult(false, request.getTransactionId(), + * "BLOCKED_COMPLIANCE", investigation.getRiskLevel(), + * investigation.getSummary(), null, null); + * } + * + * TODO 5 — Execute payment and return COMPLETED: + * String confirmation = paymentActivity.executePayment(request); + * return new PaymentResult(true, request.getTransactionId(), + * "COMPLETED", investigation.getRiskLevel(), + * investigation.getSummary(), confirmation, null); + * + * WRAP ALL STEPS in try/catch — return a FAILED result on unexpected error. + * + * LOGGING: Use Workflow.getLogger(PaymentProcessingWorkflowImpl.class), not System.out.println(). + */ +public class PaymentProcessingWorkflowImpl implements PaymentProcessingWorkflow { + + // TODO 1a: Define ACTIVITY_OPTIONS (startToCloseTimeout=30s, retryOptions) + private static final ActivityOptions ACTIVITY_OPTIONS = ActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(30)) + .setRetryOptions(RetryOptions.newBuilder() + .setMaximumAttempts(3) + .setInitialInterval(Duration.ofSeconds(1)) + .setBackoffCoefficient(2.0) + .build()) + .build(); + + // TODO 1b: Create paymentActivity stub using Workflow.newActivityStub(...) + // Hint: Workflow.newActivityStub(PaymentActivity.class, ACTIVITY_OPTIONS) + + // TODO 1c: Create complianceService Nexus stub using Workflow.newNexusServiceStub(...) + // Hint: Workflow.newNexusServiceStub(ComplianceNexusService.class, + // NexusServiceOptions.newBuilder() + // .setOperationOptions(NexusOperationOptions.newBuilder() + // .setScheduleToCloseTimeout(Duration.ofMinutes(2)) + // .build()) + // .build()) + + @Override + public PaymentResult processPayment(PaymentRequest request) { + try { + // TODO 2: Validate payment — return REJECTED if false + + // TODO 3: Build InvestigationRequest and call Workflow.startNexusOperation(...) + // Store the returned NexusOperationHandle + + // TODO 4: Call handle.getResult().get() to await result + // Return BLOCKED_COMPLIANCE if investigation.isBlocked() + + // TODO 5: Execute payment and return COMPLETED + + return null; // Remove this line once you implement the steps above + + } catch (Exception e) { + Workflow.getLogger(PaymentProcessingWorkflowImpl.class) + .error("Workflow failed for " + request.getTransactionId() + ": " + e.getMessage()); + return new PaymentResult(false, request.getTransactionId(), "FAILED", + null, null, null, e.getMessage()); + } + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentStarter.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentStarter.java new file mode 100644 index 0000000..a2e2ecc --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentStarter.java @@ -0,0 +1,92 @@ +package payments.temporal; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.serviceclient.WorkflowServiceStubs; +import payments.Shared; +import payments.domain.PaymentRequest; +import payments.domain.PaymentResult; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +/** + * [GIVEN] Starts 3 payment workflows in parallel and waits for all results. + * + * Transactions: + * TXN-ALPHA: $3,500 US → Germany ~15s investigation APPROVED / LOW + * TXN-BETA: $47,500 US → Caymans ~35s investigation APPROVED / MEDIUM + * TXN-GAMMA: $180,000 US → Iran ~60s investigation BLOCKED / CRITICAL + * + * Business ID workflow IDs: + * "payment-TXN-ALPHA", "payment-TXN-BETA", "payment-TXN-GAMMA" + * + * Runs workflows in parallel using WorkflowClient.execute(). + * All 3 investigations start at the same time. + * Total time ≈ longest investigation (~60s for TXN-GAMMA) rather than 110s sequential. + */ +public class PaymentStarter { + + public static void main(String[] args) throws Exception { + System.out.println("╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ PAYMENT STARTER — Exercise 1302: Async Nexus ║"); + System.out.println("║ Starting 3 transactions (parallel investigations) ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝\n"); + + WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); + WorkflowClient client = WorkflowClient.newInstance(service); + + PaymentRequest[] transactions = { + new PaymentRequest("TXN-ALPHA", 3500.00, "USD", "US", "Germany", + "Equipment purchase from European supplier", "ACC-001", "ACC-002"), + new PaymentRequest("TXN-BETA", 47500.00, "USD", "US", "Cayman Islands", + "Investment fund transfer to offshore account", "ACC-003", "ACC-004"), + new PaymentRequest("TXN-GAMMA", 180000.00, "USD", "US", "Iran", + "Technology export to Middle East partner", "ACC-005", "ACC-006"), + }; + + List> futures = new ArrayList<>(); + + for (PaymentRequest txn : transactions) { + String workflowId = "payment-" + txn.getTransactionId(); + System.out.println(" Launching: " + workflowId + + " ($" + String.format("%.0f", txn.getAmount()) + " -> " + txn.getReceiverCountry() + ")"); + + PaymentProcessingWorkflow workflow = client.newWorkflowStub( + PaymentProcessingWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(Shared.TASK_QUEUE) + .setWorkflowId(workflowId) + .build()); + + // WorkflowClient.execute() returns immediately — workflow runs in background + CompletableFuture future = + WorkflowClient.execute(workflow::processPayment, txn); + futures.add(future); + } + + System.out.println("\n All 3 workflows started. Waiting for results..."); + System.out.println(" (Investigations run in parallel — check Temporal UI!)\n"); + + // Wait for all workflows to complete + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); + + System.out.println("\n╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ PAYMENT RESULTS ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝"); + + for (int i = 0; i < futures.size(); i++) { + PaymentResult result = futures.get(i).get(); + String icon = "COMPLETED".equals(result.getStatus()) ? "✓" : + "BLOCKED_COMPLIANCE".equals(result.getStatus()) ? "✗" : "!"; + System.out.printf(" [%s] %-10s | %-20s | Risk: %-8s | %s%n", + icon, + result.getTransactionId(), + result.getStatus(), + result.getRiskLevel() != null ? result.getRiskLevel() : "N/A", + result.getSummary() != null ? result.getSummary().substring(0, Math.min(60, result.getSummary().length())) + "..." : ""); + } + System.out.println(); + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentStarterRerun.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentStarterRerun.java new file mode 100644 index 0000000..50ac20a --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentStarterRerun.java @@ -0,0 +1,75 @@ +package payments.temporal; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.serviceclient.WorkflowServiceStubs; +import payments.Shared; +import payments.domain.PaymentRequest; +import payments.domain.PaymentResult; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +/** + * [GIVEN] Same as PaymentStarter but uses timestamp-suffixed workflow IDs. + * Used for Checkpoint 4 (The Heist Test) — allows you to re-run without + * ALREADY_EXISTS errors. + * + * Usage: mvn compile exec:java@starter-rerun + */ +public class PaymentStarterRerun { + + public static void main(String[] args) throws Exception { + System.out.println("╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ PAYMENT STARTER RERUN — Exercise 1302: Heist Test ║"); + System.out.println("║ Starting 3 transactions (timestamp-suffixed IDs) ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝\n"); + + WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); + WorkflowClient client = WorkflowClient.newInstance(service); + + long ts = System.currentTimeMillis(); + + PaymentRequest[] transactions = { + new PaymentRequest("TXN-ALPHA-" + ts, 3500.00, "USD", "US", "Germany", + "Equipment purchase from European supplier", "ACC-001", "ACC-002"), + new PaymentRequest("TXN-BETA-" + ts, 47500.00, "USD", "US", "Cayman Islands", + "Investment fund transfer to offshore account", "ACC-003", "ACC-004"), + new PaymentRequest("TXN-GAMMA-" + ts, 180000.00, "USD", "US", "Iran", + "Technology export to Middle East partner", "ACC-005", "ACC-006"), + }; + + List> futures = new ArrayList<>(); + + for (PaymentRequest txn : transactions) { + String workflowId = "payment-" + txn.getTransactionId(); + System.out.println(" Launching: " + workflowId); + + PaymentProcessingWorkflow workflow = client.newWorkflowStub( + PaymentProcessingWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(Shared.TASK_QUEUE) + .setWorkflowId(workflowId) + .build()); + + CompletableFuture future = + WorkflowClient.execute(workflow::processPayment, txn); + futures.add(future); + } + + System.out.println("\n All 3 workflows started. Kill the compliance worker during TXN-BETA!"); + System.out.println(" Watch it resume from Phase 3 after restart.\n"); + + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); + + System.out.println("\n═══ RESULTS ═══"); + for (int i = 0; i < futures.size(); i++) { + PaymentResult result = futures.get(i).get(); + System.out.printf(" %-25s | %-20s | Risk: %s%n", + result.getTransactionId(), + result.getStatus(), + result.getRiskLevel() != null ? result.getRiskLevel() : "N/A"); + } + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentsWorkerApp.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentsWorkerApp.java new file mode 100644 index 0000000..ef0486f --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/PaymentsWorkerApp.java @@ -0,0 +1,87 @@ +package payments.temporal; + +import io.temporal.client.WorkflowClient; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; +import io.temporal.worker.WorkflowImplementationOptions; +import io.temporal.workflow.NexusServiceOptions; +import payments.PaymentGateway; +import payments.Shared; +import payments.temporal.activity.PaymentActivityImpl; + +import java.util.Collections; + +/** + * ═══════════════════════════════════════════════════════════════════ + * TODO FILE 7 of 7: PaymentsWorkerApp — CRAWL Familiar (10 min) + * ═══════════════════════════════════════════════════════════════════ + * + * PURPOSE: Wire the payments side. Same CRAWL pattern as Exercise 1301. + * Repetition reinforces the pattern. + * + * CRAWL PATTERN: + * + * C — Connect: + * WorkflowServiceStubs.newLocalServiceStubs() + * WorkflowClient.newInstance(service) + * + * R — Register workflow WITH Nexus endpoint mapping: + * WorkflowImplementationOptions maps service name → endpoint name. + * This is identical to 1301 — the workflow defines WHAT to call, + * the worker defines WHERE to find it. + * + * worker.registerWorkflowImplementationTypes( + * WorkflowImplementationOptions.newBuilder() + * .setNexusServiceOptions(Collections.singletonMap( + * "ComplianceNexusService", + * NexusServiceOptions.newBuilder() + * .setEndpoint("compliance-endpoint") + * .build())) + * .build(), + * PaymentProcessingWorkflowImpl.class); + * + * A — Register activities: + * worker.registerActivitiesImplementations(new PaymentActivityImpl(new PaymentGateway())) + * + * W — (endpoint mapping done in R step above — no separate W step needed) + * + * L — Launch and print startup banner: + * factory.start() + * Print: "Payments Worker started on payments-processing" + * + * TASK QUEUE: Shared.TASK_QUEUE ("payments-processing") + * + * REMINDER: + * The "ComplianceNexusService" string is the class name (no package). + * The "compliance-endpoint" string must match --name from the CLI command: + * temporal operator nexus endpoint create --name compliance-endpoint ... + * + * WHAT TO IMPLEMENT: + * Follow the CRAWL steps above. All necessary imports are already present. + */ +public class PaymentsWorkerApp { + + public static void main(String[] args) { + // TODO: C — Connect to Temporal + // WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); + // WorkflowClient client = WorkflowClient.newInstance(service); + + // TODO: Create WorkerFactory and Worker on Shared.TASK_QUEUE + // WorkerFactory factory = WorkerFactory.newInstance(client); + // Worker worker = factory.newWorker(Shared.TASK_QUEUE); + + // TODO: R — Register PaymentProcessingWorkflowImpl with Nexus endpoint mapping + // Map "ComplianceNexusService" → "compliance-endpoint" + // Use WorkflowImplementationOptions.newBuilder().setNexusServiceOptions(...) + + // TODO: A — Register PaymentActivityImpl with a new PaymentGateway + // worker.registerActivitiesImplementations(new PaymentActivityImpl(/* new PaymentGateway() */ )); + + // TODO: L — Launch and print startup banner + // factory.start(); + + System.out.println("PaymentsWorkerApp: not implemented yet."); + System.out.println("Implement the TODO steps above."); + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/activity/PaymentActivity.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/activity/PaymentActivity.java new file mode 100644 index 0000000..8969bc0 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/activity/PaymentActivity.java @@ -0,0 +1,18 @@ +package payments.temporal.activity; + +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import payments.domain.PaymentRequest; + +/** + * [GIVEN] Activity interface for payment operations. + * Implemented by you in PaymentActivityImpl.java. + */ +@ActivityInterface +public interface PaymentActivity { + @ActivityMethod + boolean validatePayment(PaymentRequest request); + + @ActivityMethod + String executePayment(PaymentRequest request); +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/activity/PaymentActivityImpl.java b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/activity/PaymentActivityImpl.java new file mode 100644 index 0000000..2cac1f5 --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/payments/temporal/activity/PaymentActivityImpl.java @@ -0,0 +1,56 @@ +package payments.temporal.activity; + +import payments.PaymentGateway; +import payments.domain.PaymentRequest; + +/** + * ═══════════════════════════════════════════════════════════════════ + * TODO FILE 1 of 7: PaymentActivityImpl — Warm-Up (5 min) + * ═══════════════════════════════════════════════════════════════════ + * + * PURPOSE: Muscle memory. Thin wrapper that delegates to PaymentGateway. + * Same pattern you've used since Exercise 01. + * + * WHAT TO IMPLEMENT: + * + * 1. Add a private field: PaymentGateway gateway + * + * 2. Add a constructor that accepts a PaymentGateway and assigns it + * to the field. + * + * 3. Implement validatePayment(): + * → return gateway.validatePayment(request) + * + * 4. Implement executePayment(): + * → return gateway.executePayment(request) + * + * PATTERN REMINDER: + * Activities are the "thin wrappers" that bridge Temporal to business logic. + * The activity doesn't contain business logic — it delegates to the gateway. + * This keeps PaymentGateway testable without Temporal. + * + * WHEN YOU'RE DONE: + * This file doesn't have its own checkpoint — it's used by PaymentProcessingWorkflowImpl. + * You'll test it in Checkpoint 3 when you run the full workflow. + */ +public class PaymentActivityImpl implements PaymentActivity { + + // TODO 1: Add private PaymentGateway field + + // TODO 2: Add constructor that accepts PaymentGateway + // (The no-arg constructor below is a temporary placeholder so the project compiles. + // Replace it with: public PaymentActivityImpl(PaymentGateway gateway)) + public PaymentActivityImpl() { + // Placeholder — replace with parameterized constructor + } + + @Override + public boolean validatePayment(PaymentRequest request) { + return false; // TODO 3: delegate to gateway.validatePayment(request) + } + + @Override + public String executePayment(PaymentRequest request) { + return null; // TODO 4: delegate to gateway.executePayment(request) + } +} diff --git a/exercise-1302-async-nexus/exercise/src/main/java/shared/nexus/ComplianceNexusService.java b/exercise-1302-async-nexus/exercise/src/main/java/shared/nexus/ComplianceNexusService.java new file mode 100644 index 0000000..347eaea --- /dev/null +++ b/exercise-1302-async-nexus/exercise/src/main/java/shared/nexus/ComplianceNexusService.java @@ -0,0 +1,48 @@ +package shared.nexus; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.nexusrpc.Operation; +import io.nexusrpc.Service; + +/** + * [GIVEN] Nexus Service Interface — the shared contract between Payments and Compliance teams. + * + * ═══════════════════════════════════════════════════════════════════ + * THE SAME INTERFACE PATTERN AS EXERCISE 1301 — ONE KEY DIFFERENCE + * ═══════════════════════════════════════════════════════════════════ + * + * In Exercise 1301, this interface defined a SYNC operation (checkCompliance). + * The handler ran inline and returned immediately. + * + * In Exercise 1302, the operation is identical from the CALLER's perspective — + * you still call complianceService.investigate(request) in the workflow. + * + * The difference is ENTIRELY in the handler (ComplianceNexusServiceImpl): + * 1301 handler: WorkflowClientOperationHandlers.sync(...) ← returns value inline + * 1302 handler: WorkflowClientOperationHandlers.fromWorkflowMethod(...) ← starts a workflow + * + * The interface doesn't know (or care) whether the handler is sync or async. + * That's the power of Nexus: the calling side is insulated from the implementation detail. + * + * METAPHOR: A valet parking ticket. + * You hand over your car (the request) and get a ticket (the operation handle). + * The valet might park it immediately or queue it for later — you don't care. + * You claim it when you're ready. + */ +@Service +public interface ComplianceNexusService { + + /** + * Start a 3-phase compliance investigation for the given transaction. + * + * Called by: PaymentProcessingWorkflow (Payments side) + * Handled by: ComplianceNexusServiceImpl (Compliance side) + * + * The handler starts a ComplianceInvestigationWorkflow asynchronously. + * The payment workflow holds a NexusOperationHandle and awaits the result + * when it's ready — without blocking a thread. + */ + @Operation + InvestigationResult investigate(InvestigationRequest request); +} diff --git a/exercise-1302-async-nexus/solution/pom.xml b/exercise-1302-async-nexus/solution/pom.xml new file mode 100644 index 0000000..6516ce7 --- /dev/null +++ b/exercise-1302-async-nexus/solution/pom.xml @@ -0,0 +1,99 @@ + + + 4.0.0 + + io.temporal.warmups + exercise-1302-async-nexus-solution-java + 1.0-SNAPSHOT + jar + + Exercise 1302 - Temporal Async Nexus (Java) + Upgrade from sync to async Nexus: one new concept, infinite new possibilities + + + 11 + 11 + UTF-8 + 1.27.0 + + + + + + io.temporal + temporal-sdk + ${temporal.version} + + + + + org.slf4j + slf4j-simple + 2.0.9 + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + 3.8.1 + + 11 + 11 + + + + + + org.codehaus.mojo + exec-maven-plugin + 3.0.0 + + + exercise.PaymentProcessingService + + + + + activity-test + + compliance.temporal.InvestigationActivityTest + + + + + compliance-worker + + compliance.temporal.ComplianceWorkerApp + + + + + payments-worker + + payments.temporal.PaymentsWorkerApp + + + + + starter + + payments.temporal.PaymentStarter + + + + + starter-rerun + + payments.temporal.PaymentStarterRerun + + + + + + + diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/ComplianceInvestigator.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/ComplianceInvestigator.java new file mode 100644 index 0000000..a1a1d34 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/ComplianceInvestigator.java @@ -0,0 +1,113 @@ +package compliance; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.temporal.activity.Activity; + +/** + * [GIVEN] The 3-phase forensic investigation logic. + * + * Phase 1: OFAC Screening — checks sanctioned entities and countries + * Phase 2: Pattern Analysis — detects structuring, layering, unusual routing + * Phase 3: Final Review — senior analyst sign-off with risk scoring + * + * Investigation durations per transaction: + * TXN-ALPHA ($3,500 US→Germany): ~15 seconds total + * TXN-BETA ($47,500 US→Caymans): ~35 seconds total + * TXN-GAMMA ($180,000 US→Iran): ~60 seconds total + * + * Students use this class as-is — focus is on wiring it into Temporal activities, + * not the investigation logic itself. + */ +public class ComplianceInvestigator { + + public InvestigationResult runFullInvestigation(InvestigationRequest request) { + String txn = request.getTransactionId(); + String prefix = "[INV-" + txn.replace("TXN-", "") + "]"; + + // Determine investigation profile based on transaction + int[] phaseDurations = getInvestigationProfile(request); + + try { + // Phase 1: OFAC Screening + System.out.println(prefix + " Phase 1/3: OFAC screening — checking sanctioned entities..."); + heartbeat("Phase 1/3: OFAC screening"); + sleep(phaseDurations[0]); + System.out.println(prefix + " Phase 1/3: OFAC screening complete."); + + // Phase 2: Pattern Analysis + System.out.println(prefix + " Phase 2/3: Pattern analysis — detecting structuring and layering..."); + heartbeat("Phase 2/3: Pattern analysis"); + sleep(phaseDurations[1]); + System.out.println(prefix + " Phase 2/3: Pattern analysis complete."); + + // Phase 3: Final Review + System.out.println(prefix + " Phase 3/3: Final review — senior analyst sign-off..."); + heartbeat("Phase 3/3: Final review"); + sleep(phaseDurations[2]); + System.out.println(prefix + " Phase 3/3: Final review complete."); + + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Investigation interrupted for " + txn, e); + } + + return buildResult(request); + } + + private int[] getInvestigationProfile(InvestigationRequest request) { + double amount = request.getAmount(); + String dest = request.getReceiverCountry(); + + // TXN-GAMMA: sanctioned country — full 60-second deep investigation + if (dest.equalsIgnoreCase("Iran") || dest.equalsIgnoreCase("North Korea") + || dest.equalsIgnoreCase("Cuba") || dest.equalsIgnoreCase("Syria")) { + return new int[]{15000, 20000, 25000}; // 60s total + } + // TXN-BETA: large international — 35-second investigation + if (amount >= 10000 && !request.getSenderCountry().equalsIgnoreCase(dest)) { + return new int[]{10000, 15000, 10000}; // 35s total + } + // TXN-ALPHA: routine — 15-second investigation + return new int[]{5000, 5000, 5000}; // 15s total + } + + private InvestigationResult buildResult(InvestigationRequest request) { + String dest = request.getReceiverCountry(); + double amount = request.getAmount(); + + // Sanctioned country — always blocked + if (dest.equalsIgnoreCase("Iran") || dest.equalsIgnoreCase("North Korea") + || dest.equalsIgnoreCase("Cuba") || dest.equalsIgnoreCase("Syria")) { + return new InvestigationResult(request.getTransactionId(), true, "CRITICAL", + "Transaction blocked: destination country is OFAC-sanctioned. " + + "All 3 investigation phases confirm: payment to " + dest + " violates regulatory requirements."); + } + + // Large international — medium risk, approved + if (amount >= 10000) { + return new InvestigationResult(request.getTransactionId(), false, "MEDIUM", + "Transaction approved with elevated monitoring. Amount $" + + String.format("%.0f", amount) + " to " + dest + + " triggers enhanced due diligence. No structuring patterns detected."); + } + + // Routine — low risk, approved + return new InvestigationResult(request.getTransactionId(), false, "LOW", + "Transaction approved. Routine transfer to " + dest + + ". All 3 investigation phases passed. No flags raised."); + } + + private void heartbeat(String phase) { + try { + // Heartbeat so Temporal knows the activity is still alive during long phases + Activity.getExecutionContext().heartbeat(phase); + } catch (Exception e) { + // Not running inside a Temporal activity (e.g., test runner) — ignore + } + } + + private void sleep(int millis) throws InterruptedException { + Thread.sleep(millis); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/SlowComplianceAgent.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/SlowComplianceAgent.java new file mode 100644 index 0000000..2d74f07 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/SlowComplianceAgent.java @@ -0,0 +1,42 @@ +package compliance; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; + +/** + * [GIVEN] A mock compliance agent used ONLY in Checkpoint 0. + * + * This simulates what happened before the async Nexus upgrade: + * the compliance team's new 3-phase investigation pipeline takes 30+ seconds, + * but the Payments team's sync Nexus call has a 5-second timeout. + * + * Result: every transaction times out. The Payments team pages on-call. + * + * This class is NOT used in your actual implementation — it exists to + * demonstrate the problem that async Nexus solves. + * + * Run Checkpoint 0 to see the timeout: + * mvn compile exec:java + */ +public class SlowComplianceAgent { + + public InvestigationResult runFullInvestigation(InvestigationRequest request) { + System.out.println("[SlowComplianceAgent] Starting compliance pipeline for " + + request.getTransactionId()); + System.out.println("[SlowComplianceAgent] Phase 1/3: OFAC screening (this takes a while...)"); + + try { + Thread.sleep(10000); // Phase 1: 10 seconds + System.out.println("[SlowComplianceAgent] Phase 2/3: Pattern analysis..."); + Thread.sleep(10000); // Phase 2: 10 seconds + System.out.println("[SlowComplianceAgent] Phase 3/3: Final review..."); + Thread.sleep(10000); // Phase 3: 10 seconds + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + // We never get here in practice — the caller times out first + return new InvestigationResult(request.getTransactionId(), false, "LOW", + "Investigation complete (but caller already gave up)"); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/domain/InvestigationRequest.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/domain/InvestigationRequest.java new file mode 100644 index 0000000..ca640eb --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/domain/InvestigationRequest.java @@ -0,0 +1,31 @@ +package compliance.domain; + +/** + * [GIVEN] Input to a compliance investigation. + * Sent by the Payments team via Nexus to the Compliance team. + */ +public class InvestigationRequest { + public String transactionId; + public double amount; + public String senderCountry; + public String receiverCountry; + public String description; + + public InvestigationRequest() {} + + public InvestigationRequest(String transactionId, double amount, + String senderCountry, String receiverCountry, + String description) { + this.transactionId = transactionId; + this.amount = amount; + this.senderCountry = senderCountry; + this.receiverCountry = receiverCountry; + this.description = description; + } + + public String getTransactionId() { return transactionId; } + public double getAmount() { return amount; } + public String getSenderCountry() { return senderCountry; } + public String getReceiverCountry() { return receiverCountry; } + public String getDescription() { return description; } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/domain/InvestigationResult.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/domain/InvestigationResult.java new file mode 100644 index 0000000..b6f5c2d --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/domain/InvestigationResult.java @@ -0,0 +1,38 @@ +package compliance.domain; + +/** + * [GIVEN] Result of a 3-phase compliance investigation. + * Returned by the Compliance team to the Payments team via async Nexus. + * + * Fields: + * blocked — true = block the payment, false = allow it + * riskLevel — "LOW", "MEDIUM", or "CRITICAL" + * summary — one-line summary of all 3 investigation phases + */ +public class InvestigationResult { + public String transactionId; + public boolean blocked; + public String riskLevel; // "LOW", "MEDIUM", "CRITICAL" + public String summary; + + public InvestigationResult() {} + + public InvestigationResult(String transactionId, boolean blocked, + String riskLevel, String summary) { + this.transactionId = transactionId; + this.blocked = blocked; + this.riskLevel = riskLevel; + this.summary = summary; + } + + public String getTransactionId() { return transactionId; } + public boolean isBlocked() { return blocked; } + public String getRiskLevel() { return riskLevel; } + public String getSummary() { return summary; } + + @Override + public String toString() { + return String.format("InvestigationResult{txn=%s, blocked=%s, risk=%s, summary='%s'}", + transactionId, blocked, riskLevel, summary); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceInvestigationWorkflow.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceInvestigationWorkflow.java new file mode 100644 index 0000000..e616aa6 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceInvestigationWorkflow.java @@ -0,0 +1,26 @@ +package compliance.temporal; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +/** + * [GIVEN] Workflow interface for the compliance investigation workflow. + * Implemented by you in ComplianceInvestigationWorkflowImpl.java. + * + * This workflow runs on the Compliance team's worker. It orchestrates + * the 3-phase investigation via ComplianceInvestigationActivity. + * + * IMPORTANT: The method name here must exactly match what you use in + * ComplianceNexusServiceImpl when building the fromWorkflowMethod() handler. + * + * client.newWorkflowStub(ComplianceInvestigationWorkflow.class, ...)::investigate + * ^^^^^^^^^^^ + * must match this method name + */ +@WorkflowInterface +public interface ComplianceInvestigationWorkflow { + @WorkflowMethod + InvestigationResult investigate(InvestigationRequest request); +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceInvestigationWorkflowImpl.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceInvestigationWorkflowImpl.java new file mode 100644 index 0000000..5cb19de --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceInvestigationWorkflowImpl.java @@ -0,0 +1,40 @@ +package compliance.temporal; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import compliance.temporal.activity.ComplianceInvestigationActivity; +import io.temporal.activity.ActivityOptions; +import io.temporal.common.RetryOptions; +import io.temporal.workflow.Workflow; + +import java.time.Duration; + +/** + * [SOLUTION] Orchestrates the 3-phase investigation via the activity. + * + * Heartbeat timeout is shorter than any single investigation phase (~5-15s per phase), + * so Temporal will detect a worker crash quickly and reschedule the activity. + * This is what enables the durability demo in Checkpoint 4. + */ +public class ComplianceInvestigationWorkflowImpl implements ComplianceInvestigationWorkflow { + + private static final ActivityOptions ACTIVITY_OPTIONS = ActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofMinutes(2)) + .setHeartbeatTimeout(Duration.ofSeconds(10)) + .setRetryOptions(RetryOptions.newBuilder() + .setMaximumAttempts(3) + .setInitialInterval(Duration.ofSeconds(1)) + .setBackoffCoefficient(2.0) + .build()) + .build(); + + private final ComplianceInvestigationActivity activity = + Workflow.newActivityStub(ComplianceInvestigationActivity.class, ACTIVITY_OPTIONS); + + @Override + public InvestigationResult investigate(InvestigationRequest request) { + Workflow.getLogger(ComplianceInvestigationWorkflowImpl.class) + .info("Starting 3-phase investigation for " + request.getTransactionId()); + return activity.investigate(request); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceNexusServiceImpl.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceNexusServiceImpl.java new file mode 100644 index 0000000..d18c9a1 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceNexusServiceImpl.java @@ -0,0 +1,38 @@ +package compliance.temporal; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.client.WorkflowOptions; +import io.temporal.nexus.WorkflowClientOperationHandlers; +import shared.nexus.ComplianceNexusService; + +/** + * [SOLUTION] The async Nexus handler — starts a workflow instead of running inline. + * + * Compare to Exercise 1301's sync handler: + * return WorkflowClientOperationHandlers.sync( + * (context, details, client, input) -> complianceAgent.checkCompliance(input) + * ); + * + * This handler uses fromWorkflowMethod() instead. The lambda returns a method reference + * to a workflow stub — Temporal uses this to start the workflow when a Nexus request arrives. + */ +@ServiceImpl(service = ComplianceNexusService.class) +public class ComplianceNexusServiceImpl { + + @OperationImpl + public OperationHandler investigate() { + return WorkflowClientOperationHandlers.fromWorkflowMethod( + (context, details, client, input) -> + client.newWorkflowStub( + ComplianceInvestigationWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId("investigation-" + input.getTransactionId()) + .build() + )::investigate + ); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceWorkerApp.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceWorkerApp.java new file mode 100644 index 0000000..89188c9 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/ComplianceWorkerApp.java @@ -0,0 +1,55 @@ +package compliance.temporal; + +import compliance.ComplianceInvestigator; +import compliance.temporal.activity.ComplianceInvestigationActivityImpl; +import io.temporal.client.WorkflowClient; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; + +/** + * [SOLUTION] Compliance team's worker — registers workflow, activity, and Nexus handler. + * + * CRAWL pattern extended (vs Exercise 1301's handler-only worker): + * C — Connect + * R — Register ComplianceInvestigationWorkflowImpl ← NEW + * A — Register ComplianceInvestigationActivityImpl ← NEW + * W — Wire Nexus handler + * L — Launch + * + * Task queue: "compliance-investigation" + * Must match --target-task-queue in the CLI endpoint creation command. + */ +public class ComplianceWorkerApp { + + static final String TASK_QUEUE = "compliance-investigation"; + + public static void main(String[] args) { + // C — Connect + WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); + WorkflowClient client = WorkflowClient.newInstance(service); + + WorkerFactory factory = WorkerFactory.newInstance(client); + Worker worker = factory.newWorker(TASK_QUEUE); + + // R — Register investigation workflow + worker.registerWorkflowImplementationTypes(ComplianceInvestigationWorkflowImpl.class); + + // A — Register investigation activity + worker.registerActivitiesImplementations( + new ComplianceInvestigationActivityImpl(new ComplianceInvestigator())); + + // W — Register Nexus handler + worker.registerNexusServiceImplementation(new ComplianceNexusServiceImpl()); + + // L — Launch + factory.start(); + + System.out.println("╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ Compliance Investigation Worker — ONLINE ║"); + System.out.println("║ Task queue: compliance-investigation ║"); + System.out.println("║ Registered: investigation workflow + activity + Nexus ║"); + System.out.println("║ Waiting for payment investigation requests... ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝"); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/InvestigationActivityTest.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/InvestigationActivityTest.java new file mode 100644 index 0000000..9b9c115 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/InvestigationActivityTest.java @@ -0,0 +1,64 @@ +package compliance.temporal; + +import compliance.ComplianceInvestigator; +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import compliance.temporal.activity.ComplianceInvestigationActivityImpl; + +/** + * [GIVEN] Checkpoint 1 runner — tests your ComplianceInvestigationActivityImpl. + * + * This is NOT a JUnit test. It's a simple main class that calls your + * activity implementation directly (without Temporal), so you can verify + * your delegation code is correct before wiring it into the full system. + * + * Expected output: + * [INV-ALPHA] Phase 1/3: OFAC screening — checking sanctioned entities... + * [INV-ALPHA] Phase 1/3: OFAC screening complete. + * [INV-ALPHA] Phase 2/3: Pattern analysis — detecting structuring and layering... + * ... + * Checkpoint 1 PASSED: ComplianceInvestigationActivityImpl delegates correctly. + * + * Run with: mvn compile exec:java@activity-test + */ +public class InvestigationActivityTest { + + public static void main(String[] args) { + System.out.println("═══════════════════════════════════════════════════"); + System.out.println(" Checkpoint 1: Testing ComplianceInvestigationActivityImpl"); + System.out.println(" (Running investigation inline — no Temporal needed)"); + System.out.println("═══════════════════════════════════════════════════\n"); + + ComplianceInvestigator investigator = new ComplianceInvestigator(); + ComplianceInvestigationActivityImpl activity = + new ComplianceInvestigationActivityImpl(investigator); + + InvestigationRequest request = new InvestigationRequest( + "TXN-ALPHA", 3500.00, "US", "Germany", + "Equipment purchase from European supplier"); + + System.out.println("Running investigation for TXN-ALPHA (expect ~15 seconds)...\n"); + + long start = System.currentTimeMillis(); + InvestigationResult result = activity.investigate(request); + long elapsed = (System.currentTimeMillis() - start) / 1000; + + System.out.println("\n═══════════════════════════════════════════════════"); + System.out.println(" RESULT:"); + System.out.println(" Transaction: " + result.getTransactionId()); + System.out.println(" Blocked: " + result.isBlocked()); + System.out.println(" Risk Level: " + result.getRiskLevel()); + System.out.println(" Summary: " + result.getSummary()); + System.out.println(" Time: " + elapsed + " seconds"); + + if (result.getTransactionId() != null && !result.isBlocked() && "LOW".equals(result.getRiskLevel())) { + System.out.println("\n Checkpoint 1 PASSED: ComplianceInvestigationActivityImpl delegates correctly."); + } else if (result.getTransactionId() == null) { + System.out.println("\n Checkpoint 1 FAILED: result is null — did you implement the activity?"); + System.exit(1); + } else { + System.out.println("\n Checkpoint 1 PASSED: Activity delegates to ComplianceInvestigator."); + } + System.out.println("═══════════════════════════════════════════════════\n"); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivity.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivity.java new file mode 100644 index 0000000..bab206b --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivity.java @@ -0,0 +1,19 @@ +package compliance.temporal.activity; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; + +/** + * [GIVEN] Activity interface for the 3-phase compliance investigation. + * Implemented by you in ComplianceInvestigationActivityImpl.java. + * + * This activity runs on the Compliance team's worker. It delegates + * to ComplianceInvestigator which performs the actual 3-phase logic. + */ +@ActivityInterface +public interface ComplianceInvestigationActivity { + @ActivityMethod + InvestigationResult investigate(InvestigationRequest request); +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivityImpl.java b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivityImpl.java new file mode 100644 index 0000000..a24fa44 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/compliance/temporal/activity/ComplianceInvestigationActivityImpl.java @@ -0,0 +1,23 @@ +package compliance.temporal.activity; + +import compliance.ComplianceInvestigator; +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; + +/** + * [SOLUTION] Thin activity wrapper that delegates to ComplianceInvestigator. + * Same thin-wrapper pattern as PaymentActivityImpl. + */ +public class ComplianceInvestigationActivityImpl implements ComplianceInvestigationActivity { + + private final ComplianceInvestigator investigator; + + public ComplianceInvestigationActivityImpl(ComplianceInvestigator investigator) { + this.investigator = investigator; + } + + @Override + public InvestigationResult investigate(InvestigationRequest request) { + return investigator.runFullInvestigation(request); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/exercise/PaymentProcessingService.java b/exercise-1302-async-nexus/solution/src/main/java/exercise/PaymentProcessingService.java new file mode 100644 index 0000000..8b44aa5 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/exercise/PaymentProcessingService.java @@ -0,0 +1,79 @@ +package exercise; + +import compliance.SlowComplianceAgent; +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; + +import java.util.concurrent.*; + +/** + * [GIVEN] Checkpoint 0: The Problem — sync call to a slow compliance pipeline. + * + * The Compliance team upgraded their AI pipeline from a 2-second check + * to a 3-phase forensic investigation (OFAC → pattern analysis → final review). + * Total time: 30 seconds. But the Payments team's SLA is 5 seconds. + * + * This class demonstrates what happens without async Nexus: + * - Sync call with a 5-second timeout + * - SlowComplianceAgent takes 30 seconds + * - Every transaction times out + * - No retry. No audit trail. Transaction is lost. + * + * Run this first, BEFORE implementing anything: + * mvn compile exec:java + * + * You should see a timeout exception. That's the problem you're solving. + */ +public class PaymentProcessingService { + + private static final int COMPLIANCE_TIMEOUT_SECONDS = 5; + + public static void main(String[] args) { + System.out.println("╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ CHECKPOINT 0: The Sync Timeout Problem ║"); + System.out.println("║ Payments team calling slow Compliance pipeline ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝\n"); + + SlowComplianceAgent agent = new SlowComplianceAgent(); + InvestigationRequest request = new InvestigationRequest( + "TXN-ALPHA", 3500.00, "US", "Germany", + "Equipment purchase from European supplier"); + + System.out.println(" Calling compliance team for TXN-ALPHA..."); + System.out.println(" Timeout: " + COMPLIANCE_TIMEOUT_SECONDS + " seconds"); + System.out.println(" Compliance pipeline takes: ~30 seconds"); + System.out.println(); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + Future future = executor.submit( + () -> agent.runFullInvestigation(request)); + + try { + InvestigationResult result = future.get(COMPLIANCE_TIMEOUT_SECONDS, TimeUnit.SECONDS); + // We never reach here + System.out.println(" Result: " + result); + + } catch (TimeoutException e) { + future.cancel(true); + System.out.println(); + System.out.println("╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ ERROR: java.util.concurrent.TimeoutException ║"); + System.out.println("║ Compliance check exceeded " + COMPLIANCE_TIMEOUT_SECONDS + "-second SLA ║"); + System.out.println("║ ║"); + System.out.println("║ Transaction TXN-ALPHA: LOST ║"); + System.out.println("║ Retries: 0 ║"); + System.out.println("║ Audit trail: none ║"); + System.out.println("║ On-call engineer: paged ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝\n"); + System.out.println(" This is what async Nexus solves."); + System.out.println(" Instead of blocking for 30 seconds, Temporal holds the handle"); + System.out.println(" and resumes the Payments workflow when the investigation completes."); + System.out.println(" Read the README to see how."); + + } catch (Exception e) { + System.out.println(" Unexpected error: " + e.getMessage()); + } finally { + executor.shutdownNow(); + } + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/PaymentGateway.java b/exercise-1302-async-nexus/solution/src/main/java/payments/PaymentGateway.java new file mode 100644 index 0000000..6defe82 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/PaymentGateway.java @@ -0,0 +1,48 @@ +package payments; + +import payments.domain.PaymentRequest; + +/** + * [GIVEN] Simulated payment gateway for executing transactions. + * In production this would call Stripe, PayPal, SWIFT, etc. + * + * Students use this class as-is — focus is on Temporal integration, not payment logic. + */ +public class PaymentGateway { + + public boolean validatePayment(PaymentRequest request) { + if (request.getAmount() <= 0) { + System.out.println("[PaymentGateway] REJECTED: Invalid amount for " + request.getTransactionId()); + return false; + } + if (request.getSenderAccount() == null || request.getReceiverAccount() == null) { + System.out.println("[PaymentGateway] REJECTED: Missing account info for " + request.getTransactionId()); + return false; + } + System.out.println("[PaymentGateway] Validation passed for " + request.getTransactionId()); + return true; + } + + public String executePayment(PaymentRequest request) { + System.out.println("[PaymentGateway] Processing " + request.getTransactionId() + + " | $" + String.format("%.2f", request.getAmount()) + + " | " + request.getSenderCountry() + " -> " + request.getReceiverCountry()); + + // Simulate processing time + try { + Thread.sleep(300 + (long) (Math.random() * 300)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + // Simulate occasional gateway failures (10% chance) — Temporal retries automatically + if (Math.random() < 0.10) { + throw new RuntimeException("Payment gateway timeout for " + request.getTransactionId() + + " — connection to banking network failed"); + } + + String confirmationNumber = "CONF-" + request.getTransactionId() + "-" + System.currentTimeMillis(); + System.out.println("[PaymentGateway] Payment executed: " + confirmationNumber); + return confirmationNumber; + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/Shared.java b/exercise-1302-async-nexus/solution/src/main/java/payments/Shared.java new file mode 100644 index 0000000..51d03e0 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/Shared.java @@ -0,0 +1,5 @@ +package payments; + +public interface Shared { + String TASK_QUEUE = "payments-processing"; +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/domain/PaymentRequest.java b/exercise-1302-async-nexus/solution/src/main/java/payments/domain/PaymentRequest.java new file mode 100644 index 0000000..6bbcc4e --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/domain/PaymentRequest.java @@ -0,0 +1,39 @@ +package payments.domain; + +/** + * [GIVEN] A payment transaction to be processed. + */ +public class PaymentRequest { + public String transactionId; + public double amount; + public String currency; + public String senderCountry; + public String receiverCountry; + public String description; + public String senderAccount; + public String receiverAccount; + + public PaymentRequest() {} + + public PaymentRequest(String transactionId, double amount, String currency, + String senderCountry, String receiverCountry, + String description, String senderAccount, String receiverAccount) { + this.transactionId = transactionId; + this.amount = amount; + this.currency = currency; + this.senderCountry = senderCountry; + this.receiverCountry = receiverCountry; + this.description = description; + this.senderAccount = senderAccount; + this.receiverAccount = receiverAccount; + } + + public String getTransactionId() { return transactionId; } + public double getAmount() { return amount; } + public String getCurrency() { return currency; } + public String getSenderCountry() { return senderCountry; } + public String getReceiverCountry() { return receiverCountry; } + public String getDescription() { return description; } + public String getSenderAccount() { return senderAccount; } + public String getReceiverAccount() { return receiverAccount; } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/domain/PaymentResult.java b/exercise-1302-async-nexus/solution/src/main/java/payments/domain/PaymentResult.java new file mode 100644 index 0000000..99c1cb7 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/domain/PaymentResult.java @@ -0,0 +1,48 @@ +package payments.domain; + +/** + * [GIVEN] Result of a payment workflow execution. + * + * status values: + * "COMPLETED" — payment processed successfully + * "REJECTED" — failed payment validation + * "BLOCKED_COMPLIANCE" — compliance investigation returned blocked=true + * "FAILED" — unexpected error + */ +public class PaymentResult { + public boolean success; + public String transactionId; + public String status; + public String riskLevel; // from compliance investigation: "LOW", "MEDIUM", "CRITICAL" + public String summary; // from compliance investigation AI summary + public String confirmationNumber; // set when status = COMPLETED + public String error; + + public PaymentResult() {} + + public PaymentResult(boolean success, String transactionId, String status, + String riskLevel, String summary, + String confirmationNumber, String error) { + this.success = success; + this.transactionId = transactionId; + this.status = status; + this.riskLevel = riskLevel; + this.summary = summary; + this.confirmationNumber = confirmationNumber; + this.error = error; + } + + public boolean isSuccess() { return success; } + public String getTransactionId() { return transactionId; } + public String getStatus() { return status; } + public String getRiskLevel() { return riskLevel; } + public String getSummary() { return summary; } + public String getConfirmationNumber() { return confirmationNumber; } + public String getError() { return error; } + + @Override + public String toString() { + return String.format("PaymentResult{txn=%s, status=%s, risk=%s, confirm=%s, error=%s}", + transactionId, status, riskLevel, confirmationNumber, error); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentProcessingWorkflow.java b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentProcessingWorkflow.java new file mode 100644 index 0000000..ea31733 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentProcessingWorkflow.java @@ -0,0 +1,16 @@ +package payments.temporal; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import payments.domain.PaymentRequest; +import payments.domain.PaymentResult; + +/** + * [GIVEN] Workflow interface for the payment processing workflow. + * Implemented by you in PaymentProcessingWorkflowImpl.java. + */ +@WorkflowInterface +public interface PaymentProcessingWorkflow { + @WorkflowMethod + PaymentResult processPayment(PaymentRequest request); +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentProcessingWorkflowImpl.java b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentProcessingWorkflowImpl.java new file mode 100644 index 0000000..88fd404 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentProcessingWorkflowImpl.java @@ -0,0 +1,93 @@ +package payments.temporal; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.temporal.activity.ActivityOptions; +import io.temporal.common.RetryOptions; +import io.temporal.workflow.NexusOperationHandle; +import io.temporal.workflow.NexusOperationOptions; +import io.temporal.workflow.NexusServiceOptions; +import io.temporal.workflow.Workflow; +import payments.domain.PaymentRequest; +import payments.domain.PaymentResult; +import payments.temporal.activity.PaymentActivity; +import shared.nexus.ComplianceNexusService; + +import java.time.Duration; + +/** + * [SOLUTION] Orchestrates 5 steps: validate → start investigation → await result → check → execute. + * + * The key pattern: Workflow.startNexusOperation() returns immediately with a handle. + * handle.getResult().get() is the durable await — the workflow suspends here until + * the investigation workflow completes on the Compliance side. + * + * If the Payments worker crashes between startNexusOperation() and get(), Temporal + * replays the workflow, re-attaches to the in-progress investigation, and continues. + */ +public class PaymentProcessingWorkflowImpl implements PaymentProcessingWorkflow { + + private static final ActivityOptions ACTIVITY_OPTIONS = ActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(30)) + .setRetryOptions(RetryOptions.newBuilder() + .setMaximumAttempts(3) + .setInitialInterval(Duration.ofSeconds(1)) + .setBackoffCoefficient(2.0) + .build()) + .build(); + + private final PaymentActivity paymentActivity = + Workflow.newActivityStub(PaymentActivity.class, ACTIVITY_OPTIONS); + + private final ComplianceNexusService complianceService = Workflow.newNexusServiceStub( + ComplianceNexusService.class, + NexusServiceOptions.newBuilder() + .setOperationOptions(NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofMinutes(2)) + .build()) + .build()); + + @Override + public PaymentResult processPayment(PaymentRequest request) { + try { + // Step 1: Validate payment + boolean valid = paymentActivity.validatePayment(request); + if (!valid) { + return new PaymentResult(false, request.getTransactionId(), "REJECTED", + null, null, null, "Validation failed"); + } + + // Step 2: Build investigation request and start async Nexus operation + InvestigationRequest investigationReq = new InvestigationRequest( + request.getTransactionId(), request.getAmount(), + request.getSenderCountry(), request.getReceiverCountry(), + request.getDescription()); + + NexusOperationHandle handle = + Workflow.startNexusOperation(complianceService::investigate, investigationReq); + + // Step 3: Durably await the investigation result + // Workflow suspends here — no thread blocked, fully durable + InvestigationResult investigation = handle.getResult().get(); + + // Step 4: Check investigation outcome + if (investigation.isBlocked()) { + return new PaymentResult(false, request.getTransactionId(), + "BLOCKED_COMPLIANCE", investigation.getRiskLevel(), + investigation.getSummary(), null, null); + } + + // Step 5: Execute payment + String confirmation = paymentActivity.executePayment(request); + return new PaymentResult(true, request.getTransactionId(), + "COMPLETED", investigation.getRiskLevel(), + investigation.getSummary(), confirmation, null); + + } catch (Exception e) { + Workflow.getLogger(PaymentProcessingWorkflowImpl.class) + .error("Workflow failed for " + request.getTransactionId() + ": " + e.getMessage()); + return new PaymentResult(false, request.getTransactionId(), "FAILED", + null, null, null, e.getMessage()); + } + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentStarter.java b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentStarter.java new file mode 100644 index 0000000..a2e2ecc --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentStarter.java @@ -0,0 +1,92 @@ +package payments.temporal; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.serviceclient.WorkflowServiceStubs; +import payments.Shared; +import payments.domain.PaymentRequest; +import payments.domain.PaymentResult; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +/** + * [GIVEN] Starts 3 payment workflows in parallel and waits for all results. + * + * Transactions: + * TXN-ALPHA: $3,500 US → Germany ~15s investigation APPROVED / LOW + * TXN-BETA: $47,500 US → Caymans ~35s investigation APPROVED / MEDIUM + * TXN-GAMMA: $180,000 US → Iran ~60s investigation BLOCKED / CRITICAL + * + * Business ID workflow IDs: + * "payment-TXN-ALPHA", "payment-TXN-BETA", "payment-TXN-GAMMA" + * + * Runs workflows in parallel using WorkflowClient.execute(). + * All 3 investigations start at the same time. + * Total time ≈ longest investigation (~60s for TXN-GAMMA) rather than 110s sequential. + */ +public class PaymentStarter { + + public static void main(String[] args) throws Exception { + System.out.println("╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ PAYMENT STARTER — Exercise 1302: Async Nexus ║"); + System.out.println("║ Starting 3 transactions (parallel investigations) ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝\n"); + + WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); + WorkflowClient client = WorkflowClient.newInstance(service); + + PaymentRequest[] transactions = { + new PaymentRequest("TXN-ALPHA", 3500.00, "USD", "US", "Germany", + "Equipment purchase from European supplier", "ACC-001", "ACC-002"), + new PaymentRequest("TXN-BETA", 47500.00, "USD", "US", "Cayman Islands", + "Investment fund transfer to offshore account", "ACC-003", "ACC-004"), + new PaymentRequest("TXN-GAMMA", 180000.00, "USD", "US", "Iran", + "Technology export to Middle East partner", "ACC-005", "ACC-006"), + }; + + List> futures = new ArrayList<>(); + + for (PaymentRequest txn : transactions) { + String workflowId = "payment-" + txn.getTransactionId(); + System.out.println(" Launching: " + workflowId + + " ($" + String.format("%.0f", txn.getAmount()) + " -> " + txn.getReceiverCountry() + ")"); + + PaymentProcessingWorkflow workflow = client.newWorkflowStub( + PaymentProcessingWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(Shared.TASK_QUEUE) + .setWorkflowId(workflowId) + .build()); + + // WorkflowClient.execute() returns immediately — workflow runs in background + CompletableFuture future = + WorkflowClient.execute(workflow::processPayment, txn); + futures.add(future); + } + + System.out.println("\n All 3 workflows started. Waiting for results..."); + System.out.println(" (Investigations run in parallel — check Temporal UI!)\n"); + + // Wait for all workflows to complete + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); + + System.out.println("\n╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ PAYMENT RESULTS ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝"); + + for (int i = 0; i < futures.size(); i++) { + PaymentResult result = futures.get(i).get(); + String icon = "COMPLETED".equals(result.getStatus()) ? "✓" : + "BLOCKED_COMPLIANCE".equals(result.getStatus()) ? "✗" : "!"; + System.out.printf(" [%s] %-10s | %-20s | Risk: %-8s | %s%n", + icon, + result.getTransactionId(), + result.getStatus(), + result.getRiskLevel() != null ? result.getRiskLevel() : "N/A", + result.getSummary() != null ? result.getSummary().substring(0, Math.min(60, result.getSummary().length())) + "..." : ""); + } + System.out.println(); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentStarterRerun.java b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentStarterRerun.java new file mode 100644 index 0000000..50ac20a --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentStarterRerun.java @@ -0,0 +1,75 @@ +package payments.temporal; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.serviceclient.WorkflowServiceStubs; +import payments.Shared; +import payments.domain.PaymentRequest; +import payments.domain.PaymentResult; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +/** + * [GIVEN] Same as PaymentStarter but uses timestamp-suffixed workflow IDs. + * Used for Checkpoint 4 (The Heist Test) — allows you to re-run without + * ALREADY_EXISTS errors. + * + * Usage: mvn compile exec:java@starter-rerun + */ +public class PaymentStarterRerun { + + public static void main(String[] args) throws Exception { + System.out.println("╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ PAYMENT STARTER RERUN — Exercise 1302: Heist Test ║"); + System.out.println("║ Starting 3 transactions (timestamp-suffixed IDs) ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝\n"); + + WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); + WorkflowClient client = WorkflowClient.newInstance(service); + + long ts = System.currentTimeMillis(); + + PaymentRequest[] transactions = { + new PaymentRequest("TXN-ALPHA-" + ts, 3500.00, "USD", "US", "Germany", + "Equipment purchase from European supplier", "ACC-001", "ACC-002"), + new PaymentRequest("TXN-BETA-" + ts, 47500.00, "USD", "US", "Cayman Islands", + "Investment fund transfer to offshore account", "ACC-003", "ACC-004"), + new PaymentRequest("TXN-GAMMA-" + ts, 180000.00, "USD", "US", "Iran", + "Technology export to Middle East partner", "ACC-005", "ACC-006"), + }; + + List> futures = new ArrayList<>(); + + for (PaymentRequest txn : transactions) { + String workflowId = "payment-" + txn.getTransactionId(); + System.out.println(" Launching: " + workflowId); + + PaymentProcessingWorkflow workflow = client.newWorkflowStub( + PaymentProcessingWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(Shared.TASK_QUEUE) + .setWorkflowId(workflowId) + .build()); + + CompletableFuture future = + WorkflowClient.execute(workflow::processPayment, txn); + futures.add(future); + } + + System.out.println("\n All 3 workflows started. Kill the compliance worker during TXN-BETA!"); + System.out.println(" Watch it resume from Phase 3 after restart.\n"); + + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); + + System.out.println("\n═══ RESULTS ═══"); + for (int i = 0; i < futures.size(); i++) { + PaymentResult result = futures.get(i).get(); + System.out.printf(" %-25s | %-20s | Risk: %s%n", + result.getTransactionId(), + result.getStatus(), + result.getRiskLevel() != null ? result.getRiskLevel() : "N/A"); + } + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentsWorkerApp.java b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentsWorkerApp.java new file mode 100644 index 0000000..682c363 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/PaymentsWorkerApp.java @@ -0,0 +1,62 @@ +package payments.temporal; + +import io.temporal.client.WorkflowClient; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; +import io.temporal.worker.WorkflowImplementationOptions; +import io.temporal.workflow.NexusServiceOptions; +import payments.PaymentGateway; +import payments.Shared; +import payments.temporal.activity.PaymentActivityImpl; + +import java.util.Collections; + +/** + * [SOLUTION] Payments team's worker — identical CRAWL pattern to Exercise 1301. + * + * CRAWL: + * C — Connect + * R — Register PaymentProcessingWorkflowImpl with Nexus endpoint mapping + * A — Register PaymentActivityImpl + * W — (done in R step via WorkflowImplementationOptions) + * L — Launch + * + * The endpoint mapping ("ComplianceNexusService" → "compliance-endpoint") + * tells Temporal where to route Nexus calls from this workflow. + */ +public class PaymentsWorkerApp { + + public static void main(String[] args) { + // C — Connect + WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); + WorkflowClient client = WorkflowClient.newInstance(service); + + WorkerFactory factory = WorkerFactory.newInstance(client); + Worker worker = factory.newWorker(Shared.TASK_QUEUE); + + // R + W — Register workflow with Nexus endpoint mapping + worker.registerWorkflowImplementationTypes( + WorkflowImplementationOptions.newBuilder() + .setNexusServiceOptions(Collections.singletonMap( + "ComplianceNexusService", + NexusServiceOptions.newBuilder() + .setEndpoint("compliance-endpoint") + .build())) + .build(), + PaymentProcessingWorkflowImpl.class); + + // A — Register payment activity + worker.registerActivitiesImplementations(new PaymentActivityImpl(new PaymentGateway())); + + // L — Launch + factory.start(); + + System.out.println("╔══════════════════════════════════════════════════════════╗"); + System.out.println("║ Payments Worker — ONLINE ║"); + System.out.println("║ Task queue: payments-processing ║"); + System.out.println("║ Nexus endpoint: compliance-endpoint ║"); + System.out.println("║ Ready to process payment workflows... ║"); + System.out.println("╚══════════════════════════════════════════════════════════╝"); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/activity/PaymentActivity.java b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/activity/PaymentActivity.java new file mode 100644 index 0000000..8969bc0 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/activity/PaymentActivity.java @@ -0,0 +1,18 @@ +package payments.temporal.activity; + +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import payments.domain.PaymentRequest; + +/** + * [GIVEN] Activity interface for payment operations. + * Implemented by you in PaymentActivityImpl.java. + */ +@ActivityInterface +public interface PaymentActivity { + @ActivityMethod + boolean validatePayment(PaymentRequest request); + + @ActivityMethod + String executePayment(PaymentRequest request); +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/activity/PaymentActivityImpl.java b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/activity/PaymentActivityImpl.java new file mode 100644 index 0000000..cf77498 --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/payments/temporal/activity/PaymentActivityImpl.java @@ -0,0 +1,27 @@ +package payments.temporal.activity; + +import payments.PaymentGateway; +import payments.domain.PaymentRequest; + +/** + * [SOLUTION] Thin activity wrapper that delegates to PaymentGateway. + * The activity doesn't contain business logic — it bridges Temporal to the gateway. + */ +public class PaymentActivityImpl implements PaymentActivity { + + private final PaymentGateway gateway; + + public PaymentActivityImpl(PaymentGateway gateway) { + this.gateway = gateway; + } + + @Override + public boolean validatePayment(PaymentRequest request) { + return gateway.validatePayment(request); + } + + @Override + public String executePayment(PaymentRequest request) { + return gateway.executePayment(request); + } +} diff --git a/exercise-1302-async-nexus/solution/src/main/java/shared/nexus/ComplianceNexusService.java b/exercise-1302-async-nexus/solution/src/main/java/shared/nexus/ComplianceNexusService.java new file mode 100644 index 0000000..347eaea --- /dev/null +++ b/exercise-1302-async-nexus/solution/src/main/java/shared/nexus/ComplianceNexusService.java @@ -0,0 +1,48 @@ +package shared.nexus; + +import compliance.domain.InvestigationRequest; +import compliance.domain.InvestigationResult; +import io.nexusrpc.Operation; +import io.nexusrpc.Service; + +/** + * [GIVEN] Nexus Service Interface — the shared contract between Payments and Compliance teams. + * + * ═══════════════════════════════════════════════════════════════════ + * THE SAME INTERFACE PATTERN AS EXERCISE 1301 — ONE KEY DIFFERENCE + * ═══════════════════════════════════════════════════════════════════ + * + * In Exercise 1301, this interface defined a SYNC operation (checkCompliance). + * The handler ran inline and returned immediately. + * + * In Exercise 1302, the operation is identical from the CALLER's perspective — + * you still call complianceService.investigate(request) in the workflow. + * + * The difference is ENTIRELY in the handler (ComplianceNexusServiceImpl): + * 1301 handler: WorkflowClientOperationHandlers.sync(...) ← returns value inline + * 1302 handler: WorkflowClientOperationHandlers.fromWorkflowMethod(...) ← starts a workflow + * + * The interface doesn't know (or care) whether the handler is sync or async. + * That's the power of Nexus: the calling side is insulated from the implementation detail. + * + * METAPHOR: A valet parking ticket. + * You hand over your car (the request) and get a ticket (the operation handle). + * The valet might park it immediately or queue it for later — you don't care. + * You claim it when you're ready. + */ +@Service +public interface ComplianceNexusService { + + /** + * Start a 3-phase compliance investigation for the given transaction. + * + * Called by: PaymentProcessingWorkflow (Payments side) + * Handled by: ComplianceNexusServiceImpl (Compliance side) + * + * The handler starts a ComplianceInvestigationWorkflow asynchronously. + * The payment workflow holds a NexusOperationHandle and awaits the result + * when it's ready — without blocking a thread. + */ + @Operation + InvestigationResult investigate(InvestigationRequest request); +} diff --git a/exercise-1302-async-nexus/ui/async-nexus-flow.html b/exercise-1302-async-nexus/ui/async-nexus-flow.html new file mode 100644 index 0000000..1a016c0 --- /dev/null +++ b/exercise-1302-async-nexus/ui/async-nexus-flow.html @@ -0,0 +1,695 @@ + + + + + +Async Nexus Flow — Exercise 1302 + + + + +

Async Nexus — Exercise 1302

+

How fromWorkflowMethod() replaces sync()

+ + +
+ + +
+ + +
+ Async Mode: ComplianceNexusServiceImpl uses fromWorkflowMethod() — investigation runs as a real Temporal workflow +
+
+ Sync Mode: ComplianceNexusServiceImpl uses sync() — handler blocks and times out after 5s +
+ + +
+ + + + +
+ + +
+
+ + +
+
Payments Team
+
task-queue: payments-processing
+ +
+
Payments Worker: ONLINE +
+ +
+
1
+
+
validatePayment()
+
PaymentActivity checks amount, accounts. Returns true for TXN-BETA.
+
+
+ +
+
2
+
+
Start Nexus Operation
+
+ Workflow.startNexusOperation(complianceService::investigate, req)
+ Returns a NexusOperationHandle immediately. +
+
+
NexusOperationHandle<InvestigationResult>
+
operation-token: op-TXN-BETA-1302
status: STARTED → waiting...
+
+
+
+ +
+
3
+
+
handle.getResult().get()
+
+ Workflow durably suspended. No thread blocked.
+ Temporal persists state. Worker can restart safely. +
+
+
+ +
+
6
+
+
Workflow Resumes
+
+ Investigation result arrives. Execute payment → CONF-TXN-BETA-xxx +
+
+
+ +
+ + +
+
Nexus
+
+
+ + +
+
Compliance Team
+
task-queue: compliance-investigation
+ +
+
Compliance Worker: ONLINE +
+ +
+
4
+
+
ComplianceNexusServiceImpl.investigate()
+
+ fromWorkflowMethod() starts ComplianceInvestigationWorkflow
+ workflowId: investigation-TXN-BETA +
+
+
+ +
+
5
+
+
Phase 1/3: OFAC Screening
+
Checking sanctioned entities and countries. (~10s)
+
+
+ +
+
5
+
+
Phase 2/3: Pattern Analysis
+
Detecting structuring, layering, unusual routing. (~15s)
+
+
+ +
+
5
+
+
Phase 3/3: Final Review
+
Senior analyst sign-off. Risk scoring. (~10s)
+
+
+ +
+
+ + +
+
+
Click "Start Animation" to see the async Nexus flow.
+
+
+ + + + diff --git a/exercise-1302-async-nexus/ui/sync-vs-async.html b/exercise-1302-async-nexus/ui/sync-vs-async.html new file mode 100644 index 0000000..529d3fc --- /dev/null +++ b/exercise-1302-async-nexus/ui/sync-vs-async.html @@ -0,0 +1,302 @@ + + + + + +Sync vs Async Nexus — Exercise 1302 + + + +

Sync vs Async Nexus

+

Same compliance check · Different handler · Very different outcomes

+ +
+ + +
+
Sync Nexus (Exercise 1301)
+
Handler runs inline · Caller blocks · 5s timeout · 30s investigation = CRASH
+ +
+
+ 0s + 5s + 10s + 15s + 20s +
+ +
+
Payments
+
+
+
validate
+
⏳ waiting for compliance... (BLOCKED)
+
💥 TIMEOUT
+
+
+
+ +
+
Compliance
+
+
+
+
Phase 1
+
Phase 2
+
Phase 3
+
+
+
+
+ +
+ What happens: Payments workflow times out at 5s. + The compliance investigation continues for 30s, then finishes — + but nobody is listening. Transaction is permanently lost. + No retry. No audit trail. On-call engineer paged. +
+ +
+// 1301: sync handler — runs INLINE +return WorkflowClientOperationHandlers + .sync( + (ctx, det, client, input) -> + agent.checkCompliance(input) + // ↑ This runs inside the handler. + // Caller WAITS for this to return. + ); +
+
+ + +
+
Async Nexus (Exercise 1302)
+
Handler starts a workflow · Caller holds a handle · Investigation runs independently
+ +
+
+ 0s + 5s + 10s + 15s + 20s +
+ +
+
Payments
+
+
+
validate
+
start+handle
+
💤 suspended (not blocking a thread)
+
resume
+
execute
+
+
+
+ +
+
Compliance
+
+
+
+
Phase 1
+
Phase 2
+
Phase 3 ✓
+
+
+
+
+ +
+ What happens: Payments workflow starts the Nexus operation and + suspends — no thread blocked, no timeout risk. The compliance workflow runs + all 3 phases. When done, Temporal resumes the Payments workflow with the result. + Transaction completes successfully. Full audit trail in Temporal UI. +
+ +
+// 1302: async handler — starts a WORKFLOW +return WorkflowClientOperationHandlers + .fromWorkflowMethod( + (ctx, det, client, input) -> + client.newWorkflowStub( + ComplianceInvestigationWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId("investigation-" + + input.getTransactionId()) + .build() + )::investigate + // ↑ Returns immediately. + // Temporal starts the workflow and + // gives back an operation token. + ); +
+
+ +
+ +
+ The Nexus interface (ComplianceNexusService) is identical in both exercises. + The caller doesn't know (or care) whether the handler is sync or async. + Only the handler implementation changes. +
+ + +