wip

packages/db/src/engine/compute.ts

@@ -0,0 +1,165 @@
+import { round } from '@openpanel/common';
+import { alphabetIds } from '@openpanel/constants';
+import type { IChartFormula } from '@openpanel/validation';
+import * as mathjs from 'mathjs';
+import type { ConcreteSeries } from './types';
+
+/**
+ * Compute formula series from fetched event series
+ * Formulas reference event series using alphabet IDs (A, B, C, etc.)
+ */
+export function compute(
+  fetchedSeries: ConcreteSeries[],
+  definitions: Array<{
+    type: 'event' | 'formula';
+    id?: string;
+    formula?: string;
+  }>,
+): ConcreteSeries[] {
+  const results: ConcreteSeries[] = [...fetchedSeries];
+
+  // Process formulas in order (they can reference previous formulas)
+  definitions.forEach((definition, formulaIndex) => {
+    if (definition.type !== 'formula') {
+      return;
+    }
+
+    const formula = definition as IChartFormula;
+    if (!formula.formula) {
+      return;
+    }
+
+    // Group ALL series (events + previously computed formulas) by breakdown signature
+    // Series with the same breakdown values should be computed together
+    const seriesByBreakdown = new Map<string, ConcreteSeries[]>();
+
+    // Include both fetched event series AND previously computed formulas
+    const allSeries = [
+      ...fetchedSeries,
+      ...results.filter((s) => s.definitionIndex < formulaIndex),
+    ];
+
+    allSeries.forEach((serie) => {
+      // Create breakdown signature: skip first name part (event/formula name) and use breakdown values
+      // If name.length === 1, it means no breakdowns (just event name)
+      // If name.length > 1, name[0] is event name, name[1+] are breakdown values
+      const breakdownSignature =
+        serie.name.length > 1 ? serie.name.slice(1).join(':::') : '';
+
+      if (!seriesByBreakdown.has(breakdownSignature)) {
+        seriesByBreakdown.set(breakdownSignature, []);
+      }
+      seriesByBreakdown.get(breakdownSignature)!.push(serie);
+    });
+
+    // Compute formula for each breakdown group
+    for (const [breakdownSignature, breakdownSeries] of seriesByBreakdown) {
+      // Map series by their definition index for formula evaluation
+      const seriesByIndex = new Map<number, ConcreteSeries>();
+      breakdownSeries.forEach((serie) => {
+        seriesByIndex.set(serie.definitionIndex, serie);
+      });
+
+      // Get all unique dates across all series in this breakdown group
+      const allDates = new Set<string>();
+      breakdownSeries.forEach((serie) => {
+        serie.data.forEach((item) => {
+          allDates.add(item.date);
+        });
+      });
+
+      const sortedDates = Array.from(allDates).sort(
+        (a, b) => new Date(a).getTime() - new Date(b).getTime(),
+      );
+
+      // Calculate formula for each date
+      const formulaData = sortedDates.map((date) => {
+        const scope: Record<string, number> = {};
+
+        // Build scope using alphabet IDs (A, B, C, etc.)
+        definitions.slice(0, formulaIndex).forEach((depDef, depIndex) => {
+          const readableId = alphabetIds[depIndex];
+          if (!readableId) {
+            return;
+          }
+
+          // Find the series for this dependency in the current breakdown group
+          const depSeries = seriesByIndex.get(depIndex);
+          if (depSeries) {
+            const dataPoint = depSeries.data.find((d) => d.date === date);
+            scope[readableId] = dataPoint?.count ?? 0;
+          } else {
+            // Could be a formula from a previous breakdown group - find it in results
+            // Match by definitionIndex AND breakdown signature
+            const formulaSerie = results.find(
+              (s) =>
+                s.definitionIndex === depIndex &&
+                'type' in s.definition &&
+                s.definition.type === 'formula' &&
+                s.name.slice(1).join(':::') === breakdownSignature,
+            );
+            if (formulaSerie) {
+              const dataPoint = formulaSerie.data.find((d) => d.date === date);
+              scope[readableId] = dataPoint?.count ?? 0;
+            } else {
+              scope[readableId] = 0;
+            }
+          }
+        });
+
+        // Evaluate formula
+        let count: number;
+        try {
+          count = mathjs
+            .parse(formula.formula)
+            .compile()
+            .evaluate(scope) as number;
+        } catch (error) {
+          count = 0;
+        }
+
+        return {
+          date,
+          count:
+            Number.isNaN(count) || !Number.isFinite(count)
+              ? 0
+              : round(count, 2),
+          total_count: breakdownSeries[0]?.data.find((d) => d.date === date)
+            ?.total_count,
+        };
+      });
+
+      // Create concrete series for this formula
+      const templateSerie = breakdownSeries[0]!;
+
+      // Extract breakdown values from template series name
+      // name[0] is event/formula name, name[1+] are breakdown values
+      const breakdownValues =
+        templateSerie.name.length > 1 ? templateSerie.name.slice(1) : [];
+
+      const formulaName =
+        breakdownValues.length > 0
+          ? [formula.displayName || formula.formula, ...breakdownValues]
+          : [formula.displayName || formula.formula];
+
+      const formulaSeries: ConcreteSeries = {
+        id: `formula-${formula.id ?? formulaIndex}-${breakdownSignature || 'default'}`,
+        definitionId:
+          formula.id ?? alphabetIds[formulaIndex] ?? `formula-${formulaIndex}`,
+        definitionIndex: formulaIndex,
+        name: formulaName,
+        context: {
+          filters: templateSerie.context.filters,
+          breakdownValue: templateSerie.context.breakdownValue,
+          breakdowns: templateSerie.context.breakdowns,
+        },
+        data: formulaData,
+        definition: formula,
+      };
+
+      results.push(formulaSeries);
+    }
+  });
+
+  return results;
+}

