volontariapp-implement-async-job-flow

Guide : Implémenter un Job Asynchrone End-to-End

Dans Volontariapp, un Job représente une opération asynchrone 1 : 1 (garantie qu'un seul worker l'exécutera). Il passe par le Transactional Outbox, une file BullMQ, un Worker Runner (BaseWorker + IJobHandler), puis est audité via un Trigger SQL PostgreSQL avant son nettoyage automatique.

---

Vue d'Ensemble du Cycle de Vie d'un Job

[1. Contrat SSOT]      npm-packages/packages/messaging (JobMessagingType, JobRegistry, Queues)
         ↓
[2. Émetteur]          ms-* / domain-* (JobsOutboxEntity.createJob dans jobs_outbox)
         ↓
[3. Démon Outbox]      outbox-* runner (Pousse vers BullMQ Queue Redis)
         ↓
[4. Worker Runner]     workers-runners/worker-* (BaseWorker + IJobHandler)
         ↓
[5. Table job_audit]   Worker met à jour le statut (PROCESSING -> COMPLETED / FAILED)
         ↓
[6. SQL Trigger DB]    Déclencheur auto sur job_audit -> insère dans event_queue
         ↓
[7. Post-Processor]    Nettoyage automatique (Hard Delete de jobs_outbox)

---

[!TIP] Guides Approfondis (Deep-Dive) : Si vous êtes bloqué sur une étape précise ou avez besoin d'exemples de code exhaustifs, consultez les fiches dédiées dans references/ :

• 01-job-messaging-contracts.md (references/01-job-messaging-contracts.md) (Enums, Registres, Queues, Payloads)
• 02-emitting-jobs-and-fallbacks.md (references/02-emitting-jobs-and-fallbacks.md) (Jobs classiques vs withFallback en controller)
• 03-worker-and-handlers.md (references/03-worker-and-handlers.md) (BaseWorker, IJobHandler, Idempotence)
• 04-audit-loop-and-cleanup.md (references/04-audit-loop-and-cleanup.md) (SQL Trigger, job_audit, nettoyage automatique)

---

Étape 1 : Définir le Job dans npm-packages/packages/messaging (SSOT)

📖 Détails complets : 01-job-messaging-contracts.md (references/01-job-messaging-contracts.md)

1. Ajouter la constante de Job dans packages/messaging/src/jobs/<domaine>/payloads.ts :

   export const EventsJobType = {
     // ...
     SYNC_EXTERNAL_CALENDAR: 'event.sync_external_calendar',
   } as const;
   

2. Définir l'interface du Payload dans le même fichier :

   export interface ISyncExternalCalendarPayload {
     eventId: string;
     calendarUrl: string;
     syncRequestedBy: string;
   }
   

3. Enregistrer dans JobRegistry (packages/messaging/src/jobs/index.ts) :

   export interface JobRegistry {
     // ...
     [JobMessagingType.SYNC_EXTERNAL_CALENDAR]: ISyncExternalCalendarPayload;
   }
   

4. Vérifier ou définir la Queue BullMQ (packages/messaging/src/jobs/<domaine>/queue.ts) :

   export enum EventsQueue {
     EVENTS = 'events-queue',
     FALLBACK_EVENTS = 'fallback-events-queue',
   }
   

[!CAUTION]

🛑 POINT DE BLOCAGE CRITIQUE : LE STOP IMMÉDIAT

Tu viens de modifier npm-packages/packages/messaging ? TU DOIS T'ARRÊTER.

1. Valide la compilation locale dans npm-packages : yarn build && yarn test.
2. Génère le changeset et le bump : yarn changeset add puis yarn changeset version (une seule fois par branche).
3. STOP TOTAL : Interdiction formelle de passer à l'Étape 2 (ms-*) ou l'Étape 3 (workers-runners) immédiatement !
4. Passe la main au Lead Dev pour qu'il pousse sur une PR et que la CI publie la version snapshot <version>-snap-<sha court>, publiée sous le dist-tag next et listée en commentaire de la PR (ex: @volontariapp/messaging@2.17.1-snap-a1b2c3d).
5. Ce n'est qu'après publication par la CI que tu pourras lancer yarn up @volontariapp/messaging dans les microservices et continuer les étapes ci-dessous.

