PD-43 — Plan d'implémentation : Service upload OVH S3 multipart streaming¶

Ce document est un artefact dérivé de la spécification canonique PD-43-specification.md. Toute divergence avec la spécification constitue une anomalie à corriger dans ce plan.

1. Découpage en composants¶

1.1 Architecture modulaire¶

flowchart TB
    subgraph API["API Layer (NestJS)"]
        UC[UploadController]
        AG[AuthGuard]
        TIG[TenantIsolationGuard]
    end

    subgraph Domain["Domain Layer"]
        US[UploadSession<br/>Aggregate Root]
        UP[UploadPart<br/>Value Object]
        UVR[UploadValidationRules<br/>Value Object]
    end

    subgraph Application["Application Layer"]
        USS[UploadSessionService]
        HSS[HashStreamService]
        QS[QuotaService]
        IV[IntegrityVerifier]
        CS[ConfigService]
    end

    subgraph Infrastructure["Infrastructure Layer"]
        S3A[S3MultipartAdapter<br/>OVH S3]
        SR[SessionRepository<br/>PostgreSQL]
        AL[AuditLogger]
    end

    API --> Domain
    Domain --> Application
    Application --> Infrastructure

1.2 Description des composants¶

2. Flux techniques¶

2.1 Flux N1 — Initialiser un upload multipart¶

sequenceDiagram
    participant C as Client
    participant UC as UploadController
    participant US as UploadSessionService
    participant S3 as S3MultipartAdapter

    C->>UC: POST /uploads
    UC->>UC: AuthGuard.canActivate()
    UC->>UC: TenantIsolationGuard
    UC->>US: initSession(dto)
    US->>US: QuotaService.check()
    US->>S3: createMultipartUpload()
    S3-->>US: { uploadId }
    US->>US: SessionRepository.save()
    US-->>UC: { sessionId, rules, expiresAt }
    UC-->>C: 201 Created

Données persistées (UploadSession):

{
  id: UUID,
  tenantId: UUID,
  ownerId: UUID,
  s3UploadId: string,
  s3Key: string,
  bucket: string,
  status: 'IN_PROGRESS',
  mimeType: string,
  expectedSize?: number,
  validationRules: UploadValidationRules,
  parts: [],
  hashState: null, // État SHA3-256 sérialisé
  createdAt: Date,
  expiresAt: Date,
}

2.2 Flux N2 — Uploader une part (streaming)¶

sequenceDiagram
    participant C as Client
    participant UC as UploadController
    participant US as UploadSessionService
    participant S3 as S3MultipartAdapter

    C->>UC: PUT /uploads/:id/parts/:partNumber (Stream)
    UC->>UC: Guards (Auth, Tenant, Owner)
    UC->>US: uploadPart(id, partNum, stream)
    US->>US: SessionRepository.findById()
    US->>US: validatePartRules() + idempotence check

    rect rgb(240, 248, 255)
        Note over US,S3: Streaming Pipeline
        US->>US: PassThrough Transform
        US->>US: HashStreamService (SHA3-256 incr.)
        US->>S3: S3.uploadPart(stream)
    end

    S3-->>US: { ETag, size }
    US->>US: updatePartAndHashState()
    US-->>UC: { partNumber, etag, size }
    UC-->>C: 200 OK

Mécanisme de streaming avec hash incrémental et vérification d'idempotence:

// Pseudo-code du flux streaming
async uploadPart(sessionId: string, partNumber: number, inputStream: Readable): Promise<PartResult> {
  const session = await this.sessionRepository.findById(sessionId);
  const existingPart = session.parts.find(p => p.number === partNumber);

  // Buffer temporaire pour calcul hash de vérification (taille limitée)
  // Note: Pour respecter le streaming, on utilise un hash incrémental sur le flux entrant
  const incomingHashStream = this.hashStreamService.createPartHash();
  const passThrough = new PassThrough();

  // Tee du stream: un vers hash vérification, un vers S3 (ou buffer si part existante)
  inputStream.pipe(passThrough);
  inputStream.pipe(incomingHashStream);

  // Attendre le hash complet du contenu entrant
  const incomingContentHash = await incomingHashStream.finalizePartHash();

  // === VÉRIFICATION IDEMPOTENCE (SPEC §4 Règles d'idempotence) ===
  if (existingPart) {
    if (existingPart.contentHash === incomingContentHash) {
      // Re-soumission identique → idempotence stricte (acceptée sans effet)
      // Le stream passThrough est drainé mais non traité
      await drainStream(passThrough);
      return existingPart;
    } else {
      // Re-soumission avec contenu DIFFÉRENT → REJET EXPLICITE (E4)
      await drainStream(passThrough);
      throw new UploadPartConflictException({
        code: 'UPLOAD_PART_CONFLICT',
        message: 'Part number already exists with different content',
        details: { sessionId, partNumber },
      });
    }
  }

  // === NOUVELLE PART ===
  // Création du hash stream incrémental pour le hash global
  const globalHashStream = this.hashStreamService.createIncrementalHash(session.hashState);

  // Upload vers S3 avec hash global en parallèle
  // Note: On doit re-streamer le contenu car il a été consommé pour le hash de vérification
  // Alternative: utiliser un buffer si part < seuil, sinon stocker temporairement
  const s3Result = await this.s3Adapter.uploadPart({
    bucket: session.bucket,
    key: session.s3Key,
    uploadId: session.s3UploadId,
    partNumber,
    body: passThrough,
    contentHash: incomingContentHash, // Pour vérification S3
  });

  // Mise à jour de l'état hash global
  globalHashStream.update(/* bytes de la part */);
  session.hashState = globalHashStream.getSerializedState();

  // Mise à jour session avec part
  session.addPart({
    number: partNumber,
    etag: s3Result.ETag,
    size: s3Result.size,
    contentHash: incomingContentHash, // Hash stocké pour futures vérifications idempotence
  });

  await this.sessionRepository.save(session);

  return session.parts.find(p => p.number === partNumber);
}

