ProbatioVault - Spécification Worker Réplication¶

import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bull';
import { TypeOrmModule } from '@nestjs/typeorm';

import { ReplicationStatus } from './entities/replication-status.entity';
import { ReplicationWorkerService } from './services/replication-worker.service';
import { ReplicationProcessor } from './processors/replication.processor';

@Module({
  imports: [
    // Bull Queue
    BullModule.registerQueue({
      name: 'replication',
      defaultJobOptions: {
        attempts: 3,
        backoff: {
          type: 'exponential',
          delay: 5000, // 5s, 25s, 125s
        },
        removeOnComplete: 100, // Garder 100 derniers jobs réussis
        removeOnFail: 1000,    // Garder 1000 derniers jobs échoués
      },
    }),

    // TypeORM
    TypeOrmModule.forFeature([ReplicationStatus]),
  ],
  providers: [
    ReplicationWorkerService,
    ReplicationProcessor,
  ],
  exports: [ReplicationWorkerService],
})
export class ReplicationWorkerModule {}

import { Injectable, Logger } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bull';
import { Queue } from 'bull';
import { Cron, CronExpression } from '@nestjs/schedule';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { S3Client, ListObjectsV2Command } from '@aws-sdk/client-s3';

import { ReplicationStatus } from '../entities/replication-status.entity';

@Injectable()
export class ReplicationWorkerService {
  private readonly logger = new Logger(ReplicationWorkerService.name);

  private ovhS3Client: S3Client;
  private awsS3Client: S3Client;

  constructor(
    @InjectQueue('replication') private replicationQueue: Queue,
    @InjectRepository(ReplicationStatus)
    private replicationRepo: Repository<ReplicationStatus>,
  ) {
    // Init clients S3
    this.ovhS3Client = new S3Client({
      region: process.env.OVH_S3_REGION,
      endpoint: process.env.OVH_S3_ENDPOINT,
      credentials: {
        accessKeyId: process.env.OVH_S3_ACCESS_KEY_ID,
        secretAccessKey: process.env.OVH_S3_SECRET_ACCESS_KEY,
      },
    });

    this.awsS3Client = new S3Client({
      region: process.env.AWS_S3_REGION,
      credentials: {
        accessKeyId: process.env.AWS_S3_ACCESS_KEY_ID,
        secretAccessKey: process.env.AWS_S3_SECRET_ACCESS_KEY,
      },
    });
  }

  /**
   * Cron: Détecte nouveaux fichiers OVH toutes les 5 minutes
   */
  @Cron(CronExpression.EVERY_5_MINUTES)
  async scanNewDocuments() {
    this.logger.log('Starting scan for new documents on OVH S3...');

    try {
      // Liste tous les objets OVH S3
      const listCommand = new ListObjectsV2Command({
        Bucket: process.env.OVH_S3_BUCKET_DOCUMENTS,
        MaxKeys: 1000, // Traiter par batch de 1000
      });

      const response = await this.ovhS3Client.send(listCommand);

      if (!response.Contents || response.Contents.length === 0) {
        this.logger.log('No objects found in OVH S3');
        return;
      }

      // Filtrer objets non-répliqués
      const unreplicatedKeys = await this.filterUnreplicated(
        response.Contents.map((obj) => obj.Key),
      );

      this.logger.log(`Found ${unreplicatedKeys.length} unreplicated objects`);

      // Enqueue jobs
      for (const key of unreplicatedKeys) {
        await this.replicationQueue.add('replicate-document', {
          s3_key: key,
          bucket_source: process.env.OVH_S3_BUCKET_DOCUMENTS,
          bucket_dest: process.env.AWS_S3_BUCKET_DOCUMENTS_COLD,
        });
      }

      this.logger.log(`Enqueued ${unreplicatedKeys.length} replication jobs`);
    } catch (error) {
      this.logger.error('Error scanning OVH S3', error.stack);
    }
  }

  /**
   * Filtre les clés S3 non-répliquées (pas dans table replication_status)
   */
  private async filterUnreplicated(keys: string[]): Promise<string[]> {
    const replicated = await this.replicationRepo.find({
      where: {
        s3_key: In(keys),
        status: 'replicated',
      },
      select: ['s3_key'],
    });

    const replicatedKeys = new Set(replicated.map((r) => r.s3_key));

    return keys.filter((key) => !replicatedKeys.has(key));
  }
}

