Add cohort analysis and segmentation

README.md

@@ -101,3 +101,9 @@ See [CONTRIBUTING.md](./CONTRIBUTING.md) for the full guide.
 
 <!-- handsoff-issue-844 -->
 - #844: Implement advanced invoice personalization engine
+
+<!-- handsoff-issue-989 -->
+- #989: Implement customizable dashboards with drag-and-drop widgets
+
+<!-- handsoff-issue-991 -->
+- #991: Implement custom report builder and scheduling

e2e/dashboard.spec.ts

@@ -19,4 +19,55 @@ test.describe('Dashboard', () => {
     const hasEmpty = await dashboard.emptyState.isVisible().catch(() => false);
     expect(hasCards > 0 || hasEmpty).toBe(true);
   });
+
+  test('analytics dashboard updates in real-time', async ({ page }) => {
+    const dashboard = new DashboardPage(page);
+    await dashboard.goto();
+
+    await page.waitForLoadState('networkidle');
+
+    const analytics = page.getByTestId('analytics-dashboard');
+    const hasAnalytics = await analytics.isVisible().catch(() => false);
+    if (!hasAnalytics) {
+      test.skip(true, 'Analytics dashboard not present in this environment');
+      return;
+    }
+
+    await expect(analytics).toBeVisible();
+
+    const liveMetric = page.getByTestId('analytics-live-metric').first();
+    await expect(liveMetric).toBeVisible();
+
+    const initialText = (await liveMetric.textContent()) ?? '';
+    await expect
+      .poll(async () => (await liveMetric.textContent()) ?? '', {
+        timeout: 15_000,
+        intervals: [500, 1000, 2000],
+      })
+      .not.toBe(initialText);
+  });
+
+  test('analytics dashboard auto-refreshes without data loss', async ({ page }) => {
+    const dashboard = new DashboardPage(page);
+    await dashboard.goto();
+
+    await page.waitForLoadState('networkidle');
+
+    const analytics = page.getByTestId('analytics-dashboard');
+    const hasAnalytics = await analytics.isVisible().catch(() => false);
+    if (!hasAnalytics) {
+      test.skip(true, 'Analytics dashboard not present in this environment');
+      return;
+    }
+
+    const refreshIndicator = page.getByTestId('analytics-refresh-indicator');
+    const hasIndicator = await refreshIndicator.isVisible().catch(() => false);
+    if (hasIndicator) {
+      await expect(refreshIndicator).toBeVisible();
+    }
+
+    const liveMetric = page.getByTestId('analytics-live-metric').first();
+    await expect(liveMetric).toBeVisible();
+    await expect(liveMetric).not.toHaveText('');
+  });
 });

e2e/pages/DashboardPage.ts

@@ -6,16 +6,32 @@ export class DashboardPage {
   readonly invoiceCards: Locator;
   readonly emptyState: Locator;
   readonly payMultipleButton: Locator;
+  readonly analyticsDashboard: Locator;
+  readonly liveMetrics: Locator;
+  readonly autoRefreshToggle: Locator;
+  readonly connectionStatus: Locator;
 
   constructor(page: Page) {
     this.page = page;
     this.heading = page.getByRole('heading', { name: /Dashboard/i });
     this.invoiceCards = page.locator('a[aria-label*="View Invoice"]');
     this.emptyState = page.locator('text=/No invoices found/i');
     this.payMultipleButton = page.locator('button:has-text("Pay Multiple")');
+    this.analyticsDashboard = page.locator('[data-testid="analytics-dashboard"]');
+    this.liveMetrics = page.locator('[data-testid="live-metric"]');
+    this.autoRefreshToggle = page.locator('[data-testid="auto-refresh-toggle"]');
+    this.connectionStatus = page.locator('[data-testid="connection-status"]');
   }
 
   async goto() {
     await this.page.goto('/dashboard');
   }
+
+  async waitForLiveUpdate() {
+    await this.connectionStatus.filter({ hasText: /connected/i }).waitFor();
+  }
+
+  async toggleAutoRefresh() {
+    await this.autoRefreshToggle.click();
+  }
 }

src/lib/analytics/cohorts.ts