Point critique — Streaming vs vérification d'idempotence:

Le calcul du hash du contenu entrant AVANT de décider si c'est un doublon pose un défi pour le streaming pur. Deux stratégies:

Recommandation: Stratégie A avec stockage temporaire sur disque si part > 10MB et part existante détectée. Sinon buffer mémoire.

2.3 Flux N3 — Compléter l'upload¶

sequenceDiagram
    participant C as Client
    participant UC as UploadController
    participant US as UploadSessionService
    participant S3 as S3MultipartAdapter
    participant IV as IntegrityVerifier

    C->>UC: POST /uploads/:id/complete
    UC->>UC: Guards
    UC->>US: completeSession(id)
    US->>US: validateCompletion() (status, expiry, parts)
    US->>S3: completeMultipartUpload()
    S3-->>US: { ETag, ChecksumSHA256 }
    US->>IV: verify()
    IV-->>US: OK
    US->>US: finalizeHash() (SHA3-256 final)
    US->>US: session.complete(hash)
    US-->>UC: { objectKey, size, hash, status: COMPLETED }
    UC-->>C: 200 OK

2.4 Flux N4 — Abandonner un upload¶

sequenceDiagram
    participant C as Client
    participant UC as UploadController
    participant US as UploadSessionService
    participant S3 as S3MultipartAdapter

    C->>UC: DELETE /uploads/:id
    UC->>UC: Guards
    UC->>US: abortSession(id)
    US->>US: validateAbort() (status not terminal)
    US->>S3: abortMultipartUpload()
    S3-->>US: { success }
    US->>US: session.abort()
    US-->>UC: { status: ABORTED }
    UC-->>C: 200 OK

3. Mapping invariants → mécanismes¶

3.1 Publication et consultabilité des paramètres normatifs (Spec §4, CA12, CA14)¶

La spécification exige que les paramètres de validation soient explicitement définis, consultables et publiés avant toute opération d'upload.

Endpoint de consultation¶

GET /api/v1/uploads/config
Authorization: Bearer <token>

Réponse:

interface UploadConfigResponse {
  // Identifiant unique de cette configuration (pour traçabilité)
  configVersion: string;           // ex: "v2024.1.15-tenant-premium"

  // Contexte d'application
  context: {
    tenantId: string;
    planId: string;                // ex: "premium", "standard"
    effectiveAt: string;           // ISO 8601
  };

  // Paramètres de validation des parts (Spec §4 Règles normatives)
  partConstraints: {
    minPartSize: number;           // bytes, ex: 5242880 (5MB)
    maxPartSize: number;           // bytes, ex: 104857600 (100MB)
    maxPartCount: number;          // ex: 10000
  };

  // Paramètres de validation fichier
  fileConstraints: {
    maxFileSize: number;           // bytes, ex: 5497558138880 (5TB)
  };

  // Paramètres de session
  sessionConstraints: {
    maxDuration: number;           // seconds, ex: 86400 (24h)
  };

  // Quotas applicables au tenant/plan
  quotas: {
    maxCumulativeSize: number | null;    // bytes, null = illimité
    maxConcurrentUploads: number | null; // null = illimité
    maxUploadsPerDay: number | null;     // null = illimité
  };

  // Référence aux valeurs par défaut normatives (Annexe D spec)
  normativeDefaults: {
    documentationUrl: string;      // Lien vers doc versionnée
    specVersion: string;           // ex: "PD-43-v1.0"
  };
}

Mécanismes de garantie (CA12, CA14)¶

Stockage et versioning¶

