J9 · Événements, tâches planifiées et asynchrone

Jour 9 · vendredi 20 novembre · Livrable : Notifications sur changement de statut

Spring fournit les événements applicatifs, @Scheduled et JMS ; Nest a les mêmes briques avec @nestjs/event-emitter, @nestjs/schedule et @nestjs/bullmq. La différence tient au modèle d'exécution : tout tourne sur le thread unique de Node, et une exception levée hors de la requête ne remonte à personne. À la fin de la journée, vous savez découpler un effet de bord (la notification) de la requête, planifier un traitement nocturne et déporter un travail dans une file Redis avec reprises automatiques.

Objectifs

Émettre un événement métier avec EventEmitter2 et le consommer avec @OnEvent, en mode synchrone ou asynchrone.
Planifier un traitement avec @Cron, @Interval et @Timeout.
Déporter un traitement dans une file Redis avec @nestjs/bullmq : processor, tentatives, backoff.
Savoir où part une exception levée dans un listener, un cron ou un job, et la journaliser.
Livrer dans TaskFlow : task.status.changed consommé par NotificationsListener, un cron qui marque les tâches OVERDUE, et en option une file BullMQ pour les notifications.

Spring → NestJS

Spring	NestJS	À retenir
`ApplicationEventPublisher.publishEvent(event)`	`EventEmitter2.emit('task.status.changed', event)`	L'événement est identifié par un nom (chaîne), pas par sa classe
`@EventListener`	`@OnEvent('task.status.changed')`	L'exception d'un listener ne remonte pas à l'émetteur
`@Async` + `@EventListener`	`@OnEvent(nom, { async: true })`	Même thread JS, exécuté plus tard (`setImmediate`)
`@TransactionalEventListener(phase = AFTER_COMMIT)`	`emit()` placé après `await save()`	Aucun lien automatique entre événements et transactions
`@EnableScheduling`	`ScheduleModule.forRoot()`	Importé une fois dans `AppModule`
`@Scheduled(cron = "0 0 1 * * *", zone = "Europe/Paris")`	`@Cron('0 01 * * *', { timeZone: 'Europe/Paris' })`	Chaque instance l'exécute, pas d'équivalent de ShedLock
`@Scheduled(fixedRate = 60000)`	`@Interval(60_000)`	`setInterval` : n'attend pas la fin du tour précédent
`TaskScheduler.schedule(tâche, instant)`	`@Timeout(5_000)`	Une seule exécution, après le démarrage
JMS (`@JmsListener`), Spring Batch	`@nestjs/bullmq` (`@Processor`, `WorkerHost`)	Jobs persistés dans Redis, tentatives et backoff par job

Comprendre

Événements applicatifs avec @nestjs/event-emitter

EventEmitterModule.forRoot(), importé une fois dans AppModule (le module est global par défaut), rend injectable un EventEmitter2, l'émetteur de la bibliothèque du même nom. Au démarrage, pendant la phase onApplicationBootstrap, Nest parcourt les providers et les contrôleurs et abonne chaque méthode décorée @OnEvent('nom').

emit(nom, payload) est un appel synchrone : EventEmitter2 appelle les listeners abonnés à ce nom, puis renvoie true (au moins un listener) ou false (aucun). Un listener sans option démarre donc pendant emit() et s'exécute jusqu'à son premier await. Avec { async: true }, il est reporté au tour suivant de l'event loop (setImmediate) :

text