import { Process, Processor } from '@nestjs/bull';
import { Logger } from '@nestjs/common';
import { Job } from 'bull';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import {
  S3Client,
  GetObjectCommand,
  PutObjectCommand,
  HeadObjectCommand,
} from '@aws-sdk/client-s3';

import { ReplicationStatus } from '../entities/replication-status.entity';

@Processor('replication')
export class ReplicationProcessor {
  private readonly logger = new Logger(ReplicationProcessor.name);

  private ovhS3Client: S3Client;
  private awsS3Client: S3Client;

  constructor(
    @InjectRepository(ReplicationStatus)
    private replicationRepo: Repository<ReplicationStatus>,
  ) {
    // Init clients S3 (même config que service)
    this.ovhS3Client = new S3Client({ /* ... */ });
    this.awsS3Client = new S3Client({ /* ... */ });
  }

  @Process('replicate-document')
  async handleReplication(job: Job) {
    const { s3_key, bucket_source, bucket_dest } = job.data;

    this.logger.log(`Processing replication for ${s3_key}`);

    try {
      // 1. Créer entrée replication_status (status = 'in_progress')
      let replicationStatus = await this.replicationRepo.findOne({
        where: { s3_key },
      });

      if (!replicationStatus) {
        replicationStatus = this.replicationRepo.create({
          s3_key,
          bucket_source,
          bucket_dest,
          status: 'in_progress',
        });
      } else {
        replicationStatus.status = 'in_progress';
        replicationStatus.retry_count += 1;
      }

      await this.replicationRepo.save(replicationStatus);

      // 2. Download depuis OVH S3
      const getCommand = new GetObjectCommand({
        Bucket: bucket_source,
        Key: s3_key,
      });

      const ovhResponse = await this.ovhS3Client.send(getCommand);
      const fileBuffer = await this.streamToBuffer(ovhResponse.Body);

      // 3. Parse metadata (tags, document type)
      const metadata = ovhResponse.Metadata || {};
      const documentType = metadata['document-type'] || 'general';

      // 4. Compute retention (basé sur type document)
      const retentionYears = this.getRetentionYears(documentType);
      const retainUntilDate = new Date();
      retainUntilDate.setFullYear(retainUntilDate.getFullYear() + retentionYears);

      // 5. Upload vers AWS S3 avec Object Lock
      const putCommand = new PutObjectCommand({
        Bucket: bucket_dest,
        Key: s3_key,
        Body: fileBuffer,
        Metadata: metadata,
        ObjectLockMode: 'COMPLIANCE',
        ObjectLockRetainUntilDate: retainUntilDate,
        Tagging: `DocumentType=${documentType}&RetentionYears=${retentionYears}`,
      });

      const awsResponse = await this.awsS3Client.send(putCommand);

      // 6. Vérifier intégrité (compare ETags)
      const headCommand = new HeadObjectCommand({
        Bucket: bucket_dest,
        Key: s3_key,
      });

      const awsHead = await this.awsS3Client.send(headCommand);

      if (awsHead.ETag !== ovhResponse.ETag) {
        throw new Error('ETag mismatch: integrity check failed');
      }

      // 7. Update replication_status (success)
      replicationStatus.status = 'replicated';
      replicationStatus.replicated_at = new Date();
      replicationStatus.aws_version_id = awsResponse.VersionId;
      replicationStatus.aws_etag = awsHead.ETag;
      replicationStatus.aws_object_lock_until = retainUntilDate;
      replicationStatus.ovh_etag = ovhResponse.ETag;
      replicationStatus.ovh_size = fileBuffer.length;

      await this.replicationRepo.save(replicationStatus);

      this.logger.log(`Replication successful for ${s3_key}`);

      // 8. (Optionnel) Supprimer de OVH si configuré
      if (process.env.DELETE_FROM_OVH_AFTER_REPLICATION === 'true') {
        // TODO: Delete from OVH (PD-XX)
      }
    } catch (error) {
      this.logger.error(`Replication failed for ${s3_key}`, error.stack);

      // Update status = 'failed'
      await this.replicationRepo.update(
        { s3_key },
        {
          status: 'failed',
          failed_at: new Date(),
          error_message: error.message,
        },
      );

      throw error; // Retry automatique (Bull)
    }
  }

