|
| 1 | +import { Injectable, Inject, Logger } from '@nestjs/common'; |
| 2 | +import { CACHE_MANAGER } from '@nestjs/cache-manager'; |
| 3 | +import { Cache } from 'cache-manager'; |
| 4 | + |
| 5 | +export interface WorkflowStepRecord { |
| 6 | + completedAt: string; |
| 7 | + result?: unknown; |
| 8 | +} |
| 9 | + |
| 10 | +export interface WorkflowState { |
| 11 | + workflowId: string; |
| 12 | + steps: Record<string, WorkflowStepRecord>; |
| 13 | + startedAt: string; |
| 14 | +} |
| 15 | + |
| 16 | +const WORKFLOW_KEY_PREFIX = 'workflow-idempotency'; |
| 17 | +/** Default TTL: 7 days — long enough to cover asynchronous multi-step retry windows. */ |
| 18 | +const DEFAULT_TTL_MS = 7 * 24 * 60 * 60 * 1000; |
| 19 | + |
| 20 | +/** |
| 21 | + * WorkflowIdempotencyService |
| 22 | + * |
| 23 | + * Provides step-level idempotency for multi-step workflows. Each workflow |
| 24 | + * is identified by a `workflowId`. Individual steps within the workflow |
| 25 | + * are identified by a `stepName`. On retry, callers can skip already- |
| 26 | + * completed steps rather than re-executing them. |
| 27 | + * |
| 28 | + * Usage pattern: |
| 29 | + * |
| 30 | + * const wf = workflowIdempotencyService; |
| 31 | + * const wfId = `deposit-${idempotencyKey}`; |
| 32 | + * |
| 33 | + * if (!(await wf.isStepCompleted(wfId, 'validate'))) { |
| 34 | + * await validateDeposit(payload); |
| 35 | + * await wf.markStepCompleted(wfId, 'validate'); |
| 36 | + * } |
| 37 | + * |
| 38 | + * if (!(await wf.isStepCompleted(wfId, 'reserve'))) { |
| 39 | + * const reservation = await reserveFunds(payload); |
| 40 | + * await wf.markStepCompleted(wfId, 'reserve', { reservationId: reservation.id }); |
| 41 | + * } |
| 42 | + * |
| 43 | + * // ... further steps |
| 44 | + * await wf.clearWorkflow(wfId); // optional cleanup after success |
| 45 | + * |
| 46 | + * The workflow state is stored in the shared cache (Redis in production, |
| 47 | + * in-memory in tests) under the key `workflow-idempotency:<workflowId>`. |
| 48 | + */ |
| 49 | +@Injectable() |
| 50 | +export class WorkflowIdempotencyService { |
| 51 | + private readonly logger = new Logger(WorkflowIdempotencyService.name); |
| 52 | + |
| 53 | + constructor(@Inject(CACHE_MANAGER) private readonly cache: Cache) {} |
| 54 | + |
| 55 | + /** |
| 56 | + * Returns true if the given step has already been completed for the |
| 57 | + * specified workflow. Callers should skip execution of the step when |
| 58 | + * this returns true. |
| 59 | + */ |
| 60 | + async isStepCompleted( |
| 61 | + workflowId: string, |
| 62 | + stepName: string, |
| 63 | + ): Promise<boolean> { |
| 64 | + const state = await this.getWorkflowState(workflowId); |
| 65 | + return state !== null && stepName in state.steps; |
| 66 | + } |
| 67 | + |
| 68 | + /** |
| 69 | + * Marks a workflow step as completed, optionally persisting a small |
| 70 | + * result payload (e.g. an ID or status) for downstream steps to read. |
| 71 | + * |
| 72 | + * @param workflowId Unique identifier for the workflow execution. |
| 73 | + * @param stepName Name of the step being marked complete. |
| 74 | + * @param result Optional serialisable result to store with the step. |
| 75 | + * @param ttlMs Time-to-live for the whole workflow record in ms. |
| 76 | + */ |
| 77 | + async markStepCompleted( |
| 78 | + workflowId: string, |
| 79 | + stepName: string, |
| 80 | + result?: unknown, |
| 81 | + ttlMs = DEFAULT_TTL_MS, |
| 82 | + ): Promise<void> { |
| 83 | + const state = (await this.getWorkflowState(workflowId)) ?? { |
| 84 | + workflowId, |
| 85 | + steps: {}, |
| 86 | + startedAt: new Date().toISOString(), |
| 87 | + }; |
| 88 | + |
| 89 | + state.steps[stepName] = { |
| 90 | + completedAt: new Date().toISOString(), |
| 91 | + result, |
| 92 | + }; |
| 93 | + |
| 94 | + await this.cache.set(this.cacheKey(workflowId), state, ttlMs); |
| 95 | + this.logger.debug( |
| 96 | + `Workflow ${workflowId}: step "${stepName}" marked completed`, |
| 97 | + ); |
| 98 | + } |
| 99 | + |
| 100 | + /** |
| 101 | + * Returns the stored result for a previously completed step, or |
| 102 | + * `undefined` if the step has not been completed or stored no result. |
| 103 | + */ |
| 104 | + async getStepResult<T = unknown>( |
| 105 | + workflowId: string, |
| 106 | + stepName: string, |
| 107 | + ): Promise<T | undefined> { |
| 108 | + const state = await this.getWorkflowState(workflowId); |
| 109 | + return state?.steps[stepName]?.result as T | undefined; |
| 110 | + } |
| 111 | + |
| 112 | + /** |
| 113 | + * Returns the full workflow state, or null if no state is stored for |
| 114 | + * the given workflowId. |
| 115 | + */ |
| 116 | + async getWorkflowState(workflowId: string): Promise<WorkflowState | null> { |
| 117 | + return ( |
| 118 | + ((await this.cache.get(this.cacheKey(workflowId))) as WorkflowState) ?? |
| 119 | + null |
| 120 | + ); |
| 121 | + } |
| 122 | + |
| 123 | + /** |
| 124 | + * Returns the set of step names that have been completed for the given |
| 125 | + * workflow, or an empty array if the workflow has no stored state. |
| 126 | + */ |
| 127 | + async getCompletedSteps(workflowId: string): Promise<string[]> { |
| 128 | + const state = await this.getWorkflowState(workflowId); |
| 129 | + return state ? Object.keys(state.steps) : []; |
| 130 | + } |
| 131 | + |
| 132 | + /** |
| 133 | + * Removes all stored state for a workflow. Call this after a workflow |
| 134 | + * completes successfully to free cache space; the TTL will clean it up |
| 135 | + * automatically if this is not called. |
| 136 | + */ |
| 137 | + async clearWorkflow(workflowId: string): Promise<void> { |
| 138 | + await this.cache.del(this.cacheKey(workflowId)); |
| 139 | + this.logger.debug(`Workflow ${workflowId}: state cleared`); |
| 140 | + } |
| 141 | + |
| 142 | + private cacheKey(workflowId: string): string { |
| 143 | + return `${WORKFLOW_KEY_PREFIX}:${workflowId}`; |
| 144 | + } |
| 145 | +} |
0 commit comments