emit() sans option   : listener (jusqu'à son 1er await) → retour de emit() → suite du service
emit() + async: true : retour de emit() → suite du service → listener (tour suivant de l'event loop)

Un nom, pas un type : Spring choisit les listeners d'après la classe de l'événement, ici seul le nom compte. Une faute de frappe ne lève aucune erreur, emit() renvoie simplement false. Gardez le nom dans une constante et le contenu dans une classe. Les points du nom (task.status.changed) ne sont qu'une convention, sauf si vous activez wildcard: true dans forRoot() pour écouter task.*.

Où l'analogie casse : un @EventListener Spring synchrone s'exécute dans le thread et la transaction de l'appelant, et son exception remonte à publishEvent(). Rien de tout cela ici : aucune transaction n'est associée à l'événement, et l'exception est interceptée par Nest (erreurs asynchrones). « Asynchrone » ne veut pas dire « autre thread » : un listener partage le thread unique de Node, un calcul lourd y bloque toutes les requêtes (event loop).

Tâches planifiées avec @nestjs/schedule

ScheduleModule.forRoot() joue le rôle de @EnableScheduling. Trois décorateurs se posent sur les méthodes d'un provider singleton :

Décorateur	Mécanisme	Équivalent Spring
`@Cron(expression, { name, timeZone })`	bibliothèque `cron`	`@Scheduled(cron = ..., zone = ...)`
`@Interval(ms)` ou `@Interval(nom, ms)`	`setInterval`	`@Scheduled(fixedRate = ...)`
`@Timeout(ms)` ou `@Timeout(nom, ms)`	`setTimeout`, une seule fois	pas d'équivalent direct

Les tâches sont montées au démarrage de l'application et arrêtées à sa fermeture. SchedulerRegistry (injectable) les retrouve par leur nom, par exemple getCronJob('mark-overdue-tasks'), pour les arrêter ou en ajouter à l'exécution.

Une expression cron a 5 champs (minute en premier) ou 6 (seconde en premier). CronExpression.EVERY_DAY_AT_1AM vaut '0 01 * * *' ; l'expression Spring 0 0 1 * * * (6 champs) est aussi acceptée et signifie la même chose. Sans timeZone, c'est l'heure locale du processus qui compte, souvent UTC dans un conteneur.

// Exemple hors TaskFlow : à déclarer dans les providers d'un module.
import { Injectable, Logger } from '@nestjs/common';
import { Interval, Timeout } from '@nestjs/schedule';

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

  @Timeout(5_000) // une fois, 5 s après le démarrage
  async warmUp(): Promise<void> {
    this.logger.log('cache warmed up');
  }

  @Interval('refresh-stats', 60_000) // toutes les 60 s, nommée pour SchedulerRegistry
  async refresh(): Promise<void> {
    this.logger.log('stats refreshed');
  }
}

Les méthodes planifiées s'exécutent sur l'event loop, pas dans un pool de threads. @Interval relance la méthode même si le tour précédent (asynchrone) n'est pas terminé ; pour @Cron, l'option waitForCompletion: true saute une échéance tant que la précédente tourne. Nest entoure chaque méthode d'un try/catch qui journalise l'erreur (contexte Scheduler) : une exception n'arrête ni l'application ni les exécutions suivantes, comme le gestionnaire d'erreurs par défaut de @Scheduled.

Où l'analogie casse : chaque instance exécute ses crons. Spring ne coordonne pas non plus les instances (d'où ShedLock), et Nest n'a pas d'équivalent intégré. Avec plusieurs réplicas : un traitement idempotent, un verrou en base, une seule instance qui active les crons, ou un job planifié dans une file (cron OVERDUE).

Jobs Redis avec @nestjs/bullmq

Un événement EventEmitter2 vit en mémoire : si le processus redémarre ou si l'envoi d'email échoue, la notification est perdue et rien ne la rejoue. Une file BullMQ enregistre chaque job dans Redis ; un worker le traite et le relance en cas d'échec.

Élément	Rôle	Dans TaskFlow
`BullModule.forRootAsync({ useFactory: () => ({ connection }) })`	Connexion Redis partagée	`NotificationsQueueModule`
`BullModule.registerQueue({ name })`	Déclare une file	file `notifications`
`@InjectQueue(name)` puis `queue.add(nomDuJob, data, options)`	Producteur	`NotificationsListener`
`@Processor(name)` sur une classe `extends WorkerHost`, méthode `process(job)`	Consommateur (worker)	`NotificationsProcessor`

Le cycle d'un job : si process() lève une exception (ou rejette sa promesse), le job échoue ; tant qu'il reste des tentatives (attempts), BullMQ le replanifie après le délai de backoff. En exponentiel, le délai vaut 2^(n-1) × delay : avec attempts: 5 et delay: 1_000, les reprises partent après 1 s, 2 s, 4 s puis 8 s. Après la dernière tentative, le job reste dans l'état failed dans Redis (removeOnFail n'est pas défini). removeOnComplete: 1_000 ne garde que les 1 000 derniers jobs réussis.

BullMQ garantit au moins une exécution, pas exactement une : un job interrompu (processus tué pendant le traitement) peut être repris. Écrivez un process() idempotent.

