feat: implement OpenAPI-first workflow, CQRS analytics, audit trail, and benchmarking

.github/workflows/sdk-generate.yml

@@ -0,0 +1,72 @@
+name: SDK Auto-Generation
+
+on:
+  push:
+    paths:
+      - 'spec/openapi.yaml'
+      - 'scripts/sdk-generate.sh'
+      - '.github/workflows/sdk-generate.yml'
+  pull_request:
+    paths:
+      - 'spec/openapi.yaml'
+
+jobs:
+  validate-spec:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Validate OpenAPI spec
+        uses: mbowman100/swagger-validator-action@v1
+        with:
+          files: spec/openapi.yaml
+      - name: Check breaking changes
+        uses: ponelat/oas-breaking-changes-action@v1
+        with:
+          spec-file: spec/openapi.yaml
+          old-spec-file: docs/openapi.yaml
+
+  generate-sdks:
+    needs: validate-spec
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        language: [javascript, python, go]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: '20'
+          cache: 'npm'
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+        if: matrix.language == 'python'
+      - uses: actions/setup-go@v5
+        with:
+          go-version: '1.22'
+        if: matrix.language == 'go'
+      - name: Generate ${{ matrix.language }} SDK
+        run: bash scripts/sdk-generate.sh ${{ matrix.language }}
+      - name: Check for changes
+        id: diff
+        run: |
+          if [ -n "$(git status --porcelain sdks/${{ matrix.language }}/)" ]; then
+            echo "changed=true" >> $GITHUB_OUTPUT
+          else
+            echo "changed=false" >> $GITHUB_OUTPUT
+          fi
+      - name: Create PR for SDK changes
+        if: steps.diff.outputs.changed == 'true' && github.event_name == 'push'
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          BRANCH="sdk-auto/${{ matrix.language }}-$(date +%s)"
+          git checkout -b "$BRANCH"
+          git add sdks/${{ matrix.language }}/
+          git commit -m "chore(sdk): auto-generate ${{ matrix.language }} SDK from OpenAPI spec"
+          git push origin "$BRANCH"
+          gh pr create \
+            --base main \
+            --title "chore(sdk): auto-generate ${{ matrix.language }} SDK" \
+            --body "Auto-generated ${{ matrix.language }} SDK from OpenAPI spec changes in \`spec/openapi.yaml\`." \
+            --label automated

backend/analytics/command/index.ts

@@ -0,0 +1,2 @@
+export { SubscriptionCommandHandler } from './subscriptionCommandHandler';
+export type { CreateSubscriptionCommand, CancelSubscriptionCommand } from './subscriptionCommandHandler';

backend/analytics/command/subscriptionCommandHandler.ts

@@ -0,0 +1,50 @@
+import { QueryClient } from '../../../backend/shared/query/queryRouter';
+
+export interface CreateSubscriptionCommand {
+  id: string;
+  planId: string;
+  userId: string;
+  amount: number;
+  currency: string;
+  billingCycle: string;
+  nextBillingDate: Date;
+  metadata?: Record<string, unknown>;
+}
+
+export interface CancelSubscriptionCommand {
+  id: string;
+  userId: string;
+  reason?: string;
+}
+
+export class SubscriptionCommandHandler {
+  constructor(private db: QueryClient) {}
+
+  async create(cmd: CreateSubscriptionCommand): Promise<void> {
+    await this.db.query(
+      `INSERT INTO subscriptions
+         (id, plan_id, user_id, amount, currency, billing_cycle, status, next_billing_date, metadata, created_at, updated_at)
+       VALUES ($1, $2, $3, $4, $5, $6, 'active', $7, $8, NOW(), NOW())`,
+      [
+        cmd.id,
+        cmd.planId,
+        cmd.userId,
+        cmd.amount,
+        cmd.currency,
+        cmd.billingCycle,
+        cmd.nextBillingDate,
+        cmd.metadata ? JSON.stringify(cmd.metadata) : null,
+      ],
+    );
+  }
+
+  async cancel(cmd: CancelSubscriptionCommand): Promise<void> {
+    await this.db.query(
+      `UPDATE subscriptions
+       SET status = 'cancelled', cancelled_at = NOW(), updated_at = NOW(),
+           cancellation_reason = $3
+       WHERE id = $1 AND user_id = $2`,
+      [cmd.id, cmd.userId, cmd.reason ?? null],
+    );
+  }
+}

