From 765ec37c3a07784e93e94c6184b7011ef60cc17f Mon Sep 17 00:00:00 2001 From: valentinbvro Date: Wed, 29 Jul 2026 15:02:56 +0200 Subject: [PATCH] feat: Saga Manager (Sprint 4-6) Motor generic pentru procese cu mai multi pasi care nu incap intr-o tranzactie. Garantia nu e atomicitatea, ci compensarea in ordine inversa. - saga-registry: definitii versionate in cod, backoff exponential plafonat - saga-runner: corelator outbox->instanta, step runner, timeout, retry, compensare; revendicare cu FOR UPDATE SKIP LOCKED ca doua procese sa nu avanseze aceeasi saga simultan - idempotenta la pornire prin index unic (saga, versiune, trigger_event), nu prin SELECT-apoi-INSERT care ar avea race - compensarea are propriul retry: o compensare esuata lasa sistemul mai rau decat esecul original - monitor /v1/sagas cu instante blocate si retry manual care NU sare pasi - prima saga reala: imbogatire research brief, fara pas AI (un apel AI automat per brief ar schimba profilul de cost) --- drizzle/0007_saga_manager.sql | 61 +++ drizzle/meta/_journal.json | 7 + src/app.module.ts | 2 + src/db/schema.ts | 86 ++++ src/sagas/research-brief-enrichment.saga.ts | 162 +++++++ src/sagas/saga-registry.spec.ts | 75 +++ src/sagas/saga-registry.ts | 108 +++++ src/sagas/saga-runner.service.ts | 485 ++++++++++++++++++++ src/sagas/sagas.controller.ts | 40 ++ src/sagas/sagas.module.ts | 26 ++ src/sagas/sagas.service.ts | 144 ++++++ 11 files changed, 1196 insertions(+) create mode 100644 drizzle/0007_saga_manager.sql create mode 100644 src/sagas/research-brief-enrichment.saga.ts create mode 100644 src/sagas/saga-registry.spec.ts create mode 100644 src/sagas/saga-registry.ts create mode 100644 src/sagas/saga-runner.service.ts create mode 100644 src/sagas/sagas.controller.ts create mode 100644 src/sagas/sagas.module.ts create mode 100644 src/sagas/sagas.service.ts diff --git a/drizzle/0007_saga_manager.sql b/drizzle/0007_saga_manager.sql new file mode 100644 index 0000000..450dbc8 --- /dev/null +++ b/drizzle/0007_saga_manager.sql @@ -0,0 +1,61 @@ +-- Saga Manager (Sprint 4-6). Coordoneaza procese cu mai multi pasi care nu +-- incap intr-o tranzactie. Garantia nu e atomicitatea, ci compensarea in +-- ordine inversa a pasilor deja executati. + +DO $$ BEGIN + CREATE TYPE "saga_status" AS ENUM + ('RUNNING','COMPLETED','COMPENSATING','COMPENSATED','FAILED','TIMED_OUT'); +EXCEPTION WHEN duplicate_object THEN NULL; END $$; + +DO $$ BEGIN + CREATE TYPE "saga_step_status" AS ENUM + ('PENDING','RUNNING','COMPLETED','FAILED','COMPENSATED','SKIPPED'); +EXCEPTION WHEN duplicate_object THEN NULL; END $$; + +CREATE TABLE IF NOT EXISTS "saga_instances" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "tenant_id" uuid NOT NULL, + "workspace_id" uuid, + "saga_name" text NOT NULL, + "saga_version" integer NOT NULL, + "status" "saga_status" DEFAULT 'RUNNING' NOT NULL, + "current_step" integer DEFAULT 0 NOT NULL, + "trigger_event_id" uuid NOT NULL, + "correlation_id" uuid, + "context" jsonb DEFAULT '{}'::jsonb NOT NULL, + "attempt" integer DEFAULT 0 NOT NULL, + "next_attempt_at" timestamp DEFAULT now() NOT NULL, + "timeout_at" timestamp, + "last_error" text, + "locked_until" timestamp, + "created_at" timestamp DEFAULT now() NOT NULL, + "updated_at" timestamp DEFAULT now() NOT NULL, + "completed_at" timestamp +); + +-- Un eveniment porneste o singura instanta dintr-o saga data. Aceasta +-- constrangere e ce face pornirea idempotenta, nu o verificare in cod. +CREATE UNIQUE INDEX IF NOT EXISTS "saga_trigger_uq" + ON "saga_instances" ("saga_name","saga_version","trigger_event_id"); +CREATE INDEX IF NOT EXISTS "saga_due_idx" + ON "saga_instances" ("status","next_attempt_at"); +CREATE INDEX IF NOT EXISTS "saga_tenant_idx" + ON "saga_instances" ("tenant_id","created_at"); + +CREATE TABLE IF NOT EXISTS "saga_steps" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "saga_instance_id" uuid NOT NULL, + "step_name" text NOT NULL, + "sequence" integer NOT NULL, + "status" "saga_step_status" DEFAULT 'PENDING' NOT NULL, + "attempts" integer DEFAULT 0 NOT NULL, + "output" jsonb, + "last_error" text, + "started_at" timestamp, + "completed_at" timestamp, + "compensated_at" timestamp +); + +-- Un pas apare o singura data per instanta: baza idempotentei la avans. +CREATE UNIQUE INDEX IF NOT EXISTS "saga_step_uq" + ON "saga_steps" ("saga_instance_id","sequence"); diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index adc9312..720753e 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -50,6 +50,13 @@ "when": 1785328533134, "tag": "0006_notifications_engine", "breakpoints": true + }, + { + "idx": 7, + "version": "7", + "when": 1785331800000, + "tag": "0007_saga_manager", + "breakpoints": true } ] } \ No newline at end of file diff --git a/src/app.module.ts b/src/app.module.ts index b93812c..c10f2c2 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -16,6 +16,7 @@ import { TenantsModule } from './tenants/tenants.module'; import { NavigationModule } from './navigation/navigation.module'; import { ProjectionsModule } from './projections/projections.module'; import { NotificationsModule } from './notifications/notifications.module'; +import { SagasModule } from './sagas/sagas.module'; import { IntelligenceModule } from './intelligence/intelligence.module'; import { SegmentsModule } from './segments/segments.module'; import { ResearchBriefsModule } from './research-briefs/research-briefs.module'; @@ -50,6 +51,7 @@ function parseRedisConnection(redisUrl: string | undefined) { NavigationModule, ProjectionsModule, NotificationsModule, + SagasModule, OrganizationsModule, TasksModule, IntelligenceModule, diff --git a/src/db/schema.ts b/src/db/schema.ts index cbe142c..159aa7e 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -471,6 +471,92 @@ export const notificationProcessedEvents = pgTable( (table) => [uniqueIndex('notification_processed_uq').on(table.ruleId, table.eventId)], ); +// --- Saga Manager (spec Sprint 4-6) --------------------------------------- +// O saga coordoneaza un proces cu mai multi pasi care NU incap intr-o singura +// tranzactie (apeluri externe, latenta, pasi care pot esua independent). +// Garantia nu e atomicitatea, ci: ori toti pasii reusesc, ori cei deja executati +// sunt compensati in ordine inversa. + +export const sagaStatus = pgEnum('saga_status', [ + 'RUNNING', + 'COMPLETED', + 'COMPENSATING', + 'COMPENSATED', + 'FAILED', + 'TIMED_OUT', +]); + +export const sagaStepStatus = pgEnum('saga_step_status', [ + 'PENDING', + 'RUNNING', + 'COMPLETED', + 'FAILED', + 'COMPENSATED', + 'SKIPPED', +]); + +export const sagaInstances = pgTable( + 'saga_instances', + { + id: uuid('id').defaultRandom().primaryKey(), + tenantId: uuid('tenant_id').notNull(), + workspaceId: uuid('workspace_id'), + sagaName: text('saga_name').notNull(), + sagaVersion: integer('saga_version').notNull(), + status: sagaStatus('status').notNull().default('RUNNING'), + /** Indexul pasului curent in definitie. */ + currentStep: integer('current_step').notNull().default(0), + /** Evenimentul care a pornit saga -- baza idempotentei la start. */ + triggerEventId: uuid('trigger_event_id').notNull(), + correlationId: uuid('correlation_id'), + /** Starea acumulata, citita si scrisa de pasi. */ + context: jsonb('context').notNull().default({}), + attempt: integer('attempt').notNull().default(0), + /** Cand poate fi reluata: backoff exponential dupa esec. */ + nextAttemptAt: timestamp('next_attempt_at').defaultNow().notNull(), + /** Termen absolut pentru intreaga saga, nu doar pentru pasul curent. */ + timeoutAt: timestamp('timeout_at'), + lastError: text('last_error'), + /** + * Lease de executie. Doua procese runner nu trebuie sa avanseze aceeasi + * saga simultan; randul se ia cu FOR UPDATE SKIP LOCKED, iar lockedUntil + * elibereaza saga daca procesul care o tinea a murit. + */ + lockedUntil: timestamp('locked_until'), + createdAt: timestamp('created_at').defaultNow().notNull(), + updatedAt: timestamp('updated_at').defaultNow().notNull(), + completedAt: timestamp('completed_at'), + }, + (table) => [ + // Un eveniment porneste o singura instanta dintr-o saga data. + uniqueIndex('saga_trigger_uq').on(table.sagaName, table.sagaVersion, table.triggerEventId), + index('saga_due_idx').on(table.status, table.nextAttemptAt), + index('saga_tenant_idx').on(table.tenantId, table.createdAt), + ], +); + +export const sagaSteps = pgTable( + 'saga_steps', + { + id: uuid('id').defaultRandom().primaryKey(), + sagaInstanceId: uuid('saga_instance_id').notNull(), + stepName: text('step_name').notNull(), + /** Pozitia in definitie; determina si ordinea inversa a compensarii. */ + sequence: integer('sequence').notNull(), + status: sagaStepStatus('status').notNull().default('PENDING'), + attempts: integer('attempts').notNull().default(0), + output: jsonb('output'), + lastError: text('last_error'), + startedAt: timestamp('started_at'), + completedAt: timestamp('completed_at'), + compensatedAt: timestamp('compensated_at'), + }, + (table) => [ + // Un pas apare o singura data per instanta: baza idempotentei la avans. + uniqueIndex('saga_step_uq').on(table.sagaInstanceId, table.sequence), + ], +); + // Append-oriented (spec sectiunea 29): se scrie, nu se modifica si nu se sterge. export const auditLog = pgTable( 'audit_log', diff --git a/src/sagas/research-brief-enrichment.saga.ts b/src/sagas/research-brief-enrichment.saga.ts new file mode 100644 index 0000000..49bba1c --- /dev/null +++ b/src/sagas/research-brief-enrichment.saga.ts @@ -0,0 +1,162 @@ +import { Logger } from '@nestjs/common'; +import { eq } from 'drizzle-orm'; +import { db } from '../db/client'; +import { outboxEvents, researchBriefs } from '../db/schema'; +import type { IntelligenceService } from '../intelligence/intelligence.service'; +import { registerSaga, type SagaDefinition } from './saga-registry'; + +const logger = new Logger('ResearchBriefEnrichmentSaga'); + +/** Marcaj in sources care arata ca imbogatirea a rulat. Baza idempotentei. */ +const ENRICHMENT_LABEL = 'intelligence-api (enrichment)'; + +interface SourceEntry { + label: string; + url?: string; +} + +/** + * Imbogatirea unui research brief dupa creare. + * + * De ce saga si nu un simplu handler: sunt trei pasi din care primul apeleaza + * un serviciu extern care poate fi lent sau picat. Daca pasul 2 esueaza dupa ce + * pasul 1 a scris deja in brief, un handler obisnuit ar lasa brieful marcat ca + * imbogatit fara sa fie. Saga anuleaza scrierea. + * + * NU contine pas de AI in mod deliberat: un apel AI declansat automat la fiecare + * brief creat ar schimba profilul de cost. Draftul AI ramane pe actiunea + * manuala existenta (POST .../draft). + */ +export function buildResearchBriefEnrichmentSaga( + intelligence: IntelligenceService, +): SagaDefinition { + return { + name: 'research_brief_enrichment', + version: 1, + triggerEvent: 'research_brief.created', + timeoutMinutes: 30, + + shouldStart(event) { + // Fara organizationId nu avem ce imbogati; nu pornim o saga care ar + // esua garantat la primul pas. + return typeof event.payload.organizationId === 'string' && !!event.payload.organizationId; + }, + + steps: [ + { + name: 'fetch_company_intelligence', + // Pasul care are cea mai mare sansa de esec tranzitoriu. + maxAttempts: 4, + async run(_ctx, event) { + const organizationId = String(event.payload.organizationId); + const company = await intelligence.getCompany(organizationId); + return { + context: { + company: { + name: company.organization_name ?? null, + domain: company.normalized_domain ?? null, + industries: company.industries ?? [], + hqCity: company.hq_city ?? null, + hqCountry: company.hq_country ?? null, + }, + }, + output: { organizationId, resolved: !!company.organization_name }, + }; + }, + // Citire pura: nu are efecte de anulat. + }, + + { + name: 'attach_provenance', + async run(ctx, event) { + const briefId = await resolveBriefId(event.id); + if (!briefId) throw new Error('Briefu-l declansator nu mai exista'); + + const brief = await db.query.researchBriefs.findFirst({ + where: eq(researchBriefs.id, briefId), + }); + if (!brief) throw new Error(`Brief ${briefId} nu a fost gasit`); + + const sources = (brief.sources ?? []) as SourceEntry[]; + // Idempotent: a doua rulare nu adauga a doua oara aceeasi sursa. + if (sources.some((s) => s.label === ENRICHMENT_LABEL)) { + return { context: { briefId }, output: { alreadyAttached: true } }; + } + + const company = (ctx.company ?? {}) as Record; + await db + .update(researchBriefs) + .set({ + sources: [ + ...sources, + { + label: ENRICHMENT_LABEL, + url: 'https://intelligence-api.boardmind.dev', + resolvedName: company.name ?? null, + resolvedDomain: company.domain ?? null, + }, + ], + organizationName: (company.name as string) ?? brief.organizationName, + updatedAt: new Date(), + }) + .where(eq(researchBriefs.id, briefId)); + + return { context: { briefId }, output: { attached: true } }; + }, + async compensate(ctx) { + const briefId = ctx.briefId as string | undefined; + if (!briefId) return; + const brief = await db.query.researchBriefs.findFirst({ + where: eq(researchBriefs.id, briefId), + }); + // Briefu-l poate fi sters intre timp: compensam ce gasim, nu presupunem. + if (!brief) return; + + const sources = (brief.sources ?? []) as SourceEntry[]; + await db + .update(researchBriefs) + .set({ + sources: sources.filter((s) => s.label !== ENRICHMENT_LABEL), + updatedAt: new Date(), + }) + .where(eq(researchBriefs.id, briefId)); + logger.log(`Provenance retrasa de pe briefu-l ${briefId}`); + }, + }, + + { + name: 'emit_enriched_event', + async run(ctx, event) { + const briefId = ctx.briefId as string | undefined; + if (!briefId) throw new Error('Lipseste briefId din context'); + + // Evenimentul e consumat de Notifications Engine si de proiectii; + // saga nu trimite ea notificari, ci anunta faptul. + await db.insert(outboxEvents).values({ + tenantId: event.tenantId, + workspaceId: event.workspaceId, + eventType: 'research_brief.enriched', + aggregateType: 'research_brief', + subjectId: briefId, + correlationId: event.correlationId, + causationId: event.id, + payload: { briefId, company: ctx.company ?? null }, + }); + return { output: { emitted: true } }; + }, + }, + ], + }; +} + +/** subject_id al evenimentului declansator = id-ul briefului. */ +async function resolveBriefId(triggerEventId: string): Promise { + const row = await db.query.outboxEvents.findFirst({ + where: eq(outboxEvents.id, triggerEventId), + }); + return row?.subjectId ?? null; +} + +export function registerResearchBriefEnrichment(intelligence: IntelligenceService): void { + registerSaga(buildResearchBriefEnrichmentSaga(intelligence)); +} diff --git a/src/sagas/saga-registry.spec.ts b/src/sagas/saga-registry.spec.ts new file mode 100644 index 0000000..7be4f5b --- /dev/null +++ b/src/sagas/saga-registry.spec.ts @@ -0,0 +1,75 @@ +import { + backoffDelayMs, + findSaga, + maxAttemptsFor, + registerSaga, + resetRegistry, + sagasForEvent, + triggerEventTypes, + type SagaDefinition, +} from './saga-registry'; + +const stubSaga = (over: Partial = {}): SagaDefinition => ({ + name: 'test_saga', + version: 1, + triggerEvent: 'thing.created', + timeoutMinutes: 10, + steps: [{ name: 'step_one', run: async () => {} }], + ...over, +}); + +describe('saga registry', () => { + beforeEach(() => resetRegistry()); + + it('gaseste sagas dupa evenimentul declansator', () => { + registerSaga(stubSaga()); + expect(sagasForEvent('thing.created')).toHaveLength(1); + expect(sagasForEvent('altceva.created')).toEqual([]); + }); + + it('refuza inregistrarea aceleiasi sagas de doua ori pe aceeasi versiune', () => { + // Doua definitii cu acelasi nume+versiune ar porni doua instante per + // eveniment; e o eroare de programare, deci esueaza tare la bootstrap. + registerSaga(stubSaga()); + expect(() => registerSaga(stubSaga())).toThrow(/deja inregistrata/); + }); + + it('permite versiuni diferite ale aceleiasi sagas sa coexiste', () => { + registerSaga(stubSaga({ version: 1 })); + registerSaga(stubSaga({ version: 2 })); + expect(findSaga('test_saga', 1)?.version).toBe(1); + expect(findSaga('test_saga', 2)?.version).toBe(2); + }); + + it('deduplica tipurile de evenimente la care e abonat macar o saga', () => { + registerSaga(stubSaga({ name: 'a' })); + registerSaga(stubSaga({ name: 'b' })); + expect(triggerEventTypes()).toEqual(['thing.created']); + }); + + describe('backoff', () => { + it('creste exponential intre incercari', () => { + expect(backoffDelayMs(1)).toBe(30_000); + expect(backoffDelayMs(2)).toBe(60_000); + expect(backoffDelayMs(3)).toBe(120_000); + }); + + it('se plafoneaza, ca reincercarile tarzii sa nu ajunga peste zile', () => { + expect(backoffDelayMs(50)).toBe(30 * 60_000); + }); + + it('nu intoarce delay negativ pentru incercarea zero', () => { + expect(backoffDelayMs(0)).toBeGreaterThan(0); + }); + }); + + describe('maxAttempts', () => { + it('are un default cand pasul nu specifica', () => { + expect(maxAttemptsFor({ name: 's', run: async () => {} })).toBe(3); + }); + + it('respecta valoarea explicita a pasului', () => { + expect(maxAttemptsFor({ name: 's', run: async () => {}, maxAttempts: 7 })).toBe(7); + }); + }); +}); diff --git a/src/sagas/saga-registry.ts b/src/sagas/saga-registry.ts new file mode 100644 index 0000000..f68e5a3 --- /dev/null +++ b/src/sagas/saga-registry.ts @@ -0,0 +1,108 @@ +/** + * Definitiile de saga stau in cod, versionate, ca la notification-rules si + * projection-registry. Motivul e acelasi: un proces cu mai multi pasi e logica + * de business care merita review si teste, nu configuratie editabila la runtime. + * + * Contractul unui pas: + * - run() trebuie sa fie IDEMPOTENT. Poate fi apelat de mai multe ori pentru + * acelasi pas (retry dupa timeout de retea, restart de proces). Daca a scris + * deja ceva, a doua rulare trebuie sa observe asta si sa nu dubleze. + * - compensate() anuleaza efectul lui run(). Trebuie sa fie idempotent SI + * tolerant la faptul ca run() poate sa fi esuat la jumatate -- compenseaza + * ce gaseste, nu presupune ca totul a fost creat. + */ + +export interface SagaEvent { + id: string; + tenantId: string; + workspaceId: string | null; + eventType: string; + actorId: string | null; + subjectId: string | null; + correlationId: string | null; + payload: Record; +} + +export type SagaContext = Record; + +export interface SagaStepResult { + /** Se uneste in contextul sagai si e vizibil pasilor urmatori. */ + context?: SagaContext; + output?: Record; +} + +export interface SagaStepDefinition { + name: string; + run(ctx: SagaContext, event: SagaEvent): Promise; + /** + * Lipsa lui compensate inseamna ca pasul nu are efecte de anulat (o citire, + * de exemplu). Nu inseamna "nu stim cum sa compensam". + */ + compensate?(ctx: SagaContext, event: SagaEvent): Promise; + /** Cate incercari inainte de a declara pasul esuat. */ + maxAttempts?: number; +} + +export interface SagaDefinition { + name: string; + version: number; + /** Ce eveniment porneste saga. */ + triggerEvent: string; + /** Termen absolut pentru toata saga; depasirea duce la compensare. */ + timeoutMinutes: number; + steps: SagaStepDefinition[]; + /** + * Filtru optional: chiar daca evenimentul se potriveste, saga poate decide + * ca nu o priveste (ex. lipseste un camp din payload). + */ + shouldStart?(event: SagaEvent): boolean; +} + +const DEFAULT_MAX_ATTEMPTS = 3; +const BASE_BACKOFF_MS = 30_000; +const MAX_BACKOFF_MS = 30 * 60_000; + +export function maxAttemptsFor(step: SagaStepDefinition): number { + return step.maxAttempts ?? DEFAULT_MAX_ATTEMPTS; +} + +/** + * Backoff exponential plafonat. Fara plafon, a 10-a incercare ar fi programata + * peste zile, ceea ce in practica inseamna "niciodata". + */ +export function backoffDelayMs(attempt: number): number { + return Math.min(BASE_BACKOFF_MS * 2 ** Math.max(0, attempt - 1), MAX_BACKOFF_MS); +} + +const REGISTRY: SagaDefinition[] = []; + +export function registerSaga(definition: SagaDefinition): void { + const duplicate = REGISTRY.find( + (s) => s.name === definition.name && s.version === definition.version, + ); + if (duplicate) { + throw new Error(`Saga ${definition.name}@v${definition.version} este deja inregistrata`); + } + REGISTRY.push(definition); +} + +export function allSagas(): SagaDefinition[] { + return [...REGISTRY]; +} + +export function sagasForEvent(eventType: string): SagaDefinition[] { + return REGISTRY.filter((s) => s.triggerEvent === eventType); +} + +export function findSaga(name: string, version: number): SagaDefinition | undefined { + return REGISTRY.find((s) => s.name === name && s.version === version); +} + +export function triggerEventTypes(): string[] { + return [...new Set(REGISTRY.map((s) => s.triggerEvent))]; +} + +/** Doar pentru teste: goleste registrul intre cazuri. */ +export function resetRegistry(): void { + REGISTRY.length = 0; +} diff --git a/src/sagas/saga-runner.service.ts b/src/sagas/saga-runner.service.ts new file mode 100644 index 0000000..7172434 --- /dev/null +++ b/src/sagas/saga-runner.service.ts @@ -0,0 +1,485 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { Cron, CronExpression } from '@nestjs/schedule'; +import { and, asc, desc, eq, gt, inArray, sql } from 'drizzle-orm'; +import { db } from '../db/client'; +import { outboxEvents, sagaInstances, sagaSteps } from '../db/schema'; +import { + allSagas, + backoffDelayMs, + findSaga, + maxAttemptsFor, + sagasForEvent, + triggerEventTypes, + type SagaContext, + type SagaDefinition, + type SagaEvent, +} from './saga-registry'; + +const BATCH_SIZE = 50; +const CORRELATOR_BATCH = 200; +const SAFETY_LAG_MS = 60_000; +/** Cat timp o instanta revendicata ramane a acestui proces. */ +const LEASE_MS = 5 * 60_000; + +type SagaInstanceRow = typeof sagaInstances.$inferSelect; + +@Injectable() +export class SagaRunnerService { + private readonly logger = new Logger(SagaRunnerService.name); + private lastSeenAt: Date | null = null; + + // --- Corelator: eveniment din outbox -> instanta noua de saga ------------- + + @Cron(CronExpression.EVERY_10_SECONDS) + async correlate(): Promise { + try { + await this.correlateBatch(); + } catch (error) { + this.logger.error('Saga correlator failed', error as Error); + } + } + + async correlateBatch(): Promise { + const eventTypes = triggerEventTypes(); + if (eventTypes.length === 0) return 0; + + const since = this.lastSeenAt + ? new Date(this.lastSeenAt.getTime() - SAFETY_LAG_MS) + : new Date(Date.now() - 24 * 60 * 60 * 1000); + + const events = await db + .select() + .from(outboxEvents) + .where(and(gt(outboxEvents.createdAt, since), inArray(outboxEvents.eventType, eventTypes))) + .orderBy(asc(outboxEvents.createdAt)) + .limit(CORRELATOR_BATCH); + + let started = 0; + for (const row of events) { + const event = this.toSagaEvent(row); + for (const definition of sagasForEvent(event.eventType)) { + if (definition.shouldStart && !definition.shouldStart(event)) continue; + started += await this.startInstance(definition, event); + } + this.lastSeenAt = row.createdAt; + } + + if (started > 0) this.logger.log(`Started ${started} saga instance(s)`); + return started; + } + + private async startInstance(definition: SagaDefinition, event: SagaEvent): Promise { + // Idempotenta la pornire vine din indexul unic (saga, versiune, eveniment), + // nu dintr-un SELECT-apoi-INSERT care ar avea race intre doua procese. + const inserted = await db + .insert(sagaInstances) + .values({ + tenantId: event.tenantId, + workspaceId: event.workspaceId, + sagaName: definition.name, + sagaVersion: definition.version, + triggerEventId: event.id, + correlationId: event.correlationId, + context: { event: event.payload }, + timeoutAt: new Date(Date.now() + definition.timeoutMinutes * 60_000), + }) + .onConflictDoNothing() + .returning({ id: sagaInstances.id }); + return inserted.length; + } + + // --- Step runner --------------------------------------------------------- + + @Cron(CronExpression.EVERY_10_SECONDS) + async advance(): Promise { + try { + await this.advanceBatch(); + } catch (error) { + this.logger.error('Saga runner failed', error as Error); + } + } + + async advanceBatch(): Promise { + const claimed = await this.claimDueInstances(BATCH_SIZE); + let processed = 0; + for (const instance of claimed) { + try { + await this.processInstance(instance); + } catch (error) { + this.logger.error(`Saga ${instance.sagaName}/${instance.id} crashed`, error as Error); + await this.releaseWithError(instance, error); + } + processed += 1; + } + return processed; + } + + /** + * Revendica instantele scadente intr-un singur UPDATE atomic. + * + * FOR UPDATE SKIP LOCKED e ce impiedica doua procese runner sa avanseze + * aceeasi saga simultan: al doilea proces sare peste randurile blocate in loc + * sa astepte. lockedUntil e plasa de siguranta pentru cazul in care procesul + * care detinea lease-ul moare fara sa il elibereze. + */ + private async claimDueInstances(limit: number): Promise { + const leaseUntil = new Date(Date.now() + LEASE_MS); + const result = await db.execute(sql` + UPDATE saga_instances SET locked_until = ${leaseUntil}, updated_at = now() + WHERE id IN ( + SELECT id FROM saga_instances + WHERE status IN ('RUNNING','COMPENSATING') + AND next_attempt_at <= now() + AND (locked_until IS NULL OR locked_until < now()) + ORDER BY next_attempt_at ASC + LIMIT ${limit} + FOR UPDATE SKIP LOCKED + ) + RETURNING *; + `); + return (result.rows as unknown[]).map((row) => this.mapInstance(row as Record)); + } + + private async processInstance(instance: SagaInstanceRow): Promise { + const definition = findSaga(instance.sagaName, instance.sagaVersion); + if (!definition) { + // Definitia a disparut sau a fost re-versionata sub o instanta activa. + // Nu ghicim ce ar fi trebuit sa faca: o marcam si o lasam vizibila. + await this.fail(instance, `Definitia ${instance.sagaName}@v${instance.sagaVersion} lipseste`); + return; + } + + if (instance.status === 'COMPENSATING') { + await this.compensateNext(instance, definition); + return; + } + + // Timeoutul se verifica inainte de a mai executa un pas: o saga expirata + // nu trebuie sa mai produca efecte noi, ci sa le anuleze pe cele vechi. + if (instance.timeoutAt && instance.timeoutAt < new Date()) { + this.logger.warn(`Saga ${instance.sagaName}/${instance.id} a expirat; compensez`); + await this.beginCompensation(instance, 'Saga a depasit timeout-ul'); + return; + } + + const step = definition.steps[instance.currentStep]; + if (!step) { + await this.complete(instance); + return; + } + + const event = this.eventFromContext(instance); + const context = (instance.context ?? {}) as SagaContext; + const attempts = instance.attempt + 1; + + await this.upsertStep(instance, step.name, instance.currentStep, 'RUNNING', attempts); + + try { + const result = await step.run(context, event); + const merged = { ...context, ...(result?.context ?? {}) }; + + await db.transaction(async (tx) => { + await tx + .update(sagaSteps) + .set({ + status: 'COMPLETED', + output: result?.output ?? null, + completedAt: new Date(), + lastError: null, + }) + .where( + and( + eq(sagaSteps.sagaInstanceId, instance.id), + eq(sagaSteps.sequence, instance.currentStep), + ), + ); + + await tx + .update(sagaInstances) + .set({ + currentStep: instance.currentStep + 1, + context: merged, + attempt: 0, + lastError: null, + lockedUntil: null, + nextAttemptAt: new Date(), + updatedAt: new Date(), + }) + .where(eq(sagaInstances.id, instance.id)); + }); + } catch (error) { + await this.handleStepFailure(instance, definition, step.name, attempts, error); + } + } + + private async handleStepFailure( + instance: SagaInstanceRow, + definition: SagaDefinition, + stepName: string, + attempts: number, + error: unknown, + ): Promise { + const message = error instanceof Error ? error.message : String(error); + const step = definition.steps[instance.currentStep]; + const exhausted = attempts >= maxAttemptsFor(step); + + await db + .update(sagaSteps) + .set({ status: exhausted ? 'FAILED' : 'PENDING', lastError: message }) + .where( + and(eq(sagaSteps.sagaInstanceId, instance.id), eq(sagaSteps.sequence, instance.currentStep)), + ); + + if (!exhausted) { + const retryAt = new Date(Date.now() + backoffDelayMs(attempts)); + this.logger.warn( + `Saga ${definition.name}/${instance.id} pas "${stepName}" a esuat (${attempts}/${maxAttemptsFor(step)}); reincerc la ${retryAt.toISOString()}`, + ); + await db + .update(sagaInstances) + .set({ + attempt: attempts, + lastError: message, + nextAttemptAt: retryAt, + lockedUntil: null, + updatedAt: new Date(), + }) + .where(eq(sagaInstances.id, instance.id)); + return; + } + + this.logger.error( + `Saga ${definition.name}/${instance.id} pas "${stepName}" epuizat dupa ${attempts} incercari; compensez`, + ); + await this.beginCompensation(instance, `Pasul "${stepName}" a esuat: ${message}`); + } + + // --- Compensare ---------------------------------------------------------- + + private async beginCompensation(instance: SagaInstanceRow, reason: string): Promise { + await db + .update(sagaInstances) + .set({ + status: 'COMPENSATING', + lastError: reason, + attempt: 0, + nextAttemptAt: new Date(), + lockedUntil: null, + updatedAt: new Date(), + }) + .where(eq(sagaInstances.id, instance.id)); + } + + /** + * Compenseaza un singur pas per rulare, in ordine inversa, si lasa instanta + * pentru urmatorul tur. Compensarea are propriul retry: o compensare esuata + * lasa sistemul intr-o stare mai proasta decat esecul original, deci nu o + * abandonam dupa prima incercare. + */ + private async compensateNext( + instance: SagaInstanceRow, + definition: SagaDefinition, + ): Promise { + const [pending] = await db + .select() + .from(sagaSteps) + .where(and(eq(sagaSteps.sagaInstanceId, instance.id), eq(sagaSteps.status, 'COMPLETED'))) + .orderBy(desc(sagaSteps.sequence)) + .limit(1); + + if (!pending) { + await db + .update(sagaInstances) + .set({ + status: 'COMPENSATED', + completedAt: new Date(), + lockedUntil: null, + updatedAt: new Date(), + }) + .where(eq(sagaInstances.id, instance.id)); + this.logger.log(`Saga ${definition.name}/${instance.id} compensata complet`); + return; + } + + const step = definition.steps[pending.sequence]; + const context = (instance.context ?? {}) as SagaContext; + const event = this.eventFromContext(instance); + const attempts = instance.attempt + 1; + + try { + // Un pas fara compensate() nu are efecte de anulat; il marcam si mergem + // mai departe, nu il tratam ca esec. + if (step?.compensate) { + await step.compensate(context, event); + } + await db + .update(sagaSteps) + .set({ status: 'COMPENSATED', compensatedAt: new Date() }) + .where(eq(sagaSteps.id, pending.id)); + await db + .update(sagaInstances) + .set({ attempt: 0, lockedUntil: null, nextAttemptAt: new Date(), updatedAt: new Date() }) + .where(eq(sagaInstances.id, instance.id)); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + const exhausted = attempts >= maxAttemptsFor(step ?? { name: pending.stepName, run: async () => {} }); + + if (exhausted) { + // Compensare imposibila: nu pretindem ca s-a curatat. FAILED e o stare + // care cere interventie umana, si de asta e vizibila in monitor. + this.logger.error( + `Saga ${definition.name}/${instance.id}: compensarea pasului "${pending.stepName}" a esuat definitiv`, + ); + await this.fail(instance, `Compensare esuata la "${pending.stepName}": ${message}`); + return; + } + + await db + .update(sagaSteps) + .set({ lastError: message }) + .where(eq(sagaSteps.id, pending.id)); + await db + .update(sagaInstances) + .set({ + attempt: attempts, + lastError: message, + nextAttemptAt: new Date(Date.now() + backoffDelayMs(attempts)), + lockedUntil: null, + updatedAt: new Date(), + }) + .where(eq(sagaInstances.id, instance.id)); + } + } + + // --- Tranzitii terminale ------------------------------------------------- + + private async complete(instance: SagaInstanceRow): Promise { + await db + .update(sagaInstances) + .set({ + status: 'COMPLETED', + completedAt: new Date(), + lockedUntil: null, + lastError: null, + updatedAt: new Date(), + }) + .where(eq(sagaInstances.id, instance.id)); + } + + private async fail(instance: SagaInstanceRow, reason: string): Promise { + await db + .update(sagaInstances) + .set({ + status: 'FAILED', + lastError: reason, + completedAt: new Date(), + lockedUntil: null, + updatedAt: new Date(), + }) + .where(eq(sagaInstances.id, instance.id)); + } + + private async releaseWithError(instance: SagaInstanceRow, error: unknown): Promise { + const message = error instanceof Error ? error.message : String(error); + await db + .update(sagaInstances) + .set({ + lastError: message, + lockedUntil: null, + nextAttemptAt: new Date(Date.now() + backoffDelayMs(instance.attempt + 1)), + updatedAt: new Date(), + }) + .where(eq(sagaInstances.id, instance.id)); + } + + // --- Ajutoare ------------------------------------------------------------ + + private async upsertStep( + instance: SagaInstanceRow, + stepName: string, + sequence: number, + status: 'RUNNING', + attempts: number, + ): Promise { + await db + .insert(sagaSteps) + .values({ + sagaInstanceId: instance.id, + stepName, + sequence, + status, + attempts, + startedAt: new Date(), + }) + .onConflictDoUpdate({ + target: [sagaSteps.sagaInstanceId, sagaSteps.sequence], + set: { status, attempts, startedAt: new Date() }, + }); + } + + private toSagaEvent(row: typeof outboxEvents.$inferSelect): SagaEvent { + return { + id: row.id, + tenantId: row.tenantId, + workspaceId: row.workspaceId, + eventType: row.eventType, + actorId: row.actorId, + subjectId: row.subjectId, + correlationId: row.correlationId, + payload: (row.payload ?? {}) as Record, + }; + } + + /** + * Reconstituie evenimentul declansator din contextul persistat. Nu recitim + * outbox_events: evenimentul original poate fi arhivat, iar saga trebuie sa + * ramana rulabila si dupa aceea. + */ + private eventFromContext(instance: SagaInstanceRow): SagaEvent { + const context = (instance.context ?? {}) as SagaContext; + return { + id: instance.triggerEventId, + tenantId: instance.tenantId, + workspaceId: instance.workspaceId, + eventType: '', + actorId: null, + subjectId: null, + correlationId: instance.correlationId, + payload: (context.event ?? {}) as Record, + }; + } + + private mapInstance(row: Record): SagaInstanceRow { + const date = (v: unknown) => (v ? new Date(v as string) : null); + return { + id: row.id as string, + tenantId: row.tenant_id as string, + workspaceId: (row.workspace_id as string) ?? null, + sagaName: row.saga_name as string, + sagaVersion: Number(row.saga_version), + status: row.status as SagaInstanceRow['status'], + currentStep: Number(row.current_step), + triggerEventId: row.trigger_event_id as string, + correlationId: (row.correlation_id as string) ?? null, + context: (row.context ?? {}) as SagaContext, + attempt: Number(row.attempt), + nextAttemptAt: date(row.next_attempt_at) as Date, + timeoutAt: date(row.timeout_at), + lastError: (row.last_error as string) ?? null, + lockedUntil: date(row.locked_until), + createdAt: date(row.created_at) as Date, + updatedAt: date(row.updated_at) as Date, + completedAt: date(row.completed_at), + }; + } + + /** Expus pentru monitor: cate definitii sunt incarcate. */ + registeredSagas(): { name: string; version: number; triggerEvent: string; steps: number }[] { + return allSagas().map((s) => ({ + name: s.name, + version: s.version, + triggerEvent: s.triggerEvent, + steps: s.steps.length, + })); + } +} diff --git a/src/sagas/sagas.controller.ts b/src/sagas/sagas.controller.ts new file mode 100644 index 0000000..d5fe68d --- /dev/null +++ b/src/sagas/sagas.controller.ts @@ -0,0 +1,40 @@ +import { Controller, Get, Param, ParseUUIDPipe, Post, Query } from '@nestjs/common'; +import { CurrentSession } from '../auth/session.decorator'; +import type { AuthenticatedSession } from '../auth/tenant.guard'; +import { SagasService } from './sagas.service'; + +const STATUSES = [ + 'RUNNING', + 'COMPLETED', + 'COMPENSATING', + 'COMPENSATED', + 'FAILED', + 'TIMED_OUT', +] as const; + +@Controller('sagas') +export class SagasController { + constructor(private readonly sagas: SagasService) {} + + @Get() + overview(@CurrentSession() session: AuthenticatedSession) { + return this.sagas.overview(session); + } + + @Get('instances') + list(@CurrentSession() session: AuthenticatedSession, @Query('status') status?: string) { + // Allowlist, nu interpolare: statusul ajunge intr-o comparatie SQL. + const safe = STATUSES.includes(status as (typeof STATUSES)[number]) ? status : undefined; + return this.sagas.list(session, safe); + } + + @Get('instances/:id') + detail(@CurrentSession() session: AuthenticatedSession, @Param('id', ParseUUIDPipe) id: string) { + return this.sagas.detail(session, id); + } + + @Post('instances/:id/retry') + retry(@CurrentSession() session: AuthenticatedSession, @Param('id', ParseUUIDPipe) id: string) { + return this.sagas.retry(session, id); + } +} diff --git a/src/sagas/sagas.module.ts b/src/sagas/sagas.module.ts new file mode 100644 index 0000000..cb27454 --- /dev/null +++ b/src/sagas/sagas.module.ts @@ -0,0 +1,26 @@ +import { Module, OnModuleInit } from '@nestjs/common'; +import { IntelligenceModule } from '../intelligence/intelligence.module'; +import { IntelligenceService } from '../intelligence/intelligence.service'; +import { registerResearchBriefEnrichment } from './research-brief-enrichment.saga'; +import { SagaRunnerService } from './saga-runner.service'; +import { SagasController } from './sagas.controller'; +import { SagasService } from './sagas.service'; + +@Module({ + imports: [IntelligenceModule], + controllers: [SagasController], + providers: [SagaRunnerService, SagasService], + exports: [SagaRunnerService], +}) +export class SagasModule implements OnModuleInit { + constructor(private readonly intelligence: IntelligenceService) {} + + /** + * Sagas se inregistreaza la bootstrap, nu la import: definitia are nevoie de + * servicii injectate, iar registrul trebuie sa fie complet inainte ca primul + * tur de corelator sa ruleze. + */ + onModuleInit(): void { + registerResearchBriefEnrichment(this.intelligence); + } +} diff --git a/src/sagas/sagas.service.ts b/src/sagas/sagas.service.ts new file mode 100644 index 0000000..37ee010 --- /dev/null +++ b/src/sagas/sagas.service.ts @@ -0,0 +1,144 @@ +import { Injectable, NotFoundException } from '@nestjs/common'; +import { and, asc, count, desc, eq, lt, sql } from 'drizzle-orm'; +import { db } from '../db/client'; +import { sagaInstances, sagaSteps } from '../db/schema'; +import type { AuthenticatedSession } from '../auth/tenant.guard'; +import { allSagas } from './saga-registry'; + +/** O saga blocata mai mult de atat merita atentie umana. */ +const STUCK_AFTER_MINUTES = 30; + +@Injectable() +export class SagasService { + /** + * Monitorul raspunde la o singura intrebare: ce nu merge? + * De aceea numara pe status si scoate separat instantele blocate, in loc sa + * intoarca o lista plata pe care ar trebui sa o citeasca cineva manual. + */ + async overview(session: AuthenticatedSession) { + const stuckBefore = new Date(Date.now() - STUCK_AFTER_MINUTES * 60_000); + + const [byStatus, stuck] = await Promise.all([ + db + .select({ status: sagaInstances.status, n: count() }) + .from(sagaInstances) + .where(eq(sagaInstances.tenantId, session.tenantId)) + .groupBy(sagaInstances.status), + db + .select() + .from(sagaInstances) + .where( + and( + eq(sagaInstances.tenantId, session.tenantId), + sql`${sagaInstances.status} IN ('RUNNING','COMPENSATING')`, + lt(sagaInstances.updatedAt, stuckBefore), + ), + ) + .orderBy(asc(sagaInstances.updatedAt)) + .limit(20), + ]); + + return { + definitions: allSagas().map((s) => ({ + name: s.name, + version: s.version, + triggerEvent: s.triggerEvent, + steps: s.steps.map((step) => step.name), + timeoutMinutes: s.timeoutMinutes, + })), + counts: Object.fromEntries(byStatus.map((r) => [r.status, r.n])), + stuck: stuck.map((row) => this.summarize(row)), + stuckThresholdMinutes: STUCK_AFTER_MINUTES, + }; + } + + async list(session: AuthenticatedSession, status?: string, limit = 50) { + const where = status + ? and(eq(sagaInstances.tenantId, session.tenantId), sql`${sagaInstances.status} = ${status}`) + : eq(sagaInstances.tenantId, session.tenantId); + + const rows = await db + .select() + .from(sagaInstances) + .where(where) + .orderBy(desc(sagaInstances.createdAt)) + .limit(limit); + return { instances: rows.map((row) => this.summarize(row)) }; + } + + /** Detaliul include pasii: fara ei nu se poate spune unde s-a oprit. */ + async detail(session: AuthenticatedSession, id: string) { + const instance = await db.query.sagaInstances.findFirst({ + where: and(eq(sagaInstances.id, id), eq(sagaInstances.tenantId, session.tenantId)), + }); + if (!instance) throw new NotFoundException('Saga instance not found'); + + const steps = await db + .select() + .from(sagaSteps) + .where(eq(sagaSteps.sagaInstanceId, instance.id)) + .orderBy(asc(sagaSteps.sequence)); + + return { + ...this.summarize(instance), + context: instance.context, + steps: steps.map((step) => ({ + sequence: step.sequence, + name: step.stepName, + status: step.status, + attempts: step.attempts, + output: step.output, + lastError: step.lastError, + startedAt: step.startedAt, + completedAt: step.completedAt, + compensatedAt: step.compensatedAt, + })), + }; + } + + /** + * Reia o saga blocata: reseteaza backoff-ul si elibereaza lease-ul. + * NU schimba statusul si nu sare peste pasi -- reincercarea trebuie sa treaca + * prin aceeasi logica, altfel operatorul ar putea "repara" o saga sarind + * exact pasul care esua. + */ + async retry(session: AuthenticatedSession, id: string) { + const instance = await db.query.sagaInstances.findFirst({ + where: and(eq(sagaInstances.id, id), eq(sagaInstances.tenantId, session.tenantId)), + }); + if (!instance) throw new NotFoundException('Saga instance not found'); + if (instance.status === 'COMPLETED' || instance.status === 'COMPENSATED') { + return { retried: false, reason: 'Saga s-a incheiat deja' }; + } + + await db + .update(sagaInstances) + .set({ + status: instance.status === 'FAILED' ? 'RUNNING' : instance.status, + attempt: 0, + nextAttemptAt: new Date(), + lockedUntil: null, + updatedAt: new Date(), + }) + .where(eq(sagaInstances.id, instance.id)); + return { retried: true }; + } + + private summarize(row: typeof sagaInstances.$inferSelect) { + return { + id: row.id, + sagaName: row.sagaName, + sagaVersion: row.sagaVersion, + status: row.status, + currentStep: row.currentStep, + attempt: row.attempt, + correlationId: row.correlationId, + lastError: row.lastError, + nextAttemptAt: row.nextAttemptAt, + timeoutAt: row.timeoutAt, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + completedAt: row.completedAt, + }; + } +}