feat: Sprint 2-3 -- Notifications Engine

Motorul NU trimite o notificare per eveniment. Decide daca merita notificat,
cui, ce severitate, daca se agrega, ce canal si cand.

- notifications / notification_preferences / notification_processed_events.
  Ultimul da idempotency: aceeasi regula nu proceseaza acelasi eveniment de
  doua ori (unique rule_id + event_id).
- Reguli in COD, nu in DB cu condition_expression evaluat dinamic. Un
  evaluator de expresii e o suprafata de atac si un limbaj in plus, fara ca
  nimeni sa administreze inca reguli per tenant. Contractul ramane acelasi
  cand vor migra in DB.
- Destinatarii se rezolva server-side din memberships, niciodata din payload.
  Implicit nu notificam actorul despre propria actiune.
- Deduplicare pe cheia tenant+recipient+categorie+entitate+versiune regula.
- Agregare: 47 de taskuri intr-o ora devin o notificare cu aggregated_count,
  nu 47 de notificari. Testul verifica invariantul ca fereastra de agregare
  <= fereastra de dedup, altfel s-ar crea una noua inainte sa se agrege.
- Quiet hours cu fereastra care poate traversa miezul noptii. CRITICAL le
  depaseste DOAR daca politica userului permite, nu implicit.
- Inbox: filtre (unread/action_required/critical/intelligence/system),
  mark-read, acknowledge (opreste escaladarea), dismiss, snooze.
- Preferinte per user+tenant cu UPSERT (NULLS NOT DISTINCT pe category, ca
  randul implicit "toate categoriile" sa fie unic).

drizzle/0006_notifications_engine.sql

@@ -0,0 +1,73 @@
+-- Notifications Engine (Sprint 2-3). NU o notificare per eveniment: motorul
+-- decide daca merita, cui, ce severitate, daca se agrega si cand se livreaza.
+
+DO $$ BEGIN
+  CREATE TYPE "notification_severity" AS ENUM
+    ('CRITICAL','ACTION_REQUIRED','WARNING','INFORMATION','INTELLIGENCE','SYSTEM');
+EXCEPTION WHEN duplicate_object THEN NULL; END $$;
+
+DO $$ BEGIN
+  CREATE TYPE "notification_status" AS ENUM
+    ('PENDING','SCHEDULED','DELIVERING','DELIVERED','READ','ACKNOWLEDGED',
+     'DISMISSED','FAILED','EXPIRED','ESCALATED');
+EXCEPTION WHEN duplicate_object THEN NULL; END $$;
+
+CREATE TABLE IF NOT EXISTS "notifications" (
+  "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+  "tenant_id" uuid NOT NULL,
+  "workspace_id" uuid,
+  "recipient_user_id" uuid NOT NULL,
+  "category" text NOT NULL,
+  "severity" "notification_severity" NOT NULL,
+  "title" text NOT NULL,
+  "body" text,
+  "source_event_id" uuid,
+  "source_entity_type" text,
+  "source_entity_id" text,
+  "action_url" text,
+  "action_type" text,
+  "status" "notification_status" DEFAULT 'PENDING' NOT NULL,
+  "deduplication_key" text NOT NULL,
+  "aggregation_key" text,
+  "aggregated_count" integer DEFAULT 1 NOT NULL,
+  "channels" jsonb DEFAULT '["in_app"]'::jsonb NOT NULL,
+  "scheduled_at" timestamp,
+  "expires_at" timestamp,
+  "created_at" timestamp DEFAULT now() NOT NULL,
+  "updated_at" timestamp DEFAULT now() NOT NULL,
+  "read_at" timestamp,
+  "acknowledged_at" timestamp,
+  "snoozed_until" timestamp,
+  "metadata" jsonb DEFAULT '{}'::jsonb NOT NULL
+);
+CREATE INDEX IF NOT EXISTS "notifications_recipient_idx" ON "notifications" ("recipient_user_id","status");
+CREATE INDEX IF NOT EXISTS "notifications_tenant_idx" ON "notifications" ("tenant_id","created_at");
+CREATE INDEX IF NOT EXISTS "notifications_dedup_idx" ON "notifications" ("deduplication_key","created_at");
+
+CREATE TABLE IF NOT EXISTS "notification_preferences" (
+  "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+  "tenant_id" uuid NOT NULL,
+  "user_id" uuid NOT NULL,
+  "category" text,
+  "enabled" integer DEFAULT 1 NOT NULL,
+  "minimum_severity" "notification_severity" DEFAULT 'INFORMATION' NOT NULL,
+  "quiet_hours_start" integer,
+  "quiet_hours_end" integer,
+  "critical_bypasses_quiet_hours" integer DEFAULT 1 NOT NULL,
+  "timezone" text DEFAULT 'Europe/Bucharest' NOT NULL,
+  "created_at" timestamp DEFAULT now() NOT NULL,
+  "updated_at" timestamp DEFAULT now() NOT NULL
+);
+-- NULLS NOT DISTINCT: category NULL inseamna "toate categoriile" si trebuie sa
+-- fie unic per (user, tenant), altfel ON CONFLICT n-ar prinde randul implicit.
+CREATE UNIQUE INDEX IF NOT EXISTS "notification_prefs_uq"
+  ON "notification_preferences" ("user_id","tenant_id","category") NULLS NOT DISTINCT;
+
+CREATE TABLE IF NOT EXISTS "notification_processed_events" (
+  "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+  "rule_id" text NOT NULL,
+  "event_id" uuid NOT NULL,
+  "processed_at" timestamp DEFAULT now() NOT NULL
+);
+CREATE UNIQUE INDEX IF NOT EXISTS "notification_processed_uq"
+  ON "notification_processed_events" ("rule_id","event_id");