backend/analytics/jobs/mvRefreshJob.ts

@@ -1,22 +1,8 @@
-/**
- * Materialized View Refresh Job
- *
- * Incrementally refreshes each materialized view using
- * REFRESH MATERIALIZED VIEW CONCURRENTLY so reads are never blocked.
- *
- * Runs on a configurable interval (default 60 s for real-time views).
- * Exposes a Prometheus-style metric for view freshness monitoring.
- */
-
 import { QueryClient } from '../../../backend/shared/query/queryRouter';
 
-// ── View definitions ──────────────────────────────────────────────────────────
-
 interface ViewConfig {
   name: string;
-  /** Refresh interval in ms. */
   intervalMs: number;
-  /** Last successful refresh timestamp. */
   lastRefreshedAt: Date | null;
   isRefreshing: boolean;
 }
@@ -26,19 +12,18 @@ const DEFAULT_VIEWS: ViewConfig[] = [
   { name: 'subscriber_balance_mv',        intervalMs: 60_000,  lastRefreshedAt: null, isRefreshing: false },
   { name: 'monthly_revenue_mv',           intervalMs: 300_000, lastRefreshedAt: null, isRefreshing: false },
   { name: 'churn_summary_mv',             intervalMs: 300_000, lastRefreshedAt: null, isRefreshing: false },
+  { name: 'mrr_mv',                       intervalMs: 300_000, lastRefreshedAt: null, isRefreshing: false },
+  { name: 'cohort_retention_mv',          intervalMs: 3_600_000, lastRefreshedAt: null, isRefreshing: false },
+  { name: 'ltv_mv',                       intervalMs: 86_400_000, lastRefreshedAt: null, isRefreshing: false },
 ];
 
-// ── Types ─────────────────────────────────────────────────────────────────────
-
 export interface RefreshMetric {
   viewName: string;
   lastRefreshedAt: Date | null;
   lagMs: number;
   isStale: boolean;
 }
 
