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, }; } }