  /**
   * Détermine durée rétention basée sur type document
   */
  private getRetentionYears(documentType: string): number {
    const retentionMap = {
      payslip: 50,   // Bulletins de paie (Code travail)
      invoice: 10,   // Factures (Code commerce)
      contract: 10,  // Contrats
      general: 50,   // Par défaut (sécurité maximale)
    };

    return retentionMap[documentType] || 50;
  }

  /**
   * Convert stream to buffer
   */
  private async streamToBuffer(stream: any): Promise<Buffer> {
    const chunks = [];
    for await (const chunk of stream) {
      chunks.push(chunk);
    }
    return Buffer.concat(chunks);
  }
}

describe('ReplicationProcessor', () => {
  let processor: ReplicationProcessor;
  let replicationRepo: MockRepository<ReplicationStatus>;
  let ovhS3Client: jest.Mocked<S3Client>;
  let awsS3Client: jest.Mocked<S3Client>;

  beforeEach(() => {
    // Mock repository et clients S3
    replicationRepo = createMockRepository();
    ovhS3Client = createMockS3Client();
    awsS3Client = createMockS3Client();

    processor = new ReplicationProcessor(replicationRepo);
    processor['ovhS3Client'] = ovhS3Client;
    processor['awsS3Client'] = awsS3Client;
  });

  it('should replicate document successfully', async () => {
    const job = {
      data: {
        s3_key: 'test-document.pdf.enc',
        bucket_source: 'documents-hot',
        bucket_dest: 'documents-cold',
      },
    };

    // Mock OVH GetObject
    ovhS3Client.send.mockResolvedValueOnce({
      Body: Buffer.from('encrypted content'),
      ETag: '"abc123"',
      Metadata: { 'document-type': 'payslip' },
    });

    // Mock AWS PutObject
    awsS3Client.send.mockResolvedValueOnce({
      VersionId: 'v1',
      ETag: '"abc123"',
    });

    // Mock AWS HeadObject
    awsS3Client.send.mockResolvedValueOnce({
      ETag: '"abc123"',
    });

    await processor.handleReplication(job);

    // Verify replication status saved
    expect(replicationRepo.save).toHaveBeenCalledWith(
      expect.objectContaining({
        status: 'replicated',
        aws_version_id: 'v1',
      }),
    );
  });

  it('should retry on failure', async () => {
    const job = {
      data: {
        s3_key: 'test-document.pdf.enc',
        bucket_source: 'documents-hot',
        bucket_dest: 'documents-cold',
      },
    };

    // Mock OVH GetObject failure
    ovhS3Client.send.mockRejectedValueOnce(new Error('Network error'));

    await expect(processor.handleReplication(job)).rejects.toThrow('Network error');

    // Verify status = 'failed'
    expect(replicationRepo.update).toHaveBeenCalledWith(
      { s3_key: 'test-document.pdf.enc' },
      expect.objectContaining({
        status: 'failed',
        error_message: 'Network error',
      }),
    );
  });
});

// Dashboard Bull Board (UI)
import { BullBoardModule } from 'bull-board/nestjs';

@Module({
  imports: [
    BullBoardModule.forRoot({
      route: '/admin/queues',
      adapter: ExpressAdapter,
    }),
  ],
})

// Accessible à: http://localhost:3000/admin/queues

// Logger structuré
this.logger.log({
  event: 'REPLICATION_SUCCESS',
  s3_key: key,
  duration_ms: Date.now() - startTime,
  aws_version_id: versionId,
});

// Alerte si queue > 1000 jobs
@Cron(CronExpression.EVERY_10_MINUTES)
async checkQueueSize() {
  const jobCounts = await this.replicationQueue.getJobCounts();

  if (jobCounts.waiting > 1000) {
    this.logger.error('Replication queue overloaded', { jobCounts });
    // TODO: Send alert (PD-XX)
  }
}