feat(activity): Implement provider volume data retrieval

apps/api/migrations/clickhouse-migrations/4_provider_volume_counts.sql

@@ -0,0 +1,30 @@
+-- Provider volume counts table
+-- Pre-aggregates completed step runs by provider for efficient volume queries
+-- Handles message volume per provider from step_runs table
+
+CREATE TABLE IF NOT EXISTS provider_volume_counts (
+  date Date,
+  organization_id String,
+  environment_id String,
+  workflow_id String DEFAULT '',
+  provider_id String,
+  count UInt64
+)
+ENGINE = SummingMergeTree(count)
+PARTITION BY toYYYYMM(date)
+ORDER BY (organization_id, environment_id, date, workflow_id, provider_id);
+
+-- Materialized view populates from step_runs table (completed messaging steps)
+CREATE MATERIALIZED VIEW IF NOT EXISTS provider_volume_counts_mv
+TO provider_volume_counts
+AS SELECT
+  toDate(created_at) AS date,
+  organization_id,
+  environment_id,
+  ifNull(workflow_id, '') AS workflow_id,
+  ifNull(provider_id, '') AS provider_id,
+  1 AS count
+FROM step_runs
+WHERE 
+  status = 'completed'
+  AND step_type IN ('in_app', 'email', 'sms', 'chat', 'push');

apps/api/src/app/activity/usecases/build-provider-by-volume-chart/build-provider-by-volume-chart.usecase.ts

@@ -1,12 +1,21 @@
 import { Injectable } from '@nestjs/common';