-// ── Job ───────────────────────────────────────────────────────────────────────
-
 export class MVRefreshJob {
   private db: QueryClient;
   private views: ViewConfig[];
@@ -52,10 +37,7 @@ export class MVRefreshJob {
   start(): void {
     for (const view of this.views) {
       if (this.timers.has(view.name)) continue;
-
-      // Run immediately on start, then on interval
       void this.refresh(view.name);
-
       const timer = setInterval(
         () => void this.refresh(view.name),
         view.intervalMs,
@@ -71,7 +53,6 @@ export class MVRefreshJob {
     this.timers.clear();
   }
 
-  /** Refresh a single view by name. Skips if already refreshing. */
   async refresh(viewName: string): Promise<void> {
     const view = this.views.find((v) => v.name === viewName);
     if (!view || view.isRefreshing) return;
@@ -80,7 +61,6 @@ export class MVRefreshJob {
     const start = Date.now();
 
     try {
-      // CONCURRENTLY requires a unique index on the view — see migration 002
       await this.db.query(`REFRESH MATERIALIZED VIEW CONCURRENTLY ${viewName}`);
       view.lastRefreshedAt = new Date();
       console.info(`[MVRefreshJob] Refreshed ${viewName} in ${Date.now() - start}ms`);
@@ -91,7 +71,6 @@ export class MVRefreshJob {
     }
   }
 
-  /** Return freshness metrics for all views (used by monitoring service). */
   getMetrics(): RefreshMetric[] {
     return this.views.map((view) => {
       const lagMs = view.lastRefreshedAt
@@ -106,13 +85,6 @@ export class MVRefreshJob {
     });
   }
 
-  /**
-   * Prometheus-style text format for scraping.
-   *
-   * Metrics exposed:
-   *   subtrackr_mv_lag_ms{view="..."}   – lag in milliseconds
-   *   subtrackr_mv_is_stale{view="..."}  – 1 if stale, 0 if fresh
-   */
   prometheusMetrics(): string {
     const lines: string[] = [
       '# HELP subtrackr_mv_lag_ms Materialized view refresh lag in milliseconds',

backend/analytics/query/cohortRetentionQueryHandler.ts

@@ -0,0 +1,57 @@
+import { QueryClient } from '../../../backend/shared/query/queryRouter';
+
+export interface CohortRetentionPeriod {
+  period: number;
+  retained: number;
+  retentionPct: number;
+}
+
+export interface CohortRetentionResult {
+  cohort: string;
+  periods: CohortRetentionPeriod[];
+}
+
+export class CohortRetentionQueryHandler {
+  constructor(private db: QueryClient) {}
+
+  async getCohortRetention(cohort?: string): Promise<CohortRetentionResult[]> {
+    let sql = `
+      SELECT
+        cohort,
+        period,
+        retained,
+        retention_pct AS "retentionPct"
+      FROM cohort_retention_mv
+      WHERE 1=1
+    `;
+    const params: unknown[] = [];
+    if (cohort) {
+      params.push(cohort);
+      sql += ` AND cohort = $${params.length}`;
+    }
+    sql += ' ORDER BY cohort, period';
+    const result = await this.db.query<{
+      cohort: string;
+      period: number;
+      retained: number;
+      retentionPct: number;
+    }>(sql, params);
+
+    const grouped: Map<string, CohortRetentionPeriod[]> = new Map();
+    for (const row of result.rows) {
+      if (!grouped.has(row.cohort)) {
+        grouped.set(row.cohort, []);
+      }
+      grouped.get(row.cohort)!.push({
+        period: row.period,
+        retained: row.retained,
+        retentionPct: row.retentionPct,
+      });
+    }
+
+    return Array.from(grouped.entries()).map(([cohort, periods]) => ({
+      cohort,
+      periods,
+    }));
+  }
+}

backend/analytics/query/index.ts

@@ -0,0 +1,6 @@
+export { MRRQueryHandler } from './mrrQueryHandler';
+export type { MRRQueryResult } from './mrrQueryHandler';
+export { CohortRetentionQueryHandler } from './cohortRetentionQueryHandler';
+export type { CohortRetentionResult, CohortRetentionPeriod } from './cohortRetentionQueryHandler';
+export { LTVQueryHandler } from './ltvQueryHandler';
+export type { LTVQueryResult } from './ltvQueryHandler';

backend/analytics/query/ltvQueryHandler.ts

@@ -0,0 +1,40 @@
+import { QueryClient } from '../../../backend/shared/query/queryRouter';
+
+export interface LTVQueryResult {
+  month: string;
+  averageLtv: number;
+  medianLtv: number;
+  p25Ltv: number;
+  p75Ltv: number;
+  refreshedAt: Date;
+}
+
+export class LTVQueryHandler {
+  constructor(private db: QueryClient) {}
+
+  async getLTV(from?: string, to?: string): Promise<LTVQueryResult[]> {
+    let sql = `
+      SELECT
+        month,
+        average_ltv AS "averageLtv",
+        median_ltv AS "medianLtv",
+        p25_ltv AS "p25Ltv",
+        p75_ltv AS "p75Ltv",
+        refreshed_at AS "refreshedAt"
+      FROM ltv_mv
+      WHERE 1=1
+    `;
+    const params: unknown[] = [];
+    if (from) {
+      params.push(from);
+      sql += ` AND month >= $${params.length}`;
+    }
+    if (to) {
+      params.push(to);
+      sql += ` AND month <= $${params.length}`;
+    }
+    sql += ' ORDER BY month DESC';
+    const result = await this.db.query<LTVQueryResult>(sql, params);
+    return result.rows;
+  }
+}

backend/analytics/query/mrrQueryHandler.ts

@@ -0,0 +1,42 @@
+import { QueryClient } from '../../../backend/shared/query/queryRouter';
+
+export interface MRRQueryResult {
+  month: string;
+  mrr: number;
+  newSubscriptions: number;
+  upgrades: number;
+  downgrades: number;
+  churn: number;
+  refreshedAt: Date;
+}
+
+export class MRRQueryHandler {
+  constructor(private db: QueryClient) {}
+
+  async getMRR(from?: string, to?: string): Promise<MRRQueryResult[]> {
+    let sql = `
+      SELECT
+        month,
+        mrr,
+        new_subscriptions AS "newSubscriptions",
+        upgrades,
+        downgrades,
+        churn,
+        refreshed_at AS "refreshedAt"
+      FROM mrr_mv
+      WHERE 1=1
+    `;
+    const params: unknown[] = [];
+    if (from) {
+      params.push(from);
+      sql += ` AND month >= $${params.length}`;
+    }
+    if (to) {
+      params.push(to);
+      sql += ` AND month <= $${params.length}`;
+    }
+    sql += ' ORDER BY month DESC';
+    const result = await this.db.query<MRRQueryResult>(sql, params);
+    return result.rows;
+  }
+}