JMS / Batch ↔ BullMQ : le modèle est celui d'un @JmsListener avec redélivraison (producteur, file, consommateurs). Spring Batch (lecture par lots, étapes, reprise d'un job) n'a pas d'équivalent direct. Où l'analogie casse : Redis n'est pas transactionnel avec PostgreSQL. TaskFlow ajoute le job après le commit ; si queue.add() échoue, l'erreur est journalisée et la notification est perdue. Pour une garantie forte, le motif « outbox » écrit le message dans la même transaction que la donnée, puis un processus le publie.

Installation : npm i @nestjs/bullmq bullmq ioredis. Avec BullMQ 6, ioredis est une dépendance peer optionnelle : sans installation explicite, l'application plante au chargement du module de file (piège ioredis).

Erreurs asynchrones

Une exception ne remonte qu'à celui qui attend la promesse. Personne n'attend un listener, un cron ou un job : l'erreur doit être traitée sur place, sinon elle se perd ou arrête le processus.

Où l'erreur se produit	Qui attend ?	Ce qui se passe	Dans TaskFlow
Handler HTTP	Nest	Filtre global, réponse `problem+json`	`AllExceptionsFilter`
Listener `@OnEvent`	Personne	Nest l'attrape et la journalise (contexte `Event`), l'appelant n'en sait rien	`try/catch` + log avec l'id de la tâche
`@Cron`, `@Interval`, `@Timeout`	Personne	Nest l'attrape et la journalise (contexte `Scheduler`)	`try/catch` + message explicite
`process()` d'un worker BullMQ	BullMQ	Job en échec, nouvelle tentative selon `attempts` et `backoff`	Laisser remonter, c'est le signal de reprise
Promesse lancée sans `await`	Personne	Rejet non géré : Node arrête le processus par défaut	Interdit par `no-floating-promises` (await oublié)

Un listener qui lève une exception ne remonte donc pas à l'appelant, avec ou sans async: true : PATCH /v1/tasks/:id/status répond 200 même si la notification échoue. C'est voulu (la notification ne doit pas annuler le changement de statut), mais l'erreur n'apparaît que dans les logs.

L'option @OnEvent(nom, { suppressErrors: false }) laisse ressortir l'exception. Elle n'a de sens que si l'appelant attend les listeners avec await this.events.emitAsync(...) (listeners sans async: true) ; avec un simple emit(), l'exception devient un rejet non géré qui arrête le processus. Si l'appelant doit connaître le résultat, n'utilisez pas d'événement : appelez le service directement.

Côté Spring, une méthode @Async qui lève une exception finit dans AsyncUncaughtExceptionHandler, et @Scheduled journalise puis continue : le principe est le même, personne n'attend le résultat.

Cas TaskFlow

Événement task.status.changed

L'événement : un nom constant et une classe immuable.

// Event names are plain strings: keep them in one constant.
export const TASK_STATUS_CHANGED = 'task.status.changed';

export class TaskStatusChangedEvent {
  constructor(
    readonly taskId: string,
    readonly projectId: string,
    readonly from: TaskStatus,
    readonly to: TaskStatus,
    readonly changedBy: string,
    readonly occurredAt: Date,
  ) {}
}

TasksService l'émet une fois le changement enregistré :

async updateStatus(id: string, status: TaskStatus, user: AuthUser): Promise<Task> {
  const task = await this.tasks.findOneBy({ id });
  if (!task) throw new NotFoundException(`Task ${id} not found`);
  await this.projects.assertAccess(task.projectId, user);
  if (task.status === status) return task;

  const from = task.status;
  task.status = status;
  const saved = await this.tasks.save(task);
  // Emit only once the change is saved. Listeners run separately: if one
  // fails, this request still succeeds (see NotificationsListener).
  this.events.emit(
    TASK_STATUS_CHANGED,
    new TaskStatusChangedEvent(task.id, task.projectId, from, status, user.id, this.clock()),
  );
  return saved;
}

emit() vient après await this.tasks.save(task). save() s'exécute dans sa propre transaction : l'événement part après le commit, comme avec @TransactionalEventListener(phase = AFTER_COMMIT). Émis à l'intérieur d'un dataSource.transaction(...), il pourrait déclencher une notification pour une donnée ensuite annulée (transactions).
Pas d'événement si le statut ne change pas (retour anticipé).
occurredAt vient de l'horloge injectée CLOCK (tokens), qu'un test remplace par une date fixe.
Le service ignore qui écoute : ajouter un audit ou un push ne touche pas TasksService.