-- Table de configuration par tenant/plan
CREATE TABLE upload_configs (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL,
    plan_id VARCHAR(64) NOT NULL,
    config_version VARCHAR(128) NOT NULL,
    min_part_size BIGINT NOT NULL,
    max_part_size BIGINT NOT NULL,
    max_part_count INT NOT NULL,
    max_file_size BIGINT NOT NULL,
    max_session_duration INT NOT NULL,
    quota_max_cumulative_size BIGINT,
    quota_max_concurrent_uploads INT,
    quota_max_uploads_per_day INT,
    effective_at TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    CONSTRAINT uk_tenant_plan_version UNIQUE (tenant_id, plan_id, config_version)
);

-- Audit log des changements
CREATE TABLE config_audit_log (
    id UUID PRIMARY KEY,
    config_id UUID NOT NULL REFERENCES upload_configs(id),
    changed_by VARCHAR(256) NOT NULL,
    changed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    previous_values JSONB,
    new_values JSONB,
    reason VARCHAR(512)
);

Injection dans la session¶

À l'initialisation d'une session (Flux N1), les règles actives sont :

1. Chargées depuis ConfigService.getActiveConfig(tenantId, planId)
2. Copiées dans UploadSession.validationRules (snapshot immuable)
3. Utilisées pour toutes les validations de cette session

Cela garantit que les règles restent constantes pendant toute la durée de vie de la session, même si la configuration globale change entre-temps.

4. Gestion des erreurs¶

4.1 Codes d'erreur normalisés¶

4.2 Format de réponse d'erreur¶

interface UploadErrorResponse {
  code: string;           // Code stable (ex: UPLOAD_PART_INVALID)
  message: string;        // Message non sensible
  details?: {
    sessionId?: string;   // ID session si applicable
    partNumber?: number;  // Numéro part si applicable
    constraint?: string;  // Règle violée si applicable
  };
  timestamp: string;      // ISO 8601
  requestId: string;      // Correlation ID
}

4.3 Stratégies de récupération¶

5. Impacts sécurité¶

5.1 Matrice des menaces et mitigations¶

5.2 Logging sécurisé¶

Champs autorisés dans les logs: - sessionId, tenantId, userId - partNumber, partSize, totalSize - status, errorCode - hash (empreinte finale) - timestamp, requestId, duration

Champs interdits: - Contenu du fichier (bytes) - Noms de fichiers (potentiellement sensibles) - Headers contenant des tokens (masqués) - Stack traces avec paramètres

5.3 Contrôle d'accès¶

// Hiérarchie des guards
@UseGuards(AuthGuard)           // 1. Authentification valide
@UseGuards(TenantIsolationGuard) // 2. Appartenance au tenant de la session
@UseGuards(SessionOwnerGuard)    // 3. Propriétaire de la session (par défaut)

Règle de délégation (future): - Si politique de délégation activée → SessionOwnerGuard vérifie aussi les délégations explicites - Absence de politique = owner only (comportement par défaut normatif)

6. Hypothèses techniques¶

HT-1. Stack technique backend¶

• Framework: NestJS (TypeScript)
• Runtime: Node.js 20+ (support streams natif)
• ORM: TypeORM avec PostgreSQL
• SDK S3: AWS SDK v3 (compatible OVH)

HT-2. Capacités OVH S3¶

• Support complet de l'API Multipart Upload S3
• Support des checksums via x-amz-checksum-sha256 ou équivalent
• Limites OVH: 10 000 parts max, 5GB par part, 5TB par objet

HT-3. Persistance de l'état de hash¶

• L'état interne SHA3-256 (Keccak) peut être sérialisé/désérialisé
• Utilisation de js-sha3 ou implémentation custom avec état exportable
• Stockage état en base (BYTEA PostgreSQL) entre parts

HT-4. Streams Node.js¶

• PassThrough permet de dupliquer un stream pour hash + S3
• Backpressure géré automatiquement par pipeline
• Chunk size par défaut: 64KB (configurable)

HT-5. Transactions et concurrence¶

• PostgreSQL supporte les transactions ACID
• Pas de concurrence sur même session (single owner)
• Lock optimiste via version column si nécessaire

HT-6. Configuration OVH S3¶

• Credentials via variables d'environnement ou Vault
• Endpoint OVH: s3.<region>.cloud.ovh.net
• Bucket pré-créé et configuré (hors scope)

7. Points de vigilance¶

PV-1. Sérialisation état SHA3-256¶

Risque: La plupart des implémentations SHA3 ne permettent pas de sérialiser l'état intermédiaire.

Mitigation: - Option A: Implémenter SHA3-256 avec état exportable (Keccak sponge state) - Option B: Stocker hash de chaque part + recalcul final via XOR ou tree hash - Option C: Utiliser bibliothèque supportant getState()/setState() (ex: implémentation custom)

