diff --git a/drizzle/0006_notifications_engine.sql b/drizzle/0006_notifications_engine.sql new file mode 100644 index 0000000..15b3c5b --- /dev/null +++ b/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"); diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index fa5c618..adc9312 100644 --- a/drizzle/meta/_journal.json +++ b/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 } ] } \ No newline at end of file diff --git a/src/app.module.ts b/src/app.module.ts index 1c863ac..b93812c 100644 --- a/src/app.module.ts +++ b/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, diff --git a/src/db/schema.ts b/src/db/schema.ts index 6e7caea..cbe142c 100644 --- a/src/db/schema.ts +++ b/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', diff --git a/src/notifications/dto.ts b/src/notifications/dto.ts new file mode 100644 index 0000000..195f4e1 --- /dev/null +++ b/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; +} diff --git a/src/notifications/notification-rules.spec.ts b/src/notifications/notification-rules.spec.ts new file mode 100644 index 0000000..57ee4aa --- /dev/null +++ b/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); + } + }); +}); diff --git a/src/notifications/notification-rules.ts b/src/notifications/notification-rules.ts new file mode 100644 index 0000000..1a7cbc7 --- /dev/null +++ b/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; + 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, 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 = { + 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))]; +} diff --git a/src/notifications/notification-runner.service.ts b/src/notifications/notification-runner.service.ts new file mode 100644 index 0000000..b4f4b3b --- /dev/null +++ b/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 { + try { + await this.processBatch(); + } catch (error) { + this.logger.error('Notification runner failed', error as Error); + } + } + + async processBatch(): Promise { + 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 { + // 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 { + 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 { + await db + .update(notifications) + .set({ status: 'DELIVERED', updatedAt: new Date() }) + .where( + and( + eq(notifications.status, 'SCHEDULED'), + sql`${notifications.scheduledAt} <= now()`, + ), + ); + } +} diff --git a/src/notifications/notifications.controller.ts b/src/notifications/notifications.controller.ts new file mode 100644 index 0000000..8334584 --- /dev/null +++ b/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); + } +} diff --git a/src/notifications/notifications.module.ts b/src/notifications/notifications.module.ts new file mode 100644 index 0000000..6e18d96 --- /dev/null +++ b/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 {} diff --git a/src/notifications/notifications.service.ts b/src/notifications/notifications.service.ts new file mode 100644 index 0000000..d674126 --- /dev/null +++ b/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; + + 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; + } +} diff --git a/src/research-briefs/research-briefs.module.ts b/src/research-briefs/research-briefs.module.ts index 0275682..dccb26f 100644 --- a/src/research-briefs/research-briefs.module.ts +++ b/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], }) diff --git a/src/research-briefs/research-briefs.service.ts b/src/research-briefs/research-briefs.service.ts index f4e500a..bf0abd2 100644 --- a/src/research-briefs/research-briefs.service.ts +++ b/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; }