---

Étape 2 : Émettre le Job en Base (ms-* ou domain-*)

📖 Détails complets : 02-emitting-jobs-and-fallbacks.md (references/02-emitting-jobs-and-fallbacks.md)

Le job est inséré dans la table jobs_outbox lors de la transaction PostgreSQL.

Option A : Job Standard (Opération de fond planifiée)

import { JobsOutboxEntity, JobsOutboxModel } from '@volontariapp/database';
import { JobsOutboxRepository } from '@volontariapp/outbox';
import { EventsQueue, JobMessagingType } from '@volontariapp/messaging';

const job = JobsOutboxEntity.createJob<typeof JobMessagingType.SYNC_EXTERNAL_CALENDAR>({
  type: JobMessagingType.SYNC_EXTERNAL_CALENDAR,
  emitter: 'ms-event',
  emitterId: userId,
  scheduledAt: new Date(),
  target: EventsQueue.EVENTS,
  payload: {
    eventId,
    calendarUrl,
    syncRequestedBy: userId,
  },
});

const jobsRepo = new JobsOutboxRepository(queryRunner.manager.getRepository(JobsOutboxModel));
await jobsRepo.create(job);

Option B : Fallback Job (En cas d'erreur dans un Command Controller)

Dans les contrôleurs héritant de BaseCommandController :

return await this.withFallback(
  JobMessagingType.FALLBACK_SYNC_CALENDAR,
  userId,
  { eventId, calendarUrl },
  async () => {
    return await this.calendarService.sync(eventId);
  },
);

[!NOTE] Le Runner Outbox est 100% transparent : outbox-runners/outbox-<domaine> scrute jobs_outbox avec SELECT ... FOR UPDATE SKIP LOCKED et pousse le job dans la queue BullMQ correspondante sur Redis. Aucun code supplémentaire n'est nécessaire dans l'outbox runner !

---

Étape 3 : Créer le Handler Unitaire dans workers-runners

📖 Détails complets : 03-worker-and-handlers.md (references/03-worker-and-handlers.md)

Emplacement : workers-runners/worker-<domaine>/src/workers/handlers/<nom-du-job>.handler.ts.

Chaque job possède sa propre classe implémentant IJobHandler :

import { Injectable } from '@nestjs/common';
import { Logger } from '@volontariapp/logger';
import { JobMessagingType, JobRegistry } from '@volontariapp/messaging';
import type { JobOf } from '@volontariapp/workers';
import type { IJobHandler } from './interfaces/job-handler.interface.js';

@Injectable()
export class SyncExternalCalendarHandler implements IJobHandler<typeof JobMessagingType.SYNC_EXTERNAL_CALENDAR> {
  private readonly logger = new Logger({ context: SyncExternalCalendarHandler.name });
  
  readonly jobType = JobMessagingType.SYNC_EXTERNAL_CALENDAR;

  async handle(job: JobOf<typeof JobMessagingType.SYNC_EXTERNAL_CALENDAR>): Promise<{
    originalPayload: JobRegistry[typeof JobMessagingType.SYNC_EXTERNAL_CALENDAR];
  }> {
    const { eventId, calendarUrl } = job.data.payload;
    this.logger.info(`Syncing external calendar for event ${eventId}`, { calendarUrl });

    // Exécution du traitement lourd (ex: parsing iCal, appel réseau)
    await this.performSync(eventId, calendarUrl);

    this.logger.info(`Calendar synced successfully for ${eventId}`);
    
    // Renvoyer obligatoirement le payload d'origine pour l'enregistrement d'audit
    return { originalPayload: job.data.payload };
  }

  private async performSync(eventId: string, url: string): Promise<void> {
    // Logique métier
  }
}

---

Étape 4 : Raccorder le Handler au Worker NestJS

Dans le Worker principal du domaine (workers-runners/worker-<domaine>/src/workers/<domaine>.worker.ts) :

1. Injecter le nouveau handler dans la handlerMap :

   @Injectable()
   @Processor(EventsQueue.EVENTS)
   export class EventWorker extends BaseWorker<JobMessagingType> {
     private readonly handlerMap: Map<JobMessagingType, IJobHandler>;
   
     constructor(
       @Inject(JobAuditRepository) auditRepo: JobAuditRepository,
       @Inject(PublishEventHandler) publishHandler: PublishEventHandler,
       @Inject(SyncExternalCalendarHandler) syncHandler: SyncExternalCalendarHandler, // ← Nouveau
     ) {
       super(auditRepo);
       const handlers: IJobHandler[] = [publishHandler, syncHandler];
       this.handlerMap = new Map(handlers.map((h) => [h.jobType, h]));
     }
   
     protected async processJob(job: JobOf<JobMessagingType>) {
       const handler = this.handlerMap.get(job.name as JobMessagingType);
       if (!handler) {
         throw new InternalServerError(`Unhandled job type: ${job.name}`, 'UNHANDLED_JOB_TYPE', { jobName: job.name });
       }
       return handler.handle(job);
     }
   }
   

2. Déclarer le handler comme Provider dans app.module.ts de ce worker.

---

Étape 5 : La Boucle d'Audit et le Nettoyage Automatique

📖 Détails complets : 04-audit-loop-and-cleanup.md (references/04-audit-loop-and-cleanup.md)

Vous n'avez aucun code de nettoyage manuel à écrire :

1. BaseWorker passe le job en statut PROCESSING dans la table job_audit dès qu'il le prend en charge.
2. Dès que le handler retourne son résultat, BaseWorker passe le statut à COMPLETED (ou FAILED en cas d'erreur).
3. Le Trigger SQL PostgreSQL job_audit_status_trigger se déclenche automatiquement sur job_audit et insère un événement <domaine>:job:outbox:success (ou :failure) dans la table event_queue.
4. L'outbox-runner pousse cet événement dans Redis Stream.
5. Un post-processor central écoute ce stream et exécute le Hard Delete de la ligne d'origine dans jobs_outbox.

---

Faits vérifiés dans le code (2026-10-06)

Sujet	Réalité	Conséquence
IJobHandler	Interface locale à chaque worker, pas exportée par @volontariapp/workers : worker-user/src/handlers/interfaces/, worker-post et worker-social/src/workers/job-handler.interface.ts	Importer celle du worker concerné. Un handler vit sous src/handlers/ dans worker-user, sous src/workers/handlers/ ailleurs.
withFallback	Dupliqué dans chaque ms-* (BaseCommandController), aucun job sur une erreur 400/404/409, sinon job FALLBACK_* vers la queue FALLBACK_* puis FALLBACK_ACTIVATED^base-cmd	Les queries n'ont pas de fallback. Le payload est { userId, payload }, posé avec un @ts-expect-error existant : dette à ne pas recopier. Piège : un ForbiddenError (403) crée un job puis renvoie 206 dans tout ms-* dont withFallback ne l'exclut pas (corrigé dans ms-event, vérifié par test ; non vérifié dans les autres ms-*). Un job rejoué doit re-vérifier les droits : ms-event passe extraPayload à withFallback (ex. userRole dans IFallbackDeleteEventJobPayload).
Statut de jobs_outbox	PROCESSING au pull, COMPLETED dès le push BullMQ[^dispatcher]	La ligne reste en base jusqu'au DELETE de JobOutboxSuccessPostProcessor[^pp-success]. COMPLETED ne veut pas dire « exécuté ».
Échec d'un job	Le trigger émet <domaine>:job:outbox:failure, JobOutboxFailedPostProcessor n'accepte que :job:outbox:failed, et son UPDATE ... WHERE status = PENDING ne toucherait pas une ligne déjà COMPLETED[^pp-failed]	Bug P4 : aucun feedback d'échec de job sur la plateforme. Ne pas s'appuyer sur ce chemin.
Idempotence	BaseWorker saute un job déjà COMPLETED dans job_audit[^base-worker]	Un rejeu BullMQ d'un job terminé est sans effet ; un job interrompu en PROCESSING est rejoué.
Workers existants	throw new Error et cast as Promise<...> dans processJob	Dette : utiliser InternalServerError (@volontariapp/errors) comme dans l'exemple ci-dessus.
Handlers sans producteur	8 handlers (FALLBACK_GET_*, FALLBACK_SIGN_UP) n'ont aucun producteur ; 5 jobs (PUBLISH_EVENT, SEND_WELCOME_EMAIL...) ne sont produits que par ms-user/.../controllers/tests/user.test.controller.ts	Vérifier avec le catalogue avant de supposer qu'un job tourne en production.
worker-ranking (2026-10-28)	Trois workers BaseWorker dans src/workers/ : ranking.worker.ts (RankingQueue.RANKING, handlers ComputeUserRankingHandler et FanoutRankingHandler), ranking-feed.worker.ts (RankingQueue.FEED, ComputeUserFeedHandler) et fallback-ranking.worker.ts (RankingQueue.FALLBACK_RANKING), chacun avec concurrency: config.ranking.workerConcurrency. Handlers sous src/workers/handlers/, types de job dans handlers/types/.	Plusieurs queues pour un même domaine : un worker par queue, pas une handlerMap unique.
Producteurs de jobs ranking (2026-10-28)	job_catalogue.py affiche « aucun » producteur pour COMPUTE_USER_RANKING, COMPUTE_USER_FEED et FANOUT_RANKING, alors qu'ils sont écrits dans jobs_outbox : le type vient d'une table [RankingListKind] -> JobMessagingType (ms-ranking/.../ranking-job.constants.ts, post-processor-ranking/src/domain/constants/ranking-job.constants.ts, worker-ranking/src/domain/constants/ranking-job.constants.ts) et FANOUT_RANKING par post-processor-ranking/src/domain/ranking-job-writer.ts:85.	Le script ne résout pas un type de job indirect (R3 d'AGENTS.md) : « aucun producteur » du catalogue n'est pas une preuve d'absence pour ces jobs.
Drapeau Jev (2026-10-28)	RANKING_JEV_ENABLED (worker-ranking/src/config/helpers/boolean-flag.helpers.ts, scoring/adapters/jev-ranking-scorer.ts, disjoncteur circuit-breaker.ts) ; activé par défaut dans deploy/apps/base/ms-ranking/deployment-worker.yaml.	Un handler peut dépendre d'un adaptateur externe derrière un drapeau et un disjoncteur : prévoir le repli structurel.

Catalogue à jour, calculé depuis le code[^catalogue] :

python3 .agents/skills/volontariapp-implement-async-job-flow/scripts/job_catalogue.py            # tous les jobs
python3 .agents/skills/volontariapp-implement-async-job-flow/scripts/job_catalogue.py --orphans  # sans producteur ou sans handler

Contexte de trace des jobs

JobsOutboxEntity.createJob capture le contexte OpenTelemetry actif dans traceparent ; JobsOutboxPusher le place dans l'enveloppe BullMQ (JobEnvelope.traceparent?, absent sans span actif), en push unitaire et en bulk. Les handlers lisent JobEnvelope, pas la ligne outbox. BaseWorker.process exécute le job dans un span job <type> enfant de ce traceparent (runInTraceContext) : un handler n'a rien à faire pour s'y rattacher. Un écrivain qui reconstruit lui-même la ligne outbox (RankingJobWriter dans worker-ranking) doit recopier traceparent du job source (traceparent: job.traceparent, workers-runners#45), sinon la trace est coupée au job suivant. Détails : skill volontariapp-trace-async-flow.

Règles d'Or & Checklist Anti-Bugs

• [ ] 1 Job = 1 Exécution : Si vous avez besoin de notifier plusieurs services, créez un Event (EventQueueEntity), pas un Job.
• [ ] Toujours retourner { originalPayload: job.data.payload } à la fin du handler pour que BaseWorker puisse auditer l'exécution.
• [ ] Pas d'appel Redis direct : Toujours créer le job via JobsOutboxEntity.createJob() dans la transaction PostgreSQL.
• [ ] Idempotence : Un worker peut redémarrer en cours de traitement ; concevez le handler pour qu'il puisse être rejoué sans corrompre les données.

[^dispatcher]: OutboxDispatcher : COMPLETED après le push [^pp-success]: JobOutboxSuccessPostProcessor [^pp-failed]: JobOutboxFailedPostProcessor [^base-worker]: BaseWorker (garde d'idempotence job_audit) [^catalogue]: job_catalogue.py