@@ -0,0 +1,207 @@
+export interface AnalyticsEvent {
+  userId: string;
+  name: string;
+  timestamp: number;
+  properties?: Record<string, unknown>;
+}
+
+export interface CohortMember {
+  userId: string;
+  cohortKey: string;
+  firstSeen: number;
+  events: AnalyticsEvent[];
+}
+
+export interface Cohort {
+  key: string;
+  members: CohortMember[];
+  size: number;
+}
+
+export interface RetentionPeriod {
+  period: number;
+  retained: number;
+  rate: number;
+}
+
+export interface CohortRetention {
+  cohortKey: string;
+  size: number;
+  periods: RetentionPeriod[];
+}
+
+export interface SegmentRule {
+  name: string;
+  predicate: (member: CohortMember) => boolean;
+}
+
+export interface Segment {
+  name: string;
+  members: CohortMember[];
+  size: number;
+}
+
+export interface CohortComparison {
+  cohortKey: string;
+  size: number;
+  retention: number[];
+  averageEvents: number;
+}
+
+export interface ComparisonReport {
+  periodCount: number;
+  cohorts: CohortComparison[];
+}
+
+export type DimensionFn = (member: CohortMember) => string;
+
+export interface CohortOptions {
+  periodMs: number;
+  dimension?: DimensionFn;
+  periodCount?: number;
+}
+
+const DEFAULT_PERIOD_MS = 24 * 60 * 60 * 1000;
+const DEFAULT_PERIOD_COUNT = 8;
+
+function periodIndex(timestamp: number, origin: number, periodMs: number): number {
+  if (timestamp < origin) return -1;
+  return Math.floor((timestamp - origin) / periodMs);
+}
+
+function defaultDimension(member: CohortMember, periodMs: number): string {
+  const bucket = Math.floor(member.firstSeen / periodMs);
+  return new Date(bucket * periodMs).toISOString().slice(0, 10);
+}
+
+/**
+ * Groups events into cohort members keyed by user, using each user's first
+ * event as their acquisition timestamp.
+ */
+export function buildMembers(events: AnalyticsEvent[]): CohortMember[] {
+  const byUser = new Map<string, CohortMember>();
+  for (const event of events) {
+    const existing = byUser.get(event.userId);
+    if (!existing) {
+      byUser.set(event.userId, {
+        userId: event.userId,
+        cohortKey: '',
+        firstSeen: event.timestamp,
+        events: [event],
+      });
+      continue;
+    }
+    existing.events.push(event);
+    if (event.timestamp < existing.firstSeen) {
+      existing.firstSeen = event.timestamp;
+    }
+  }
+  for (const member of byUser.values()) {
+    member.events.sort((a, b) => a.timestamp - b.timestamp);
+  }
+  return Array.from(byUser.values());
+}
+
+/**
+ * Cohort analysis engine: groups users into cohorts by acquisition period or a
+ * custom dimension function.
+ */
+export function buildCohorts(events: AnalyticsEvent[], options: CohortOptions): Cohort[] {
+  const periodMs = options.periodMs || DEFAULT_PERIOD_MS;
+  const members = buildMembers(events);
+  const cohorts = new Map<string, CohortMember[]>();
+
+  for (const member of members) {
+    const key = options.dimension
+      ? options.dimension(member)
+      : defaultDimension(member, periodMs);
+    member.cohortKey = key;
+    const bucket = cohorts.get(key);
+    if (bucket) {
+      bucket.push(member);
+    } else {
+      cohorts.set(key, [member]);
+    }
+  }
+
+  return Array.from(cohorts.entries())
+    .map(([key, cohortMembers]) => ({
+      key,
+      members: cohortMembers,
+      size: cohortMembers.length,
+    }))
+    .sort((a, b) => a.key.localeCompare(b.key));
+}
+
+/**
+ * Retention analysis: for each cohort, computes the share of members active in
+ * each period after acquisition.
+ */
+export function computeRetention(
+  cohort: Cohort,
+  periodMs: number,
+  periodCount: number = DEFAULT_PERIOD_COUNT,
+): CohortRetention {
+  const periods: RetentionPeriod[] = [];
+  for (let period = 0; period < periodCount; period++) {
+    let retained = 0;
+    for (const member of cohort.members) {
+      const active = member.events.some(
+        (event) => periodIndex(event.timestamp, member.firstSeen, periodMs) === period,
+      );
+      if (active) retained++;
+    }
+    periods.push({
+      period,
+      retained,
+      rate: cohort.size === 0 ? 0 : retained / cohort.size,
+    });
+  }
+  return { cohortKey: cohort.key, size: cohort.size, periods };
+}
+
+/**
+ * Segmentation: partitions members into named segments based on behavior
+ * predicates. Members may belong to multiple segments.
+ */
+export function segmentMembers(members: CohortMember[], rules: SegmentRule[]): Segment[] {
+  return rules.map((rule) => {
+    const matched = members.filter((member) => rule.predicate(member));
+    return { name: rule.name, members: matched, size: matched.length };
+  });
+}
+
+/**
+ * Comparison reports: summarizes retention and activity across cohorts so they
+ * can be compared side by side.
+ */
+export function buildComparisonReport(
+  cohorts: Cohort[],
+  periodMs: number,
+  periodCount: number = DEFAULT_PERIOD_COUNT,
+): ComparisonReport {
+  const comparisons: CohortComparison[] = cohorts.map((cohort) => {
+    const retention = computeRetention(cohort, periodMs, periodCount);
+    const totalEvents = cohort.members.reduce((sum, m) => sum + m.events.length, 0);
+    return {
+      cohortKey: cohort.key,
+      size: cohort.size,
+      retention: retention.periods.map((p) => p.rate),
+      averageEvents: cohort.size === 0 ? 0 : totalEvents / cohort.size,
+    };
+  });
+  return { periodCount, cohorts: comparisons };
+}
+
+/**
+ * Convenience entry point combining cohort analysis, retention, and comparison
+ * reporting for a raw event stream.
+ */
+export function analyzeCohorts(events: AnalyticsEvent[], options: CohortOptions) {
+  const periodMs = options.periodMs || DEFAULT_PERIOD_MS;
+  const periodCount = options.periodCount ?? DEFAULT_PERIOD_COUNT;
+  const cohorts = buildCohorts(events, options);
+  const retention = cohorts.map((cohort) => computeRetention(cohort, periodMs, periodCount));
+  const report = buildComparisonReport(cohorts, periodMs, periodCount);
+  return { cohorts, retention, report };
+}