packages/db/src/engine/fetch.ts

@@ -0,0 +1,151 @@
+import type { ISerieDataItem } from '@openpanel/common';
+import { groupByLabels } from '@openpanel/common';
+import { alphabetIds } from '@openpanel/constants';
+import type { IGetChartDataInput } from '@openpanel/validation';
+import { chQuery } from '../clickhouse/client';
+import { getChartSql } from '../services/chart.service';
+import type { ConcreteSeries, Plan } from './types';
+
+/**
+ * Fetch data for all event series in the plan
+ * This handles breakdown expansion automatically via groupByLabels
+ */
+export async function fetch(plan: Plan): Promise<ConcreteSeries[]> {
+  const results: ConcreteSeries[] = [];
+
+  // Process each event definition
+  for (let i = 0; i < plan.definitions.length; i++) {
+    const definition = plan.definitions[i]!;
+
+    if (definition.type !== 'event') {
+      // Skip formulas - they'll be handled in compute stage
+      continue;
+    }
+
+    const event = definition as typeof definition & { type: 'event' };
+
+    // Find the corresponding concrete series placeholder
+    const placeholder = plan.concreteSeries.find(
+      (cs) => cs.definitionId === definition.id,
+    );
+
+    if (!placeholder) {
+      continue;
+    }
+
+    // Build query input
+    const queryInput: IGetChartDataInput = {
+      event: {
+        id: event.id,
+        name: event.name,
+        segment: event.segment,
+        filters: event.filters,
+        displayName: event.displayName,
+        property: event.property,
+      },
+      projectId: plan.input.projectId,
+      startDate: plan.input.startDate,
+      endDate: plan.input.endDate,
+      breakdowns: plan.input.breakdowns,
+      interval: plan.input.interval,
+      chartType: plan.input.chartType,
+      metric: plan.input.metric,
+      previous: plan.input.previous ?? false,
+      limit: plan.input.limit,
+      offset: plan.input.offset,
+      criteria: plan.input.criteria,
+      funnelGroup: plan.input.funnelGroup,
+      funnelWindow: plan.input.funnelWindow,
+    };
+
+    // Execute query
+    let queryResult = await chQuery<ISerieDataItem>(
+      getChartSql({ ...queryInput, timezone: plan.timezone }),
+      {
+        session_timezone: plan.timezone,
+      },
+    );
+
+    // Fallback: if no results with breakdowns, try without breakdowns
+    if (queryResult.length === 0 && plan.input.breakdowns.length > 0) {
+      queryResult = await chQuery<ISerieDataItem>(
+        getChartSql({
+          ...queryInput,
+          breakdowns: [],
+          timezone: plan.timezone,
+        }),
+        {
+          session_timezone: plan.timezone,
+        },
+      );
+    }
+
+    // Group by labels (handles breakdown expansion)
+    const groupedSeries = groupByLabels(queryResult);
+
+    // Create concrete series for each grouped result
+    groupedSeries.forEach((grouped) => {
+      // Extract breakdown value from name array
+      // If breakdowns exist, name[0] is event name, name[1+] are breakdown values
+      const breakdownValue =
+        plan.input.breakdowns.length > 0 && grouped.name.length > 1
+          ? grouped.name.slice(1).join(' - ')
+          : undefined;
+
+      // Build breakdowns object: { country: 'SE', path: '/ewoqmepwq' }
+      const breakdowns: Record<string, string> | undefined =
+        plan.input.breakdowns.length > 0 && grouped.name.length > 1
+          ? {}
+          : undefined;
+
+      if (breakdowns) {
+        plan.input.breakdowns.forEach((breakdown, idx) => {
+          const breakdownNamePart = grouped.name[idx + 1];
+          if (breakdownNamePart) {
+            breakdowns[breakdown.name] = breakdownNamePart;
+          }
+        });
+      }
+
+      // Build filters including breakdown value
+      const filters = [...event.filters];
+      if (breakdownValue && plan.input.breakdowns.length > 0) {
+        // Add breakdown filter
+        plan.input.breakdowns.forEach((breakdown, idx) => {
+          const breakdownNamePart = grouped.name[idx + 1];
+          if (breakdownNamePart) {
+            filters.push({
+              id: `breakdown-${idx}`,
+              name: breakdown.name,
+              operator: 'is',
+              value: [breakdownNamePart],
+            });
+          }
+        });
+      }
+
+      const concrete: ConcreteSeries = {
+        id: `${placeholder.id}-${grouped.name.join('-')}`,
+        definitionId: definition.id ?? alphabetIds[i] ?? `series-${i}`,
+        definitionIndex: i,
+        name: grouped.name,
+        context: {
+          event: event.name,
+          filters,
+          breakdownValue,
+          breakdowns,
+        },
+        data: grouped.data.map((item) => ({
+          date: item.date,
+          count: item.count,
+          total_count: item.total_count,
+        })),
+        definition,
+      };
+
+      results.push(concrete);
+    });
+  }
+
+  return results;
+}