drizzle/meta/_journal.json

@@ -43,6 +43,13 @@
       "when": 1785319371761,
       "tag": "0005_projection_kernel",
       "breakpoints": true
+    },
+    {
+      "idx": 6,
+      "version": "7",
+      "when": 1785328533134,
+      "tag": "0006_notifications_engine",
+      "breakpoints": true
     }
   ]
 }

src/app.module.ts

@@ -15,6 +15,7 @@ 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 { NotificationsModule } from './notifications/notifications.module';
 import { IntelligenceModule } from './intelligence/intelligence.module';
 import { SegmentsModule } from './segments/segments.module';
 import { ResearchBriefsModule } from './research-briefs/research-briefs.module';
@@ -48,6 +49,7 @@ function parseRedisConnection(redisUrl: string | undefined) {
     TenantsModule,
     NavigationModule,
     ProjectionsModule,
+    NotificationsModule,
     OrganizationsModule,
     TasksModule,
     IntelligenceModule,

src/db/schema.ts

@@ -370,6 +370,107 @@ export const executiveDashboardProjection = pgTable(
   ],
 );
 
+// --- Notifications Engine (spec sectiunile 8-15) ---------------------------
+// Regula centrala: NU o notificare per eveniment. Motorul decide daca merita
+// notificat, cui, ce severitate, daca se agrega, ce canal si cand.
+
+export const notificationSeverity = pgEnum('notification_severity', [
+  'CRITICAL',
+  'ACTION_REQUIRED',
+  'WARNING',
+  'INFORMATION',
+  'INTELLIGENCE',
+  'SYSTEM',
+]);
+
+export const notificationStatus = pgEnum('notification_status', [
+  'PENDING',
+  'SCHEDULED',
+  'DELIVERING',
+  'DELIVERED',
+  'READ',
+  'ACKNOWLEDGED',
+  'DISMISSED',
+  'FAILED',
+  'EXPIRED',
+  'ESCALATED',
+]);
+
+export const notifications = pgTable(
+  'notifications',
+  {
+    id: uuid('id').defaultRandom().primaryKey(),
+    tenantId: uuid('tenant_id').notNull(),
+    workspaceId: uuid('workspace_id'),
+    recipientUserId: uuid('recipient_user_id').notNull(),
+    category: text('category').notNull(),
+    severity: notificationSeverity('severity').notNull(),
+    title: text('title').notNull(),
+    body: text('body'),
+    /** Fiecare notificare are eveniment-sursa (criteriu de acceptare #1). */
+    sourceEventId: uuid('source_event_id'),
+    sourceEntityType: text('source_entity_type'),
+    sourceEntityId: text('source_entity_id'),
+    actionUrl: text('action_url'),
+    /** Quick action: ce command executa, nu doar marcheaza notificarea. */
+    actionType: text('action_type'),
+    status: notificationStatus('status').notNull().default('PENDING'),
+    /** tenant + recipient + categorie + entitate: impiedica retrimiterea. */
+    deduplicationKey: text('deduplication_key').notNull(),
+    /** Mai multe evenimente similare intr-o singura notificare. */
+    aggregationKey: text('aggregation_key'),
+    aggregatedCount: integer('aggregated_count').notNull().default(1),
+    channels: jsonb('channels').notNull().default(['in_app']),
+    scheduledAt: timestamp('scheduled_at'),
+    expiresAt: timestamp('expires_at'),
+    createdAt: timestamp('created_at').defaultNow().notNull(),
+    updatedAt: timestamp('updated_at').defaultNow().notNull(),
+    readAt: timestamp('read_at'),
+    acknowledgedAt: timestamp('acknowledged_at'),
+    snoozedUntil: timestamp('snoozed_until'),
+    metadata: jsonb('metadata').notNull().default({}),
+  },
+  (table) => [
+    index('notifications_recipient_idx').on(table.recipientUserId, table.status),
+    index('notifications_tenant_idx').on(table.tenantId, table.createdAt),
+    index('notifications_dedup_idx').on(table.deduplicationKey, table.createdAt),
+  ],
+);
+
+export const notificationPreferences = pgTable(
+  'notification_preferences',
+  {
+    id: uuid('id').defaultRandom().primaryKey(),
+    tenantId: uuid('tenant_id').notNull(),
+    userId: uuid('user_id').notNull(),
+    /** NULL = se aplica tuturor categoriilor. */
+    category: text('category'),
+    enabled: integer('enabled').notNull().default(1),
+    minimumSeverity: notificationSeverity('minimum_severity').notNull().default('INFORMATION'),
+    /** Ore locale 0-23. Egale = fara quiet hours. */
+    quietHoursStart: integer('quiet_hours_start'),
+    quietHoursEnd: integer('quiet_hours_end'),
+    /** CRITICAL poate ignora quiet hours doar daca politica permite. */
+    criticalBypassesQuietHours: integer('critical_bypasses_quiet_hours').notNull().default(1),
+    timezone: text('timezone').notNull().default('Europe/Bucharest'),
+    createdAt: timestamp('created_at').defaultNow().notNull(),
+    updatedAt: timestamp('updated_at').defaultNow().notNull(),
+  },
+  (table) => [uniqueIndex('notification_prefs_uq').on(table.userId, table.tenantId, table.category)],
+);
+
+/** Idempotency: acelasi eveniment nu produce notificari duplicate. */
+export const notificationProcessedEvents = pgTable(
+  'notification_processed_events',
+  {
+    id: uuid('id').defaultRandom().primaryKey(),
+    ruleId: text('rule_id').notNull(),
+    eventId: uuid('event_id').notNull(),
+    processedAt: timestamp('processed_at').defaultNow().notNull(),
+  },
+  (table) => [uniqueIndex('notification_processed_uq').on(table.ruleId, table.eventId)],
+);
+
 // Append-oriented (spec sectiunea 29): se scrie, nu se modifica si nu se sterge.
 export const auditLog = pgTable(
   'audit_log',

src/notifications/dto.ts

@@ -0,0 +1,50 @@
+import { ArrayMaxSize, IsArray, IsBoolean, IsIn, IsInt, IsOptional, IsUUID, Max, Min } from 'class-validator';
+
+const SEVERITIES = [
+  'CRITICAL',
+  'ACTION_REQUIRED',
+  'WARNING',
+  'INFORMATION',
+  'INTELLIGENCE',
+  'SYSTEM',
+] as const;
+
+export class MarkReadDto {
+  @IsArray()
+  @ArrayMaxSize(200)
+  @IsUUID('4', { each: true })
+  ids!: string[];
+}
+
+export class SnoozeDto {
+  @IsInt()
+  @Min(1)
+  @Max(10080)
+  minutes!: number;
+}
+
+export class UpdatePreferencesDto {
+  @IsOptional()
+  @IsBoolean()
+  enabled?: boolean;
+
+  @IsOptional()
+  @IsIn(SEVERITIES as unknown as string[])
+  minimumSeverity?: (typeof SEVERITIES)[number];
+
+  @IsOptional()
+  @IsInt()
+  @Min(0)
+  @Max(23)
+  quietHoursStart?: number | null;
+
+  @IsOptional()
+  @IsInt()
+  @Min(0)
+  @Max(23)
+  quietHoursEnd?: number | null;
+
+  @IsOptional()
+  @IsBoolean()
+  criticalBypassesQuietHours?: boolean;
+}

src/notifications/notification-rules.spec.ts

@@ -0,0 +1,40 @@
+import { SEVERITY_ORDER, rulesFor, subscribedEventTypes } from './notification-rules';
+
+describe('notification rules', () => {
+  it('ordoneaza severitatile de la SYSTEM la CRITICAL', () => {
+    expect(SEVERITY_ORDER.SYSTEM).toBeLessThan(SEVERITY_ORDER.INFORMATION);
+    expect(SEVERITY_ORDER.INFORMATION).toBeLessThan(SEVERITY_ORDER.ACTION_REQUIRED);
+    expect(SEVERITY_ORDER.ACTION_REQUIRED).toBeLessThan(SEVERITY_ORDER.CRITICAL);
+  });
+
+  it('nu intoarce reguli pentru evenimente la care nimeni nu s-a abonat', () => {
+    expect(rulesFor('inventat.nu_exista')).toEqual([]);
+  });
+
+  it('fiecare regula are o cheie de deduplicare pozitiva', () => {
+    for (const eventType of subscribedEventTypes()) {
+      for (const rule of rulesFor(eventType)) {
+        expect(rule.deduplicationWindowMinutes).toBeGreaterThan(0);
+      }
+    }
+  });
+
+  it('regulile agregabile au fereastra de agregare mai mica sau egala cu cea de dedup', () => {
+    // Altfel s-ar crea o notificare noua inainte sa se poata agrega in cea veche.
+    for (const eventType of subscribedEventTypes()) {
+      for (const rule of rulesFor(eventType)) {
+        if (rule.aggregationWindowMinutes) {
+          expect(rule.aggregationWindowMinutes).toBeLessThanOrEqual(
+            rule.deduplicationWindowMinutes,
+          );
+        }
+      }
+    }
+  });
+
+  it('nu notifica actorul despre propria actiune, in afara de evenimentele de sistem', () => {
+    for (const rule of rulesFor('task.created')) {
+      expect(rule.notifyActor).toBe(false);
+    }
+  });
+});

src/notifications/notification-rules.ts

@@ -0,0 +1,131 @@
+/**
+ * Registrul de reguli de notificare (spec §11.2).
+ *
+ * Alegere deliberata: regulile sunt in cod, nu in baza cu `condition_expression`
+ * evaluat dinamic. Un evaluator de expresii din DB e o suprafata de atac si un
+ * limbaj de programat in plus, fara ca cineva sa administreze inca reguli per
+ * tenant. Contractul catre restul sistemului e acelasi cand vor migra in DB.
+ */
+export type Severity =
+  | 'CRITICAL'
+  | 'ACTION_REQUIRED'
+  | 'WARNING'
+  | 'INFORMATION'
+  | 'INTELLIGENCE'
+  | 'SYSTEM';
+
+/** Cui i se trimite. Rezolvat server-side, niciodata din payload-ul clientului. */
+export type RecipientStrategy = 'actor' | 'owners_and_admins' | 'all_members';
+
+export interface NotificationEvent {
+  id: string;
+  tenantId: string;
+  workspaceId: string | null;
+  eventType: string;
+  actorId: string | null;
+  subjectId: string | null;
+  aggregateType: string | null;
+  payload: Record<string, unknown>;
+  createdAt: Date;
+}
+
+export interface NotificationRule {
+  id: string;
+  eventType: string;
+  category: string;
+  severity: Severity;
+  recipientStrategy: RecipientStrategy;
+  channels: string[];
+  /** Fereastra in care o notificare cu aceeasi cheie nu se retrimite. */
+  deduplicationWindowMinutes: number;
+  /** Daca e setat, notificarile din fereastra se contopesc intr-una singura. */
+  aggregationWindowMinutes?: number;
+  /** Nu notifica actorul despre propria actiune, decat daca e explicit dorit. */
+  notifyActor: boolean;
+  isActive: boolean;
+  title(event: NotificationEvent): string;
+  body?(event: NotificationEvent): string | undefined;
+  actionUrl?(event: NotificationEvent): string | undefined;
+  /** Quick action: ce command real executa butonul din inbox. */
+  actionType?: string;
+}
+
+function str(payload: Record<string, unknown>, key: string, fallback = ''): string {
+  const value = payload[key];
+  return typeof value === 'string' ? value : fallback;
+}
+
+export const NOTIFICATION_RULES: NotificationRule[] = [
+  {
+    id: 'task-created@v1',
+    eventType: 'task.created',
+    category: 'tasks',
+    severity: 'INFORMATION',
+    recipientStrategy: 'owners_and_admins',
+    channels: ['in_app'],
+    deduplicationWindowMinutes: 60,
+    // Multe taskuri create intr-o ora devin o singura notificare, nu 47.
+    aggregationWindowMinutes: 60,
+    notifyActor: false,
+    isActive: true,
+    title: (event) => `Task nou: ${str(event.payload, 'title', 'fara titlu')}`,
+    actionUrl: () => '/dashboard/tasks',
+  },
+  {
+    id: 'organization-created@v1',
+    eventType: 'organization.created',
+    category: 'business',
+    severity: 'INFORMATION',
+    recipientStrategy: 'owners_and_admins',
+    channels: ['in_app'],
+    deduplicationWindowMinutes: 60,
+    aggregationWindowMinutes: 60,
+    notifyActor: false,
+    isActive: true,
+    title: (event) => `Companie adaugata: ${str(event.payload, 'name', 'fara nume')}`,
+    actionUrl: () => '/dashboard/organizations',
+  },
+  {
+    id: 'research-brief-created@v1',
+    eventType: 'research_brief.created',
+    category: 'intelligence',
+    severity: 'INTELLIGENCE',
+    recipientStrategy: 'owners_and_admins',
+    channels: ['in_app'],
+    deduplicationWindowMinutes: 60,
+    notifyActor: false,
+    isActive: true,
+    title: (event) => `Research brief nou: ${str(event.payload, 'title', '')}`,
+    actionUrl: () => '/dashboard/research',
+  },
+  {
+    id: 'user-onboarded@v1',
+    eventType: 'user.onboarded',
+    category: 'system',
+    severity: 'SYSTEM',
+    recipientStrategy: 'owners_and_admins',
+    channels: ['in_app'],
+    deduplicationWindowMinutes: 1440,
+    notifyActor: true,
+    isActive: true,
+    title: (event) => `Workspace creat: ${str(event.payload, 'tenantName', '')}`,
+    actionUrl: () => '/dashboard',
+  },
+];
+
+export const SEVERITY_ORDER: Record<Severity, number> = {
+  SYSTEM: 0,
+  INFORMATION: 1,
+  INTELLIGENCE: 2,
+  WARNING: 3,
+  ACTION_REQUIRED: 4,
+  CRITICAL: 5,
+};
+
+export function rulesFor(eventType: string): NotificationRule[] {
+  return NOTIFICATION_RULES.filter((rule) => rule.isActive && rule.eventType === eventType);
+}
+
+export function subscribedEventTypes(): string[] {
+  return [...new Set(NOTIFICATION_RULES.filter((r) => r.isActive).map((r) => r.eventType))];
+}

src/notifications/notification-runner.service.ts

@@ -0,0 +1,267 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { Cron, CronExpression } from '@nestjs/schedule';
+import { and, asc, desc, eq, gt, gte, inArray, isNull, sql } from 'drizzle-orm';
+import { db } from '../db/client';
+import {
+  memberships,
+  notificationPreferences,
+  notificationProcessedEvents,
+  notifications,
+  outboxEvents,
+} from '../db/schema';
+import {
+  SEVERITY_ORDER,
+  rulesFor,
+  subscribedEventTypes,
+  type NotificationEvent,
+  type NotificationRule,
+  type Severity,
+} from './notification-rules';
+
+const BATCH_SIZE = 200;
+const SAFETY_LAG_MS = 60_000;
+
+@Injectable()
+export class NotificationRunnerService {
+  private readonly logger = new Logger(NotificationRunnerService.name);
+  private lastSeenAt: Date | null = null;
+
+  @Cron(CronExpression.EVERY_10_SECONDS)
+  async run(): Promise<void> {
+    try {
+      await this.processBatch();
+    } catch (error) {
+      this.logger.error('Notification runner failed', error as Error);
+    }
+  }
+
+  async processBatch(): Promise<number> {
+    const eventTypes = subscribedEventTypes();
+    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({
+        id: outboxEvents.id,
+        tenantId: outboxEvents.tenantId,
+        workspaceId: outboxEvents.workspaceId,
+        eventType: outboxEvents.eventType,
+        actorId: outboxEvents.actorId,
+        subjectId: outboxEvents.subjectId,
+        aggregateType: outboxEvents.aggregateType,
+        payload: outboxEvents.payload,
+        createdAt: outboxEvents.createdAt,
+      })
+      .from(outboxEvents)
+      .where(and(gt(outboxEvents.createdAt, since), inArray(outboxEvents.eventType, eventTypes)))
+      .orderBy(asc(outboxEvents.createdAt))
+      .limit(BATCH_SIZE);
+
+    if (events.length === 0) return 0;
+
+    let created = 0;
+    for (const event of events) {
+      for (const rule of rulesFor(event.eventType)) {
+        created += await this.applyRule(rule, event as NotificationEvent);
+      }
+      this.lastSeenAt = event.createdAt;
+    }
+
+    if (created > 0) {
+      this.logger.log(`Created ${created} notification(s)`);
+    }
+    return created;
+  }
+
+  private async applyRule(rule: NotificationRule, event: NotificationEvent): Promise<number> {
+    // Idempotency: aceeasi regula nu proceseaza acelasi eveniment de doua ori.
+    const claimed = await db
+      .insert(notificationProcessedEvents)
+      .values({ ruleId: rule.id, eventId: event.id })
+      .onConflictDoNothing()
+      .returning({ id: notificationProcessedEvents.id });
+    if (claimed.length === 0) return 0;
+
+    const recipients = await this.resolveRecipients(rule, event);
+    let created = 0;
+
+    for (const recipientUserId of recipients) {
+      const decision = await this.shouldDeliver(rule, event, recipientUserId);
+      if (!decision.deliver) continue;
+
+      const dedupKey = this.deduplicationKey(rule, event, recipientUserId);
+
+      // Agregare: daca exista deja o notificare necitita cu aceeasi cheie in
+      // fereastra, o incrementam in loc sa cream a doua (spec §12).
+      if (rule.aggregationWindowMinutes) {
+        const windowStart = new Date(Date.now() - rule.aggregationWindowMinutes * 60_000);
+        const existing = await db.query.notifications.findFirst({
+          where: and(
+            eq(notifications.deduplicationKey, dedupKey),
+            eq(notifications.recipientUserId, recipientUserId),
+            gte(notifications.createdAt, windowStart),
+            isNull(notifications.readAt),
+          ),
+          orderBy: desc(notifications.createdAt),
+        });
+        if (existing) {
+          const nextCount = existing.aggregatedCount + 1;
+          await db
+            .update(notifications)
+            .set({
+              aggregatedCount: nextCount,
+              title: `${nextCount} ${rule.category === 'tasks' ? 'taskuri noi' : 'actualizari'}`,
+              updatedAt: new Date(),
+            })
+            .where(eq(notifications.id, existing.id));
+          continue;
+        }
+      } else {
+        // Fara agregare: deduplicare simpla in fereastra.
+        const windowStart = new Date(Date.now() - rule.deduplicationWindowMinutes * 60_000);
+        const duplicate = await db.query.notifications.findFirst({
+          where: and(
+            eq(notifications.deduplicationKey, dedupKey),
+            eq(notifications.recipientUserId, recipientUserId),
+            gte(notifications.createdAt, windowStart),
+          ),
+        });
+        if (duplicate) continue;
+      }
+
+      await db.insert(notifications).values({
+        tenantId: event.tenantId,
+        workspaceId: event.workspaceId,
+        recipientUserId,
+        category: rule.category,
+        severity: rule.severity,
+        title: rule.title(event),
+        body: rule.body?.(event),
+        sourceEventId: event.id,
+        sourceEntityType: event.aggregateType,
+        sourceEntityId: event.subjectId,
+        actionUrl: rule.actionUrl?.(event),
+        actionType: rule.actionType,
+        deduplicationKey: dedupKey,
+        aggregationKey: rule.aggregationWindowMinutes ? dedupKey : null,
+        channels: rule.channels,
+        // In-app e livrat imediat. Alte canale ar trece prin adaptere care nu
+        // exista inca -- vezi nota din raport, nu pretindem ca s-au trimis.
+        status: decision.scheduledAt ? 'SCHEDULED' : 'DELIVERED',
+        scheduledAt: decision.scheduledAt,
+        metadata: { ruleId: rule.id },
+      });
+      created += 1;
+    }
+
+    return created;
+  }
+
+  /** Destinatarii se rezolva server-side din memberships (criteriu #2). */
+  private async resolveRecipients(
+    rule: NotificationRule,
+    event: NotificationEvent,
+  ): Promise<string[]> {
+    if (rule.recipientStrategy === 'actor') {
+      return event.actorId ? [event.actorId] : [];
+    }
+
+    const rows = await db
+      .select({ userId: memberships.userId, role: memberships.role })
+      .from(memberships)
+      .where(eq(memberships.tenantId, event.tenantId));
+
+    const filtered =
+      rule.recipientStrategy === 'owners_and_admins'
+        ? rows.filter((r) => r.role === 'owner' || r.role === 'admin')
+        : rows;
+
+    return filtered
+      .map((r) => r.userId)
+      .filter((userId) => rule.notifyActor || userId !== event.actorId);
+  }
+
+  /**
+   * Preferinte + quiet hours. CRITICAL poate depasi quiet hours doar daca
+   * politica userului permite (spec §11.3) -- nu implicit.
+   */
+  private async shouldDeliver(
+    rule: NotificationRule,
+    event: NotificationEvent,
+    recipientUserId: string,
+  ): Promise<{ deliver: boolean; scheduledAt: Date | null }> {
+    const prefs = await db.query.notificationPreferences.findFirst({
+      where: and(
+        eq(notificationPreferences.userId, recipientUserId),
+        eq(notificationPreferences.tenantId, event.tenantId),
+      ),
+    });
+
+    if (!prefs) {
+      return { deliver: true, scheduledAt: null };
+    }
+    if (prefs.enabled === 0) {
+      return { deliver: false, scheduledAt: null };
+    }
+    if (SEVERITY_ORDER[rule.severity] < SEVERITY_ORDER[prefs.minimumSeverity as Severity]) {
+      return { deliver: false, scheduledAt: null };
+    }
+
+    const scheduledAt = this.quietHoursDelay(
+      prefs.quietHoursStart,
+      prefs.quietHoursEnd,
+      rule.severity,
+      prefs.criticalBypassesQuietHours === 1,
+    );
+    return { deliver: true, scheduledAt };
+  }
+
+  private quietHoursDelay(
+    start: number | null,
+    end: number | null,
+    severity: Severity,
+    criticalBypasses: boolean,
+  ): Date | null {
+    if (start === null || end === null || start === end) return null;
+    if (severity === 'CRITICAL' && criticalBypasses) return null;
+
+    const now = new Date();
+    const hour = now.getUTCHours();
+    // Fereastra poate traversa miezul noptii (ex. 22 -> 7).
+    const inQuiet = start < end ? hour >= start && hour < end : hour >= start || hour < end;
+    if (!inQuiet) return null;
+
+    const deliverAt = new Date(now);
+    deliverAt.setUTCMinutes(0, 0, 0);
+    if (hour >= end) deliverAt.setUTCDate(deliverAt.getUTCDate() + 1);
+    deliverAt.setUTCHours(end);
+    return deliverAt;
+  }
+
+  private deduplicationKey(
+    rule: NotificationRule,
+    event: NotificationEvent,
+    recipientUserId: string,
+  ): string {
+    // tenant + recipient + categorie + entitate + versiunea regulii (spec §12).
+    const entity = rule.aggregationWindowMinutes ? 'aggregate' : (event.subjectId ?? 'none');
+    return `${event.tenantId}:${recipientUserId}:${rule.category}:${entity}:${rule.id}`;
+  }
+
+  /** Livreaza notificarile programate carora le-a trecut ora. */
+  @Cron(CronExpression.EVERY_MINUTE)
+  async releaseScheduled(): Promise<void> {
+    await db
+      .update(notifications)
+      .set({ status: 'DELIVERED', updatedAt: new Date() })
+      .where(
+        and(
+          eq(notifications.status, 'SCHEDULED'),
+          sql`${notifications.scheduledAt} <= now()`,
+        ),
+      );
+  }
+}

src/notifications/notifications.controller.ts

@@ -0,0 +1,65 @@
+import { Body, Controller, Get, Param, ParseUUIDPipe, Patch, Post, Query } from '@nestjs/common';
+import { CurrentSession } from '../auth/session.decorator';
+import type { AuthenticatedSession } from '../auth/tenant.guard';
+import { MarkReadDto, SnoozeDto, UpdatePreferencesDto } from './dto';
+import { NotificationsService, type InboxFilter } from './notifications.service';
+
+const FILTERS: InboxFilter[] = [
+  'all',
+  'unread',
+  'action_required',
+  'critical',
+  'intelligence',
+  'system',
+];
+
+@Controller('notifications')
+export class NotificationsController {
+  constructor(private readonly notificationsService: NotificationsService) {}
+
+  @Get()
+  list(@CurrentSession() session: AuthenticatedSession, @Query('filter') filter?: string) {
+    const safe = FILTERS.includes(filter as InboxFilter) ? (filter as InboxFilter) : 'all';
+    return this.notificationsService.list(session, safe);
+  }
+
+  @Post('mark-read')
+  markRead(@CurrentSession() session: AuthenticatedSession, @Body() dto: MarkReadDto) {
+    return this.notificationsService.markRead(session, dto.ids);
+  }
+
+  @Post(':id/acknowledge')
+  acknowledge(
+    @CurrentSession() session: AuthenticatedSession,
+    @Param('id', ParseUUIDPipe) id: string,
+  ) {
+    return this.notificationsService.acknowledge(session, id);
+  }
+
+  @Post(':id/dismiss')
+  dismiss(@CurrentSession() session: AuthenticatedSession, @Param('id', ParseUUIDPipe) id: string) {
+    return this.notificationsService.dismiss(session, id);
+  }
+
+  @Post(':id/snooze')
+  snooze(
+    @CurrentSession() session: AuthenticatedSession,
+    @Param('id', ParseUUIDPipe) id: string,
+    @Body() dto: SnoozeDto,
+  ) {
+    return this.notificationsService.snooze(session, id, dto.minutes);
+  }
+
+  @Get('preferences')
+  getPreferences(@CurrentSession() session: AuthenticatedSession) {
+    return this.notificationsService.getPreferences(session);
+  }
+
+  @Patch('preferences')
+  updatePreferences(
+    @CurrentSession() session: AuthenticatedSession,
+    @Body() dto: UpdatePreferencesDto,
+  ) {
+    return this.notificationsService.updatePreferences(session, dto);
+  }
+}

src/notifications/notifications.module.ts

@@ -0,0 +1,11 @@
+import { Module } from '@nestjs/common';
+import { NotificationRunnerService } from './notification-runner.service';
+import { NotificationsController } from './notifications.controller';
+import { NotificationsService } from './notifications.service';
+
+@Module({
+  controllers: [NotificationsController],
+  providers: [NotificationRunnerService, NotificationsService],
+  exports: [NotificationRunnerService],
+})
+export class NotificationsModule {}

src/notifications/notifications.service.ts

@@ -0,0 +1,196 @@
+import { Injectable, NotFoundException } from '@nestjs/common';
+import { and, count, desc, eq, inArray, isNull, ne } from 'drizzle-orm';
+import { db } from '../db/client';
+import { notificationPreferences, notifications } from '../db/schema';
+import { AuditService } from '../audit/audit.service';
+import type { AuthenticatedSession } from '../auth/tenant.guard';
+import type { Severity } from './notification-rules';
+
+export type InboxFilter = 'all' | 'unread' | 'action_required' | 'critical' | 'intelligence' | 'system';
+
+@Injectable()
+export class NotificationsService {
+  constructor(private readonly audit: AuditService) {}
+
+  async list(session: AuthenticatedSession, filter: InboxFilter = 'all', limit = 50) {
+    // Notificarile sunt strict ale destinatarului: filtrul pe recipient +
+    // tenant e obligatoriu, nu optional.
+    const base = and(
+      eq(notifications.tenantId, session.tenantId),
+      eq(notifications.recipientUserId, session.userId),
+      ne(notifications.status, 'DISMISSED'),
+    );
+
+    const filters = {
+      all: base,
+      unread: and(base, isNull(notifications.readAt)),
+      action_required: and(base, eq(notifications.severity, 'ACTION_REQUIRED')),
+      critical: and(base, eq(notifications.severity, 'CRITICAL')),
+      intelligence: and(base, eq(notifications.severity, 'INTELLIGENCE')),
+      system: and(base, eq(notifications.severity, 'SYSTEM')),
+    } satisfies Record<InboxFilter, unknown>;
+
+    const rows = await db.query.notifications.findMany({
+      where: filters[filter] as never,
+      orderBy: desc(notifications.createdAt),
+      limit,
+    });
+
+    const [unread] = await db
+      .select({ n: count() })
+      .from(notifications)
+      .where(and(base, isNull(notifications.readAt)));
+
+    return {
+      unreadCount: unread?.n ?? 0,
+      notifications: rows.map((row) => ({
+        id: row.id,
+        category: row.category,
+        severity: row.severity,
+        title: row.title,
+        body: row.body,
+        actionUrl: row.actionUrl,
+        actionType: row.actionType,
+        status: row.status,
+        aggregatedCount: row.aggregatedCount,
+        sourceEntityType: row.sourceEntityType,
+        sourceEntityId: row.sourceEntityId,
+        createdAt: row.createdAt,
+        readAt: row.readAt,
+        acknowledgedAt: row.acknowledgedAt,
+        snoozedUntil: row.snoozedUntil,
+      })),
+    };
+  }
+
+  async markRead(session: AuthenticatedSession, ids: string[]) {
+    if (ids.length === 0) return { updated: 0 };
+    const updated = await db
+      .update(notifications)
+      .set({ status: 'READ', readAt: new Date(), updatedAt: new Date() })
+      .where(and(this.ownedBy(session), inArray(notifications.id, ids), isNull(notifications.readAt)))
+      .returning({ id: notifications.id });
+    return { updated: updated.length };
+  }
+
+  /** Acknowledge = user a luat act si opreste escaladarea (spec §14). */
+  async acknowledge(session: AuthenticatedSession, id: string) {
+    const row = await this.getOwned(session, id);
+    await db
+      .update(notifications)
+      .set({
+        status: 'ACKNOWLEDGED',
+        acknowledgedAt: new Date(),
+        readAt: row.readAt ?? new Date(),
+        updatedAt: new Date(),
+      })
+      .where(eq(notifications.id, row.id));
+
+    await this.audit.record(null, {
+      tenantId: session.tenantId,
+      actorId: session.userId,
+      action: 'notification.acknowledged',
+      resource: `notification:${id}`,
+    });
+    return { acknowledged: true };
+  }
+
+  async dismiss(session: AuthenticatedSession, id: string) {
+    const row = await this.getOwned(session, id);
+    await db
+      .update(notifications)
+      .set({ status: 'DISMISSED', updatedAt: new Date() })
+      .where(eq(notifications.id, row.id));
+    return { dismissed: true };
+  }
+
+  async snooze(session: AuthenticatedSession, id: string, minutes: number) {
+    const row = await this.getOwned(session, id);
+    const until = new Date(Date.now() + minutes * 60_000);
+    await db
+      .update(notifications)
+      .set({ snoozedUntil: until, status: 'SCHEDULED', scheduledAt: until, updatedAt: new Date() })
+      .where(eq(notifications.id, row.id));
+    return { snoozedUntil: until };
+  }
+
+  async getPreferences(session: AuthenticatedSession) {
+    const prefs = await db.query.notificationPreferences.findFirst({
+      where: and(
+        eq(notificationPreferences.userId, session.userId),
+        eq(notificationPreferences.tenantId, session.tenantId),
+      ),
+    });
+    return (
+      prefs ?? {
+        enabled: 1,
+        minimumSeverity: 'INFORMATION' as Severity,
+        quietHoursStart: null,
+        quietHoursEnd: null,
+        criticalBypassesQuietHours: 1,
+        timezone: session.timezone,
+        isDefault: true,
+      }
+    );
+  }
+
+  async updatePreferences(
+    session: AuthenticatedSession,
+    input: {
+      enabled?: boolean;
+      minimumSeverity?: Severity;
+      quietHoursStart?: number | null;
+      quietHoursEnd?: number | null;
+      criticalBypassesQuietHours?: boolean;
+    },
+  ) {
+    const values = {
+      tenantId: session.tenantId,
+      userId: session.userId,
+      category: null,
+      enabled: input.enabled === undefined ? 1 : input.enabled ? 1 : 0,
+      minimumSeverity: input.minimumSeverity ?? ('INFORMATION' as Severity),
+      quietHoursStart: input.quietHoursStart ?? null,
+      quietHoursEnd: input.quietHoursEnd ?? null,
+      criticalBypassesQuietHours:
+        input.criticalBypassesQuietHours === undefined
+          ? 1
+          : input.criticalBypassesQuietHours
+            ? 1
+            : 0,
+      timezone: session.timezone,
+      updatedAt: new Date(),
+    };
+
+    const [row] = await db
+      .insert(notificationPreferences)
+      .values(values)
+      .onConflictDoUpdate({
+        target: [
+          notificationPreferences.userId,
+          notificationPreferences.tenantId,
+          notificationPreferences.category,
+        ],
+        set: values,
+      })
+      .returning();
+    return row;
+  }
+
+  private ownedBy(session: AuthenticatedSession) {
+    return and(
+      eq(notifications.tenantId, session.tenantId),
+      eq(notifications.recipientUserId, session.userId),
+    );
+  }
+
+  private async getOwned(session: AuthenticatedSession, id: string) {
+    const row = await db.query.notifications.findFirst({
+      where: and(eq(notifications.id, id), this.ownedBy(session)),
+    });
+    if (!row) {
+      throw new NotFoundException('Notification not found');
+    }
+    return row;
+  }
+}

src/research-briefs/research-briefs.module.ts

@@ -1,11 +1,12 @@
 import { Module } from '@nestjs/common';
 import { AiGatewayModule } from '../ai-gateway/ai-gateway.module';
+import { EventsModule } from '../events/events.module';
 import { IntelligenceModule } from '../intelligence/intelligence.module';
 import { ResearchBriefsController } from './research-briefs.controller';
 import { ResearchBriefsService } from './research-briefs.service';
 
 @Module({
-  imports: [AiGatewayModule, IntelligenceModule],
+  imports: [AiGatewayModule, EventsModule, IntelligenceModule],
   controllers: [ResearchBriefsController],
   providers: [ResearchBriefsService],
 })

src/research-briefs/research-briefs.service.ts

@@ -3,6 +3,7 @@ import { and, desc, eq } from 'drizzle-orm';
 import { db } from '../db/client';
 import { researchBriefs } from '../db/schema';
 import { AuditService } from '../audit/audit.service';
+import { OutboxService } from '../events/outbox.service';
 import { AiGatewayService } from '../ai-gateway/ai-gateway.service';
 import { IntelligenceService } from '../intelligence/intelligence.service';
 import type { AuthenticatedSession } from '../auth/tenant.guard';
@@ -16,6 +17,7 @@ export class ResearchBriefsService {
     private readonly audit: AuditService,
     private readonly aiGateway: AiGatewayService,
     private readonly intelligence: IntelligenceService,
+    private readonly outbox: OutboxService,
   ) {}
 
   /**
@@ -121,6 +123,18 @@ export class ResearchBriefsService {
       action: 'research_brief.created',
       resource: `research_brief:${row.id}`,
     });
+    await this.outbox.withTransaction(async (tx) => {
+      await this.outbox.record(tx, {
+        tenantId: session.tenantId,
+        workspaceId: session.workspaceId,
+        actorId: session.userId,
+        eventType: 'research_brief.created',
+        aggregateType: 'research_brief',
+        subjectId: row.id,
+        correlationId: session.correlationId,
+        payload: { title: row.title, organizationId: row.organizationId },
+      });
+    });
     return row;
   }