All files / src/platform/events outbox-dispatcher.service.ts

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
import { Injectable, Logger } from '@nestjs/common';
import { InjectModel } from '@nestjs/mongoose';
import { Model } from 'mongoose';
import { OutboxEvent } from './schemas/outbox-event.schema';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
import { Interval } from '@nestjs/schedule';
 
@Injectable()
export class OutboxDispatcherService {
  private readonly logger = new Logger(OutboxDispatcherService.name);
  private isProcessing = false;
 
  constructor(
    @InjectModel(OutboxEvent.name)
    private readonly outboxModel: Model<OutboxEvent>,
    @InjectQueue('finance.event-posting')
    private readonly financeQueue: Queue,
  ) {}
 
  @Interval(2000) // Poll every 2 seconds
  async pollAndDispatch() {
    if (this.isProcessing) return;
    this.isProcessing = true;
 
    try {
      // Find pending outbox events
      const pendingEvents = await this.outboxModel.find({ status: 'PENDING' }).sort({ createdAt: 1 }).limit(20).exec();
 
      for (const event of pendingEvents) {
        // Multi-instance safety: Atomic CAS to mark as QUEUED
        const lockedEvent = await this.outboxModel.findOneAndUpdate(
          { _id: event._id, status: 'PENDING' },
          { status: 'QUEUED' },
          { new: true }
        ).exec();
 
        if (!lockedEvent) {
          // Another instance already picked it up
          continue;
        }
 
        try {
          // Route event to appropriate BullMQ queue
          let targetQueue = this.financeQueue;
          // In the future we can route dynamically based on prefix
          
          await targetQueue.add(event.eventName, {
            eventId: event._id.toString(),
            tenantId: event.tenantId,
            eventName: event.eventName,
            payload: event.payload,
            correlationId: event.correlationId || event._id.toString(),
            causationId: event.causationId
          }, {
            jobId: event._id.toString(), // Ensures absolute idempotency at queue level
            attempts: 5,
            backoff: {
              type: 'exponential',
              delay: 1000 // 1s, 2s, 4s, 8s, 16s...
            }
          });
 
          // Mark as DISPATCHED in Outbox
          lockedEvent.status = 'PUBLISHED';
          lockedEvent.processedAt = new Date();
          await lockedEvent.save();
 
          this.logger.log(`Outbox event ${event.eventName} (${event._id}) durably dispatched to BullMQ.`);
        } catch (dispatchErr) {
          this.logger.error(`Failed to dispatch event ${event._id}: ${dispatchErr.message}`);
          lockedEvent.status = 'PENDING'; // Release lock on failure so it can retry
          lockedEvent.error = dispatchErr.message;
          await lockedEvent.save();
        }
      }
    } catch (err) {
      this.logger.error(`Error in outbox dispatcher poll loop: ${err.message}`);
    } finally {
      this.isProcessing = false;
    }
  }
}