packages/db/src/engine/format.ts

@@ -0,0 +1,141 @@
+import {
+  average,
+  getPreviousMetric,
+  max,
+  min,
+  round,
+  slug,
+  sum,
+} from '@openpanel/common';
+import { alphabetIds } from '@openpanel/constants';
+import type { FinalChart } from '@openpanel/validation';
+import type { ConcreteSeries } from './types';
+
+/**
+ * Format concrete series into FinalChart format (backward compatible)
+ * TODO: Migrate frontend to use cleaner ChartResponse format
+ */
+export function format(
+  concreteSeries: ConcreteSeries[],
+  definitions: Array<{
+    id?: string;
+    type: 'event' | 'formula';
+    displayName?: string;
+    formula?: string;
+    name?: string;
+  }>,
+  includeAlphaIds: boolean,
+  previousSeries: ConcreteSeries[] | null = null,
+): FinalChart {
+  const series = concreteSeries.map((cs) => {
+    // Find definition for this series
+    const definition = definitions[cs.definitionIndex];
+    const alphaId = includeAlphaIds
+      ? alphabetIds[cs.definitionIndex]
+      : undefined;
+
+    // Build display name with optional alpha ID
+    let displayName: string[];
+
+    // Replace the first name (which is the event name) with the display name if it exists
+    const names = cs.name.slice(0);
+    if (cs.definition.displayName) {
+      names.splice(0, 1, cs.definition.displayName);
+    }
+    // Add the alpha ID to the first name if it exists
+    if (alphaId) {
+      displayName = [`(${alphaId}) ${names[0]}`, ...names.slice(1)];
+    } else {
+      displayName = names;
+    }
+
+    // Calculate metrics for this series
+    const counts = cs.data.map((d) => d.count);
+    const metrics = {
+      sum: sum(counts),
+      average: round(average(counts), 2),
+      min: min(counts),
+      max: max(counts),
+      count: cs.data.find((item) => !!item.total_count)?.total_count,
+    };
+
+    // Build event object for compatibility
+    const eventName =
+      definition?.type === 'formula'
+        ? definition.displayName || definition.formula || 'Formula'
+        : definition?.name || cs.context.event || 'unknown';
+
+    // Find matching previous series
+    const previousSerie = previousSeries?.find(
+      (ps) =>
+        ps.definitionIndex === cs.definitionIndex &&
+        ps.name.slice(1).join(':::') === cs.name.slice(1).join(':::'),
+    );
+
+    return {
+      id: cs.id,
+      names: displayName,
+      // TODO: Do we need this now?
+      event: {
+        id: definition?.id,
+        name: eventName,
+        breakdowns: cs.context.breakdowns,
+      },
+      metrics: {
+        ...metrics,
+        ...(previousSerie
+          ? {
+              previous: {
+                sum: getPreviousMetric(
+                  metrics.sum,
+                  sum(previousSerie.data.map((d) => d.count)),
+                ),
+                average: getPreviousMetric(
+                  metrics.average,
+                  round(average(previousSerie.data.map((d) => d.count)), 2),
+                ),
+                min: getPreviousMetric(
+                  metrics.min,
+                  min(previousSerie.data.map((d) => d.count)),
+                ),
+                max: getPreviousMetric(
+                  metrics.max,
+                  max(previousSerie.data.map((d) => d.count)),
+                ),
+                count: getPreviousMetric(
+                  metrics.count ?? 0,
+                  previousSerie.data.find((item) => !!item.total_count)
+                    ?.total_count ?? null,
+                ),
+              },
+            }
+          : {}),
+      },
+      data: cs.data.map((item, index) => ({
+        date: item.date,
+        count: item.count,
+        previous: previousSerie?.data[index]
+          ? getPreviousMetric(
+              item.count,
+              previousSerie.data[index]?.count ?? null,
+            )
+          : undefined,
+      })),
+    };
+  });
+
+  // Calculate global metrics
+  const allValues = concreteSeries.flatMap((cs) => cs.data.map((d) => d.count));
+  const globalMetrics = {
+    sum: sum(allValues),
+    average: round(average(allValues), 2),
+    min: min(allValues),
+    max: max(allValues),
+    count: undefined as number | undefined,
+  };
+
+  return {
+    series,
+    metrics: globalMetrics,
+  };
+}