Le listener :

// @EventListener equivalent. { async: true } runs it after the emitting
// request has moved on: an exception here never reaches TasksService.
@Injectable()
export class NotificationsListener {
  private readonly logger = new Logger(NotificationsListener.name);

  constructor(
    // Present only when NotificationsQueueModule is loaded.
    @Optional() @InjectQueue(NOTIFICATIONS_QUEUE) private readonly queue?: Queue<NotificationJob>,
  ) {}

  @OnEvent(TASK_STATUS_CHANGED, { async: true })
  async onStatusChanged(event: TaskStatusChangedEvent): Promise<void> {
    try {
      if (!this.queue) {
        this.logger.log(`[simulated email] task ${event.taskId}: ${event.from} -> ${event.to}`);
        return;
      }
      await this.queue.add(
        'task-status-changed',
        { taskId: event.taskId, from: event.from, to: event.to, recipientId: event.changedBy },
        { attempts: 5, backoff: { type: 'exponential', delay: 1_000 }, removeOnComplete: 1_000 },
      );
    } catch (error) {
      // Nobody awaits this listener, so log here or the error is lost.
      this.logger.error(`Notification for task ${event.taskId} failed`, error instanceof Error ? error.stack : String(error));
    }
  }
}

{ async: true } : la méthode s'exécute au tour suivant de l'event loop, une fois que updateStatus() a rendu la main.
Sans file (pas de REDIS_URL), il journalise un « email simulé ». Avec la file, il ajoute un job avec 5 tentatives et un backoff exponentiel (file BullMQ).
@Optional() correspond à @Autowired(required = false) : la file n'est injectée que si NotificationsQueueModule est chargé.
Le try/catch journalise l'erreur avec l'id de la tâche ; sans lui, Nest la journaliserait quand même, mais sans ce contexte.

Le module qui l'enregistre (la file n'est chargée que si REDIS_URL est défini) :

@Module({
  imports: [
    // A function, not the string 'REDIS_URL': the string form only checks
    // that the value is not "false", so an unset variable would still load
    // the queue module and make it try to reach Redis.
    ConditionalModule.registerWhen(NotificationsQueueModule, (env) => Boolean(env.REDIS_URL)),
  ],
  providers: [NotificationsListener],
})
export class NotificationsModule {}

EventEmitterModule.forRoot() et ScheduleModule.forRoot() sont importés dans AppModule. Un changement de statut :

http

PATCH /v1/tasks/2b9d…/status
Authorization: Bearer eyJ…
Content-Type: application/json

{ "status": "DONE" }

La réponse est 200 avec la tâche ("status": "DONE"), et le log du contexte NotificationsListener contient [simulated email] task 2b9d…: TODO -> DONE. Le test unitaire vérifie l'émission, avec le bon contenu, sans démarrer de listener (tests unitaires) :

it('saves the new status, then emits task.status.changed', async () => {
  repository.findOneBy.mockResolvedValue(Object.assign(new Task(), { id: 't1', projectId: 'p1', status: TaskStatus.Todo }));

  const task = await service.updateStatus('t1', TaskStatus.Done, user);

  expect(task.status).toBe(TaskStatus.Done);
  expect(repository.save).toHaveBeenCalledTimes(1);
  expect(events.emit).toHaveBeenCalledWith(
    TASK_STATUS_CHANGED,
    new TaskStatusChangedEvent('t1', 'p1', TaskStatus.Todo, TaskStatus.Done, 'u1', now),
  );
});

Cron quotidien OVERDUE

// @Scheduled(cron = "0 0 1 * * *", zone = "Europe/Paris"). Every running
// instance fires it: with several replicas, use a queue or a lock instead.
@Injectable()
export class TasksScheduler {
  private readonly logger = new Logger(TasksScheduler.name);

  constructor(private readonly tasks: TasksService) {}

  @Cron(CronExpression.EVERY_DAY_AT_1AM, { name: 'mark-overdue-tasks', timeZone: APP_TIME_ZONE })
  async markOverdue(): Promise<void> {
    try {
      const count = await this.tasks.markOverdue();
      this.logger.log(`${count} task(s) marked OVERDUE`);
    } catch (error) {
      // Nobody awaits a cron job: catch, or the error is only half visible.
      this.logger.error('Overdue job failed', error instanceof Error ? error.stack : String(error));
    }
  }
}