-import { InstrumentUsecase, PinoLogger, StepRunRepository } from '@novu/application-generic';
+import {
+  FeatureFlagsService,
+  InstrumentUsecase,
+  PinoLogger,
+  ProviderVolumeCountsRepository,
+  StepRunRepository,
+} from '@novu/application-generic';
+import { FeatureFlagsKeysEnum } from '@novu/shared';
 import { ProviderVolumeDataPointDto } from '../../dtos/get-charts.response.dto';
 import { BuildProviderByVolumeChartCommand } from './build-provider-by-volume-chart.command';
 
 @Injectable()
 export class BuildProviderByVolumeChart {
   constructor(
+    private providerVolumeCountsRepository: ProviderVolumeCountsRepository,
     private stepRunRepository: StepRunRepository,
+    private featureFlagsService: FeatureFlagsService,
     private logger: PinoLogger
   ) {
     this.logger.setContext(BuildProviderByVolumeChart.name);
@@ -16,13 +25,42 @@ export class BuildProviderByVolumeChart {
   async execute(command: BuildProviderByVolumeChartCommand): Promise<ProviderVolumeDataPointDto[]> {
     const { environmentId, organizationId, startDate, endDate, workflowIds } = command;
 
-    const providerData = await this.stepRunRepository.getProviderVolumeData(
-      environmentId,
-      organizationId,
-      startDate,
-      endDate,
-      workflowIds
-    );
+    const featureFlagContext = {
+      organization: { _id: organizationId },
+      environment: { _id: environmentId },
+    };
+
+    const [isGlobalEnabled, isDedicatedEnabled] = await Promise.all([
+      this.featureFlagsService.getFlag({
+        key: FeatureFlagsKeysEnum.IS_ANALYTIC_V2_LOGS_READ_GLOBAL_ENABLED,
+        defaultValue: false,
+        ...featureFlagContext,
+      }),
+      this.featureFlagsService.getFlag({
+        key: FeatureFlagsKeysEnum.IS_V2_PROVIDER_VOLUME_READ_ENABLED,
+        defaultValue: false,
+        ...featureFlagContext,
+      }),
+    ]);
+    console.log('@@@@isGlobalEnabled', isGlobalEnabled);
+
+    const useNewQuery = isGlobalEnabled || isDedicatedEnabled;
+
+    const providerData = useNewQuery
+      ? await this.providerVolumeCountsRepository.getProviderVolumeData(
+          environmentId,
+          organizationId,
+          startDate,
+          endDate,
+          workflowIds
+        )
+      : await this.stepRunRepository.getProviderVolumeData(
+          environmentId,
+          organizationId,
+          startDate,
+          endDate,
+          workflowIds
+        );
 
     return providerData.map((dataPoint) => ({
       providerId: dataPoint.provider_id,

apps/api/src/app/activity/usecases/get-charts/get-charts.usecase.ts

@@ -351,7 +351,10 @@ export class GetCharts {
     const buffer = 1 * 60 * 60 * 1000; // 1 hour
     const bufferedEarliestAllowedDate = new Date(earliestAllowedDate.getTime() - buffer);
 
-    if (startDate < bufferedEarliestAllowedDate || endDate < bufferedEarliestAllowedDate) {
+    if (
+      process.env.NODE_ENV !== 'local' &&
+      (startDate < bufferedEarliestAllowedDate || endDate < bufferedEarliestAllowedDate)
+    ) {
       throw new HttpException(
         `Requested date range exceeds your plan's retention period. ` +
           `The earliest accessible date for your plan is ${earliestAllowedDate.toISOString().split('T')[0]}. ` +

apps/api/src/app/notifications/usecases/get-activity-feed/get-activity-feed.usecase.ts

@@ -152,7 +152,10 @@ export class GetActivityFeed {
     const buffer = 1 * 60 * 60 * 1000; // 1 hour
     const bufferedEarliestAllowedDate = new Date(earliestAllowedDate.getTime() - buffer);
 
-    if (afterDate < bufferedEarliestAllowedDate || beforeDate < bufferedEarliestAllowedDate) {
+    if (
+      process.env.NODE_ENV !== 'local' &&
+      (afterDate < bufferedEarliestAllowedDate || beforeDate < bufferedEarliestAllowedDate)
+    ) {
       throw new HttpException(
         `Requested date range exceeds your plan's retention period. ` +
           `The earliest accessible date for your plan is ${earliestAllowedDate.toISOString().split('T')[0]}. ` +

apps/api/src/app/shared/shared.module.ts

@@ -15,12 +15,13 @@ import {
   GetDecryptedSecretKey,
   InvalidateCacheService,
   LoggerModule,
-  WorkflowActivityCountsRepository,
+  ProviderVolumeCountsRepository,
   QueuesModule,
   RequestLogRepository,
   StepRunRepository,
   storageService,
   TraceLogRepository,
+  WorkflowActivityCountsRepository,
   WorkflowRunRepository,
 } from '@novu/application-generic';
 import {
@@ -122,6 +123,7 @@ const ANALYTICS_PROVIDERS = [
   StepRunRepository,
   WorkflowRunRepository,
   WorkflowActivityCountsRepository,
+  ProviderVolumeCountsRepository,
 
   // Services
   clickHouseService,

libs/application-generic/src/services/analytic-logs/index.ts

@@ -1,6 +1,7 @@
 export { createClient as createClickHouseClient } from '@clickhouse/client';
 export * from './clickhouse.service';
 export * from './log.repository';
+export * from './provider-volume-counts';
 export * from './request-log';
 export { StepRun, StepRunFinalStatus, StepRunNonFinalStatus, StepRunRepository, StepRunStatus } from './step-run';
 export { EventType, mapEventTypeToTitle, Trace, TraceLogRepository, TraceStatus, traceLogSchema } from './trace-log';

libs/application-generic/src/services/analytic-logs/provider-volume-counts/index.ts

@@ -0,0 +1,2 @@
+export * from './provider-volume-counts.repository';
+export * from './provider-volume-counts.schema';

libs/application-generic/src/services/analytic-logs/provider-volume-counts/provider-volume-counts.repository.ts

@@ -0,0 +1,83 @@
+import { Injectable } from '@nestjs/common';
+import { PinoLogger } from 'nestjs-pino';
+import { FeatureFlagsService } from '../../feature-flags/feature-flags.service';
+import { ClickHouseService } from '../clickhouse.service';
+import { LogRepository } from '../log.repository';
+import {
+  PROVIDER_VOLUME_COUNTS_ORDER_BY,
+  PROVIDER_VOLUME_COUNTS_TABLE_NAME,
+  ProviderVolumeCount,
+  providerVolumeCountsSchema,
+} from './provider-volume-counts.schema';
+
+@Injectable()
+export class ProviderVolumeCountsRepository extends LogRepository<
+  typeof providerVolumeCountsSchema,
+  ProviderVolumeCount
+> {
+  public readonly table = PROVIDER_VOLUME_COUNTS_TABLE_NAME;
+  public readonly identifierPrefix = 'pvc_';
+
+  constructor(
+    protected readonly clickhouseService: ClickHouseService,
+    protected readonly logger: PinoLogger,
+    protected readonly featureFlagsService: FeatureFlagsService
+  ) {
+    super(
+      clickhouseService,
+      logger,
+      providerVolumeCountsSchema,
+      PROVIDER_VOLUME_COUNTS_ORDER_BY,
+      featureFlagsService
+    );
+    this.logger.setContext(this.constructor.name);
+  }
+
+  async getProviderVolumeData(
+    environmentId: string,
+    organizationId: string,
+    startDate: Date,
+    endDate: Date,
+    workflowIds?: string[]
+  ): Promise<Array<{ provider_id: string; count: string }>> {
+    const workflowFilter =
+      workflowIds && workflowIds.length > 0 ? 'AND workflow_id IN {workflowIds:Array(String)}' : '';
+
+    const query = `
+      SELECT 
+        provider_id,
+        sum(count) as count
+      FROM ${PROVIDER_VOLUME_COUNTS_TABLE_NAME}
+      WHERE 
+        environment_id = {environmentId:String} 
+        AND organization_id = {organizationId:String}
+        AND date >= {startDate:Date}
+        AND date <= {endDate:Date}
+        ${workflowFilter}
+      GROUP BY provider_id
+      ORDER BY count DESC
+      LIMIT 5
+    `;
+
+    const params: Record<string, unknown> = {
+      environmentId,
+      organizationId,
+      startDate: startDate.toISOString().split('T')[0],
+      endDate: endDate.toISOString().split('T')[0],
+    };
+
+    if (workflowIds && workflowIds.length > 0) {
+      params.workflowIds = workflowIds;
+    }
+
+    const result = await this.clickhouseService.query<{
+      provider_id: string;
+      count: string;
+    }>({
+      query,
+      params,
+    });
+
+    return result.data;
+  }
+}

libs/application-generic/src/services/analytic-logs/provider-volume-counts/provider-volume-counts.schema.ts

@@ -0,0 +1,38 @@
+import { CHDate, CHString, CHUInt64, ClickhouseSchema } from 'clickhouse-schema';
+
+export const PROVIDER_VOLUME_COUNTS_TABLE_NAME = 'provider_volume_counts';
+
+const schemaDefinition = {
+  date: { type: CHDate() },
+  organization_id: { type: CHString() },
+  environment_id: { type: CHString() },
+  workflow_id: { type: CHString() },
+  provider_id: { type: CHString() },
+  count: { type: CHUInt64() },
+};
+
+export const PROVIDER_VOLUME_COUNTS_ORDER_BY: (keyof typeof schemaDefinition)[] = [
+  'organization_id',
+  'environment_id',
+  'date',
+  'workflow_id',
+  'provider_id',
+];
+
+const clickhouseSchemaOptions = {
+  table_name: PROVIDER_VOLUME_COUNTS_TABLE_NAME,
+  engine: 'SummingMergeTree',
+  order_by: `(${PROVIDER_VOLUME_COUNTS_ORDER_BY.join(', ')})` as any,
+  additional_options: ['PARTITION BY toYYYYMM(date)'],
+};
+
+export const providerVolumeCountsSchema = new ClickhouseSchema(schemaDefinition, clickhouseSchemaOptions);
+
+export type ProviderVolumeCount = {
+  date: string;
+  organization_id: string;
+  environment_id: string;
+  workflow_id: string;
+  provider_id: string;
+  count: number;
+};

packages/shared/src/types/feature-flags.ts

@@ -77,6 +77,7 @@ export enum FeatureFlagsKeysEnum {
   IS_V2_MESSAGE_SENT_COUNTS_READ_ENABLED = 'IS_V2_MESSAGE_SENT_COUNTS_READ_ENABLED',
   IS_V2_SUBSCRIBER_ACTIVITY_READ_ENABLED = 'IS_V2_SUBSCRIBER_ACTIVITY_READ_ENABLED',
   IS_V2_AVG_MESSAGES_PER_SUBSCRIBER_READ_ENABLED = 'IS_V2_AVG_MESSAGES_PER_SUBSCRIBER_READ_ENABLED',
+  IS_V2_PROVIDER_VOLUME_READ_ENABLED = 'IS_V2_PROVIDER_VOLUME_READ_ENABLED',
 
   // String flags
   CF_SCHEDULER_MODE = 'CF_SCHEDULER_MODE', // Values: "off" | "shadow" | "live" | "complete"