feat: read model rebuild data ingestion metrics

apps/backend/src/app.module.ts

@@ -68,6 +68,7 @@ import { ReviewMetricsModule } from './review-metrics/review-metrics.module';
 import { BotAuthModule } from './bot-auth/bot-auth.module';
 import { DemoBootstrapModule } from './demo-bootstrap/demo-bootstrap.module';
 import { ContributorFeedModule } from './contributor-feed/contributor-feed.module';
+import { ReadModelRebuildModule } from './read-model-rebuild';
 
 @Module({
   imports: [
@@ -145,6 +146,7 @@ import { ContributorFeedModule } from './contributor-feed/contributor-feed.modul
     BotAuthModule,
     DemoBootstrapModule,
     ContributorFeedModule,
+    ReadModelRebuildModule,
   ],
   controllers: [AppController, TestController, TestExceptionController],
   providers: [

apps/backend/src/database/migrations/1820000000000-CreateReadModelRebuildJobs.ts

@@ -0,0 +1,179 @@
+import { MigrationInterface, QueryRunner, Table, TableIndex } from 'typeorm';
+
+export class CreateReadModelRebuildJobs1820000000000 implements MigrationInterface {
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.createTable(
+      new Table({
+        name: 'read_model_rebuild_jobs',
+        columns: [
+          {
+            name: 'id',
+            type: 'uuid',
+            isPrimary: true,
+            default: 'gen_random_uuid()',
+          },
+          {
+            name: 'dataset',
+            type: 'enum',
+            enum: [
+              'kpi_snapshots',
+              'project_views',
+              'contract_events',
+              'daily_metrics',
+              'all',
+            ],
+            isNullable: false,
+          },
+          {
+            name: 'contract_id',
+            type: 'varchar',
+            length: '255',
+            isNullable: true,
+          },
+          {
+            name: 'status',
+            type: 'enum',
+            enum: [
+              'pending',
+              'in_progress',
+              'completed',
+              'failed',
+              'cancelled',
+            ],
+            default: "'pending'",
+          },
+          {
+            name: 'trigger_reason',
+            type: 'text',
+            isNullable: true,
+          },
+          {
+            name: 'triggered_by',
+            type: 'varchar',
+            length: '255',
+            isNullable: true,
+          },
+          {
+            name: 'total_items',
+            type: 'int',
+            default: 0,
+          },
+          {
+            name: 'processed_items',
+            type: 'int',
+            default: 0,
+          },
+          {
+            name: 'failed_items',
+            type: 'int',
+            default: 0,
+          },
+          {
+            name: 'progress_details',
+            type: 'jsonb',
+            isNullable: true,
+          },
+          {
+            name: 'error_message',
+            type: 'text',
+            isNullable: true,
+          },
+          {
+            name: 'error_stack',
+            type: 'text',
+            isNullable: true,
+          },
+          {
+            name: 'started_at',
+            type: 'timestamp with time zone',
+            isNullable: true,
+          },
+          {
+            name: 'completed_at',
+            type: 'timestamp with time zone',
+            isNullable: true,
+          },
+          {
+            name: 'idempotency_key',
+            type: 'varchar',
+            length: '255',
+            isNullable: true,
+          },
+          {
+            name: 'rebuild_version',
+            type: 'varchar',
+            length: '50',
+            isNullable: true,
+          },
+          {
+            name: 'created_at',
+            type: 'timestamp with time zone',
+            default: 'CURRENT_TIMESTAMP',
+          },
+          {
+            name: 'updated_at',
+            type: 'timestamp with time zone',
+            default: 'CURRENT_TIMESTAMP',
+            onUpdate: 'CURRENT_TIMESTAMP',
+          },
+        ],
+      }),
+      true,
+    );
+
+    // Create indexes
+    await queryRunner.createIndex(
+      'read_model_rebuild_jobs',
+      new TableIndex({
+        name: 'idx_rebuild_jobs_status_created',
+        columnNames: ['status', 'created_at'],
+      }),
+    );
+
+    await queryRunner.createIndex(
+      'read_model_rebuild_jobs',
+      new TableIndex({
+        name: 'idx_rebuild_jobs_dataset_status',
+        columnNames: ['dataset', 'status'],
+      }),
+    );
+
+    await queryRunner.createIndex(
+      'read_model_rebuild_jobs',
+      new TableIndex({
+        name: 'idx_rebuild_jobs_contract_status',
+        columnNames: ['contract_id', 'status'],
+      }),
+    );
+
+    await queryRunner.createIndex(
+      'read_model_rebuild_jobs',
+      new TableIndex({
+        name: 'idx_rebuild_jobs_idempotency_key',
+        columnNames: ['idempotency_key'],
+        isUnique: true,
+        where: '"idempotency_key" IS NOT NULL',
+      }),
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.dropIndex(
+      'read_model_rebuild_jobs',
+      'idx_rebuild_jobs_idempotency_key',
+    );
+    await queryRunner.dropIndex(
+      'read_model_rebuild_jobs',
+      'idx_rebuild_jobs_contract_status',
+    );
+    await queryRunner.dropIndex(
+      'read_model_rebuild_jobs',
+      'idx_rebuild_jobs_dataset_status',
+    );
+    await queryRunner.dropIndex(
+      'read_model_rebuild_jobs',
+      'idx_rebuild_jobs_status_created',
+    );
+    await queryRunner.dropTable('read_model_rebuild_jobs');
+  }
+}

apps/backend/src/health/latency-budget.health.service.ts

@@ -72,8 +72,7 @@ export class LatencyBudgetHealthService {
    */
   async getLatencyBudgetReport(): Promise<LatencyBudgetReport> {
     const network = (process.env.STELLAR_NETWORK ?? 'testnet') as
-      | 'testnet'
-      | 'mainnet';
+      'testnet' | 'mainnet';
 
     const horizonUrl =
       process.env.STELLAR_HORIZON_URL ?? DEFAULT_HORIZON_URLS[network];

apps/backend/src/portfolio/dto/portfolio-snapshot.dto.ts

@@ -132,11 +132,7 @@ export class PortfolioSnapshotBatchStatusDto {
     example: 'running',
   })
   status:
-    | 'queued'
-    | 'running'
-    | 'completed'
-    | 'completed_with_errors'
-    | 'failed';
+    'queued' | 'running' | 'completed' | 'completed_with_errors' | 'failed';
 
   @ApiProperty({
     description: 'Total users scheduled for snapshot generation',
@@ -208,11 +204,7 @@ export class TriggerSnapshotBatchResponseDto {
     example: 'queued',
   })
   status:
-    | 'queued'
-    | 'running'
-    | 'completed'
-    | 'completed_with_errors'
-    | 'failed';
+    'queued' | 'running' | 'completed' | 'completed_with_errors' | 'failed';
 
   @ApiProperty({
     description: 'Total users scheduled for snapshot generation',

apps/backend/src/portfolio/queue/portfolio-snapshot.types.ts

@@ -13,11 +13,7 @@ export interface PortfolioSnapshotUserJobData {
 export interface PortfolioSnapshotBatchStatus {
   batchId: string;
   status:
-    | 'queued'
-    | 'running'
-    | 'completed'
-    | 'completed_with_errors'
-    | 'failed';
+    'queued' | 'running' | 'completed' | 'completed_with_errors' | 'failed';
   total: number;
   completed: number;
   failed: number;

apps/backend/src/read-model-rebuild/dto/rebuild-request.dto.ts

@@ -0,0 +1,48 @@
+import { ApiProperty } from '@nestjs/swagger';
+import { IsEnum, IsOptional, IsString } from 'class-validator';
+import { RebuildDataset } from '../entities/read-model-rebuild-job.entity';
+
+export class RebuildRequestDto {
+  @ApiProperty({
+    description: 'Dataset to rebuild',
+    enum: RebuildDataset,
+    example: RebuildDataset.KPI_SNAPSHOTS,
+  })
+  @IsEnum(RebuildDataset)
+  dataset: RebuildDataset;
+
+  @ApiProperty({
+    description: 'Optional contract ID to scope the rebuild',
+    required: false,
+    example: 'CBBQW7T65XBDPIPXEIIPJVJEEIBSPC566HMEU2LTBAULLKCNUFRFBKRO',
+  })
+  @IsOptional()
+  @IsString()
+  contractId?: string;
+
+  @ApiProperty({
+    description: 'Reason for triggering the rebuild (for audit)',
+    required: false,
+    example: 'KPI computation logic updated to handle corrections',
+  })
+  @IsOptional()
+  @IsString()
+  reason?: string;
+
+  @ApiProperty({
+    description: 'Idempotency key to prevent duplicate rebuilds',
+    required: false,
+    example: 'rebuild-kpi-snapshots-2024-01-01',
+  })
+  @IsOptional()
+  @IsString()
+  idempotencyKey?: string;
+
+  @ApiProperty({
+    description: 'Whether to force rebuild even if in progress',
+    required: false,
+    default: false,
+  })
+  @IsOptional()
+  force?: boolean;
+}

apps/backend/src/read-model-rebuild/dto/rebuild-response.dto.ts

@@ -0,0 +1,95 @@
+import { ApiProperty } from '@nestjs/swagger';
+import {
+  RebuildStatus,
+  RebuildDataset,
+} from '../entities/read-model-rebuild-job.entity';
+
+export class RebuildResponseDto {
+  @ApiProperty({ description: 'Job ID for tracking' })
+  id: string;
+
+  @ApiProperty({ enum: RebuildDataset })
+  dataset: RebuildDataset;
+
+  @ApiProperty({ nullable: true })
+  contractId: string | null;
+
+  @ApiProperty({ enum: RebuildStatus })
+  status: RebuildStatus;
+
+  @ApiProperty({ nullable: true })
+  triggerReason: string | null;
+
+  @ApiProperty({ nullable: true })
+  triggeredBy: string | null;
+
+  @ApiProperty()
+  totalItems: number;
+
+  @ApiProperty()
+  processedItems: number;
+
+  @ApiProperty()
+  failedItems: number;
+
+  @ApiProperty({ nullable: true })
+  progressDetails: Record<string, unknown> | null;
+
+  @ApiProperty({ nullable: true })
+  errorMessage: string | null;
+
+  @ApiProperty({ nullable: true })
+  startedAt: Date | null;
+
+  @ApiProperty({ nullable: true })
+  completedAt: Date | null;
+
+  @ApiProperty()
+  createdAt: Date;
+
+  @ApiProperty()
+  updatedAt: Date;
+
+  @ApiProperty({ description: 'Progress percentage (0-100)' })
+  get progressPercentage(): number {
+    if (this.totalItems === 0) return 0;
+    return Math.round((this.processedItems / this.totalItems) * 100);
+  }
+
+  @ApiProperty({ description: 'Estimated time remaining in seconds' })
+  get estimatedTimeRemaining(): number | null {
+    if (this.status !== RebuildStatus.IN_PROGRESS || !this.startedAt) {
+      return null;
+    }
+    const elapsed = Date.now() - this.startedAt.getTime();
+    if (this.processedItems === 0) return null;
+    const avgTimePerItem = elapsed / this.processedItems;
+    const remaining = this.totalItems - this.processedItems;
+    return Math.round((avgTimePerItem * remaining) / 1000);
+  }
+}
+
+export class RebuildStatusResponseDto extends RebuildResponseDto {
+  @ApiProperty({ description: 'Whether the job is in a terminal state' })
+  isTerminal: boolean;
+
+  @ApiProperty({ description: 'Human-readable status message' })
+  statusMessage: string;
+}
+
+export class RebuildTriggerResponseDto {
+  @ApiProperty({ description: 'Job ID for tracking' })
+  jobId: string;
+
+  @ApiProperty({ description: 'Status of the trigger request' })
+  status: string;
+
+  @ApiProperty({ description: 'Message describing the result' })
+  message: string;
+
+  @ApiProperty({ description: 'Whether a new rebuild was started' })
+  started: boolean;
+
+  @ApiProperty({ description: 'Reference to existing job if duplicate' })
+  existingJobId?: string;
+}