feat(analytic-logs): Implement feature flag logic for message deliver…

…y data retrieval

apps/api/src/app/activity/usecases/build-messages-delivered-chart/build-messages-delivered-chart.usecase.ts

@@ -1,12 +1,21 @@
 import { Injectable } from '@nestjs/common';
-import { InstrumentUsecase, MessageSentCountsRepository, PinoLogger } from '@novu/application-generic';
+import {
+  FeatureFlagsService,
+  InstrumentUsecase,
+  MessageSentCountsRepository,
+  PinoLogger,
+  StepRunRepository,
+} from '@novu/application-generic';
+import { FeatureFlagsKeysEnum } from '@novu/shared';
 import { MessagesDeliveredDataPointDto } from '../../dtos/get-charts.response.dto';
 import { BuildMessagesDeliveredChartCommand } from './build-messages-delivered-chart.command';
 
 @Injectable()
 export class BuildMessagesDeliveredChart {
   constructor(
     private messageSentCountsRepository: MessageSentCountsRepository,
+    private stepRunRepository: StepRunRepository,
+    private featureFlagsService: FeatureFlagsService,
     private logger: PinoLogger
   ) {
     this.logger.setContext(BuildMessagesDeliveredChart.name);
@@ -20,15 +29,45 @@ export class BuildMessagesDeliveredChart {
     const previousEndDate = new Date(startDate.getTime() - 1);
     const previousStartDate = new Date(previousEndDate.getTime() - periodDuration);
 
-    const result = await this.messageSentCountsRepository.getMessagesSentData(
-      environmentId,
-      organizationId,
-      startDate,
-      endDate,
-      previousStartDate,
-      previousEndDate,
-      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_MESSAGE_SENT_COUNTS_READ_ENABLED,
+        defaultValue: false,
+        ...featureFlagContext,
+      }),
+    ]);
+
+    const useNewLogic = isGlobalEnabled || isDedicatedEnabled;
+
+    const result = useNewLogic
+      ? await this.messageSentCountsRepository.getMessagesSentData(
+          environmentId,
+          organizationId,
+          startDate,
+          endDate,
+          previousStartDate,
+          previousEndDate,
+          workflowIds
+        )
+      : await this.stepRunRepository.getMessagesDeliveredData(
+          environmentId,
+          organizationId,
+          startDate,
+          endDate,
+          previousStartDate,
+          previousEndDate,
+          workflowIds
+        );
 
     return {
       currentPeriod: result.currentPeriod,

libs/application-generic/src/services/analytic-logs/step-run/step-run.repository.ts

@@ -214,6 +214,83 @@ export class StepRunRepository extends LogRepository<typeof stepRunSchema, StepR
     return result.data;
   }
 
+  async getMessagesDeliveredData(
+    environmentId: string,
+    organizationId: string,
+    startDate: Date,
+    endDate: Date,
+    previousStartDate: Date,
+    previousEndDate: Date,
+    workflowIds?: string[]
+  ): Promise<{ currentPeriod: number; previousPeriod: number }> {
+    const workflowFilter =
+      workflowIds && workflowIds.length > 0 ? `AND workflow_id IN {workflowIds:Array(String)}` : '';
+
+    // Query for current period
+    const currentPeriodQuery = `
+      SELECT count(*) as count
+      FROM step_runs FINAL
+      WHERE 
+        environment_id = {environmentId:String} 
+        AND organization_id = {organizationId:String}
+        AND created_at >= {startDate:DateTime64(3)}
+        AND created_at <= {endDate:DateTime64(3)}
+        AND step_type IN ('in_app', 'email', 'sms', 'chat', 'push')
+        AND status = 'completed'
+        ${workflowFilter}
+    `;
+
+    // Query for previous period
+    const previousPeriodQuery = `
+      SELECT count(*) as count
+      FROM step_runs FINAL
+      WHERE 
+        environment_id = {environmentId:String} 
+        AND organization_id = {organizationId:String}
+        AND created_at >= {previousStartDate:DateTime64(3)}
+        AND created_at <= {previousEndDate:DateTime64(3)}
+        AND step_type IN ('in_app', 'email', 'sms', 'chat', 'push')
+        AND status = 'completed'
+        ${workflowFilter}
+    `;
+
+    const baseParams: Record<string, unknown> = {
+      environmentId,
+      organizationId,
+    };
+
+    if (workflowIds && workflowIds.length > 0) {
+      baseParams.workflowIds = workflowIds;
+    }
+
+    const [currentResult, previousResult] = await Promise.all([
+      this.clickhouseService.query<{ count: string }>({
+        query: currentPeriodQuery,
+        params: {
+          ...baseParams,
+          startDate: LogRepository.formatDateTime64(startDate),
+          endDate: LogRepository.formatDateTime64(endDate),
+        },
+      }),
+      this.clickhouseService.query<{ count: string }>({
+        query: previousPeriodQuery,
+        params: {
+          ...baseParams,
+          previousStartDate: LogRepository.formatDateTime64(previousStartDate),
+          previousEndDate: LogRepository.formatDateTime64(previousEndDate),
+        },
+      }),
+    ]);
+
+    const currentPeriod = parseInt(currentResult.data[0]?.count || '0', 10);
+    const previousPeriod = parseInt(previousResult.data[0]?.count || '0', 10);
+
+    return {
+      currentPeriod,
+      previousPeriod,
+    };
+  }
+
   async getAvgMessagesPerSubscriberData(
     environmentId: string,
     organizationId: string,

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

@@ -73,6 +73,8 @@ export enum FeatureFlagsKeysEnum {
   IS_CONTEXTUAL_HELP_DRAWER_ENABLED = 'IS_CONTEXTUAL_HELP_DRAWER_ENABLED',
   IS_SUBSCRIPTION_PREFERENCES_ENABLED = 'IS_SUBSCRIPTION_PREFERENCES_ENABLED',
   IS_LRU_CACHE_ENABLED = 'IS_LRU_CACHE_ENABLED',
+  IS_ANALYTIC_V2_LOGS_READ_GLOBAL_ENABLED = 'IS_ANALYTIC_V2_LOGS_READ_GLOBAL_ENABLED',
+  IS_V2_MESSAGE_SENT_COUNTS_READ_ENABLED = 'IS_V2_MESSAGE_SENT_COUNTS_READ_ENABLED',
 
   // String flags
   CF_SCHEDULER_MODE = 'CF_SCHEDULER_MODE', // Values: "off" | "shadow" | "live" | "complete"