Recommandation: Option A avec implémentation Keccak pure JS modifiée.

PV-2. Cohérence état en cas de crash¶

Risque: Crash entre upload S3 et persistance DB = état incohérent.

Mitigation: 1. Transaction DB autour de chaque opération 2. Pattern "write-ahead": persister intention avant action S3 3. Job de réconciliation périodique (list S3 parts vs DB)

PV-3. Vérification intégrité S3¶

Risque: OVH S3 peut ne pas supporter tous les checksums natifs AWS.

Mitigation: 1. Vérifier support ChecksumSHA256 sur OVH 2. Fallback: télécharger et recalculer hash après complétion (coûteux) 3. Alternative: stocker MD5 parts (ETag) + vérifier cohérence liste

Recommandation: Tester en environnement OVH. Prévoir fallback.

PV-4. Gestion mémoire streaming¶

Risque: Accumulation de buffers si backpressure mal géré.

Mitigation: 1. Utiliser pipeline() qui gère backpressure 2. Limiter highWaterMark des streams 3. Monitorer heap usage en production

PV-5. Expiration et nettoyage¶

Risque: Sessions expirées avec parts orphelines sur S3 = coûts.

Mitigation: 1. Job CRON de nettoyage sessions expirées 2. Appel AbortMultipartUpload S3 pour chaque session expirée 3. Lifecycle rule S3 en backup (abort incomplete uploads > 7j)

PV-6. Taille des parts et performance¶

Risque: Parts trop petites = overhead réseau. Parts trop grandes = risque timeout.

Mitigation: 1. Taille minimale recommandée: 5MB (sauf dernière part) 2. Taille maximale: 100MB (compromis) 3. Documentation des contraintes pour clients

PV-7. Ordre des parts pour hash¶

Risque: Hash calculé dans mauvais ordre = empreinte incorrecte.

Mitigation: 1. Stocker parts avec numéro explicite 2. À la complétion, trier parts par numéro 3. Concaténer hashs partiels dans l'ordre (si tree hash) ou recalculer

PV-8. Rollback complétion S3¶

Risque: S3 CompleteMultipartUpload n'est pas annulable.

Mitigation: 1. Vérifier intégrité AVANT complétion impossible (parts non accessibles) 2. Vérifier APRÈS complétion, marquer FAILED si KO 3. Objet S3 existant mais non référencé = orphelin (cleanup batch)

Annexe A — Schéma de données¶

-- Table principale des sessions
CREATE TABLE upload_sessions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL,
    owner_id UUID NOT NULL,
    s3_upload_id VARCHAR(256) NOT NULL,
    s3_bucket VARCHAR(128) NOT NULL,
    s3_key VARCHAR(512) NOT NULL,
    status VARCHAR(32) NOT NULL DEFAULT 'IN_PROGRESS',
    mime_type VARCHAR(128),
    expected_size BIGINT,
    final_size BIGINT,
    final_hash VARCHAR(64), -- SHA3-256 hex
    hash_state BYTEA, -- État Keccak sérialisé
    validation_rules JSONB NOT NULL,
    expires_at TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    completed_at TIMESTAMPTZ,

    CONSTRAINT chk_status CHECK (status IN ('IN_PROGRESS', 'COMPLETED', 'ABORTED', 'FAILED', 'EXPIRED'))
);

-- Table des parts
CREATE TABLE upload_parts (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    session_id UUID NOT NULL REFERENCES upload_sessions(id) ON DELETE CASCADE,
    part_number INT NOT NULL,
    size BIGINT NOT NULL,
    etag VARCHAR(128) NOT NULL,
    content_hash VARCHAR(64), -- Hash SHA3-256 de cette part seule
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    CONSTRAINT uk_session_part UNIQUE (session_id, part_number),
    CONSTRAINT chk_part_number CHECK (part_number >= 1 AND part_number <= 10000)
);

-- Index pour requêtes fréquentes
CREATE INDEX idx_sessions_tenant ON upload_sessions(tenant_id);
CREATE INDEX idx_sessions_owner ON upload_sessions(owner_id);
CREATE INDEX idx_sessions_status ON upload_sessions(status) WHERE status = 'IN_PROGRESS';
CREATE INDEX idx_sessions_expires ON upload_sessions(expires_at) WHERE status = 'IN_PROGRESS';

Annexe B — Contrat API (OpenAPI summary)¶

Annexe C — Dépendances NPM¶

{
  "@aws-sdk/client-s3": "^3.x",
  "@aws-sdk/lib-storage": "^3.x",
  "js-sha3": "^0.9.x",
  "uuid": "^9.x",
  "class-validator": "^0.14.x",
  "class-transformer": "^0.5.x"
}

Annexe D — Valeurs par défaut normatives (référence spec)¶