From eb652446f6a86dcd832d2c7c4bb4efa28965e5f4 Mon Sep 17 00:00:00 2001 From: valentinbvro Date: Wed, 29 Jul 2026 12:03:34 +0200 Subject: [PATCH] feat: Sprint 1 -- Projection Kernel (read models pentru dashboard) Repara criteriul de acceptare "dashboardurile citesc read models, nu interogheaza haotic toate modulele". - projection_processed_events (unique: name+version+event_id) = garantia de idempotency. Checkpointul e doar optimizare de scanare, nu corectitudine: rescanam cu un safety lag de 60s si sarim ce s-a aplicat deja, ca sa nu pierdem evenimente comise dupa unul cu created_at mai mare. - ProjectionRegistry cu ProjectionDefinition (name, version, subscribedEvents, rebuildStrategy, apply, rebuildTenant). - executive_dashboard foloseste rebuildStrategy 'canonical-tables': recalculeaza contorii din tabelele canonice, nu incrementeaza. Contorii incrementali pot deriva daca un eveniment se pierde sau se dubleaza; recalcularea e corecta prin constructie si idempotenta natural. Costul e o interogare per eveniment relevant -- ok la volumul actual. - GET /v1/dashboard/executive citeste proiectia. Cand proiectia inca n-a rulat pentru workspace, intoarce status 'building' explicit, NU zerouri care ar parea date reale. - POST /v1/dashboard/executive/rebuild -- owner/admin, auditat. - Serviciile de taskuri/organizatii/segmente emit acum evenimente in outbox (in aceeasi tranzactie cu scrierea). Fara ele proiectia era cod mort. - Campurile din spec care depind de module neconstruite (documents, transactions, approvals) raman 0 explicit, nu inventate. --- drizzle/0005_projection_kernel.sql | 47 ++++++ drizzle/meta/_journal.json | 7 + src/app.module.ts | 2 + src/db/schema.ts | 75 +++++++++ src/events/outbox.service.ts | 16 ++ src/organizations/organizations.service.ts | 14 ++ src/projections/dashboard.controller.ts | 19 +++ src/projections/dashboard.service.ts | 92 ++++++++++ src/projections/projection-registry.ts | 136 +++++++++++++++ src/projections/projection-runner.service.ts | 168 +++++++++++++++++++ src/projections/projections.module.ts | 11 ++ src/segments/segments.module.ts | 3 +- src/segments/segments.service.ts | 68 +++++--- src/tasks/tasks.service.ts | 30 ++++ 14 files changed, 666 insertions(+), 22 deletions(-) create mode 100644 drizzle/0005_projection_kernel.sql create mode 100644 src/projections/dashboard.controller.ts create mode 100644 src/projections/dashboard.service.ts create mode 100644 src/projections/projection-registry.ts create mode 100644 src/projections/projection-runner.service.ts create mode 100644 src/projections/projections.module.ts diff --git a/drizzle/0005_projection_kernel.sql b/drizzle/0005_projection_kernel.sql new file mode 100644 index 0000000..48adad7 --- /dev/null +++ b/drizzle/0005_projection_kernel.sql @@ -0,0 +1,47 @@ +-- Projection Kernel (Sprint 1). Read models: NU sunt sursa de adevar, pot fi +-- sterse si reconstruite oricand din tabelele canonice. + +-- Garantia de idempotency: acelasi eveniment nu se aplica de doua ori. +CREATE TABLE IF NOT EXISTS "projection_processed_events" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "projection_name" text NOT NULL, + "projection_version" integer NOT NULL, + "event_id" uuid NOT NULL, + "processed_at" timestamp DEFAULT now() NOT NULL +); +CREATE UNIQUE INDEX IF NOT EXISTS "projection_processed_uq" + ON "projection_processed_events" ("projection_name","projection_version","event_id"); + +-- Checkpoint = optimizare (de unde reia scanarea), nu garantia de corectitudine. +CREATE TABLE IF NOT EXISTS "projection_checkpoints" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "projection_name" text NOT NULL, + "projection_version" integer NOT NULL, + "last_event_at" timestamp, + "last_event_id" uuid, + "status" text DEFAULT 'active' NOT NULL, + "last_error" text, + "updated_at" timestamp DEFAULT now() NOT NULL +); +CREATE UNIQUE INDEX IF NOT EXISTS "projection_checkpoint_uq" + ON "projection_checkpoints" ("projection_name","projection_version"); + +CREATE TABLE IF NOT EXISTS "executive_dashboard_projection" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "tenant_id" uuid NOT NULL, + "workspace_id" uuid NOT NULL, + "open_tasks_count" integer DEFAULT 0 NOT NULL, + "overdue_tasks_count" integer DEFAULT 0 NOT NULL, + "deadlines_next_7_days" integer DEFAULT 0 NOT NULL, + "organizations_count" integer DEFAULT 0 NOT NULL, + "segments_count" integer DEFAULT 0 NOT NULL, + "research_briefs_count" integer DEFAULT 0 NOT NULL, + "documents_missing_count" integer DEFAULT 0 NOT NULL, + "transactions_unclassified_count" integer DEFAULT 0 NOT NULL, + "pending_approvals_count" integer DEFAULT 0 NOT NULL, + "generated_at" timestamp DEFAULT now() NOT NULL, + "projection_version" integer DEFAULT 1 NOT NULL +); +-- Cheia de UPSERT (spec §5.3): o linie per tenant+workspace. +CREATE UNIQUE INDEX IF NOT EXISTS "executive_dashboard_tenant_ws_uq" + ON "executive_dashboard_projection" ("tenant_id","workspace_id"); diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index cfb8481..fa5c618 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -36,6 +36,13 @@ "when": 1785318305227, "tag": "0004_platform_kernel", "breakpoints": true + }, + { + "idx": 5, + "version": "7", + "when": 1785319371761, + "tag": "0005_projection_kernel", + "breakpoints": true } ] } \ No newline at end of file diff --git a/src/app.module.ts b/src/app.module.ts index fa4fe4c..1c863ac 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -14,6 +14,7 @@ import { OrganizationsModule } from './organizations/organizations.module'; import { TasksModule } from './tasks/tasks.module'; import { TenantsModule } from './tenants/tenants.module'; import { NavigationModule } from './navigation/navigation.module'; +import { ProjectionsModule } from './projections/projections.module'; import { IntelligenceModule } from './intelligence/intelligence.module'; import { SegmentsModule } from './segments/segments.module'; import { ResearchBriefsModule } from './research-briefs/research-briefs.module'; @@ -46,6 +47,7 @@ function parseRedisConnection(redisUrl: string | undefined) { AgentNichesModule, TenantsModule, NavigationModule, + ProjectionsModule, OrganizationsModule, TasksModule, IntelligenceModule, diff --git a/src/db/schema.ts b/src/db/schema.ts index 5e9dd23..6e7caea 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -295,6 +295,81 @@ export const outboxEvents = pgTable( // --- Audit (blueprint sectiunea 31.4) --- +// --- Projection Kernel (spec Read Models sectiunile 2-6) ------------------- +// Read models NU sunt sursa de adevar: pot fi sterse si reconstruite oricand +// din tabelele canonice. Userul nu le modifica direct. + +/** + * Garantia de idempotency: acelasi eveniment procesat de doua ori de aceeasi + * proiectie nu produce efect dublu. Cheia unica e (proiectie, versiune, event). + */ +export const projectionProcessedEvents = pgTable( + 'projection_processed_events', + { + id: uuid('id').defaultRandom().primaryKey(), + projectionName: text('projection_name').notNull(), + projectionVersion: integer('projection_version').notNull(), + eventId: uuid('event_id').notNull(), + processedAt: timestamp('processed_at').defaultNow().notNull(), + }, + (table) => [ + uniqueIndex('projection_processed_uq').on( + table.projectionName, + table.projectionVersion, + table.eventId, + ), + ], +); + +/** + * Checkpoint = optimizare (de unde reia scanarea), NU garantia de corectitudine. + * Corectitudinea vine din projection_processed_events: chiar daca checkpointul + * e in urma si se rescaneaza, evenimentele deja procesate sunt sarite. + */ +export const projectionCheckpoints = pgTable( + 'projection_checkpoints', + { + id: uuid('id').defaultRandom().primaryKey(), + projectionName: text('projection_name').notNull(), + projectionVersion: integer('projection_version').notNull(), + lastEventAt: timestamp('last_event_at'), + lastEventId: uuid('last_event_id'), + /** active | paused | rebuilding | failed */ + status: text('status').notNull().default('active'), + lastError: text('last_error'), + updatedAt: timestamp('updated_at').defaultNow().notNull(), + }, + (table) => [ + uniqueIndex('projection_checkpoint_uq').on(table.projectionName, table.projectionVersion), + ], +); + +export const executiveDashboardProjection = pgTable( + 'executive_dashboard_projection', + { + id: uuid('id').defaultRandom().primaryKey(), + tenantId: uuid('tenant_id').notNull(), + workspaceId: uuid('workspace_id').notNull(), + openTasksCount: integer('open_tasks_count').notNull().default(0), + overdueTasksCount: integer('overdue_tasks_count').notNull().default(0), + deadlinesNext7Days: integer('deadlines_next_7_days').notNull().default(0), + organizationsCount: integer('organizations_count').notNull().default(0), + segmentsCount: integer('segments_count').notNull().default(0), + researchBriefsCount: integer('research_briefs_count').notNull().default(0), + // Campurile de mai jos exista in spec dar depind de module neconstruite + // (Documents, Transactions, Approvals, Opportunities). Raman 0 explicit, + // ca sa nu para date reale cand modulele nu exista. + documentsMissingCount: integer('documents_missing_count').notNull().default(0), + transactionsUnclassifiedCount: integer('transactions_unclassified_count').notNull().default(0), + pendingApprovalsCount: integer('pending_approvals_count').notNull().default(0), + generatedAt: timestamp('generated_at').defaultNow().notNull(), + projectionVersion: integer('projection_version').notNull().default(1), + }, + (table) => [ + uniqueIndex('executive_dashboard_tenant_ws_uq').on(table.tenantId, table.workspaceId), + ], +); + // Append-oriented (spec sectiunea 29): se scrie, nu se modifica si nu se sterge. export const auditLog = pgTable( 'audit_log', diff --git a/src/events/outbox.service.ts b/src/events/outbox.service.ts index 5d60f40..a450d1c 100644 --- a/src/events/outbox.service.ts +++ b/src/events/outbox.service.ts @@ -6,10 +6,19 @@ import * as schema from '../db/schema'; export interface OutboxEventInput { tenantId: string; + /** Fara workspace, proiectiile nu stiu in ce partitie sa scrie si sar evenimentul. */ + workspaceId?: string; eventType: string; subjectId?: string; + aggregateType?: string; + actorId?: string; payload: Record; correlationId?: string; + /** Ce comanda/eveniment a produs acest eveniment. */ + causationId?: string; + /** C0-C4; controleaza catre ce servicii poate fi distribuit. */ + classification?: string; + provenance?: Record; eventVersion?: number; } @@ -30,11 +39,18 @@ export class OutboxService { async record(tx: Transaction, event: OutboxEventInput): Promise { await tx.insert(outboxEvents).values({ tenantId: event.tenantId, + workspaceId: event.workspaceId, eventType: event.eventType, eventVersion: event.eventVersion ?? 1, + occurredAt: new Date(), + actorId: event.actorId, + aggregateType: event.aggregateType, subjectId: event.subjectId, payload: event.payload, correlationId: event.correlationId, + causationId: event.causationId, + classification: event.classification ?? 'c2', + provenance: event.provenance ?? {}, }); } } diff --git a/src/organizations/organizations.service.ts b/src/organizations/organizations.service.ts index 84d696d..cbcb621 100644 --- a/src/organizations/organizations.service.ts +++ b/src/organizations/organizations.service.ts @@ -69,6 +69,10 @@ export class OrganizationsService { }); await this.outbox.record(tx, { tenantId: session.tenantId, + workspaceId: session.workspaceId, + actorId: session.userId, + aggregateType: 'organization', + correlationId: session.correlationId, eventType: 'organization.created', subjectId: row.id, payload: { name: row.name, country: row.country }, @@ -118,6 +122,16 @@ export class OrganizationsService { action: 'organization.deleted', resource: `organization:${id}`, }); + await this.outbox.record(tx, { + tenantId: session.tenantId, + workspaceId: session.workspaceId, + actorId: session.userId, + eventType: 'organization.deleted', + aggregateType: 'organization', + subjectId: id, + correlationId: session.correlationId, + payload: {}, + }); }); return { deleted: true }; } diff --git a/src/projections/dashboard.controller.ts b/src/projections/dashboard.controller.ts new file mode 100644 index 0000000..81e199b --- /dev/null +++ b/src/projections/dashboard.controller.ts @@ -0,0 +1,19 @@ +import { Controller, Get, Post } from '@nestjs/common'; +import { CurrentSession } from '../auth/session.decorator'; +import type { AuthenticatedSession } from '../auth/tenant.guard'; +import { DashboardService } from './dashboard.service'; + +@Controller('dashboard') +export class DashboardController { + constructor(private readonly dashboardService: DashboardService) {} + + @Get('executive') + getExecutive(@CurrentSession() session: AuthenticatedSession) { + return this.dashboardService.getExecutive(session); + } + + @Post('executive/rebuild') + rebuildExecutive(@CurrentSession() session: AuthenticatedSession) { + return this.dashboardService.rebuildExecutive(session); + } +} diff --git a/src/projections/dashboard.service.ts b/src/projections/dashboard.service.ts new file mode 100644 index 0000000..49bb2a9 --- /dev/null +++ b/src/projections/dashboard.service.ts @@ -0,0 +1,92 @@ +import { ForbiddenException, Injectable } from '@nestjs/common'; +import { and, eq } from 'drizzle-orm'; +import { db } from '../db/client'; +import { executiveDashboardProjection, projectionCheckpoints } from '../db/schema'; +import { AuditService } from '../audit/audit.service'; +import type { AuthenticatedSession } from '../auth/tenant.guard'; +import { EXECUTIVE_DASHBOARD_PROJECTION } from './projection-registry'; +import { ProjectionRunnerService } from './projection-runner.service'; + +@Injectable() +export class DashboardService { + constructor( + private readonly runner: ProjectionRunnerService, + private readonly audit: AuditService, + ) {} + + /** + * Citeste read model-ul, nu interogheaza modulele. tenant/workspace vin din + * ExecutionContext, niciodata din query (spec §7). + */ + async getExecutive(session: AuthenticatedSession) { + const row = await db.query.executiveDashboardProjection.findFirst({ + where: and( + eq(executiveDashboardProjection.tenantId, session.tenantId), + eq(executiveDashboardProjection.workspaceId, session.workspaceId), + ), + }); + + const checkpoint = await db.query.projectionCheckpoints.findFirst({ + where: and( + eq(projectionCheckpoints.projectionName, EXECUTIVE_DASHBOARD_PROJECTION.projectionName), + eq(projectionCheckpoints.projectionVersion, EXECUTIVE_DASHBOARD_PROJECTION.version), + ), + }); + + if (!row) { + // Proiectia inca nu a rulat pentru acest workspace. Spunem asta explicit, + // in loc sa intoarcem zerouri care ar parea date reale. + return { + status: 'building' as const, + generatedAt: null, + counters: null, + projection: { + name: EXECUTIVE_DASHBOARD_PROJECTION.projectionName, + version: EXECUTIVE_DASHBOARD_PROJECTION.version, + checkpointStatus: checkpoint?.status ?? 'active', + }, + }; + } + + return { + status: 'ready' as const, + generatedAt: row.generatedAt, + counters: { + openTasks: row.openTasksCount, + overdueTasks: row.overdueTasksCount, + deadlinesNext7Days: row.deadlinesNext7Days, + organizations: row.organizationsCount, + segments: row.segmentsCount, + researchBriefs: row.researchBriefsCount, + }, + projection: { + name: EXECUTIVE_DASHBOARD_PROJECTION.projectionName, + version: row.projectionVersion, + checkpointStatus: checkpoint?.status ?? 'active', + lastEventAt: checkpoint?.lastEventAt ?? null, + }, + }; + } + + /** Rebuild din tabelele canonice. Operatiune administrativa -> owner/admin + audit. */ + async rebuildExecutive(session: AuthenticatedSession) { + if (session.role !== 'owner' && session.role !== 'admin') { + throw new ForbiddenException('Rebuilding a projection requires owner or admin role'); + } + + const workspacesRebuilt = await this.runner.rebuildTenant( + EXECUTIVE_DASHBOARD_PROJECTION, + session.tenantId, + ); + + await this.audit.record(null, { + tenantId: session.tenantId, + actorId: session.userId, + action: 'projection.rebuilt', + resource: `projection:${EXECUTIVE_DASHBOARD_PROJECTION.projectionName}`, + reason: `workspaces=${workspacesRebuilt}`, + }); + + return { rebuilt: true, workspaces: workspacesRebuilt }; + } +} diff --git a/src/projections/projection-registry.ts b/src/projections/projection-registry.ts new file mode 100644 index 0000000..008ea49 --- /dev/null +++ b/src/projections/projection-registry.ts @@ -0,0 +1,136 @@ +import { and, count, eq, gte, isNull, lt, ne } from 'drizzle-orm'; +import type { NodePgDatabase } from 'drizzle-orm/node-postgres'; +import * as schema from '../db/schema'; +import { + executiveDashboardProjection, + organizations, + researchBriefs, + savedSegments, + tasks, +} from '../db/schema'; + +type Tx = NodePgDatabase; + +export interface ProjectionEvent { + id: string; + tenantId: string; + workspaceId: string | null; + eventType: string; + createdAt: Date; +} + +/** + * rebuildStrategy explica de unde se poate reconstrui proiectia: + * - 'canonical-tables' = recalculeaza din tabelele canonice (mereu corect) + * - 'events' = aplica incremental evenimentele (mai rapid, poate deriva) + * - 'hybrid' = snapshot + evenimente + */ +export interface ProjectionDefinition { + projectionName: string; + version: number; + subscribedEvents: string[]; + rebuildStrategy: 'events' | 'canonical-tables' | 'hybrid'; + /** Aplica efectul unui eveniment. Trebuie sa fie idempotent. */ + apply(tx: Tx, event: ProjectionEvent): Promise; + /** Recalculeaza complet pentru un tenant (folosit de rebuild). */ + rebuildTenant(tx: Tx, tenantId: string, workspaceId: string): Promise; +} + +const SEVEN_DAYS_MS = 7 * 24 * 60 * 60 * 1000; + +/** + * Recalculeaza contorii din tabelele canonice pentru un tenant+workspace. + * + * Alegere deliberata: recalculare, nu incrementare. Contorii incrementali pot + * deriva daca un eveniment se pierde sau se dubleaza; recalcularea e corecta + * prin constructie si idempotenta natural. Costul e o interogare per eveniment + * relevant -- acceptabil la volumul actual, de reevaluat cand creste. + */ +async function recomputeExecutiveDashboard( + tx: Tx, + tenantId: string, + workspaceId: string, +): Promise { + const now = new Date(); + const weekFromNow = new Date(now.getTime() + SEVEN_DAYS_MS); + const activeTask = and( + eq(tasks.tenantId, tenantId), + isNull(tasks.deletedAt), + ne(tasks.status, 'done'), + ne(tasks.status, 'cancelled'), + ); + + const [openTasks, overdueTasks, upcoming, orgs, segments, briefs] = await Promise.all([ + tx.select({ n: count() }).from(tasks).where(activeTask), + tx.select({ n: count() }).from(tasks).where(and(activeTask, lt(tasks.dueAt, now))), + tx + .select({ n: count() }) + .from(tasks) + .where(and(activeTask, gte(tasks.dueAt, now), lt(tasks.dueAt, weekFromNow))), + tx + .select({ n: count() }) + .from(organizations) + .where(and(eq(organizations.tenantId, tenantId), isNull(organizations.deletedAt))), + tx.select({ n: count() }).from(savedSegments).where(eq(savedSegments.tenantId, tenantId)), + tx.select({ n: count() }).from(researchBriefs).where(eq(researchBriefs.tenantId, tenantId)), + ]); + + const row = { + tenantId, + workspaceId, + openTasksCount: openTasks[0]?.n ?? 0, + overdueTasksCount: overdueTasks[0]?.n ?? 0, + deadlinesNext7Days: upcoming[0]?.n ?? 0, + organizationsCount: orgs[0]?.n ?? 0, + segmentsCount: segments[0]?.n ?? 0, + researchBriefsCount: briefs[0]?.n ?? 0, + generatedAt: new Date(), + projectionVersion: 1, + }; + + // UPSERT, nu insert orb (spec §5.3). + await tx + .insert(executiveDashboardProjection) + .values(row) + .onConflictDoUpdate({ + target: [executiveDashboardProjection.tenantId, executiveDashboardProjection.workspaceId], + set: row, + }); +} + +export const EXECUTIVE_DASHBOARD_PROJECTION: ProjectionDefinition = { + projectionName: 'executive_dashboard', + version: 1, + rebuildStrategy: 'canonical-tables', + subscribedEvents: [ + 'task.created', + 'task.updated', + 'task.status_changed', + 'task.deleted', + 'organization.created', + 'organization.updated', + 'organization.deleted', + 'segment.created', + 'segment.deleted', + 'research_brief.created', + 'research_brief.deleted', + 'user.onboarded', + ], + async apply(tx, event) { + if (!event.workspaceId) { + // Evenimente scrise inainte de Platform Kernel nu au workspace; le sarim + // in loc sa ghicim un workspace si sa scriem intr-o partitie gresita. + return; + } + await recomputeExecutiveDashboard(tx, event.tenantId, event.workspaceId); + }, + async rebuildTenant(tx, tenantId, workspaceId) { + await recomputeExecutiveDashboard(tx, tenantId, workspaceId); + }, +}; + +export const PROJECTIONS: ProjectionDefinition[] = [EXECUTIVE_DASHBOARD_PROJECTION]; + +export function findProjection(name: string): ProjectionDefinition | undefined { + return PROJECTIONS.find((projection) => projection.projectionName === name); +} diff --git a/src/projections/projection-runner.service.ts b/src/projections/projection-runner.service.ts new file mode 100644 index 0000000..fc853ab --- /dev/null +++ b/src/projections/projection-runner.service.ts @@ -0,0 +1,168 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { Cron, CronExpression } from '@nestjs/schedule'; +import { and, asc, eq, gt, inArray } from 'drizzle-orm'; +import { db } from '../db/client'; +import { + outboxEvents, + projectionCheckpoints, + projectionProcessedEvents, + workspaces, +} from '../db/schema'; +import { PROJECTIONS, type ProjectionDefinition } from './projection-registry'; + +const BATCH_SIZE = 200; +/** Cat de mult inapoi rescanam fata de checkpoint, ca sa prindem evenimente + * care s-au comis dupa unul cu created_at mai mare (tranzactii concurente). + * Rescanarea e sigura: processed_events face reprocesarea inofensiva. */ +const SAFETY_LAG_MS = 60_000; + +@Injectable() +export class ProjectionRunnerService { + private readonly logger = new Logger(ProjectionRunnerService.name); + + @Cron(CronExpression.EVERY_10_SECONDS) + async runAll(): Promise { + for (const projection of PROJECTIONS) { + try { + await this.runProjection(projection); + } catch (error) { + this.logger.error(`Projection ${projection.projectionName} failed`, error as Error); + await this.markFailed(projection, error as Error); + } + } + } + + async runProjection(projection: ProjectionDefinition): Promise { + const checkpoint = await this.loadCheckpoint(projection); + if (checkpoint.status === 'paused') { + return 0; + } + + const since = checkpoint.lastEventAt + ? new Date(checkpoint.lastEventAt.getTime() - SAFETY_LAG_MS) + : new Date(0); + + const candidates = await db + .select({ + id: outboxEvents.id, + tenantId: outboxEvents.tenantId, + workspaceId: outboxEvents.workspaceId, + eventType: outboxEvents.eventType, + createdAt: outboxEvents.createdAt, + }) + .from(outboxEvents) + .where( + and( + gt(outboxEvents.createdAt, since), + inArray(outboxEvents.eventType, projection.subscribedEvents), + ), + ) + .orderBy(asc(outboxEvents.createdAt)) + .limit(BATCH_SIZE); + + if (candidates.length === 0) { + return 0; + } + + let applied = 0; + let lastEvent = candidates[candidates.length - 1]; + + for (const event of candidates) { + // Idempotency: daca inserarea in processed_events da conflict, evenimentul + // a fost deja aplicat de aceasta proiectie -> il sarim. + const claimed = await db + .insert(projectionProcessedEvents) + .values({ + projectionName: projection.projectionName, + projectionVersion: projection.version, + eventId: event.id, + }) + .onConflictDoNothing() + .returning({ id: projectionProcessedEvents.id }); + + if (claimed.length === 0) { + continue; + } + + await projection.apply(db, event); + applied += 1; + lastEvent = event; + } + + await this.saveCheckpoint(projection, lastEvent.createdAt, lastEvent.id); + if (applied > 0) { + this.logger.log(`Projection ${projection.projectionName}: applied ${applied} event(s)`); + } + return applied; + } + + /** Reconstruieste proiectia pentru un tenant, din tabelele canonice. */ + async rebuildTenant(projection: ProjectionDefinition, tenantId: string): Promise { + const tenantWorkspaces = await db + .select({ id: workspaces.id }) + .from(workspaces) + .where(eq(workspaces.tenantId, tenantId)); + + for (const workspace of tenantWorkspaces) { + await projection.rebuildTenant(db, tenantId, workspace.id); + } + return tenantWorkspaces.length; + } + + private async loadCheckpoint(projection: ProjectionDefinition) { + const existing = await db.query.projectionCheckpoints.findFirst({ + where: and( + eq(projectionCheckpoints.projectionName, projection.projectionName), + eq(projectionCheckpoints.projectionVersion, projection.version), + ), + }); + if (existing) { + return existing; + } + + const [created] = await db + .insert(projectionCheckpoints) + .values({ + projectionName: projection.projectionName, + projectionVersion: projection.version, + status: 'active', + }) + .onConflictDoNothing() + .returning(); + + return ( + created ?? { + lastEventAt: null as Date | null, + status: 'active' as string, + } + ); + } + + private async saveCheckpoint( + projection: ProjectionDefinition, + lastEventAt: Date, + lastEventId: string, + ): Promise { + await db + .update(projectionCheckpoints) + .set({ lastEventAt, lastEventId, status: 'active', lastError: null, updatedAt: new Date() }) + .where( + and( + eq(projectionCheckpoints.projectionName, projection.projectionName), + eq(projectionCheckpoints.projectionVersion, projection.version), + ), + ); + } + + private async markFailed(projection: ProjectionDefinition, error: Error): Promise { + await db + .update(projectionCheckpoints) + .set({ status: 'failed', lastError: error.message.slice(0, 500), updatedAt: new Date() }) + .where( + and( + eq(projectionCheckpoints.projectionName, projection.projectionName), + eq(projectionCheckpoints.projectionVersion, projection.version), + ), + ); + } +} diff --git a/src/projections/projections.module.ts b/src/projections/projections.module.ts new file mode 100644 index 0000000..5f2dde8 --- /dev/null +++ b/src/projections/projections.module.ts @@ -0,0 +1,11 @@ +import { Module } from '@nestjs/common'; +import { DashboardController } from './dashboard.controller'; +import { DashboardService } from './dashboard.service'; +import { ProjectionRunnerService } from './projection-runner.service'; + +@Module({ + controllers: [DashboardController], + providers: [ProjectionRunnerService, DashboardService], + exports: [ProjectionRunnerService], +}) +export class ProjectionsModule {} diff --git a/src/segments/segments.module.ts b/src/segments/segments.module.ts index 00a7965..2950303 100644 --- a/src/segments/segments.module.ts +++ b/src/segments/segments.module.ts @@ -1,10 +1,11 @@ import { Module } from '@nestjs/common'; +import { EventsModule } from '../events/events.module'; import { IntelligenceModule } from '../intelligence/intelligence.module'; import { SegmentsController } from './segments.controller'; import { SegmentsService } from './segments.service'; @Module({ - imports: [IntelligenceModule], + imports: [EventsModule, IntelligenceModule], controllers: [SegmentsController], providers: [SegmentsService], }) diff --git a/src/segments/segments.service.ts b/src/segments/segments.service.ts index 13efb30..8e89e21 100644 --- a/src/segments/segments.service.ts +++ b/src/segments/segments.service.ts @@ -3,6 +3,7 @@ import { and, desc, eq } from 'drizzle-orm'; import { db } from '../db/client'; import { savedSegments } from '../db/schema'; import { AuditService } from '../audit/audit.service'; +import { OutboxService } from '../events/outbox.service'; import { IntelligenceService } from '../intelligence/intelligence.service'; import type { AuthenticatedSession } from '../auth/tenant.guard'; import type { CreateSegmentDto } from './dto'; @@ -12,6 +13,7 @@ export class SegmentsService { constructor( private readonly audit: AuditService, private readonly intelligence: IntelligenceService, + private readonly outbox: OutboxService, ) {} async list(session: AuthenticatedSession) { @@ -22,23 +24,35 @@ export class SegmentsService { } async create(session: AuthenticatedSession, dto: CreateSegmentDto) { - const [row] = await db - .insert(savedSegments) - .values({ + return this.outbox.withTransaction(async (tx) => { + const [row] = await tx + .insert(savedSegments) + .values({ + tenantId: session.tenantId, + createdByUserId: session.userId, + name: dto.name, + query: dto.query, + resultLimit: dto.resultLimit, + }) + .returning(); + await this.audit.record(tx, { tenantId: session.tenantId, - createdByUserId: session.userId, - name: dto.name, - query: dto.query, - resultLimit: dto.resultLimit, - }) - .returning(); - await this.audit.record(null, { - tenantId: session.tenantId, - actorId: session.userId, - action: 'segment.created', - resource: `segment:${row.id}`, + actorId: session.userId, + action: 'segment.created', + resource: `segment:${row.id}`, + }); + await this.outbox.record(tx, { + tenantId: session.tenantId, + workspaceId: session.workspaceId, + actorId: session.userId, + eventType: 'segment.created', + aggregateType: 'segment', + subjectId: row.id, + correlationId: session.correlationId, + payload: { name: row.name }, + }); + return row; }); - return row; } /** Re-ruleaza cautarea salvata contra intelligence-api si intoarce rezultate proaspete. */ @@ -50,12 +64,24 @@ export class SegmentsService { async remove(session: AuthenticatedSession, id: string) { const segment = await this.getById(session, id); - await db.delete(savedSegments).where(eq(savedSegments.id, segment.id)); - await this.audit.record(null, { - tenantId: session.tenantId, - actorId: session.userId, - action: 'segment.deleted', - resource: `segment:${id}`, + await this.outbox.withTransaction(async (tx) => { + await tx.delete(savedSegments).where(eq(savedSegments.id, segment.id)); + await this.audit.record(tx, { + tenantId: session.tenantId, + actorId: session.userId, + action: 'segment.deleted', + resource: `segment:${id}`, + }); + await this.outbox.record(tx, { + tenantId: session.tenantId, + workspaceId: session.workspaceId, + actorId: session.userId, + eventType: 'segment.deleted', + aggregateType: 'segment', + subjectId: id, + correlationId: session.correlationId, + payload: {}, + }); }); return { deleted: true }; } diff --git a/src/tasks/tasks.service.ts b/src/tasks/tasks.service.ts index bf969ae..027dba9 100644 --- a/src/tasks/tasks.service.ts +++ b/src/tasks/tasks.service.ts @@ -55,6 +55,16 @@ export class TasksService { action: 'task.created', resource: `task:${row.id}`, }); + await this.outbox.record(tx, { + tenantId: session.tenantId, + workspaceId: session.workspaceId, + actorId: session.userId, + eventType: 'task.created', + aggregateType: 'task', + subjectId: row.id, + correlationId: session.correlationId, + payload: { title: row.title, priority: row.priority }, + }); return row; }); } @@ -80,6 +90,16 @@ export class TasksService { action: dto.status !== undefined ? `task.status_changed:${dto.status}` : 'task.updated', resource: `task:${id}`, }); + await this.outbox.record(tx, { + tenantId: session.tenantId, + workspaceId: session.workspaceId, + actorId: session.userId, + eventType: dto.status !== undefined ? 'task.status_changed' : 'task.updated', + aggregateType: 'task', + subjectId: id, + correlationId: session.correlationId, + payload: { status: row?.status }, + }); return row; }); } @@ -97,6 +117,16 @@ export class TasksService { action: 'task.deleted', resource: `task:${id}`, }); + await this.outbox.record(tx, { + tenantId: session.tenantId, + workspaceId: session.workspaceId, + actorId: session.userId, + eventType: 'task.deleted', + aggregateType: 'task', + subjectId: id, + correlationId: session.correlationId, + payload: {}, + }); }); return { deleted: true }; }