packages/db/src/engine/index.ts

@@ -0,0 +1,77 @@
+import { getPreviousMetric } from '@openpanel/common';
+
+import type { FinalChart, IChartInput } from '@openpanel/validation';
+import { getChartPrevStartEndDate } from '../services/chart.service';
+import {
+  getOrganizationSubscriptionChartEndDate,
+  getSettingsForProject,
+} from '../services/organization.service';
+import { compute } from './compute';
+import { fetch } from './fetch';
+import { format } from './format';
+import { normalize } from './normalize';
+import { plan } from './plan';
+import type { ConcreteSeries } from './types';
+
+/**
+ * Chart Engine - Main entry point
+ * Executes the pipeline: normalize -> plan -> fetch -> compute -> format
+ */
+export async function executeChart(input: IChartInput): Promise<FinalChart> {
+  const { timezone } = await getSettingsForProject(input.projectId);
+
+  // Stage 1: Normalize input
+  const normalized = await normalize(input);
+
+  // Handle subscription end date limit
+  const endDate = await getOrganizationSubscriptionChartEndDate(
+    input.projectId,
+    normalized.endDate,
+  );
+  if (endDate) {
+    normalized.endDate = endDate;
+  }
+
+  // Stage 2: Create execution plan
+  const executionPlan = await plan(normalized);
+
+  // Stage 3: Fetch data for event series (current period)
+  const fetchedSeries = await fetch(executionPlan);
+
+  // Stage 4: Compute formula series
+  const computedSeries = compute(fetchedSeries, executionPlan.definitions);
+
+  // Stage 5: Fetch previous period if requested
+  let previousSeries: ConcreteSeries[] | null = null;
+  if (input.previous) {
+    const currentPeriod = {
+      startDate: normalized.startDate,
+      endDate: normalized.endDate,
+    };
+    const previousPeriod = getChartPrevStartEndDate(currentPeriod);
+
+    const previousPlan = await plan({
+      ...normalized,
+      ...previousPeriod,
+    });
+
+    const previousFetched = await fetch(previousPlan);
+    previousSeries = compute(previousFetched, previousPlan.definitions);
+  }
+
+  // Stage 6: Format final output with previous period data
+  const includeAlphaIds = executionPlan.definitions.length > 1;
+  const response = format(
+    computedSeries,
+    executionPlan.definitions,
+    includeAlphaIds,
+    previousSeries,
+  );
+
+  return response;
+}
+
+// Export as ChartEngine for backward compatibility
+export const ChartEngine = {
+  execute: executeChart,
+};

