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

@@ -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");

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
     }
   ]
 }

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,

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',

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<string, unknown>;
   correlationId?: string;
+  /** Ce comanda/eveniment a produs acest eveniment. */
+  causationId?: string;
+  /** C0-C4; controleaza catre ce servicii poate fi distribuit. */
+  classification?: string;
+  provenance?: Record<string, unknown>;
   eventVersion?: number;
 }
 
@@ -30,11 +39,18 @@ export class OutboxService {
   async record(tx: Transaction, event: OutboxEventInput): Promise<void> {
     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 ?? {},
     });
   }
 }

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

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);
+  }
+}

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

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<typeof schema>;
+
+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<void>;
+  /** Recalculeaza complet pentru un tenant (folosit de rebuild). */
+  rebuildTenant(tx: Tx, tenantId: string, workspaceId: string): Promise<void>;
+}
+
+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<void> {
+  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);
+}

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<void> {
+    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<number> {
+    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<number> {
+    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<void> {
+    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<void> {
+    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),
+        ),
+      );
+  }
+}

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 {}

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],
 })

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

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