feat: insights
* fix: migration for newly created self-hosting instances * fix: build script * wip * wip * wip * fix: tailwind css
This commit is contained in:
Carl-Gerhard Lindesvärd
committed by
GitHub
GitHub
parent
1e4f02fb5e
commit
5f38560373
@@ -43,7 +43,7 @@ class Expression {
|
||||
}
|
||||
|
||||
export class Query<T = any> {
|
||||
private _select: string[] = [];
|
||||
private _select: (string | Expression)[] = [];
|
||||
private _except: string[] = [];
|
||||
private _from?: string | Expression;
|
||||
private _where: WhereCondition[] = [];
|
||||
@@ -81,17 +81,19 @@ export class Query<T = any> {
|
||||
|
||||
// Select methods
|
||||
select<U>(
|
||||
columns: (string | null | undefined | false)[],
|
||||
columns: (string | Expression | null | undefined | false)[],
|
||||
type: 'merge' | 'replace' = 'replace',
|
||||
): Query<U> {
|
||||
if (this._skipNext) return this as unknown as Query<U>;
|
||||
if (type === 'merge') {
|
||||
this._select = [
|
||||
...this._select,
|
||||
...columns.filter((col): col is string => Boolean(col)),
|
||||
...columns.filter((col): col is string | Expression => Boolean(col)),
|
||||
];
|
||||
} else {
|
||||
this._select = columns.filter((col): col is string => Boolean(col));
|
||||
this._select = columns.filter((col): col is string | Expression =>
|
||||
Boolean(col),
|
||||
);
|
||||
}
|
||||
return this as unknown as Query<U>;
|
||||
}
|
||||
@@ -372,7 +374,14 @@ export class Query<T = any> {
|
||||
if (this._select.length > 0) {
|
||||
parts.push(
|
||||
'SELECT',
|
||||
this._select.map((col) => this.escapeDate(col)).join(', '),
|
||||
this._select
|
||||
// Important: Expressions are treated as raw SQL; do not run escapeDate()
|
||||
// on them, otherwise any embedded date strings get double-escaped
|
||||
// (e.g. ''2025-12-16 23:59:59'') which ClickHouse rejects.
|
||||
.map((col) =>
|
||||
col instanceof Expression ? col.toString() : this.escapeDate(col),
|
||||
)
|
||||
.join(', '),
|
||||
);
|
||||
} else {
|
||||
parts.push('SELECT *');
|
||||
|
||||
@@ -42,11 +42,11 @@ const getPrismaClient = () => {
|
||||
operation === 'update' ||
|
||||
operation === 'delete'
|
||||
) {
|
||||
logger.info('Prisma operation', {
|
||||
operation,
|
||||
args,
|
||||
model,
|
||||
});
|
||||
// logger.info('Prisma operation', {
|
||||
// operation,
|
||||
// args,
|
||||
// model,
|
||||
// });
|
||||
}
|
||||
return query(args);
|
||||
},
|
||||
|
||||
68
packages/db/src/services/insights/cached-clix.ts
Normal file
68
packages/db/src/services/insights/cached-clix.ts
Normal file
@@ -0,0 +1,68 @@
|
||||
import crypto from 'node:crypto';
|
||||
import type { ClickHouseClient } from '@clickhouse/client';
|
||||
import {
|
||||
type Query,
|
||||
clix as originalClix,
|
||||
} from '../../clickhouse/query-builder';
|
||||
|
||||
/**
|
||||
* Creates a cached wrapper around clix that automatically caches query results
|
||||
* based on query hash. This eliminates duplicate queries within the same module/window context.
|
||||
*
|
||||
* @param client - ClickHouse client
|
||||
* @param cache - Optional cache Map to store query results
|
||||
* @param timezone - Timezone for queries (defaults to UTC)
|
||||
* @returns A function that creates cached Query instances (compatible with clix API)
|
||||
*/
|
||||
export function createCachedClix(
|
||||
client: ClickHouseClient,
|
||||
cache?: Map<string, any>,
|
||||
timezone?: string,
|
||||
) {
|
||||
function clixCached(): Query {
|
||||
const query = originalClix(client, timezone);
|
||||
const queryTimezone = timezone ?? 'UTC';
|
||||
|
||||
// Override execute() method to add caching
|
||||
const originalExecute = query.execute.bind(query);
|
||||
query.execute = async () => {
|
||||
// Build the query SQL string
|
||||
const querySQL = query.toSQL();
|
||||
|
||||
// Create cache key from query SQL + timezone
|
||||
const cacheKey = crypto
|
||||
.createHash('sha256')
|
||||
.update(`${querySQL}|${queryTimezone}`)
|
||||
.digest('hex');
|
||||
|
||||
// Check cache first
|
||||
if (cache?.has(cacheKey)) {
|
||||
return cache.get(cacheKey);
|
||||
}
|
||||
|
||||
// Execute query
|
||||
const result = await originalExecute();
|
||||
|
||||
// Cache the result
|
||||
if (cache) {
|
||||
cache.set(cacheKey, result);
|
||||
}
|
||||
|
||||
return result;
|
||||
};
|
||||
|
||||
return query;
|
||||
}
|
||||
|
||||
// Copy static methods from original clix
|
||||
clixCached.exp = originalClix.exp;
|
||||
clixCached.date = originalClix.date;
|
||||
clixCached.datetime = originalClix.datetime;
|
||||
clixCached.dynamicDatetime = originalClix.dynamicDatetime;
|
||||
clixCached.toStartOf = originalClix.toStartOf;
|
||||
clixCached.toStartOfInterval = originalClix.toStartOfInterval;
|
||||
clixCached.toInterval = originalClix.toInterval;
|
||||
clixCached.toDate = originalClix.toDate;
|
||||
|
||||
return clixCached;
|
||||
}
|
||||
303
packages/db/src/services/insights/engine.ts
Normal file
303
packages/db/src/services/insights/engine.ts
Normal file
@@ -0,0 +1,303 @@
|
||||
import { createCachedClix } from './cached-clix';
|
||||
import { materialDecision } from './material';
|
||||
import { defaultImpactScore, severityBand } from './scoring';
|
||||
import type {
|
||||
Cadence,
|
||||
ComputeContext,
|
||||
ComputeResult,
|
||||
InsightModule,
|
||||
InsightStore,
|
||||
WindowKind,
|
||||
} from './types';
|
||||
import { resolveWindow } from './windows';
|
||||
|
||||
const DEFAULT_WINDOWS: WindowKind[] = [
|
||||
'yesterday',
|
||||
'rolling_7d',
|
||||
'rolling_30d',
|
||||
];
|
||||
|
||||
export interface EngineConfig {
|
||||
keepTopNPerModuleWindow: number; // e.g. 5
|
||||
closeStaleAfterDays: number; // e.g. 7
|
||||
dimensionBatchSize: number; // e.g. 50
|
||||
globalThresholds: {
|
||||
minTotal: number; // e.g. 200
|
||||
minAbsDelta: number; // e.g. 80
|
||||
minPct: number; // e.g. 0.15
|
||||
};
|
||||
}
|
||||
|
||||
/** Simple gating to cut noise; modules can override via thresholds. */
|
||||
function passesThresholds(
|
||||
r: ComputeResult,
|
||||
mod: InsightModule,
|
||||
cfg: EngineConfig,
|
||||
): boolean {
|
||||
const t = mod.thresholds ?? {};
|
||||
const minTotal = t.minTotal ?? cfg.globalThresholds.minTotal;
|
||||
const minAbsDelta = t.minAbsDelta ?? cfg.globalThresholds.minAbsDelta;
|
||||
const minPct = t.minPct ?? cfg.globalThresholds.minPct;
|
||||
const cur = r.currentValue ?? 0;
|
||||
const cmp = r.compareValue ?? 0;
|
||||
const total = cur + cmp;
|
||||
const absDelta = Math.abs(cur - cmp);
|
||||
const pct = Math.abs(r.changePct ?? 0);
|
||||
if (total < minTotal) return false;
|
||||
if (absDelta < minAbsDelta) return false;
|
||||
if (pct < minPct) return false;
|
||||
return true;
|
||||
}
|
||||
|
||||
function chunk<T>(arr: T[], size: number): T[][] {
|
||||
if (size <= 0) return [arr];
|
||||
const out: T[][] = [];
|
||||
for (let i = 0; i < arr.length; i += size) out.push(arr.slice(i, i + size));
|
||||
return out;
|
||||
}
|
||||
|
||||
export function createEngine(args: {
|
||||
store: InsightStore;
|
||||
modules: InsightModule[];
|
||||
db: any;
|
||||
logger?: Pick<Console, 'info' | 'warn' | 'error'>;
|
||||
config: EngineConfig;
|
||||
}) {
|
||||
const { store, modules, db, config } = args;
|
||||
const logger = args.logger ?? console;
|
||||
|
||||
function isProjectOldEnoughForWindow(
|
||||
projectCreatedAt: Date | null | undefined,
|
||||
baselineStart: Date,
|
||||
): boolean {
|
||||
if (!projectCreatedAt) return true; // best-effort; don't block if unknown
|
||||
return projectCreatedAt.getTime() <= baselineStart.getTime();
|
||||
}
|
||||
|
||||
async function runProject(opts: {
|
||||
projectId: string;
|
||||
cadence: Cadence;
|
||||
now: Date;
|
||||
projectCreatedAt?: Date | null;
|
||||
}): Promise<void> {
|
||||
const { projectId, cadence, now, projectCreatedAt } = opts;
|
||||
const projLogger = logger;
|
||||
const eligible = modules.filter((m) => m.cadence.includes(cadence));
|
||||
|
||||
for (const mod of eligible) {
|
||||
const windows = mod.windows ?? DEFAULT_WINDOWS;
|
||||
for (const windowKind of windows) {
|
||||
let window: ReturnType<typeof resolveWindow>;
|
||||
let ctx: ComputeContext;
|
||||
try {
|
||||
window = resolveWindow(windowKind, now);
|
||||
if (
|
||||
!isProjectOldEnoughForWindow(projectCreatedAt, window.baselineStart)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
// Initialize cache for this module+window combination.
|
||||
// Cache is automatically garbage collected when context goes out of scope.
|
||||
const cache = new Map<string, any>();
|
||||
ctx = {
|
||||
projectId,
|
||||
window,
|
||||
db,
|
||||
now,
|
||||
logger: projLogger,
|
||||
clix: createCachedClix(db, cache),
|
||||
};
|
||||
} catch (e) {
|
||||
projLogger.error('[insights] failed to create compute context', {
|
||||
projectId,
|
||||
module: mod.key,
|
||||
windowKind,
|
||||
err: e,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
// 1) enumerate dimensions
|
||||
let dims: string[] = [];
|
||||
try {
|
||||
dims = mod.enumerateDimensions
|
||||
? await mod.enumerateDimensions(ctx)
|
||||
: [];
|
||||
} catch (e) {
|
||||
// Important: enumeration failures should not abort the whole project run.
|
||||
// Also avoid lifecycle close/suppression when we didn't actually evaluate dims.
|
||||
projLogger.error('[insights] module enumerateDimensions failed', {
|
||||
projectId,
|
||||
module: mod.key,
|
||||
windowKind,
|
||||
err: e,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const maxDims = mod.thresholds?.maxDims ?? 25;
|
||||
if (dims.length > maxDims) dims = dims.slice(0, maxDims);
|
||||
|
||||
if (dims.length === 0) {
|
||||
// Still do lifecycle close / suppression based on "nothing emitted"
|
||||
await store.closeMissingActiveInsights({
|
||||
projectId,
|
||||
moduleKey: mod.key,
|
||||
windowKind,
|
||||
seenDimensionKeys: [],
|
||||
now,
|
||||
staleDays: config.closeStaleAfterDays,
|
||||
});
|
||||
|
||||
await store.applySuppression({
|
||||
projectId,
|
||||
moduleKey: mod.key,
|
||||
windowKind,
|
||||
keepTopN: config.keepTopNPerModuleWindow,
|
||||
now,
|
||||
});
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
// 2) compute in batches
|
||||
const seen: string[] = [];
|
||||
const dimBatches = chunk(dims, config.dimensionBatchSize);
|
||||
for (const batch of dimBatches) {
|
||||
let results: ComputeResult[] = [];
|
||||
try {
|
||||
results = await mod.computeMany(ctx, batch);
|
||||
} catch (e) {
|
||||
projLogger.error('[insights] module computeMany failed', {
|
||||
projectId,
|
||||
module: mod.key,
|
||||
windowKind,
|
||||
err: e,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const r of results) {
|
||||
if (!r?.ok) continue;
|
||||
if (!r.dimensionKey) continue;
|
||||
|
||||
// 3) gate noise
|
||||
if (!passesThresholds(r, mod, config)) continue;
|
||||
|
||||
// 4) score
|
||||
const impact = mod.score
|
||||
? mod.score(r, ctx)
|
||||
: defaultImpactScore(r);
|
||||
const sev = severityBand(r.changePct);
|
||||
|
||||
// 5) dedupe/material change requires loading prev identity
|
||||
const prev = await store.getActiveInsightByIdentity({
|
||||
projectId,
|
||||
moduleKey: mod.key,
|
||||
dimensionKey: r.dimensionKey,
|
||||
windowKind,
|
||||
});
|
||||
|
||||
const decision = materialDecision(prev, {
|
||||
changePct: r.changePct,
|
||||
direction: r.direction,
|
||||
});
|
||||
|
||||
// 6) render
|
||||
const card = mod.render(r, ctx);
|
||||
|
||||
// 7) upsert
|
||||
const persisted = await store.upsertInsight({
|
||||
projectId,
|
||||
moduleKey: mod.key,
|
||||
dimensionKey: r.dimensionKey,
|
||||
window,
|
||||
card,
|
||||
metrics: {
|
||||
direction: r.direction,
|
||||
impactScore: impact,
|
||||
severityBand: sev,
|
||||
},
|
||||
now,
|
||||
decision,
|
||||
prev,
|
||||
});
|
||||
|
||||
seen.push(r.dimensionKey);
|
||||
|
||||
// 8) events only when material
|
||||
if (!prev) {
|
||||
await store.insertEvent({
|
||||
projectId,
|
||||
insightId: persisted.id,
|
||||
moduleKey: mod.key,
|
||||
dimensionKey: r.dimensionKey,
|
||||
windowKind,
|
||||
eventKind: 'created',
|
||||
changeFrom: null,
|
||||
changeTo: {
|
||||
title: card.title,
|
||||
changePct: r.changePct,
|
||||
direction: r.direction,
|
||||
impact,
|
||||
severityBand: sev,
|
||||
},
|
||||
now,
|
||||
});
|
||||
} else if (decision.material) {
|
||||
const eventKind =
|
||||
decision.reason === 'direction_flip'
|
||||
? 'direction_flip'
|
||||
: decision.reason === 'severity_change'
|
||||
? sev && prev.severityBand && sev > prev.severityBand
|
||||
? 'severity_up'
|
||||
: 'severity_down'
|
||||
: 'updated';
|
||||
|
||||
await store.insertEvent({
|
||||
projectId,
|
||||
insightId: persisted.id,
|
||||
moduleKey: mod.key,
|
||||
dimensionKey: r.dimensionKey,
|
||||
windowKind,
|
||||
eventKind,
|
||||
changeFrom: {
|
||||
direction: prev.direction,
|
||||
impactScore: prev.impactScore,
|
||||
severityBand: prev.severityBand,
|
||||
},
|
||||
changeTo: {
|
||||
changePct: r.changePct,
|
||||
direction: r.direction,
|
||||
impactScore: impact,
|
||||
severityBand: sev,
|
||||
},
|
||||
now,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 10) lifecycle: close missing insights for this module/window
|
||||
await store.closeMissingActiveInsights({
|
||||
projectId,
|
||||
moduleKey: mod.key,
|
||||
windowKind,
|
||||
seenDimensionKeys: seen,
|
||||
now,
|
||||
staleDays: config.closeStaleAfterDays,
|
||||
});
|
||||
|
||||
// 11) suppression: keep top N
|
||||
await store.applySuppression({
|
||||
projectId,
|
||||
moduleKey: mod.key,
|
||||
windowKind,
|
||||
keepTopN: config.keepTopNPerModuleWindow,
|
||||
now,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return { runProject };
|
||||
}
|
||||
8
packages/db/src/services/insights/index.ts
Normal file
8
packages/db/src/services/insights/index.ts
Normal file
@@ -0,0 +1,8 @@
|
||||
export * from './types';
|
||||
export * from './windows';
|
||||
export * from './scoring';
|
||||
export * from './material';
|
||||
export * from './engine';
|
||||
export * from './store';
|
||||
export * from './utils';
|
||||
export * from './modules';
|
||||
43
packages/db/src/services/insights/material.ts
Normal file
43
packages/db/src/services/insights/material.ts
Normal file
@@ -0,0 +1,43 @@
|
||||
import { severityBand as band } from './scoring';
|
||||
import type { MaterialDecision, PersistedInsight } from './types';
|
||||
|
||||
export function materialDecision(
|
||||
prev: PersistedInsight | null,
|
||||
next: {
|
||||
changePct?: number;
|
||||
direction?: 'up' | 'down' | 'flat';
|
||||
},
|
||||
): MaterialDecision {
|
||||
const nextBand = band(next.changePct);
|
||||
if (!prev) {
|
||||
return { material: true, reason: 'created', newSeverityBand: nextBand };
|
||||
}
|
||||
|
||||
// direction flip is always meaningful
|
||||
const prevDir = (prev.direction ?? 'flat') as any;
|
||||
const nextDir = next.direction ?? 'flat';
|
||||
if (prevDir !== nextDir && (nextDir === 'up' || nextDir === 'down')) {
|
||||
return {
|
||||
material: true,
|
||||
reason: 'direction_flip',
|
||||
newSeverityBand: nextBand,
|
||||
};
|
||||
}
|
||||
|
||||
// severity band change
|
||||
const prevBand = (prev.severityBand ?? null) as any;
|
||||
if (prevBand !== nextBand && nextBand !== null) {
|
||||
return {
|
||||
material: true,
|
||||
reason: 'severity_change',
|
||||
newSeverityBand: nextBand,
|
||||
};
|
||||
}
|
||||
|
||||
// Otherwise: treat as non-material (silent refresh). You can add deadband crossing here if you store prior changePct.
|
||||
return {
|
||||
material: false,
|
||||
reason: 'none',
|
||||
newSeverityBand: prevBand ?? nextBand,
|
||||
};
|
||||
}
|
||||
275
packages/db/src/services/insights/modules/devices.module.ts
Normal file
275
packages/db/src/services/insights/modules/devices.module.ts
Normal file
@@ -0,0 +1,275 @@
|
||||
import { TABLE_NAMES, formatClickhouseDate } from '../../../clickhouse/client';
|
||||
import type {
|
||||
ComputeContext,
|
||||
ComputeResult,
|
||||
InsightModule,
|
||||
RenderedCard,
|
||||
} from '../types';
|
||||
import {
|
||||
buildLookupMap,
|
||||
computeChangePct,
|
||||
computeDirection,
|
||||
computeMedian,
|
||||
getEndOfDay,
|
||||
getWeekday,
|
||||
selectTopDimensions,
|
||||
} from '../utils';
|
||||
|
||||
async function fetchDeviceAggregates(ctx: ComputeContext): Promise<{
|
||||
currentMap: Map<string, number>;
|
||||
baselineMap: Map<string, number>;
|
||||
totalCurrent: number;
|
||||
totalBaseline: number;
|
||||
}> {
|
||||
if (ctx.window.kind === 'yesterday') {
|
||||
const [currentResults, baselineResults, totals] = await Promise.all([
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ device: string; cnt: number }>(['device', 'count(*) as cnt'])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.start,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.groupBy(['device'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ date: string; device: string; cnt: number }>([
|
||||
'toDate(created_at) as date',
|
||||
'device',
|
||||
'count(*) as cnt',
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.baselineEnd),
|
||||
])
|
||||
.groupBy(['date', 'device'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ cur_total: number }>([
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${formatClickhouseDate(ctx.window.start)}' AND '${formatClickhouseDate(getEndOfDay(ctx.window.end))}') as cur_total`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.execute(),
|
||||
]);
|
||||
|
||||
const currentMap = buildLookupMap(currentResults, (r) => r.device);
|
||||
|
||||
const targetWeekday = getWeekday(ctx.window.start);
|
||||
const aggregated = new Map<string, { date: string; cnt: number }[]>();
|
||||
for (const r of baselineResults) {
|
||||
if (!aggregated.has(r.device)) {
|
||||
aggregated.set(r.device, []);
|
||||
}
|
||||
const entries = aggregated.get(r.device)!;
|
||||
const existing = entries.find((e) => e.date === r.date);
|
||||
if (existing) {
|
||||
existing.cnt += Number(r.cnt ?? 0);
|
||||
} else {
|
||||
entries.push({ date: r.date, cnt: Number(r.cnt ?? 0) });
|
||||
}
|
||||
}
|
||||
|
||||
const baselineMap = new Map<string, number>();
|
||||
for (const [deviceType, entries] of aggregated) {
|
||||
const sameWeekdayValues = entries
|
||||
.filter((e) => getWeekday(new Date(e.date)) === targetWeekday)
|
||||
.map((e) => e.cnt)
|
||||
.sort((a, b) => a - b);
|
||||
|
||||
if (sameWeekdayValues.length > 0) {
|
||||
baselineMap.set(deviceType, computeMedian(sameWeekdayValues));
|
||||
}
|
||||
}
|
||||
|
||||
const totalCurrent = totals[0]?.cur_total ?? 0;
|
||||
const totalBaseline =
|
||||
baselineMap.size > 0
|
||||
? Array.from(baselineMap.values()).reduce((sum, val) => sum + val, 0)
|
||||
: 0;
|
||||
|
||||
return { currentMap, baselineMap, totalCurrent, totalBaseline };
|
||||
}
|
||||
|
||||
const curStart = formatClickhouseDate(ctx.window.start);
|
||||
const curEnd = formatClickhouseDate(getEndOfDay(ctx.window.end));
|
||||
const baseStart = formatClickhouseDate(ctx.window.baselineStart);
|
||||
const baseEnd = formatClickhouseDate(getEndOfDay(ctx.window.baselineEnd));
|
||||
|
||||
const [results, totals] = await Promise.all([
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ device: string; cur: number; base: number }>([
|
||||
'device',
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${curStart}' AND '${curEnd}') as cur`,
|
||||
),
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${baseStart}' AND '${baseEnd}') as base`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.groupBy(['device'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ cur_total: number; base_total: number }>([
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${curStart}' AND '${curEnd}') as cur_total`,
|
||||
),
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${baseStart}' AND '${baseEnd}') as base_total`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.execute(),
|
||||
]);
|
||||
|
||||
const currentMap = buildLookupMap(
|
||||
results,
|
||||
(r) => r.device,
|
||||
(r) => Number(r.cur ?? 0),
|
||||
);
|
||||
|
||||
const baselineMap = buildLookupMap(
|
||||
results,
|
||||
(r) => r.device,
|
||||
(r) => Number(r.base ?? 0),
|
||||
);
|
||||
|
||||
const totalCurrent = totals[0]?.cur_total ?? 0;
|
||||
const totalBaseline = totals[0]?.base_total ?? 0;
|
||||
|
||||
return { currentMap, baselineMap, totalCurrent, totalBaseline };
|
||||
}
|
||||
|
||||
export const devicesModule: InsightModule = {
|
||||
key: 'devices',
|
||||
cadence: ['daily'],
|
||||
thresholds: { minTotal: 100, minAbsDelta: 0, minPct: 0.08, maxDims: 5 },
|
||||
|
||||
async enumerateDimensions(ctx) {
|
||||
const { currentMap, baselineMap } = await fetchDeviceAggregates(ctx);
|
||||
const topDims = selectTopDimensions(
|
||||
currentMap,
|
||||
baselineMap,
|
||||
this.thresholds?.maxDims ?? 5,
|
||||
);
|
||||
return topDims.map((dim) => `device:${dim}`);
|
||||
},
|
||||
|
||||
async computeMany(ctx, dimensionKeys): Promise<ComputeResult[]> {
|
||||
const { currentMap, baselineMap, totalCurrent, totalBaseline } =
|
||||
await fetchDeviceAggregates(ctx);
|
||||
const results: ComputeResult[] = [];
|
||||
|
||||
for (const dimKey of dimensionKeys) {
|
||||
if (!dimKey.startsWith('device:')) continue;
|
||||
const deviceType = dimKey.replace('device:', '');
|
||||
|
||||
const currentValue = currentMap.get(deviceType) ?? 0;
|
||||
const compareValue = baselineMap.get(deviceType) ?? 0;
|
||||
|
||||
const currentShare = totalCurrent > 0 ? currentValue / totalCurrent : 0;
|
||||
const compareShare = totalBaseline > 0 ? compareValue / totalBaseline : 0;
|
||||
|
||||
const shareShiftPp = (currentShare - compareShare) * 100;
|
||||
const changePct = computeChangePct(currentValue, compareValue);
|
||||
const direction = computeDirection(changePct);
|
||||
|
||||
results.push({
|
||||
ok: true,
|
||||
dimensionKey: dimKey,
|
||||
currentValue,
|
||||
compareValue,
|
||||
changePct,
|
||||
direction,
|
||||
extra: {
|
||||
shareShiftPp,
|
||||
currentShare,
|
||||
compareShare,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
return results;
|
||||
},
|
||||
|
||||
render(result, ctx): RenderedCard {
|
||||
const device = result.dimensionKey.replace('device:', '');
|
||||
const changePct = result.changePct ?? 0;
|
||||
const isIncrease = changePct >= 0;
|
||||
|
||||
const sessionsCurrent = result.currentValue ?? 0;
|
||||
const sessionsCompare = result.compareValue ?? 0;
|
||||
const shareCurrent = Number(result.extra?.currentShare ?? 0);
|
||||
const shareCompare = Number(result.extra?.compareShare ?? 0);
|
||||
|
||||
return {
|
||||
title: `${device} ${isIncrease ? '↑' : '↓'} ${Math.abs(changePct * 100).toFixed(0)}%`,
|
||||
summary: `${ctx.window.label}. Device traffic change.`,
|
||||
displayName: device,
|
||||
payload: {
|
||||
kind: 'insight_v1',
|
||||
dimensions: [{ key: 'device', value: device, displayName: device }],
|
||||
primaryMetric: 'sessions',
|
||||
metrics: {
|
||||
sessions: {
|
||||
current: sessionsCurrent,
|
||||
compare: sessionsCompare,
|
||||
delta: sessionsCurrent - sessionsCompare,
|
||||
changePct: sessionsCompare > 0 ? (result.changePct ?? 0) : null,
|
||||
direction: result.direction ?? 'flat',
|
||||
unit: 'count',
|
||||
},
|
||||
share: {
|
||||
current: shareCurrent,
|
||||
compare: shareCompare,
|
||||
delta: shareCurrent - shareCompare,
|
||||
changePct:
|
||||
shareCompare > 0
|
||||
? (shareCurrent - shareCompare) / shareCompare
|
||||
: null,
|
||||
direction:
|
||||
shareCurrent - shareCompare > 0.0005
|
||||
? 'up'
|
||||
: shareCurrent - shareCompare < -0.0005
|
||||
? 'down'
|
||||
: 'flat',
|
||||
unit: 'ratio',
|
||||
},
|
||||
},
|
||||
extra: {
|
||||
// keep module-specific flags/fields if needed later
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
287
packages/db/src/services/insights/modules/entry-pages.module.ts
Normal file
287
packages/db/src/services/insights/modules/entry-pages.module.ts
Normal file
@@ -0,0 +1,287 @@
|
||||
import { TABLE_NAMES, formatClickhouseDate } from '../../../clickhouse/client';
|
||||
import type {
|
||||
ComputeContext,
|
||||
ComputeResult,
|
||||
InsightModule,
|
||||
RenderedCard,
|
||||
} from '../types';
|
||||
import {
|
||||
buildLookupMap,
|
||||
computeChangePct,
|
||||
computeDirection,
|
||||
computeWeekdayMedians,
|
||||
getEndOfDay,
|
||||
getWeekday,
|
||||
selectTopDimensions,
|
||||
} from '../utils';
|
||||
|
||||
const DELIMITER = '|||';
|
||||
|
||||
async function fetchEntryPageAggregates(ctx: ComputeContext): Promise<{
|
||||
currentMap: Map<string, number>;
|
||||
baselineMap: Map<string, number>;
|
||||
totalCurrent: number;
|
||||
totalBaseline: number;
|
||||
}> {
|
||||
if (ctx.window.kind === 'yesterday') {
|
||||
const [currentResults, baselineResults, totals] = await Promise.all([
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ entry_origin: string; entry_path: string; cnt: number }>([
|
||||
'entry_origin',
|
||||
'entry_path',
|
||||
'count(*) as cnt',
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.start,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.groupBy(['entry_origin', 'entry_path'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{
|
||||
date: string;
|
||||
entry_origin: string;
|
||||
entry_path: string;
|
||||
cnt: number;
|
||||
}>([
|
||||
'toDate(created_at) as date',
|
||||
'entry_origin',
|
||||
'entry_path',
|
||||
'count(*) as cnt',
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.baselineEnd),
|
||||
])
|
||||
.groupBy(['date', 'entry_origin', 'entry_path'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ cur_total: number }>([
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${formatClickhouseDate(ctx.window.start)}' AND '${formatClickhouseDate(getEndOfDay(ctx.window.end))}') as cur_total`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.execute(),
|
||||
]);
|
||||
|
||||
const currentMap = buildLookupMap(
|
||||
currentResults,
|
||||
(r) => `${r.entry_origin || ''}${DELIMITER}${r.entry_path || '/'}`,
|
||||
);
|
||||
|
||||
const targetWeekday = getWeekday(ctx.window.start);
|
||||
const baselineMap = computeWeekdayMedians(
|
||||
baselineResults,
|
||||
targetWeekday,
|
||||
(r) => `${r.entry_origin || ''}${DELIMITER}${r.entry_path || '/'}`,
|
||||
);
|
||||
|
||||
const totalCurrent = totals[0]?.cur_total ?? 0;
|
||||
const totalBaseline = Array.from(baselineMap.values()).reduce(
|
||||
(sum, val) => sum + val,
|
||||
0,
|
||||
);
|
||||
|
||||
return { currentMap, baselineMap, totalCurrent, totalBaseline };
|
||||
}
|
||||
|
||||
const curStart = formatClickhouseDate(ctx.window.start);
|
||||
const curEnd = formatClickhouseDate(getEndOfDay(ctx.window.end));
|
||||
const baseStart = formatClickhouseDate(ctx.window.baselineStart);
|
||||
const baseEnd = formatClickhouseDate(getEndOfDay(ctx.window.baselineEnd));
|
||||
|
||||
const [results, totals] = await Promise.all([
|
||||
ctx
|
||||
.clix()
|
||||
.select<{
|
||||
entry_origin: string;
|
||||
entry_path: string;
|
||||
cur: number;
|
||||
base: number;
|
||||
}>([
|
||||
'entry_origin',
|
||||
'entry_path',
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${curStart}' AND '${curEnd}') as cur`,
|
||||
),
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${baseStart}' AND '${baseEnd}') as base`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.groupBy(['entry_origin', 'entry_path'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ cur_total: number; base_total: number }>([
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${curStart}' AND '${curEnd}') as cur_total`,
|
||||
),
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${baseStart}' AND '${baseEnd}') as base_total`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.execute(),
|
||||
]);
|
||||
|
||||
const currentMap = buildLookupMap(
|
||||
results,
|
||||
(r) => `${r.entry_origin || ''}${DELIMITER}${r.entry_path || '/'}`,
|
||||
(r) => Number(r.cur ?? 0),
|
||||
);
|
||||
|
||||
const baselineMap = buildLookupMap(
|
||||
results,
|
||||
(r) => `${r.entry_origin || ''}${DELIMITER}${r.entry_path || '/'}`,
|
||||
(r) => Number(r.base ?? 0),
|
||||
);
|
||||
|
||||
const totalCurrent = totals[0]?.cur_total ?? 0;
|
||||
const totalBaseline = totals[0]?.base_total ?? 0;
|
||||
|
||||
return { currentMap, baselineMap, totalCurrent, totalBaseline };
|
||||
}
|
||||
|
||||
export const entryPagesModule: InsightModule = {
|
||||
key: 'entry-pages',
|
||||
cadence: ['daily'],
|
||||
thresholds: { minTotal: 100, minAbsDelta: 30, minPct: 0.2, maxDims: 100 },
|
||||
|
||||
async enumerateDimensions(ctx) {
|
||||
const { currentMap, baselineMap } = await fetchEntryPageAggregates(ctx);
|
||||
const topDims = selectTopDimensions(
|
||||
currentMap,
|
||||
baselineMap,
|
||||
this.thresholds?.maxDims ?? 100,
|
||||
);
|
||||
return topDims.map((dim) => `entry:${dim}`);
|
||||
},
|
||||
|
||||
async computeMany(ctx, dimensionKeys): Promise<ComputeResult[]> {
|
||||
const { currentMap, baselineMap, totalCurrent, totalBaseline } =
|
||||
await fetchEntryPageAggregates(ctx);
|
||||
const results: ComputeResult[] = [];
|
||||
|
||||
for (const dimKey of dimensionKeys) {
|
||||
if (!dimKey.startsWith('entry:')) continue;
|
||||
const originPath = dimKey.replace('entry:', '');
|
||||
|
||||
const currentValue = currentMap.get(originPath) ?? 0;
|
||||
const compareValue = baselineMap.get(originPath) ?? 0;
|
||||
|
||||
const currentShare = totalCurrent > 0 ? currentValue / totalCurrent : 0;
|
||||
const compareShare = totalBaseline > 0 ? compareValue / totalBaseline : 0;
|
||||
|
||||
const shareShiftPp = (currentShare - compareShare) * 100;
|
||||
const changePct = computeChangePct(currentValue, compareValue);
|
||||
const direction = computeDirection(changePct);
|
||||
|
||||
results.push({
|
||||
ok: true,
|
||||
dimensionKey: dimKey,
|
||||
currentValue,
|
||||
compareValue,
|
||||
changePct,
|
||||
direction,
|
||||
extra: {
|
||||
shareShiftPp,
|
||||
currentShare,
|
||||
compareShare,
|
||||
isNew: compareValue === 0 && currentValue > 0,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
return results;
|
||||
},
|
||||
|
||||
render(result, ctx): RenderedCard {
|
||||
const originPath = result.dimensionKey.replace('entry:', '');
|
||||
const [origin, path] = originPath.split(DELIMITER);
|
||||
const displayValue = origin ? `${origin}${path}` : path || '/';
|
||||
const pct = ((result.changePct ?? 0) * 100).toFixed(1);
|
||||
const isIncrease = (result.changePct ?? 0) >= 0;
|
||||
const isNew = result.extra?.isNew as boolean | undefined;
|
||||
|
||||
const title = isNew
|
||||
? `New entry page: ${displayValue}`
|
||||
: `Entry page ${displayValue} ${isIncrease ? '↑' : '↓'} ${Math.abs(Number(pct))}%`;
|
||||
|
||||
const sessionsCurrent = result.currentValue ?? 0;
|
||||
const sessionsCompare = result.compareValue ?? 0;
|
||||
const shareCurrent = Number(result.extra?.currentShare ?? 0);
|
||||
const shareCompare = Number(result.extra?.compareShare ?? 0);
|
||||
|
||||
return {
|
||||
title,
|
||||
summary: `${ctx.window.label}. Sessions ${sessionsCurrent} vs ${sessionsCompare}.`,
|
||||
displayName: displayValue,
|
||||
payload: {
|
||||
kind: 'insight_v1',
|
||||
dimensions: [
|
||||
{ key: 'origin', value: origin ?? '', displayName: origin ?? '' },
|
||||
{ key: 'path', value: path ?? '', displayName: path ?? '' },
|
||||
],
|
||||
primaryMetric: 'sessions',
|
||||
metrics: {
|
||||
sessions: {
|
||||
current: sessionsCurrent,
|
||||
compare: sessionsCompare,
|
||||
delta: sessionsCurrent - sessionsCompare,
|
||||
changePct: sessionsCompare > 0 ? (result.changePct ?? 0) : null,
|
||||
direction: result.direction ?? 'flat',
|
||||
unit: 'count',
|
||||
},
|
||||
share: {
|
||||
current: shareCurrent,
|
||||
compare: shareCompare,
|
||||
delta: shareCurrent - shareCompare,
|
||||
changePct:
|
||||
shareCompare > 0
|
||||
? (shareCurrent - shareCompare) / shareCompare
|
||||
: null,
|
||||
direction:
|
||||
shareCurrent - shareCompare > 0.0005
|
||||
? 'up'
|
||||
: shareCurrent - shareCompare < -0.0005
|
||||
? 'down'
|
||||
: 'flat',
|
||||
unit: 'ratio',
|
||||
},
|
||||
},
|
||||
extra: {
|
||||
isNew: result.extra?.isNew,
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
271
packages/db/src/services/insights/modules/geo.module.ts
Normal file
271
packages/db/src/services/insights/modules/geo.module.ts
Normal file
@@ -0,0 +1,271 @@
|
||||
import { getCountry } from '@openpanel/constants';
|
||||
import { TABLE_NAMES, formatClickhouseDate } from '../../../clickhouse/client';
|
||||
import type {
|
||||
ComputeContext,
|
||||
ComputeResult,
|
||||
InsightModule,
|
||||
RenderedCard,
|
||||
} from '../types';
|
||||
import {
|
||||
buildLookupMap,
|
||||
computeChangePct,
|
||||
computeDirection,
|
||||
computeWeekdayMedians,
|
||||
getEndOfDay,
|
||||
getWeekday,
|
||||
selectTopDimensions,
|
||||
} from '../utils';
|
||||
|
||||
async function fetchGeoAggregates(ctx: ComputeContext): Promise<{
|
||||
currentMap: Map<string, number>;
|
||||
baselineMap: Map<string, number>;
|
||||
totalCurrent: number;
|
||||
totalBaseline: number;
|
||||
}> {
|
||||
if (ctx.window.kind === 'yesterday') {
|
||||
const [currentResults, baselineResults, totals] = await Promise.all([
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ country: string; cnt: number }>([
|
||||
'country',
|
||||
'count(*) as cnt',
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.start,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.groupBy(['country'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ date: string; country: string; cnt: number }>([
|
||||
'toDate(created_at) as date',
|
||||
'country',
|
||||
'count(*) as cnt',
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.baselineEnd),
|
||||
])
|
||||
.groupBy(['date', 'country'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ cur_total: number }>([
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${formatClickhouseDate(ctx.window.start)}' AND '${formatClickhouseDate(getEndOfDay(ctx.window.end))}') as cur_total`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.execute(),
|
||||
]);
|
||||
|
||||
const currentMap = buildLookupMap(
|
||||
currentResults,
|
||||
(r) => r.country || 'unknown',
|
||||
);
|
||||
|
||||
const targetWeekday = getWeekday(ctx.window.start);
|
||||
const baselineMap = computeWeekdayMedians(
|
||||
baselineResults,
|
||||
targetWeekday,
|
||||
(r) => r.country || 'unknown',
|
||||
);
|
||||
|
||||
const totalCurrent = totals[0]?.cur_total ?? 0;
|
||||
const totalBaseline = Array.from(baselineMap.values()).reduce(
|
||||
(sum, val) => sum + val,
|
||||
0,
|
||||
);
|
||||
|
||||
return { currentMap, baselineMap, totalCurrent, totalBaseline };
|
||||
}
|
||||
|
||||
const curStart = formatClickhouseDate(ctx.window.start);
|
||||
const curEnd = formatClickhouseDate(getEndOfDay(ctx.window.end));
|
||||
const baseStart = formatClickhouseDate(ctx.window.baselineStart);
|
||||
const baseEnd = formatClickhouseDate(getEndOfDay(ctx.window.baselineEnd));
|
||||
|
||||
const [results, totals] = await Promise.all([
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ country: string; cur: number; base: number }>([
|
||||
'country',
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${curStart}' AND '${curEnd}') as cur`,
|
||||
),
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${baseStart}' AND '${baseEnd}') as base`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.groupBy(['country'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ cur_total: number; base_total: number }>([
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${curStart}' AND '${curEnd}') as cur_total`,
|
||||
),
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${baseStart}' AND '${baseEnd}') as base_total`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.execute(),
|
||||
]);
|
||||
|
||||
const currentMap = buildLookupMap(
|
||||
results,
|
||||
(r) => r.country || 'unknown',
|
||||
(r) => Number(r.cur ?? 0),
|
||||
);
|
||||
|
||||
const baselineMap = buildLookupMap(
|
||||
results,
|
||||
(r) => r.country || 'unknown',
|
||||
(r) => Number(r.base ?? 0),
|
||||
);
|
||||
|
||||
const totalCurrent = totals[0]?.cur_total ?? 0;
|
||||
const totalBaseline = totals[0]?.base_total ?? 0;
|
||||
|
||||
return { currentMap, baselineMap, totalCurrent, totalBaseline };
|
||||
}
|
||||
|
||||
export const geoModule: InsightModule = {
|
||||
key: 'geo',
|
||||
cadence: ['daily'],
|
||||
thresholds: { minTotal: 100, minAbsDelta: 0, minPct: 0.08, maxDims: 30 },
|
||||
|
||||
async enumerateDimensions(ctx) {
|
||||
const { currentMap, baselineMap } = await fetchGeoAggregates(ctx);
|
||||
const topDims = selectTopDimensions(
|
||||
currentMap,
|
||||
baselineMap,
|
||||
this.thresholds?.maxDims ?? 30,
|
||||
);
|
||||
return topDims.map((dim) => `country:${dim}`);
|
||||
},
|
||||
|
||||
async computeMany(ctx, dimensionKeys): Promise<ComputeResult[]> {
|
||||
const { currentMap, baselineMap, totalCurrent, totalBaseline } =
|
||||
await fetchGeoAggregates(ctx);
|
||||
const results: ComputeResult[] = [];
|
||||
|
||||
for (const dimKey of dimensionKeys) {
|
||||
if (!dimKey.startsWith('country:')) continue;
|
||||
const country = dimKey.replace('country:', '');
|
||||
|
||||
const currentValue = currentMap.get(country) ?? 0;
|
||||
const compareValue = baselineMap.get(country) ?? 0;
|
||||
|
||||
const currentShare = totalCurrent > 0 ? currentValue / totalCurrent : 0;
|
||||
const compareShare = totalBaseline > 0 ? compareValue / totalBaseline : 0;
|
||||
|
||||
const shareShiftPp = (currentShare - compareShare) * 100;
|
||||
const changePct = computeChangePct(currentValue, compareValue);
|
||||
const direction = computeDirection(changePct);
|
||||
|
||||
results.push({
|
||||
ok: true,
|
||||
dimensionKey: dimKey,
|
||||
currentValue,
|
||||
compareValue,
|
||||
changePct,
|
||||
direction,
|
||||
extra: {
|
||||
shareShiftPp,
|
||||
currentShare,
|
||||
compareShare,
|
||||
isNew: compareValue === 0 && currentValue > 0,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
return results;
|
||||
},
|
||||
|
||||
render(result, ctx): RenderedCard {
|
||||
const country = result.dimensionKey.replace('country:', '');
|
||||
const changePct = result.changePct ?? 0;
|
||||
const isIncrease = changePct >= 0;
|
||||
const isNew = result.extra?.isNew as boolean | undefined;
|
||||
const displayName = getCountry(country);
|
||||
|
||||
const title = isNew
|
||||
? `New traffic from: ${displayName}`
|
||||
: `${displayName} ${isIncrease ? '↑' : '↓'} ${Math.abs(changePct * 100).toFixed(0)}%`;
|
||||
|
||||
const sessionsCurrent = result.currentValue ?? 0;
|
||||
const sessionsCompare = result.compareValue ?? 0;
|
||||
const shareCurrent = Number(result.extra?.currentShare ?? 0);
|
||||
const shareCompare = Number(result.extra?.compareShare ?? 0);
|
||||
|
||||
return {
|
||||
title,
|
||||
summary: `${ctx.window.label}. Traffic change from ${displayName}.`,
|
||||
displayName,
|
||||
payload: {
|
||||
kind: 'insight_v1',
|
||||
dimensions: [
|
||||
{ key: 'country', value: country, displayName: displayName },
|
||||
],
|
||||
primaryMetric: 'sessions',
|
||||
metrics: {
|
||||
sessions: {
|
||||
current: sessionsCurrent,
|
||||
compare: sessionsCompare,
|
||||
delta: sessionsCurrent - sessionsCompare,
|
||||
changePct: sessionsCompare > 0 ? (result.changePct ?? 0) : null,
|
||||
direction: result.direction ?? 'flat',
|
||||
unit: 'count',
|
||||
},
|
||||
share: {
|
||||
current: shareCurrent,
|
||||
compare: shareCompare,
|
||||
delta: shareCurrent - shareCompare,
|
||||
changePct:
|
||||
shareCompare > 0
|
||||
? (shareCurrent - shareCompare) / shareCompare
|
||||
: null,
|
||||
direction:
|
||||
shareCurrent - shareCompare > 0.0005
|
||||
? 'up'
|
||||
: shareCurrent - shareCompare < -0.0005
|
||||
? 'down'
|
||||
: 'flat',
|
||||
unit: 'ratio',
|
||||
},
|
||||
},
|
||||
extra: {
|
||||
isNew: result.extra?.isNew,
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
5
packages/db/src/services/insights/modules/index.ts
Normal file
5
packages/db/src/services/insights/modules/index.ts
Normal file
@@ -0,0 +1,5 @@
|
||||
export { referrersModule } from './referrers.module';
|
||||
export { entryPagesModule } from './entry-pages.module';
|
||||
export { pageTrendsModule } from './page-trends.module';
|
||||
export { geoModule } from './geo.module';
|
||||
export { devicesModule } from './devices.module';
|
||||
298
packages/db/src/services/insights/modules/page-trends.module.ts
Normal file
298
packages/db/src/services/insights/modules/page-trends.module.ts
Normal file
@@ -0,0 +1,298 @@
|
||||
import { TABLE_NAMES, formatClickhouseDate } from '../../../clickhouse/client';
|
||||
import type {
|
||||
ComputeContext,
|
||||
ComputeResult,
|
||||
InsightModule,
|
||||
RenderedCard,
|
||||
} from '../types';
|
||||
import {
|
||||
buildLookupMap,
|
||||
computeChangePct,
|
||||
computeDirection,
|
||||
computeWeekdayMedians,
|
||||
getEndOfDay,
|
||||
getWeekday,
|
||||
selectTopDimensions,
|
||||
} from '../utils';
|
||||
|
||||
const DELIMITER = '|||';
|
||||
|
||||
async function fetchPageTrendAggregates(ctx: ComputeContext): Promise<{
|
||||
currentMap: Map<string, number>;
|
||||
baselineMap: Map<string, number>;
|
||||
totalCurrent: number;
|
||||
totalBaseline: number;
|
||||
}> {
|
||||
if (ctx.window.kind === 'yesterday') {
|
||||
const [currentResults, baselineResults, totals] = await Promise.all([
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ origin: string; path: string; cnt: number }>([
|
||||
'origin',
|
||||
'path',
|
||||
'count(*) as cnt',
|
||||
])
|
||||
.from(TABLE_NAMES.events)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('name', '=', 'screen_view')
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.start,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.groupBy(['origin', 'path'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ date: string; origin: string; path: string; cnt: number }>([
|
||||
'toDate(created_at) as date',
|
||||
'origin',
|
||||
'path',
|
||||
'count(*) as cnt',
|
||||
])
|
||||
.from(TABLE_NAMES.events)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('name', '=', 'screen_view')
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.baselineEnd),
|
||||
])
|
||||
.groupBy(['date', 'origin', 'path'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ cur_total: number }>([
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${formatClickhouseDate(ctx.window.start)}' AND '${formatClickhouseDate(getEndOfDay(ctx.window.end))}') as cur_total`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.events)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('name', '=', 'screen_view')
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.execute(),
|
||||
]);
|
||||
|
||||
const currentMap = buildLookupMap(
|
||||
currentResults,
|
||||
(r) => `${r.origin || ''}${DELIMITER}${r.path || '/'}`,
|
||||
);
|
||||
|
||||
const targetWeekday = getWeekday(ctx.window.start);
|
||||
const baselineMap = computeWeekdayMedians(
|
||||
baselineResults,
|
||||
targetWeekday,
|
||||
(r) => `${r.origin || ''}${DELIMITER}${r.path || '/'}`,
|
||||
);
|
||||
|
||||
const totalCurrent = totals[0]?.cur_total ?? 0;
|
||||
const totalBaseline = Array.from(baselineMap.values()).reduce(
|
||||
(sum, val) => sum + val,
|
||||
0,
|
||||
);
|
||||
|
||||
return { currentMap, baselineMap, totalCurrent, totalBaseline };
|
||||
}
|
||||
|
||||
const curStart = formatClickhouseDate(ctx.window.start);
|
||||
const curEnd = formatClickhouseDate(getEndOfDay(ctx.window.end));
|
||||
const baseStart = formatClickhouseDate(ctx.window.baselineStart);
|
||||
const baseEnd = formatClickhouseDate(getEndOfDay(ctx.window.baselineEnd));
|
||||
|
||||
const [results, totals] = await Promise.all([
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ origin: string; path: string; cur: number; base: number }>([
|
||||
'origin',
|
||||
'path',
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${curStart}' AND '${curEnd}') as cur`,
|
||||
),
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${baseStart}' AND '${baseEnd}') as base`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.events)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('name', '=', 'screen_view')
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.groupBy(['origin', 'path'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ cur_total: number; base_total: number }>([
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${curStart}' AND '${curEnd}') as cur_total`,
|
||||
),
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${baseStart}' AND '${baseEnd}') as base_total`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.events)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('name', '=', 'screen_view')
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.execute(),
|
||||
]);
|
||||
|
||||
const currentMap = buildLookupMap(
|
||||
results,
|
||||
(r) => `${r.origin || ''}${DELIMITER}${r.path || '/'}`,
|
||||
(r) => Number(r.cur ?? 0),
|
||||
);
|
||||
|
||||
const baselineMap = buildLookupMap(
|
||||
results,
|
||||
(r) => `${r.origin || ''}${DELIMITER}${r.path || '/'}`,
|
||||
(r) => Number(r.base ?? 0),
|
||||
);
|
||||
|
||||
const totalCurrent = totals[0]?.cur_total ?? 0;
|
||||
const totalBaseline = totals[0]?.base_total ?? 0;
|
||||
|
||||
return { currentMap, baselineMap, totalCurrent, totalBaseline };
|
||||
}
|
||||
|
||||
export const pageTrendsModule: InsightModule = {
|
||||
key: 'page-trends',
|
||||
cadence: ['daily'],
|
||||
// Share-based thresholds (values in basis points: 100 = 1%)
|
||||
// minTotal: require at least 0.5% combined share (current + baseline)
|
||||
// minAbsDelta: require at least 0.5 percentage point shift
|
||||
// minPct: require at least 25% relative change in share
|
||||
thresholds: { minTotal: 50, minAbsDelta: 50, minPct: 0.25, maxDims: 100 },
|
||||
|
||||
async enumerateDimensions(ctx) {
|
||||
const { currentMap, baselineMap } = await fetchPageTrendAggregates(ctx);
|
||||
const topDims = selectTopDimensions(
|
||||
currentMap,
|
||||
baselineMap,
|
||||
this.thresholds?.maxDims ?? 100,
|
||||
);
|
||||
return topDims.map((dim) => `page:${dim}`);
|
||||
},
|
||||
|
||||
async computeMany(ctx, dimensionKeys): Promise<ComputeResult[]> {
|
||||
const { currentMap, baselineMap, totalCurrent, totalBaseline } =
|
||||
await fetchPageTrendAggregates(ctx);
|
||||
const results: ComputeResult[] = [];
|
||||
|
||||
for (const dimKey of dimensionKeys) {
|
||||
if (!dimKey.startsWith('page:')) continue;
|
||||
const originPath = dimKey.replace('page:', '');
|
||||
|
||||
const pageviewsCurrent = currentMap.get(originPath) ?? 0;
|
||||
const pageviewsCompare = baselineMap.get(originPath) ?? 0;
|
||||
|
||||
const currentShare =
|
||||
totalCurrent > 0 ? pageviewsCurrent / totalCurrent : 0;
|
||||
const compareShare =
|
||||
totalBaseline > 0 ? pageviewsCompare / totalBaseline : 0;
|
||||
|
||||
// Use share values in basis points (100 = 1%) for thresholding
|
||||
// This makes thresholds intuitive: minAbsDelta=50 means 0.5pp shift
|
||||
const currentShareBp = currentShare * 10000;
|
||||
const compareShareBp = compareShare * 10000;
|
||||
|
||||
const shareShiftPp = (currentShare - compareShare) * 100;
|
||||
// changePct is relative change in share, not absolute pageviews
|
||||
const shareChangePct = computeChangePct(currentShare, compareShare);
|
||||
const direction = computeDirection(shareChangePct);
|
||||
|
||||
results.push({
|
||||
ok: true,
|
||||
dimensionKey: dimKey,
|
||||
// Use share in basis points for threshold checks
|
||||
currentValue: currentShareBp,
|
||||
compareValue: compareShareBp,
|
||||
changePct: shareChangePct,
|
||||
direction,
|
||||
extra: {
|
||||
// Keep absolute values for display
|
||||
pageviewsCurrent,
|
||||
pageviewsCompare,
|
||||
shareShiftPp,
|
||||
currentShare,
|
||||
compareShare,
|
||||
isNew: pageviewsCompare === 0 && pageviewsCurrent > 0,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
return results;
|
||||
},
|
||||
|
||||
render(result, ctx): RenderedCard {
|
||||
const originPath = result.dimensionKey.replace('page:', '');
|
||||
const [origin, path] = originPath.split(DELIMITER);
|
||||
const displayValue = origin ? `${origin}${path}` : path || '/';
|
||||
|
||||
// Get absolute pageviews from extra (currentValue/compareValue are now share-based)
|
||||
const pageviewsCurrent = Number(result.extra?.pageviewsCurrent ?? 0);
|
||||
const pageviewsCompare = Number(result.extra?.pageviewsCompare ?? 0);
|
||||
const shareCurrent = Number(result.extra?.currentShare ?? 0);
|
||||
const shareCompare = Number(result.extra?.compareShare ?? 0);
|
||||
const shareShiftPp = Number(result.extra?.shareShiftPp ?? 0);
|
||||
const isNew = result.extra?.isNew as boolean | undefined;
|
||||
|
||||
// Display share shift in percentage points
|
||||
const isIncrease = shareShiftPp >= 0;
|
||||
const shareShiftDisplay = Math.abs(shareShiftPp).toFixed(1);
|
||||
|
||||
const title = isNew
|
||||
? `New page getting views: ${displayValue}`
|
||||
: `Page ${displayValue} share ${isIncrease ? '↑' : '↓'} ${shareShiftDisplay}pp`;
|
||||
|
||||
return {
|
||||
title,
|
||||
summary: `${ctx.window.label}. Share ${(shareCurrent * 100).toFixed(1)}% vs ${(shareCompare * 100).toFixed(1)}%.`,
|
||||
displayName: displayValue,
|
||||
payload: {
|
||||
kind: 'insight_v1',
|
||||
dimensions: [
|
||||
{ key: 'origin', value: origin ?? '', displayName: origin ?? '' },
|
||||
{ key: 'path', value: path ?? '', displayName: path ?? '' },
|
||||
],
|
||||
primaryMetric: 'share',
|
||||
metrics: {
|
||||
pageviews: {
|
||||
current: pageviewsCurrent,
|
||||
compare: pageviewsCompare,
|
||||
delta: pageviewsCurrent - pageviewsCompare,
|
||||
changePct:
|
||||
pageviewsCompare > 0
|
||||
? (pageviewsCurrent - pageviewsCompare) / pageviewsCompare
|
||||
: null,
|
||||
direction:
|
||||
pageviewsCurrent > pageviewsCompare
|
||||
? 'up'
|
||||
: pageviewsCurrent < pageviewsCompare
|
||||
? 'down'
|
||||
: 'flat',
|
||||
unit: 'count',
|
||||
},
|
||||
share: {
|
||||
current: shareCurrent,
|
||||
compare: shareCompare,
|
||||
delta: shareCurrent - shareCompare,
|
||||
changePct: result.changePct ?? null, // This is now share-based
|
||||
direction: result.direction ?? 'flat',
|
||||
unit: 'ratio',
|
||||
},
|
||||
},
|
||||
extra: {
|
||||
isNew: result.extra?.isNew,
|
||||
shareShiftPp,
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
275
packages/db/src/services/insights/modules/referrers.module.ts
Normal file
275
packages/db/src/services/insights/modules/referrers.module.ts
Normal file
@@ -0,0 +1,275 @@
|
||||
import { TABLE_NAMES, formatClickhouseDate } from '../../../clickhouse/client';
|
||||
import type {
|
||||
ComputeContext,
|
||||
ComputeResult,
|
||||
InsightModule,
|
||||
RenderedCard,
|
||||
} from '../types';
|
||||
import {
|
||||
buildLookupMap,
|
||||
computeChangePct,
|
||||
computeDirection,
|
||||
computeWeekdayMedians,
|
||||
getEndOfDay,
|
||||
getWeekday,
|
||||
selectTopDimensions,
|
||||
} from '../utils';
|
||||
|
||||
async function fetchReferrerAggregates(ctx: ComputeContext): Promise<{
|
||||
currentMap: Map<string, number>;
|
||||
baselineMap: Map<string, number>;
|
||||
totalCurrent: number;
|
||||
totalBaseline: number;
|
||||
}> {
|
||||
if (ctx.window.kind === 'yesterday') {
|
||||
const [currentResults, baselineResults, totals] = await Promise.all([
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ referrer_name: string; cnt: number }>([
|
||||
'referrer_name',
|
||||
'count(*) as cnt',
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.start,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.groupBy(['referrer_name'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ date: string; referrer_name: string; cnt: number }>([
|
||||
'toDate(created_at) as date',
|
||||
'referrer_name',
|
||||
'count(*) as cnt',
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.baselineEnd),
|
||||
])
|
||||
.groupBy(['date', 'referrer_name'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ cur_total: number }>([
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${formatClickhouseDate(ctx.window.start)}' AND '${formatClickhouseDate(getEndOfDay(ctx.window.end))}') as cur_total`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.execute(),
|
||||
]);
|
||||
|
||||
const currentMap = buildLookupMap(
|
||||
currentResults,
|
||||
(r) => r.referrer_name || 'direct',
|
||||
);
|
||||
|
||||
const targetWeekday = getWeekday(ctx.window.start);
|
||||
const baselineMap = computeWeekdayMedians(
|
||||
baselineResults,
|
||||
targetWeekday,
|
||||
(r) => r.referrer_name || 'direct',
|
||||
);
|
||||
|
||||
const totalCurrent = totals[0]?.cur_total ?? 0;
|
||||
const totalBaseline = Array.from(baselineMap.values()).reduce(
|
||||
(sum, val) => sum + val,
|
||||
0,
|
||||
);
|
||||
|
||||
return { currentMap, baselineMap, totalCurrent, totalBaseline };
|
||||
}
|
||||
|
||||
const curStart = formatClickhouseDate(ctx.window.start);
|
||||
const curEnd = formatClickhouseDate(getEndOfDay(ctx.window.end));
|
||||
const baseStart = formatClickhouseDate(ctx.window.baselineStart);
|
||||
const baseEnd = formatClickhouseDate(getEndOfDay(ctx.window.baselineEnd));
|
||||
|
||||
const [results, totals] = await Promise.all([
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ referrer_name: string; cur: number; base: number }>([
|
||||
'referrer_name',
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${curStart}' AND '${curEnd}') as cur`,
|
||||
),
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${baseStart}' AND '${baseEnd}') as base`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.groupBy(['referrer_name'])
|
||||
.execute(),
|
||||
ctx
|
||||
.clix()
|
||||
.select<{ cur_total: number; base_total: number }>([
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${curStart}' AND '${curEnd}') as cur_total`,
|
||||
),
|
||||
ctx.clix.exp(
|
||||
`countIf(created_at BETWEEN '${baseStart}' AND '${baseEnd}') as base_total`,
|
||||
),
|
||||
])
|
||||
.from(TABLE_NAMES.sessions)
|
||||
.where('project_id', '=', ctx.projectId)
|
||||
.where('sign', '=', 1)
|
||||
.where('created_at', 'BETWEEN', [
|
||||
ctx.window.baselineStart,
|
||||
getEndOfDay(ctx.window.end),
|
||||
])
|
||||
.execute(),
|
||||
]);
|
||||
|
||||
const currentMap = buildLookupMap(
|
||||
results,
|
||||
(r) => r.referrer_name || 'direct',
|
||||
(r) => Number(r.cur ?? 0),
|
||||
);
|
||||
|
||||
const baselineMap = buildLookupMap(
|
||||
results,
|
||||
(r) => r.referrer_name || 'direct',
|
||||
(r) => Number(r.base ?? 0),
|
||||
);
|
||||
|
||||
const totalCurrent = totals[0]?.cur_total ?? 0;
|
||||
const totalBaseline = totals[0]?.base_total ?? 0;
|
||||
|
||||
return { currentMap, baselineMap, totalCurrent, totalBaseline };
|
||||
}
|
||||
|
||||
export const referrersModule: InsightModule = {
|
||||
key: 'referrers',
|
||||
cadence: ['daily'],
|
||||
thresholds: { minTotal: 100, minAbsDelta: 20, minPct: 0.15, maxDims: 50 },
|
||||
|
||||
async enumerateDimensions(ctx) {
|
||||
const { currentMap, baselineMap } = await fetchReferrerAggregates(ctx);
|
||||
const topDims = selectTopDimensions(
|
||||
currentMap,
|
||||
baselineMap,
|
||||
this.thresholds?.maxDims ?? 50,
|
||||
);
|
||||
return topDims.map((dim) => `referrer:${dim}`);
|
||||
},
|
||||
|
||||
async computeMany(ctx, dimensionKeys): Promise<ComputeResult[]> {
|
||||
const { currentMap, baselineMap, totalCurrent, totalBaseline } =
|
||||
await fetchReferrerAggregates(ctx);
|
||||
const results: ComputeResult[] = [];
|
||||
|
||||
for (const dimKey of dimensionKeys) {
|
||||
if (!dimKey.startsWith('referrer:')) continue;
|
||||
const referrerName = dimKey.replace('referrer:', '');
|
||||
|
||||
const currentValue = currentMap.get(referrerName) ?? 0;
|
||||
const compareValue = baselineMap.get(referrerName) ?? 0;
|
||||
|
||||
const currentShare = totalCurrent > 0 ? currentValue / totalCurrent : 0;
|
||||
const compareShare = totalBaseline > 0 ? compareValue / totalBaseline : 0;
|
||||
|
||||
const shareShiftPp = (currentShare - compareShare) * 100;
|
||||
const changePct = computeChangePct(currentValue, compareValue);
|
||||
const direction = computeDirection(changePct);
|
||||
|
||||
results.push({
|
||||
ok: true,
|
||||
dimensionKey: dimKey,
|
||||
currentValue,
|
||||
compareValue,
|
||||
changePct,
|
||||
direction,
|
||||
extra: {
|
||||
shareShiftPp,
|
||||
currentShare,
|
||||
compareShare,
|
||||
isNew: compareValue === 0 && currentValue > 0,
|
||||
isGone: currentValue === 0 && compareValue > 0,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
return results;
|
||||
},
|
||||
|
||||
render(result, ctx): RenderedCard {
|
||||
const referrer = result.dimensionKey.replace('referrer:', '');
|
||||
const pct = ((result.changePct ?? 0) * 100).toFixed(1);
|
||||
const isIncrease = (result.changePct ?? 0) >= 0;
|
||||
const isNew = result.extra?.isNew as boolean | undefined;
|
||||
|
||||
const title = isNew
|
||||
? `New traffic source: ${referrer}`
|
||||
: `Traffic from ${referrer} ${isIncrease ? '↑' : '↓'} ${Math.abs(Number(pct))}%`;
|
||||
|
||||
const sessionsCurrent = result.currentValue ?? 0;
|
||||
const sessionsCompare = result.compareValue ?? 0;
|
||||
const shareCurrent = Number(result.extra?.currentShare ?? 0);
|
||||
const shareCompare = Number(result.extra?.compareShare ?? 0);
|
||||
|
||||
return {
|
||||
title,
|
||||
summary: `${ctx.window.label}. Sessions ${sessionsCurrent} vs ${sessionsCompare}.`,
|
||||
displayName: referrer,
|
||||
payload: {
|
||||
kind: 'insight_v1',
|
||||
dimensions: [
|
||||
{
|
||||
key: 'referrer_name',
|
||||
value: referrer,
|
||||
displayName: referrer,
|
||||
},
|
||||
],
|
||||
primaryMetric: 'sessions',
|
||||
metrics: {
|
||||
sessions: {
|
||||
current: sessionsCurrent,
|
||||
compare: sessionsCompare,
|
||||
delta: sessionsCurrent - sessionsCompare,
|
||||
changePct: sessionsCompare > 0 ? (result.changePct ?? 0) : null,
|
||||
direction: result.direction ?? 'flat',
|
||||
unit: 'count',
|
||||
},
|
||||
share: {
|
||||
current: shareCurrent,
|
||||
compare: shareCompare,
|
||||
delta: shareCurrent - shareCompare,
|
||||
changePct:
|
||||
shareCompare > 0
|
||||
? (shareCurrent - shareCompare) / shareCompare
|
||||
: null,
|
||||
direction:
|
||||
shareCurrent - shareCompare > 0.0005
|
||||
? 'up'
|
||||
: shareCurrent - shareCompare < -0.0005
|
||||
? 'down'
|
||||
: 'flat',
|
||||
unit: 'ratio',
|
||||
},
|
||||
},
|
||||
extra: {
|
||||
isNew: result.extra?.isNew,
|
||||
isGone: result.extra?.isGone,
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
18
packages/db/src/services/insights/scoring.ts
Normal file
18
packages/db/src/services/insights/scoring.ts
Normal file
@@ -0,0 +1,18 @@
|
||||
import type { ComputeResult } from './types';
|
||||
|
||||
export function defaultImpactScore(r: ComputeResult): number {
|
||||
const vol = (r.currentValue ?? 0) + (r.compareValue ?? 0);
|
||||
const pct = Math.abs(r.changePct ?? 0);
|
||||
// stable-ish: bigger change + bigger volume => higher impact
|
||||
return Math.log1p(vol) * (pct * 100);
|
||||
}
|
||||
|
||||
export function severityBand(
|
||||
changePct?: number | null,
|
||||
): 'low' | 'moderate' | 'severe' | null {
|
||||
const p = Math.abs(changePct ?? 0);
|
||||
if (p < 0.1) return null;
|
||||
if (p < 0.5) return 'low';
|
||||
if (p < 1) return 'moderate';
|
||||
return 'severe';
|
||||
}
|
||||
343
packages/db/src/services/insights/store.ts
Normal file
343
packages/db/src/services/insights/store.ts
Normal file
@@ -0,0 +1,343 @@
|
||||
import { Prisma, db } from '../../prisma-client';
|
||||
import type {
|
||||
Cadence,
|
||||
InsightStore,
|
||||
PersistedInsight,
|
||||
RenderedCard,
|
||||
WindowKind,
|
||||
WindowRange,
|
||||
} from './types';
|
||||
|
||||
export const insightStore: InsightStore = {
|
||||
async listProjectIdsForCadence(cadence: Cadence): Promise<string[]> {
|
||||
const projects = await db.project.findMany({
|
||||
where: {
|
||||
deleteAt: null,
|
||||
eventsCount: { gt: 10_000 },
|
||||
updatedAt: { gt: new Date(Date.now() - 1000 * 60 * 60 * 24) },
|
||||
organization: {
|
||||
subscriptionStatus: 'active',
|
||||
},
|
||||
},
|
||||
select: { id: true },
|
||||
});
|
||||
return projects.map((p) => p.id);
|
||||
},
|
||||
|
||||
async getProjectCreatedAt(projectId: string): Promise<Date | null> {
|
||||
const project = await db.project.findFirst({
|
||||
where: { id: projectId, deleteAt: null },
|
||||
select: { createdAt: true },
|
||||
});
|
||||
return project?.createdAt ?? null;
|
||||
},
|
||||
|
||||
async getActiveInsightByIdentity({
|
||||
projectId,
|
||||
moduleKey,
|
||||
dimensionKey,
|
||||
windowKind,
|
||||
}): Promise<PersistedInsight | null> {
|
||||
const insight = await db.projectInsight.findFirst({
|
||||
where: {
|
||||
projectId,
|
||||
moduleKey,
|
||||
dimensionKey,
|
||||
windowKind,
|
||||
state: 'active',
|
||||
},
|
||||
});
|
||||
|
||||
if (!insight) return null;
|
||||
|
||||
return {
|
||||
id: insight.id,
|
||||
projectId: insight.projectId,
|
||||
moduleKey: insight.moduleKey,
|
||||
dimensionKey: insight.dimensionKey,
|
||||
windowKind: insight.windowKind as WindowKind,
|
||||
state: insight.state as 'active' | 'suppressed' | 'closed',
|
||||
version: insight.version,
|
||||
impactScore: insight.impactScore,
|
||||
lastSeenAt: insight.lastSeenAt,
|
||||
lastUpdatedAt: insight.lastUpdatedAt,
|
||||
direction: insight.direction,
|
||||
severityBand: insight.severityBand,
|
||||
};
|
||||
},
|
||||
|
||||
async upsertInsight({
|
||||
projectId,
|
||||
moduleKey,
|
||||
dimensionKey,
|
||||
window,
|
||||
card,
|
||||
metrics,
|
||||
now,
|
||||
decision,
|
||||
prev,
|
||||
}): Promise<PersistedInsight> {
|
||||
const baseData = {
|
||||
projectId,
|
||||
moduleKey,
|
||||
dimensionKey,
|
||||
windowKind: window.kind,
|
||||
state: prev?.state === 'closed' ? 'active' : (prev?.state ?? 'active'),
|
||||
title: card.title,
|
||||
summary: card.summary ?? null,
|
||||
displayName: card.displayName,
|
||||
payload: card.payload,
|
||||
direction: metrics.direction ?? null,
|
||||
impactScore: metrics.impactScore,
|
||||
severityBand: metrics.severityBand ?? null,
|
||||
version: prev ? (decision.material ? prev.version + 1 : prev.version) : 1,
|
||||
windowStart: window.start,
|
||||
windowEnd: window.end,
|
||||
lastSeenAt: now,
|
||||
lastUpdatedAt: now,
|
||||
};
|
||||
|
||||
// Try to find existing insight first
|
||||
const existing = prev
|
||||
? await db.projectInsight.findFirst({
|
||||
where: {
|
||||
projectId,
|
||||
moduleKey,
|
||||
dimensionKey,
|
||||
windowKind: window.kind,
|
||||
state: prev.state,
|
||||
},
|
||||
})
|
||||
: null;
|
||||
|
||||
let insight: any;
|
||||
if (existing) {
|
||||
// Update existing
|
||||
insight = await db.projectInsight.update({
|
||||
where: { id: existing.id },
|
||||
data: {
|
||||
...baseData,
|
||||
threadId: existing.threadId, // Preserve threadId
|
||||
},
|
||||
});
|
||||
} else {
|
||||
// Create new - need to check if there's a closed/suppressed one to reopen
|
||||
const closed = await db.projectInsight.findFirst({
|
||||
where: {
|
||||
projectId,
|
||||
moduleKey,
|
||||
dimensionKey,
|
||||
windowKind: window.kind,
|
||||
state: { in: ['closed', 'suppressed'] },
|
||||
},
|
||||
orderBy: { lastUpdatedAt: 'desc' },
|
||||
});
|
||||
|
||||
if (closed) {
|
||||
// Reopen and update
|
||||
insight = await db.projectInsight.update({
|
||||
where: { id: closed.id },
|
||||
data: {
|
||||
...baseData,
|
||||
state: 'active',
|
||||
threadId: closed.threadId, // Preserve threadId
|
||||
},
|
||||
});
|
||||
} else {
|
||||
// Create new
|
||||
insight = await db.projectInsight.create({
|
||||
data: {
|
||||
...baseData,
|
||||
firstDetectedAt: now,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
id: insight.id,
|
||||
projectId: insight.projectId,
|
||||
moduleKey: insight.moduleKey,
|
||||
dimensionKey: insight.dimensionKey,
|
||||
windowKind: insight.windowKind as WindowKind,
|
||||
state: insight.state as 'active' | 'suppressed' | 'closed',
|
||||
version: insight.version,
|
||||
impactScore: insight.impactScore,
|
||||
lastSeenAt: insight.lastSeenAt,
|
||||
lastUpdatedAt: insight.lastUpdatedAt,
|
||||
direction: insight.direction,
|
||||
severityBand: insight.severityBand,
|
||||
};
|
||||
},
|
||||
|
||||
async insertEvent({
|
||||
projectId,
|
||||
insightId,
|
||||
moduleKey,
|
||||
dimensionKey,
|
||||
windowKind,
|
||||
eventKind,
|
||||
changeFrom,
|
||||
changeTo,
|
||||
now,
|
||||
}): Promise<void> {
|
||||
await db.insightEvent.create({
|
||||
data: {
|
||||
insightId,
|
||||
eventKind,
|
||||
changeFrom: changeFrom
|
||||
? (changeFrom as Prisma.InputJsonValue)
|
||||
: Prisma.DbNull,
|
||||
changeTo: changeTo
|
||||
? (changeTo as Prisma.InputJsonValue)
|
||||
: Prisma.DbNull,
|
||||
createdAt: now,
|
||||
},
|
||||
});
|
||||
},
|
||||
|
||||
async closeMissingActiveInsights({
|
||||
projectId,
|
||||
moduleKey,
|
||||
windowKind,
|
||||
seenDimensionKeys,
|
||||
now,
|
||||
staleDays,
|
||||
}): Promise<number> {
|
||||
const staleDate = new Date(now);
|
||||
staleDate.setDate(staleDate.getDate() - staleDays);
|
||||
|
||||
const result = await db.projectInsight.updateMany({
|
||||
where: {
|
||||
projectId,
|
||||
moduleKey,
|
||||
windowKind,
|
||||
state: 'active',
|
||||
lastSeenAt: { lt: staleDate },
|
||||
dimensionKey: { notIn: seenDimensionKeys },
|
||||
},
|
||||
data: {
|
||||
state: 'closed',
|
||||
lastUpdatedAt: now,
|
||||
},
|
||||
});
|
||||
|
||||
return result.count;
|
||||
},
|
||||
|
||||
async applySuppression({
|
||||
projectId,
|
||||
moduleKey,
|
||||
windowKind,
|
||||
keepTopN,
|
||||
now,
|
||||
}): Promise<{ suppressed: number; unsuppressed: number }> {
|
||||
// Get all active insights for this module/window, ordered by impactScore desc
|
||||
const insights = await db.projectInsight.findMany({
|
||||
where: {
|
||||
projectId,
|
||||
moduleKey,
|
||||
windowKind,
|
||||
state: { in: ['active', 'suppressed'] },
|
||||
},
|
||||
orderBy: { impactScore: 'desc' },
|
||||
});
|
||||
|
||||
if (insights.length === 0) {
|
||||
return { suppressed: 0, unsuppressed: 0 };
|
||||
}
|
||||
|
||||
let suppressed = 0;
|
||||
let unsuppressed = 0;
|
||||
|
||||
// For "yesterday" insights, suppress any that are stale (windowEnd is not actually yesterday)
|
||||
// This prevents showing confusing insights like "Yesterday traffic dropped" when it's from 2+ days ago
|
||||
if (windowKind === 'yesterday') {
|
||||
const yesterday = new Date(now);
|
||||
yesterday.setUTCHours(0, 0, 0, 0);
|
||||
yesterday.setUTCDate(yesterday.getUTCDate() - 1);
|
||||
const yesterdayTime = yesterday.getTime();
|
||||
|
||||
for (const insight of insights) {
|
||||
// If windowEnd is null, consider it stale
|
||||
const isStale = insight.windowEnd
|
||||
? new Date(insight.windowEnd).setUTCHours(0, 0, 0, 0) !==
|
||||
yesterdayTime
|
||||
: true;
|
||||
|
||||
if (isStale && insight.state === 'active') {
|
||||
await db.projectInsight.update({
|
||||
where: { id: insight.id },
|
||||
data: { state: 'suppressed', lastUpdatedAt: now },
|
||||
});
|
||||
suppressed++;
|
||||
}
|
||||
}
|
||||
|
||||
// Filter to only non-stale insights for top-N logic
|
||||
const freshInsights = insights.filter((insight) => {
|
||||
if (!insight.windowEnd) return false;
|
||||
const windowEndTime = new Date(insight.windowEnd).setUTCHours(
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
);
|
||||
return windowEndTime === yesterdayTime;
|
||||
});
|
||||
|
||||
const topN = freshInsights.slice(0, keepTopN);
|
||||
const belowN = freshInsights.slice(keepTopN);
|
||||
|
||||
for (const insight of belowN) {
|
||||
if (insight.state === 'active') {
|
||||
await db.projectInsight.update({
|
||||
where: { id: insight.id },
|
||||
data: { state: 'suppressed', lastUpdatedAt: now },
|
||||
});
|
||||
suppressed++;
|
||||
}
|
||||
}
|
||||
|
||||
for (const insight of topN) {
|
||||
if (insight.state === 'suppressed') {
|
||||
await db.projectInsight.update({
|
||||
where: { id: insight.id },
|
||||
data: { state: 'active', lastUpdatedAt: now },
|
||||
});
|
||||
unsuppressed++;
|
||||
}
|
||||
}
|
||||
|
||||
return { suppressed, unsuppressed };
|
||||
}
|
||||
|
||||
// For non-yesterday windows, apply standard top-N suppression
|
||||
const topN = insights.slice(0, keepTopN);
|
||||
const belowN = insights.slice(keepTopN);
|
||||
|
||||
// Suppress those below top N
|
||||
for (const insight of belowN) {
|
||||
if (insight.state === 'active') {
|
||||
await db.projectInsight.update({
|
||||
where: { id: insight.id },
|
||||
data: { state: 'suppressed', lastUpdatedAt: now },
|
||||
});
|
||||
suppressed++;
|
||||
}
|
||||
}
|
||||
|
||||
// Unsuppress those in top N
|
||||
for (const insight of topN) {
|
||||
if (insight.state === 'suppressed') {
|
||||
await db.projectInsight.update({
|
||||
where: { id: insight.id },
|
||||
data: { state: 'active', lastUpdatedAt: now },
|
||||
});
|
||||
unsuppressed++;
|
||||
}
|
||||
}
|
||||
|
||||
return { suppressed, unsuppressed };
|
||||
},
|
||||
};
|
||||
191
packages/db/src/services/insights/types.ts
Normal file
191
packages/db/src/services/insights/types.ts
Normal file
@@ -0,0 +1,191 @@
|
||||
import type {
|
||||
InsightDimension,
|
||||
InsightMetricEntry,
|
||||
InsightMetricKey,
|
||||
InsightPayload,
|
||||
} from '@openpanel/validation';
|
||||
|
||||
export type Cadence = 'daily';
|
||||
|
||||
export type WindowKind = 'yesterday' | 'rolling_7d' | 'rolling_30d';
|
||||
|
||||
export interface WindowRange {
|
||||
kind: WindowKind;
|
||||
start: Date; // inclusive
|
||||
end: Date; // inclusive (or exclusive, but be consistent)
|
||||
baselineStart: Date;
|
||||
baselineEnd: Date;
|
||||
label: string; // e.g. "Yesterday" / "Last 7 days"
|
||||
}
|
||||
|
||||
export interface ComputeContext {
|
||||
projectId: string;
|
||||
window: WindowRange;
|
||||
db: any; // your DB client
|
||||
now: Date;
|
||||
logger: Pick<Console, 'info' | 'warn' | 'error'>;
|
||||
/**
|
||||
* Cached clix function that automatically caches query results based on query hash.
|
||||
* This eliminates duplicate queries within the same module+window context.
|
||||
* Use this instead of importing clix directly to benefit from automatic caching.
|
||||
*/
|
||||
clix: ReturnType<typeof import('./cached-clix').createCachedClix>;
|
||||
}
|
||||
|
||||
export interface ComputeResult {
|
||||
ok: boolean;
|
||||
dimensionKey: string; // e.g. "referrer:instagram" / "page:/pricing"
|
||||
currentValue?: number;
|
||||
compareValue?: number;
|
||||
changePct?: number; // -0.15 = -15%
|
||||
direction?: 'up' | 'down' | 'flat';
|
||||
extra?: Record<string, unknown>; // share delta pp, rank, sparkline, etc.
|
||||
}
|
||||
|
||||
// Types imported from @openpanel/validation:
|
||||
// - InsightMetricKey
|
||||
// - InsightMetricEntry
|
||||
// - InsightDimension
|
||||
// - InsightPayload
|
||||
|
||||
/**
|
||||
* Render should be deterministic and safe to call multiple times.
|
||||
* Returns the shape that matches ProjectInsight create input.
|
||||
* The payload contains all metric data and display metadata.
|
||||
*/
|
||||
export interface RenderedCard {
|
||||
title: string;
|
||||
summary?: string;
|
||||
displayName: string;
|
||||
payload: InsightPayload; // Contains dimensions, primaryMetric, metrics, extra
|
||||
}
|
||||
|
||||
/** Optional per-module thresholds (the engine can still apply global defaults) */
|
||||
export interface ModuleThresholds {
|
||||
minTotal?: number; // min current+baseline
|
||||
minAbsDelta?: number; // min abs(current-compare)
|
||||
minPct?: number; // min abs(changePct)
|
||||
maxDims?: number; // cap enumerateDimensions
|
||||
}
|
||||
|
||||
export interface InsightModule {
|
||||
key: string;
|
||||
cadence: Cadence[];
|
||||
/** Optional per-module override; engine applies a default if omitted. */
|
||||
windows?: WindowKind[];
|
||||
thresholds?: ModuleThresholds;
|
||||
enumerateDimensions?(ctx: ComputeContext): Promise<string[]>;
|
||||
/** Preferred path: batch compute many dimensions in one go. */
|
||||
computeMany(
|
||||
ctx: ComputeContext,
|
||||
dimensionKeys: string[],
|
||||
): Promise<ComputeResult[]>;
|
||||
/** Must not do DB reads; just format output. */
|
||||
render(result: ComputeResult, ctx: ComputeContext): RenderedCard;
|
||||
/** Score decides what to show (top-N). */
|
||||
score?(result: ComputeResult, ctx: ComputeContext): number;
|
||||
/** Optional: compute "drivers" for AI explain step */
|
||||
drivers?(
|
||||
result: ComputeResult,
|
||||
ctx: ComputeContext,
|
||||
): Promise<Record<string, unknown>>;
|
||||
}
|
||||
|
||||
/** Insight row shape returned from persistence (minimal fields engine needs). */
|
||||
export interface PersistedInsight {
|
||||
id: string;
|
||||
projectId: string;
|
||||
moduleKey: string;
|
||||
dimensionKey: string;
|
||||
windowKind: WindowKind;
|
||||
state: 'active' | 'suppressed' | 'closed';
|
||||
version: number;
|
||||
impactScore: number;
|
||||
lastSeenAt: Date;
|
||||
lastUpdatedAt: Date;
|
||||
direction?: string | null;
|
||||
severityBand?: string | null;
|
||||
}
|
||||
|
||||
/** Material change decision used for events/notifications. */
|
||||
export type MaterialReason =
|
||||
| 'created'
|
||||
| 'direction_flip'
|
||||
| 'severity_change'
|
||||
| 'cross_deadband'
|
||||
| 'reopened'
|
||||
| 'none';
|
||||
|
||||
export interface MaterialDecision {
|
||||
material: boolean;
|
||||
reason: MaterialReason;
|
||||
newSeverityBand?: 'low' | 'moderate' | 'severe' | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Persistence interface: implement with Postgres.
|
||||
* Keep engine independent of query builder choice.
|
||||
*/
|
||||
export interface InsightStore {
|
||||
listProjectIdsForCadence(cadence: Cadence): Promise<string[]>;
|
||||
/** Used by the engine/worker to decide if a window has enough baseline history. */
|
||||
getProjectCreatedAt(projectId: string): Promise<Date | null>;
|
||||
getActiveInsightByIdentity(args: {
|
||||
projectId: string;
|
||||
moduleKey: string;
|
||||
dimensionKey: string;
|
||||
windowKind: WindowKind;
|
||||
}): Promise<PersistedInsight | null>;
|
||||
upsertInsight(args: {
|
||||
projectId: string;
|
||||
moduleKey: string;
|
||||
dimensionKey: string;
|
||||
window: WindowRange;
|
||||
card: RenderedCard;
|
||||
metrics: {
|
||||
direction?: 'up' | 'down' | 'flat';
|
||||
impactScore: number;
|
||||
severityBand?: string | null;
|
||||
};
|
||||
now: Date;
|
||||
decision: MaterialDecision;
|
||||
prev: PersistedInsight | null;
|
||||
}): Promise<PersistedInsight>;
|
||||
insertEvent(args: {
|
||||
projectId: string;
|
||||
insightId: string;
|
||||
moduleKey: string;
|
||||
dimensionKey: string;
|
||||
windowKind: WindowKind;
|
||||
eventKind:
|
||||
| 'created'
|
||||
| 'updated'
|
||||
| 'severity_up'
|
||||
| 'severity_down'
|
||||
| 'direction_flip'
|
||||
| 'closed'
|
||||
| 'reopened'
|
||||
| 'suppressed'
|
||||
| 'unsuppressed';
|
||||
changeFrom?: Record<string, unknown> | null;
|
||||
changeTo?: Record<string, unknown> | null;
|
||||
now: Date;
|
||||
}): Promise<void>;
|
||||
/** Mark insights as not seen this run if you prefer lifecycle via closeMissing() */
|
||||
closeMissingActiveInsights(args: {
|
||||
projectId: string;
|
||||
moduleKey: string;
|
||||
windowKind: WindowKind;
|
||||
seenDimensionKeys: string[];
|
||||
now: Date;
|
||||
staleDays: number; // close if not seen for X days
|
||||
}): Promise<number>; // count closed
|
||||
/** Enforce top-N display by suppressing below-threshold insights. */
|
||||
applySuppression(args: {
|
||||
projectId: string;
|
||||
moduleKey: string;
|
||||
windowKind: WindowKind;
|
||||
keepTopN: number;
|
||||
now: Date;
|
||||
}): Promise<{ suppressed: number; unsuppressed: number }>;
|
||||
}
|
||||
151
packages/db/src/services/insights/utils.ts
Normal file
151
packages/db/src/services/insights/utils.ts
Normal file
@@ -0,0 +1,151 @@
|
||||
/**
|
||||
* Shared utilities for insight modules
|
||||
*/
|
||||
|
||||
/**
|
||||
* Get UTC weekday (0 = Sunday, 6 = Saturday)
|
||||
*/
|
||||
export function getWeekday(date: Date): number {
|
||||
return date.getUTCDay();
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute median of a sorted array of numbers
|
||||
*/
|
||||
export function computeMedian(sortedValues: number[]): number {
|
||||
if (sortedValues.length === 0) return 0;
|
||||
const mid = Math.floor(sortedValues.length / 2);
|
||||
return sortedValues.length % 2 === 0
|
||||
? ((sortedValues[mid - 1] ?? 0) + (sortedValues[mid] ?? 0)) / 2
|
||||
: (sortedValues[mid] ?? 0);
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute weekday medians from daily breakdown data.
|
||||
* Groups by dimension, filters to matching weekday, computes median per dimension.
|
||||
*
|
||||
* @param data - Array of { date, dimension, cnt } rows
|
||||
* @param targetWeekday - Weekday to filter to (0-6)
|
||||
* @param getDimension - Function to extract normalized dimension from row
|
||||
* @returns Map of dimension -> median value
|
||||
*/
|
||||
export function computeWeekdayMedians<T>(
|
||||
data: T[],
|
||||
targetWeekday: number,
|
||||
getDimension: (row: T) => string,
|
||||
): Map<string, number> {
|
||||
// Group by dimension, filtered to target weekday
|
||||
const byDimension = new Map<string, number[]>();
|
||||
|
||||
for (const row of data) {
|
||||
const rowWeekday = getWeekday(new Date((row as any).date));
|
||||
if (rowWeekday !== targetWeekday) continue;
|
||||
|
||||
const dim = getDimension(row);
|
||||
const values = byDimension.get(dim) ?? [];
|
||||
values.push(Number((row as any).cnt ?? 0));
|
||||
byDimension.set(dim, values);
|
||||
}
|
||||
|
||||
// Compute median per dimension
|
||||
const result = new Map<string, number>();
|
||||
for (const [dim, values] of byDimension) {
|
||||
values.sort((a, b) => a - b);
|
||||
result.set(dim, computeMedian(values));
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute change percentage between current and compare values
|
||||
*/
|
||||
export function computeChangePct(
|
||||
currentValue: number,
|
||||
compareValue: number,
|
||||
): number {
|
||||
return compareValue > 0
|
||||
? (currentValue - compareValue) / compareValue
|
||||
: currentValue > 0
|
||||
? 1
|
||||
: 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Determine direction based on change percentage
|
||||
*/
|
||||
export function computeDirection(
|
||||
changePct: number,
|
||||
threshold = 0.05,
|
||||
): 'up' | 'down' | 'flat' {
|
||||
return changePct > threshold
|
||||
? 'up'
|
||||
: changePct < -threshold
|
||||
? 'down'
|
||||
: 'flat';
|
||||
}
|
||||
|
||||
/**
|
||||
* Get end of day timestamp (23:59:59.999) for a given date.
|
||||
* Used to ensure BETWEEN queries include the full day.
|
||||
*/
|
||||
export function getEndOfDay(date: Date): Date {
|
||||
const end = new Date(date);
|
||||
end.setUTCHours(23, 59, 59, 999);
|
||||
return end;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a lookup map from query results.
|
||||
* Aggregates counts by key, handling duplicate keys by summing values.
|
||||
*
|
||||
* @param results - Array of result rows
|
||||
* @param getKey - Function to extract the key from each row
|
||||
* @param getCount - Function to extract the count from each row (defaults to 'cnt' field)
|
||||
* @returns Map of key -> aggregated count
|
||||
*/
|
||||
export function buildLookupMap<T>(
|
||||
results: T[],
|
||||
getKey: (row: T) => string,
|
||||
getCount: (row: T) => number = (row) => Number((row as any).cnt ?? 0),
|
||||
): Map<string, number> {
|
||||
const map = new Map<string, number>();
|
||||
for (const row of results) {
|
||||
const key = getKey(row);
|
||||
const cnt = getCount(row);
|
||||
map.set(key, (map.get(key) ?? 0) + cnt);
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
/**
|
||||
* Select top-N dimensions by ranking on greatest(current, baseline).
|
||||
* This preserves union behavior: dimensions with high values in either period are included.
|
||||
*
|
||||
* @param currentMap - Map of dimension -> current value
|
||||
* @param baselineMap - Map of dimension -> baseline value
|
||||
* @param maxDims - Maximum number of dimensions to return
|
||||
* @returns Array of dimension keys, ranked by greatest(current, baseline)
|
||||
*/
|
||||
export function selectTopDimensions(
|
||||
currentMap: Map<string, number>,
|
||||
baselineMap: Map<string, number>,
|
||||
maxDims: number,
|
||||
): string[] {
|
||||
// Merge all dimensions from both maps
|
||||
const allDims = new Set<string>();
|
||||
for (const dim of currentMap.keys()) allDims.add(dim);
|
||||
for (const dim of baselineMap.keys()) allDims.add(dim);
|
||||
|
||||
// Rank by greatest(current, baseline)
|
||||
const ranked = Array.from(allDims)
|
||||
.map((dim) => ({
|
||||
dim,
|
||||
maxValue: Math.max(currentMap.get(dim) ?? 0, baselineMap.get(dim) ?? 0),
|
||||
}))
|
||||
.sort((a, b) => b.maxValue - a.maxValue)
|
||||
.slice(0, maxDims)
|
||||
.map((x) => x.dim);
|
||||
|
||||
return ranked;
|
||||
}
|
||||
59
packages/db/src/services/insights/windows.ts
Normal file
59
packages/db/src/services/insights/windows.ts
Normal file
@@ -0,0 +1,59 @@
|
||||
import type { WindowKind, WindowRange } from './types';
|
||||
|
||||
function atUtcMidnight(d: Date) {
|
||||
const x = new Date(d);
|
||||
x.setUTCHours(0, 0, 0, 0);
|
||||
return x;
|
||||
}
|
||||
|
||||
function addDays(d: Date, days: number) {
|
||||
const x = new Date(d);
|
||||
x.setUTCDate(x.getUTCDate() + days);
|
||||
return x;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convention: end is inclusive (end of day). If you prefer exclusive, adapt consistently.
|
||||
*/
|
||||
export function resolveWindow(kind: WindowKind, now: Date): WindowRange {
|
||||
const today0 = atUtcMidnight(now);
|
||||
const yesterday0 = addDays(today0, -1);
|
||||
if (kind === 'yesterday') {
|
||||
const start = yesterday0;
|
||||
const end = yesterday0;
|
||||
// Baseline: median of last 4 same weekdays -> engine/module implements the median.
|
||||
// Here we just define the candidate range; module queries last 28 days and filters weekday.
|
||||
const baselineStart = addDays(yesterday0, -28);
|
||||
const baselineEnd = addDays(yesterday0, -1);
|
||||
return { kind, start, end, baselineStart, baselineEnd, label: 'Yesterday' };
|
||||
}
|
||||
if (kind === 'rolling_7d') {
|
||||
const end = yesterday0;
|
||||
const start = addDays(end, -6); // 7 days inclusive
|
||||
const baselineEnd = addDays(start, -1);
|
||||
const baselineStart = addDays(baselineEnd, -6);
|
||||
return {
|
||||
kind,
|
||||
start,
|
||||
end,
|
||||
baselineStart,
|
||||
baselineEnd,
|
||||
label: 'Last 7 days',
|
||||
};
|
||||
}
|
||||
// rolling_30d
|
||||
{
|
||||
const end = yesterday0;
|
||||
const start = addDays(end, -29);
|
||||
const baselineEnd = addDays(start, -1);
|
||||
const baselineStart = addDays(baselineEnd, -29);
|
||||
return {
|
||||
kind,
|
||||
start,
|
||||
end,
|
||||
baselineStart,
|
||||
baselineEnd,
|
||||
label: 'Last 30 days',
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -180,11 +180,11 @@ export function sessionConsistency() {
|
||||
|
||||
// For write operations with session: cache WAL LSN after write
|
||||
if (isWriteOperation(operation)) {
|
||||
logger.info('Prisma operation', {
|
||||
operation,
|
||||
args,
|
||||
model,
|
||||
});
|
||||
// logger.info('Prisma operation', {
|
||||
// operation,
|
||||
// args,
|
||||
// model,
|
||||
// });
|
||||
|
||||
const result = await query(args);
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ import type {
|
||||
IIntegrationConfig,
|
||||
INotificationRuleConfig,
|
||||
IProjectFilters,
|
||||
InsightPayload,
|
||||
} from '@openpanel/validation';
|
||||
import type {
|
||||
IClickhouseBotEvent,
|
||||
@@ -18,6 +19,7 @@ declare global {
|
||||
type IPrismaIntegrationConfig = IIntegrationConfig;
|
||||
type IPrismaNotificationPayload = INotificationPayload;
|
||||
type IPrismaProjectFilters = IProjectFilters[];
|
||||
type IPrismaProjectInsightPayload = InsightPayload;
|
||||
type IPrismaClickhouseEvent = IClickhouseEvent;
|
||||
type IPrismaClickhouseProfile = IClickhouseProfile;
|
||||
type IPrismaClickhouseBotEvent = IClickhouseBotEvent;
|
||||
|
||||
Reference in New Issue
Block a user