packages/db/src/engine/normalize.ts

@@ -0,0 +1,66 @@
+import { alphabetIds } from '@openpanel/constants';
+import type {
+  IChartEvent,
+  IChartEventItem,
+  IChartInput,
+  IChartInputWithDates,
+} from '@openpanel/validation';
+import { getChartStartEndDate } from '../services/chart.service';
+import { getSettingsForProject } from '../services/organization.service';
+import type { SeriesDefinition } from './types';
+
+export type NormalizedInput = Awaited<ReturnType<typeof normalize>>;
+
+/**
+ * Normalize a chart input into a clean structure with dates and normalized series
+ */
+export async function normalize(
+  input: IChartInput,
+): Promise<IChartInputWithDates & { series: SeriesDefinition[] }> {
+  const { timezone } = await getSettingsForProject(input.projectId);
+  const { startDate, endDate } = getChartStartEndDate(
+    {
+      range: input.range,
+      startDate: input.startDate ?? undefined,
+      endDate: input.endDate ?? undefined,
+    },
+    timezone,
+  );
+
+  // Get series from input (handles both 'series' and 'events' fields)
+  // The schema preprocessing should have already converted 'events' to 'series', but handle both for safety
+  const rawSeries = (input as any).series ?? (input as any).events ?? [];
+
+  // Normalize each series item
+  const normalizedSeries: SeriesDefinition[] = rawSeries.map(
+    (item: any, index: number) => {
+      // If item already has type field, it's the new format
+      if (item && typeof item === 'object' && 'type' in item) {
+        return {
+          ...item,
+          id: item.id ?? alphabetIds[index] ?? `series-${index}`,
+        } as SeriesDefinition;
+      }
+
+      // Old format without type field - assume it's an event
+      const event = item as Partial<IChartEvent>;
+      return {
+        type: 'event',
+        id: event.id ?? alphabetIds[index] ?? `series-${index}`,
+        name: event.name || 'unknown_event',
+        segment: event.segment ?? 'event',
+        filters: event.filters ?? [],
+        displayName: event.displayName,
+        property: event.property,
+      } as SeriesDefinition;
+    },
+  );
+
+  return {
+    ...input,
+    series: normalizedSeries,
+    startDate,
+    endDate,
+  };
+}
+

packages/db/src/engine/plan.ts