/** One UPDATE for every late task; returns how many changed. */
async markOverdue(): Promise<number> {
  // The Paris calendar day, like the cron schedule: toISOString() would give
  // the UTC day, one day behind at 01:00 in summer.
  const today = localDate(this.clock());
  const result = await this.tasks.update(
    {
      dueDate: LessThan(today),
      status: In([TaskStatus.Todo, TaskStatus.InProgress]),
      archived: false,
    },
    { status: TaskStatus.Overdue },
  );
  return result.affected ?? 0;
}

Tous les jours à 1 h, heure de Paris (APP_TIME_ZONE). Le nom mark-overdue-tasks permet de retrouver la tâche dans SchedulerRegistry.
Un seul UPDATE sur toutes les tâches en retard (échéance passée, statut TODO ou IN_PROGRESS, non archivées), au lieu de charger chaque tâche. affected donne le nombre de lignes modifiées.
dueDate est une colonne date : TypeORM la lit et la compare comme une chaîne 'YYYY-MM-DD'.
Le traitement est idempotent : relancé, il ne trouve plus rien à modifier, puisque les tâches déjà OVERDUE sont exclues. Exécuté par deux réplicas, il ne coûte qu'une requête de plus. Un job non idempotent (envoyer un récapitulatif) demanderait un verrou (pg_try_advisory_lock dans PostgreSQL), une instance dédiée ou un job planifié dans BullMQ.
L'UPDATE en masse ne passe pas par updateStatus() : le passage automatique en OVERDUE n'émet pas task.status.changed. Si ce passage doit être notifié, émettez un événement après l'UPDATE.
try/catch : Nest journaliserait l'erreur seul, mais sans dire quel traitement a échoué.

Pour observer le cron sans attendre la nuit, remplacez temporairement l'expression par CronExpression.EVERY_30_SECONDS ('*/30 * * * * *').

Fuseau du cron et date du jour

Le cron se déclenche à 1 h à Paris. Calculer today avec toISOString() donnerait la date UTC : en hiver (UTC+1), 1 h à Paris correspond à 0 h UTC, la date est la même ; en été (UTC+2), c'est 23 h UTC la veille, et une tâche arrivée à échéance la veille ne passerait OVERDUE qu'un jour plus tard. TaskFlow calcule donc la date dans le fuseau du cron : APP_TIME_ZONE et localDate(), dans src/common/time-zone.ts (Intl.DateTimeFormat avec timeZone). Un test unitaire fixe l'horloge au 30 juin à 23 h UTC (1 h le 1er juillet à Paris) et vérifie que la date comparée est 2026-07-01 (J10).

OVERDUE : un statut ou un calcul ?

Le plan stocke OVERDUE comme un statut, et TaskFlow le respecte. Ce statut mélange pourtant un fait dérivé du temps et l'avancement : une tâche IN_PROGRESS qui devient OVERDUE perd l'information « en cours », et rien n'empêche un client d'envoyer OVERDUE dans PATCH /v1/tasks/:id/status. L'alternative : ne stocker que l'avancement et calculer « en retard » à la lecture (échéance passée et statut différent de DONE), sans cron. Voir le modèle de domaine.

File BullMQ pour les notifications (optionnel)

Redis est un service du Compose, activé par le profil queue :

docker-compose.yml complet

yaml

