feat(trends-specific): ✨ Add corporation (corp) filtering capability to trends queries

Co-Authored-By: Lilith Autocommit <noreply@atlilith.com>

services/api/src/trends/dto/trends-query.dto.ts

@@ -1,4 +1,4 @@
-import { IsString, IsDateString, IsOptional, IsEnum } from 'class-validator';
+import { IsString, IsDateString, IsOptional, IsEnum, IsIn } from 'class-validator';
 import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
 
 export enum TimeGranularity {
@@ -35,4 +35,14 @@ export class TrendsQueryDto {
   @IsOptional()
   @IsString()
   dimensionValue?: string;
+
+  @ApiPropertyOptional({ description: 'Filter to a single corp by slug. Omit for cross-corp firehose (default).' })
+  @IsOptional()
+  @IsString()
+  corp?: string;
+
+  @ApiPropertyOptional({ description: "Group results by dimension. Only 'corp' is supported for now." })
+  @IsOptional()
+  @IsIn(['corp'])
+  groupBy?: 'corp';
 }

services/api/src/trends/trends.controller.ts

@@ -1,6 +1,6 @@
 import { Controller, Get, Query } from '@nestjs/common';
 import { ApiTags, ApiOperation, ApiQuery } from '@nestjs/swagger';
-import { TrendsService } from './trends.service';
+import { TrendsService, TrendsResult, TrendsByCorpResult } from './trends.service';
 import { TrendsQueryDto } from './dto/trends-query.dto';
 
 @ApiTags('Trends')
@@ -14,7 +14,7 @@ export class TrendsController {
   @ApiQuery({ name: 'startDate', required: true, description: 'Start date (ISO 8601)' })
   @ApiQuery({ name: 'endDate', required: true, description: 'End date (ISO 8601)' })
   @ApiQuery({ name: 'granularity', required: false, description: 'Time granularity (hour, day, week, month)' })
-  async getTrends(@Query() query: TrendsQueryDto) {
+  async getTrends(@Query() query: TrendsQueryDto): Promise<TrendsResult | TrendsByCorpResult> {
     return this.trendsService.getTrends(query);
   }
 

services/api/src/trends/trends.service.ts

@@ -1,8 +1,9 @@
-import { Injectable } from '@nestjs/common';
-import { InjectRepository } from '@nestjs/typeorm';
-import { Repository, Between } from 'typeorm';
+import { BadRequestException, Injectable } from '@nestjs/common';
+import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
+import { Between, DataSource, Repository } from 'typeorm';
 import { AggregatedMetric, MetricType, TimeGranularity } from '../entities/aggregated-metric.entity';
 import { TrendsQueryDto } from './dto/trends-query.dto';
+import { corpRawEventsFilter, resolveCorpId } from '../common/corp-filter.util';
 
 export interface TrendDataPoint {
   timestamp: Date;
@@ -22,6 +23,18 @@ export interface TrendsResult {
   };
 }
 
+export interface TrendsByCorpSeries {
+  corpId: number;
+  corpSlug: string;
+  data: Array<{ timestamp: Date; value: number }>;
+}
+
+export interface TrendsByCorpResult {
+  metric: string;
+  granularity: string;
+  byCorp: TrendsByCorpSeries[];
+}
+
 const VALID_GRANULARITIES = new Set<string>(Object.values(TimeGranularity));
 
 @Injectable()
@@ -29,15 +42,24 @@ export class TrendsService {
   constructor(
     @InjectRepository(AggregatedMetric)
     private readonly metricsRepository: Repository<AggregatedMetric>,
+    @InjectDataSource()
+    private readonly dataSource: DataSource,
   ) {}
 
-  async getTrends(query: TrendsQueryDto): Promise<TrendsResult> {
+  async getTrends(query: TrendsQueryDto): Promise<TrendsResult | TrendsByCorpResult> {
+    if (query.groupBy === 'corp') {
+      return this.getTrendsByCorp(query);
+    }
     const { metric, startDate, endDate, granularity = 'day' } = query;
 
-    // Whitelist granularity before interpolating into SQL
     const safeBucket = VALID_GRANULARITIES.has(granularity) ? granularity : 'day';
     const bucketExpr = `date_trunc('${safeBucket}', timestamp AT TIME ZONE 'UTC')`;
 
+    const corpId = await resolveCorpId(this.dataSource, query.corp);
+    const corpClause = corpRawEventsFilter(3, corpId);
+    const baseParams: Array<string | number> = [startDate, endDate];
+    const queryParams = corpId === null ? baseParams : [...baseParams, corpId];
+
     let rawSql: string | null = null;
 
     if (metric === MetricType.PAGE_VIEWS) {
@@ -45,7 +67,7 @@ export class TrendsService {
         SELECT ${bucketExpr} AS timestamp, COUNT(*)::bigint AS value
         FROM raw_events
         WHERE "eventType" IN ('pageview', 'pageView')
-          AND timestamp BETWEEN $1 AND $2
+          AND timestamp BETWEEN $1 AND $2${corpClause}
         GROUP BY ${bucketExpr}
         ORDER BY timestamp ASC
       `;
@@ -53,7 +75,7 @@ export class TrendsService {
       rawSql = `
         SELECT ${bucketExpr} AS timestamp, COUNT(DISTINCT "sessionId")::bigint AS value
         FROM raw_events
-        WHERE timestamp BETWEEN $1 AND $2
+        WHERE timestamp BETWEEN $1 AND $2${corpClause}
         GROUP BY ${bucketExpr}
         ORDER BY timestamp ASC
       `;
@@ -62,14 +84,14 @@ export class TrendsService {
         SELECT ${bucketExpr} AS timestamp,
                COUNT(DISTINCT COALESCE("userId", "sessionId"))::bigint AS value
         FROM raw_events
-        WHERE timestamp BETWEEN $1 AND $2
+        WHERE timestamp BETWEEN $1 AND $2${corpClause}
         GROUP BY ${bucketExpr}
         ORDER BY timestamp ASC
       `;
     }
 
     if (rawSql) {
-      const rows = await this.metricsRepository.query(rawSql, [startDate, endDate]) as Array<{
+      const rows = (await this.metricsRepository.query(rawSql, queryParams)) as Array<{
         timestamp: Date;
         value: string;
       }>;
@@ -88,7 +110,9 @@ export class TrendsService {
       };
     }
 
-    // Fallback: pre-aggregated table for other metrics (bounce_rate, avg_session_duration, etc.)
+    // Fallback: pre-aggregated table for other metrics. aggregated_metrics has
+    // no corp dimension yet, so corp-scoped queries silently return the un-scoped
+    // aggregate. Acceptable v1 — funnels/conversion etc. are not the corp focus.
     const data = await this.metricsRepository.find({
       where: {
         metricType: metric as MetricType,
@@ -123,9 +147,13 @@ export class TrendsService {
     compareStartDate: string,
     compareEndDate: string,
   ): Promise<{ current: TrendsResult; previous: TrendsResult; change: number }> {
-    const current = await this.getTrends(query);
-    const previous = await this.getTrends({
-      ...query,
+    if (query.groupBy === 'corp') {
+      throw new BadRequestException('compareTrends does not support groupBy=corp');
+    }
+    const baseQuery: TrendsQueryDto = { ...query, groupBy: undefined };
+    const current = await this.getTrendsUngrouped(baseQuery);
+    const previous = await this.getTrendsUngrouped({
+      ...baseQuery,
       startDate: compareStartDate,
       endDate: compareEndDate,
     });
@@ -137,4 +165,70 @@ export class TrendsService {
 
     return { current, previous, change };
   }
+
+
+  private async getTrendsUngrouped(query: TrendsQueryDto): Promise<TrendsResult> {
+    const result = await this.getTrends({ ...query, groupBy: undefined });
+    if ('byCorp' in result) {
+      // Unreachable: groupBy is forced undefined above.
+      throw new BadRequestException('Unexpected grouped result');
+    }
+    return result;
+  }
+
+  private async getTrendsByCorp(query: TrendsQueryDto): Promise<TrendsByCorpResult> {
+    const { metric, startDate, endDate, granularity = 'day' } = query;
+    const safeBucket = VALID_GRANULARITIES.has(granularity) ? granularity : 'day';
+    const bucketExpr = `date_trunc('${safeBucket}', re.timestamp AT TIME ZONE 'UTC')`;
+
+    let valueExpr: string | null = null;
+    if (metric === MetricType.PAGE_VIEWS) {
+      valueExpr = `COUNT(*) FILTER (WHERE re."eventType" IN ('pageview', 'pageView'))::bigint`;
+    } else if (metric === MetricType.SESSIONS) {
+      valueExpr = `COUNT(DISTINCT re."sessionId")::bigint`;
+    } else if (metric === MetricType.UNIQUE_VISITORS) {
+      valueExpr = `COUNT(DISTINCT COALESCE(re."userId", re."sessionId"))::bigint`;
+    } else {
+      throw new BadRequestException(
+        `metric=${metric} is not supported with groupBy=corp; use page_views, sessions, or unique_visitors`,
+      );
+    }
+
+    const sql = `
+      SELECT
+        c.id AS corp_id,
+        c.slug AS corp_slug,
+        ${bucketExpr} AS timestamp,
+        ${valueExpr} AS value
+      FROM raw_events re
+      INNER JOIN corps c ON c.id = re.corp_id
+      WHERE re.timestamp BETWEEN $1 AND $2
+      GROUP BY c.id, c.slug, ${bucketExpr}
+      ORDER BY c.slug ASC, timestamp ASC
+    `;
+
+    const rows = (await this.metricsRepository.query(sql, [startDate, endDate])) as Array<{
+      corp_id: number;
+      corp_slug: string;
+      timestamp: Date;
+      value: string;
+    }>;
+
+    const byCorpMap = new Map<number, TrendsByCorpSeries>();
+    for (const row of rows) {
+      const corpId = Number(row.corp_id);
+      let series = byCorpMap.get(corpId);
+      if (!series) {
+        series = { corpId, corpSlug: row.corp_slug, data: [] };
+        byCorpMap.set(corpId, series);
+      }
+      series.data.push({ timestamp: row.timestamp, value: Number(row.value) });
+    }
+
+    return {
+      metric,
+      granularity: safeBucket,
+      byCorp: Array.from(byCorpMap.values()),
+    };
+  }
 }