@@ -0,0 +1,59 @@
+import { slug } from '@openpanel/common';
+import { alphabetIds } from '@openpanel/constants';
+import type {
+  IChartBreakdown,
+  IChartEvent,
+  IChartEventItem,
+} from '@openpanel/validation';
+import { getSettingsForProject } from '../services/organization.service';
+import type { ConcreteSeries, Plan } from './types';
+import type { NormalizedInput } from './normalize';
+
+/**
+ * Create an execution plan from normalized input
+ * This sets up ConcreteSeries placeholders - actual breakdown expansion happens during fetch
+ */
+export async function plan(
+  normalized: NormalizedInput,
+): Promise<Plan> {
+  const { timezone } = await getSettingsForProject(normalized.projectId);
+
+  const concreteSeries: ConcreteSeries[] = [];
+
+  // Create concrete series placeholders for each definition
+  normalized.series.forEach((definition, index) => {
+    if (definition.type === 'event') {
+      const event = definition as IChartEventItem & { type: 'event' };
+      
+      // For events, create a placeholder
+      // If breakdowns exist, fetch will return multiple series (one per breakdown value)
+      // If no breakdowns, fetch will return one series
+      const concrete: ConcreteSeries = {
+        id: `${slug(event.name)}-${event.id ?? index}`,
+        definitionId: event.id ?? alphabetIds[index] ?? `series-${index}`,
+        definitionIndex: index,
+        name: [event.displayName || event.name],
+        context: {
+          event: event.name,
+          filters: [...event.filters],
+        },
+        data: [], // Will be populated by fetch stage
+        definition,
+      };
+      concreteSeries.push(concrete);
+    } else {
+      // For formulas, we'll create placeholders during compute stage
+      // Formulas depend on event series, so we skip them here
+    }
+  });
+
+  return {
+    concreteSeries,
+    definitions: normalized.series,
+    input: normalized,
+    timezone,
+  };
+}
+
+export type NormalizedInput = Awaited<ReturnType<typeof import('./normalize').normalize>>;
+

packages/db/src/engine/types.ts

@@ -0,0 +1,85 @@
+import type {
+  IChartBreakdown,
+  IChartEvent,
+  IChartEventFilter,
+  IChartEventItem,
+  IChartFormula,
+  IChartInput,
+  IChartInputWithDates,
+} from '@openpanel/validation';
+
+/**
+ * Series Definition - The input representation of what the user wants
+ * This is what comes from the frontend (events or formulas)
+ */
+export type SeriesDefinition = IChartEventItem;
+
+/**
+ * Concrete Series - A resolved series that will be displayed as a line/bar on the chart
+ * When breakdowns exist, one SeriesDefinition can expand into multiple ConcreteSeries
+ */
+export type ConcreteSeries = {
+  id: string;
+  definitionId: string; // ID of the SeriesDefinition this came from
+  definitionIndex: number; // Index in the original series array (for A, B, C references)
+  name: string[]; // Display name parts: ["Session Start", "Chrome"] or ["Formula 1"]
+  
+  // Context for Drill-down / Profiles
+  // This contains everything needed to query 'who are these users?'
+  context: {
+    event?: string; // Event name (if this is an event series)
+    filters: IChartEventFilter[]; // All filters including breakdown value
+    breakdownValue?: string; // The breakdown value for this concrete series (deprecated, use breakdowns instead)
+    breakdowns?: Record<string, string>; // Breakdown keys and values: { country: 'SE', path: '/ewoqmepwq' }
+  };
+
+  // Data points for this series
+  data: Array<{
+    date: string;
+    count: number;
+    total_count?: number;
+  }>;
+
+  // The original definition (event or formula)
+  definition: SeriesDefinition;
+};
+
+/**
+ * Plan - The execution plan after normalization and expansion
+ */
+export type Plan = {
+  concreteSeries: ConcreteSeries[];
+  definitions: SeriesDefinition[];
+  input: IChartInputWithDates;
+  timezone: string;
+};
+
+/**
+ * Chart Response - The final output format
+ */
+export type ChartResponse = {
+  series: Array<{
+    id: string;
+    name: string[];
+    data: Array<{
+      date: string;
+      value: number;
+      previous?: number;
+    }>;
+    summary: {
+      total: number;
+      average: number;
+      min: number;
+      max: number;
+      count?: number;
+    };
+    context?: ConcreteSeries['context']; // Include context for drill-down
+  }>;
+  summary: {
+    total: number;
+    average: number;
+    min: number;
+    max: number;
+  };
+};
+