# docker compose up -d                 -> PostgreSQL + Adminer for `npm run start:dev`
# docker compose --profile queue up -d -> + Redis for the BullMQ queue (J9)
# docker compose --profile app up -d   -> + the API itself, built from the Dockerfile (J11)
services:
  db:
    image: postgres:16-alpine
    container_name: taskflow-db
    environment:
      POSTGRES_USER: taskflow
      POSTGRES_PASSWORD: taskflow
      POSTGRES_DB: taskflow
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U taskflow -d taskflow"]
      interval: 5s
      timeout: 3s
      retries: 10

  # Throwaway database for e2e tests: data lives in memory (tmpfs).
  db-test:
    image: postgres:16-alpine
    container_name: taskflow-db-test
    environment:
      POSTGRES_USER: taskflow
      POSTGRES_PASSWORD: taskflow
      POSTGRES_DB: taskflow_test
    ports:
      - "5433:5432"
    tmpfs:
      - /var/lib/postgresql/data

  adminer:
    image: adminer
    ports:
      - "8081:8080"
    depends_on:
      - db

  redis:
    image: redis:7-alpine
    profiles: [queue]
    ports:
      - "6379:6379"

  app:
    build: .
    profiles: [app]
    environment:
      NODE_ENV: production
      PORT: 3000
      DB_HOST: db
      DB_PORT: 5432
      DB_USER: taskflow
      DB_PASSWORD: taskflow
      DB_NAME: taskflow
      # Empty unless exported: the app then refuses to start (validateEnv), while
      # the other services still work without it.
      JWT_SECRET: ${JWT_SECRET:-}
    ports:
      - "3000:3000"
    depends_on:
      db:
        condition: service_healthy

volumes:
  pgdata:

bash

docker compose --profile queue up -d                     # PostgreSQL, Adminer et Redis
echo 'REDIS_URL=redis://localhost:6379' >> .env.development
npm run start:dev

Le module de file, chargé seulement quand REDIS_URL est défini (voir NotificationsModule plus haut) :

// Only loaded when REDIS_URL is set (see NotificationsModule).
@Module({
  imports: [
    BullModule.forRootAsync({
      inject: [ConfigService],
      useFactory: (config: ConfigService) => ({
        connection: { url: config.getOrThrow<string>('REDIS_URL') },
      }),
    }),
    BullModule.registerQueue({ name: NOTIFICATIONS_QUEUE }),
  ],
  providers: [NotificationsProcessor],
  exports: [BullModule],
})
export class NotificationsQueueModule {}

forRootAsync lit REDIS_URL dans ConfigService ; getOrThrow échoue au démarrage si la variable manque (ConfigModule).
exports: [BullModule] rend la file notifications injectable dans NotificationsModule, donc dans le listener.

Le consommateur :

// The consumer side, like a @JmsListener: runs in the background with
// retries and backoff configured on each job (see NotificationsListener).
@Processor(NOTIFICATIONS_QUEUE)
export class NotificationsProcessor extends WorkerHost {
  private readonly logger = new Logger(NotificationsProcessor.name);

  async process(job: Job<NotificationJob>): Promise<void> {
    // A real implementation would call an email provider here; throwing makes
    // BullMQ retry the job.
    this.logger.log(`Email to ${job.data.recipientId}: task ${job.data.taskId} is now ${job.data.to} (attempt ${job.attemptsMade + 1})`);
  }
}

Le worker tourne dans le même processus que l'API : un traitement lourd y bloquerait aussi les requêtes. BullMQ permet de le déplacer plus tard dans un processus séparé, sans changer le producteur. Le parcours a été vérifié de bout en bout avec Redis : changement de statut → événement → listener → file → log de NotificationsProcessor (Email to …: task … is now DONE (attempt 1)).

Pour voir les reprises, faites lever une exception à process() : le log affiche attempt 1 à attempt 5, espacés de 1, 2, 4 puis 8 s, puis le job passe en failed. Pour journaliser l'échec définitif :

// Exemple hors TaskFlow : méthode à ajouter dans NotificationsProcessor
// (import { OnWorkerEvent } from '@nestjs/bullmq').
@OnWorkerEvent('failed')
onFailed(job: Job<NotificationJob> | undefined, error: Error) {
  this.logger.warn(`Job ${job?.id} failed after ${job?.attemptsMade} attempt(s): ${error.message}`);
}

ConditionalModule : passez une fonction, pas une chaîne

ConditionalModule.registerWhen(Module, 'REDIS_URL') vérifie seulement que la valeur n'est pas "false" : une variable absente charge quand même le module, qui tente de joindre Redis. TaskFlow passe une fonction, (env) => Boolean(env.REDIS_URL). Voir le piège.

Situations

