feat(projection): wire transactionsUnclassifiedCount to executive dashboard

This commit is contained in:
admin-valentin 2026-07-31 10:46:11 +00:00
parent bdcf63e2a6
commit 281fb9e780

View file

@ -3,10 +3,12 @@ import type { NodePgDatabase } from 'drizzle-orm/node-postgres';
import * as schema from '../db/schema'; import * as schema from '../db/schema';
import { import {
executiveDashboardProjection, executiveDashboardProjection,
goals,
organizations, organizations,
researchBriefs, researchBriefs,
savedSegments, savedSegments,
tasks, tasks,
transactions,
} from '../db/schema'; } from '../db/schema';
type Tx = NodePgDatabase<typeof schema>; type Tx = NodePgDatabase<typeof schema>;
@ -19,33 +21,17 @@ export interface ProjectionEvent {
createdAt: Date; 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 { export interface ProjectionDefinition {
projectionName: string; projectionName: string;
version: number; version: number;
subscribedEvents: string[]; subscribedEvents: string[];
rebuildStrategy: 'events' | 'canonical-tables' | 'hybrid'; rebuildStrategy: 'events' | 'canonical-tables' | 'hybrid';
/** Aplica efectul unui eveniment. Trebuie sa fie idempotent. */
apply(tx: Tx, event: ProjectionEvent): Promise<void>; apply(tx: Tx, event: ProjectionEvent): Promise<void>;
/** Recalculeaza complet pentru un tenant (folosit de rebuild). */
rebuildTenant(tx: Tx, tenantId: string, workspaceId: string): Promise<void>; rebuildTenant(tx: Tx, tenantId: string, workspaceId: string): Promise<void>;
} }
const SEVEN_DAYS_MS = 7 * 24 * 60 * 60 * 1000; 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( async function recomputeExecutiveDashboard(
tx: Tx, tx: Tx,
tenantId: string, tenantId: string,
@ -60,7 +46,15 @@ async function recomputeExecutiveDashboard(
ne(tasks.status, 'cancelled'), ne(tasks.status, 'cancelled'),
); );
const [openTasks, overdueTasks, upcoming, orgs, segments, briefs] = await Promise.all([ const [
openTasks,
overdueTasks,
upcoming,
orgs,
segments,
briefs,
txUnclassified,
] = await Promise.all([
tx.select({ n: count() }).from(tasks).where(activeTask), 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, lt(tasks.dueAt, now))),
tx tx
@ -73,6 +67,16 @@ async function recomputeExecutiveDashboard(
.where(and(eq(organizations.tenantId, tenantId), isNull(organizations.deletedAt))), .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(savedSegments).where(eq(savedSegments.tenantId, tenantId)),
tx.select({ n: count() }).from(researchBriefs).where(eq(researchBriefs.tenantId, tenantId)), tx.select({ n: count() }).from(researchBriefs).where(eq(researchBriefs.tenantId, tenantId)),
tx
.select({ n: count() })
.from(transactions)
.where(
and(
eq(transactions.tenantId, tenantId),
isNull(transactions.deletedAt),
eq(transactions.evidenceStatus, 'missing'),
),
),
]); ]);
const row = { const row = {
@ -84,11 +88,11 @@ async function recomputeExecutiveDashboard(
organizationsCount: orgs[0]?.n ?? 0, organizationsCount: orgs[0]?.n ?? 0,
segmentsCount: segments[0]?.n ?? 0, segmentsCount: segments[0]?.n ?? 0,
researchBriefsCount: briefs[0]?.n ?? 0, researchBriefsCount: briefs[0]?.n ?? 0,
transactionsUnclassifiedCount: txUnclassified[0]?.n ?? 0,
generatedAt: new Date(), generatedAt: new Date(),
projectionVersion: 1, projectionVersion: 1,
}; };
// UPSERT, nu insert orb (spec §5.3).
await tx await tx
.insert(executiveDashboardProjection) .insert(executiveDashboardProjection)
.values(row) .values(row)
@ -115,11 +119,13 @@ export const EXECUTIVE_DASHBOARD_PROJECTION: ProjectionDefinition = {
'research_brief.created', 'research_brief.created',
'research_brief.deleted', 'research_brief.deleted',
'user.onboarded', 'user.onboarded',
'transaction.created',
'transaction.evidence_updated',
'goal.created',
'goal.deleted',
], ],
async apply(tx, event) { async apply(tx, event) {
if (!event.workspaceId) { 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; return;
} }
await recomputeExecutiveDashboard(tx, event.tenantId, event.workspaceId); await recomputeExecutiveDashboard(tx, event.tenantId, event.workspaceId);