Vous devez notifier après un changement sans ralentir la requête → emit() après l'enregistrement et @OnEvent(nom, { async: true }) (événement).
Un traitement doit tourner chaque nuit → @Cron avec timeZone explicite et un nom (cron).
L'envoi doit survivre à un redémarrage et être retenté → job BullMQ avec attempts et backoff (file BullMQ).
Votre cron tourne trois fois avec trois réplicas → traitement idempotent, verrou en base ou job planifié dans une file (schedule).
Une erreur de listener n'apparaît nulle part → try/catch et log avec le contexte métier dans le listener (erreurs asynchrones).
Redis n'existe que dans certains environnements → ConditionalModule.registerWhen avec une fonction et @Optional() côté consommateur (file BullMQ).
L'appelant doit savoir si l'effet de bord a échoué → pas d'événement : appel direct du service (erreurs asynchrones).
Vous voulez tester qu'un événement est émis → mock de EventEmitter2 et toHaveBeenCalledWith (tests unitaires).

Pièges

Événement émis avant le commit

Un emit() placé dans un dataSource.transaction(...) part avant le commit : le listener peut notifier une modification qui sera annulée. Émettez après la fin de la transaction.

BullMQ 6 sans ioredis

« BullMQ could not load the optional 'ioredis' package » : installez ioredis explicitement. TaskFlow fixe ioredis 5.11.1, la version 5 étant celle qu'accepte TypeORM 1.1 (piège ioredis).

Exception de listener invisible pour l'appelant. C'est le comportement par défaut. Journalisez dans le listener ; si l'appelant doit réagir, appelez le service au lieu d'émettre un événement.
suppressErrors: false avec emit(). L'exception devient un rejet non géré qui arrête Node. Réservez cette option à await emitAsync().
Nom d'événement mal orthographié. Aucune erreur, aucun listener : emit() renvoie false. Utilisez une constante partagée (TASK_STATUS_CHANGED).
Événement émis dans onModuleInit. Les listeners sont abonnés plus tard, à onApplicationBootstrap : l'événement est perdu (cycle de vie).
Cron exécuté sur chaque instance. Rendez le traitement idempotent ou coordonnez les instances.
@Interval qui se chevauche. Si un tour dure plus longtemps que l'intervalle, deux exécutions tournent en même temps. Pour un cron, waitForCompletion: true.
Fuseaux horaires. Fixez timeZone sur le cron et calculez les dates dans le même fuseau (cron OVERDUE).
Promesse non attendue dans un listener ou un cron (this.queue.add(...) sans await) : le try/catch ne voit pas l'erreur. La règle no-floating-promises d'oxlint la signale.
Travail CPU dans un listener ou un processor. Il bloque toutes les requêtes du processus (event loop).

Exercices

[ ] Installer @nestjs/event-emitter et @nestjs/schedule, importer EventEmitterModule.forRoot() et ScheduleModule.forRoot() dans AppModule.
[ ] Émettre task.status.changed depuis TasksService après l'enregistrement, et le consommer dans un NotificationsListener qui journalise un email simulé.
[ ] Faire lever une exception au listener : vérifier que PATCH /v1/tasks/:id/status répond toujours 200 et repérer l'erreur dans les logs. Recommencer sans async: true et comparer l'ordre des logs.
[ ] Écrire le cron quotidien qui marque les tâches en retard OVERDUE, puis le tester avec une horloge fixe (J10), dont une horloge à 1 h du matin, heure d'été de Paris.
[ ] Passer temporairement le cron à CronExpression.EVERY_30_SECONDS et observer le log task(s) marked OVERDUE.
[ ] Optionnel : ajouter Redis au Compose (profil queue), installer @nestjs/bullmq bullmq ioredis et déporter l'envoi des notifications dans une file BullMQ.
[ ] Optionnel : faire échouer process() et observer les 5 tentatives et le backoff exponentiel.
[ ] Retirer REDIS_URL et vérifier que l'application démarre sans Redis.
[ ] Noter dans LEARNINGS.md ce qui diffère de @EventListener (transaction, propagation des exceptions) et de @Scheduled (plusieurs instances), puis git tag jour-09.

J9 · Événements, tâches planifiées et asynchrone ​

Objectifs ​

Spring → NestJS ​

Comprendre ​

Événements applicatifs avec @nestjs/event-emitter ​

Tâches planifiées avec @nestjs/schedule ​

Jobs Redis avec @nestjs/bullmq ​

Erreurs asynchrones ​

Cas TaskFlow ​

Événement task.status.changed ​

Cron quotidien OVERDUE ​

File BullMQ pour les notifications (optionnel) ​

Situations ​

Pièges ​

Exercices ​

Pour aller plus loin ​