Triển khai Saga Pattern trong microservices với NodeJS và Choreography-Based Saga

// booking-service.ts
import express from 'express'
import amqp from 'amqplib'

const app = express()
const PORT = 3001

app.use(express.json())

let channel: amqp.Channel

const paymentQueue = 'payment_queue'

app.post('/booking', async (req, res) => {
  const { userId, eventId, numberOfSeats } = req.body

  // Pre-step 1: Validate booking request

  // Pre-step 2: Save booking request to application database
  const booking = {
    userId,
    eventId,
    numberOfSeats,
    bookingReservedSuccessfully: true,
  }

  /**
   * Step 1: Send Booking Request to PaymentService
   */
  if (booking.bookingReservedSuccessfully) {
    await sendMessageToQueue(paymentQueue, { booking })
  }

  res.json({ message: 'Booking request sent successfully' })
})

async function connectQueue(queue: string) {
  const connection = await amqp.connect('amqp://localhost')

  channel = await connection.createChannel()

  await channel.assertQueue(queue)
}

async function sendMessageToQueue(queue: string, message: unknown) {
  await channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)))
}

app.listen(PORT, async () => {
  console.log(`BookingService is running on http://localhost:${PORT}`)

  await connectQueue(paymentQueue)
})

// booking-service.ts
import amqp from 'amqplib'

let channel: amqp.Channel

const paymentQueue = 'payment_queue'
const seatUpdatingQueue = 'seat_update_queue'
const compensationQueue = 'compensation_queue'

async function processPayment() {
  const connection = await amqp.connect('amqp://localhost')

  channel = await connection.createChannel()

  await channel.assertQueue(paymentQueue)
  await channel.assertQueue(seatUpdatingQueue)
  await channel.assertQueue(compensationQueue)

  channel.consume(paymentQueue, async (msg) => {
    const { booking } = JSON.parse(msg.content.toString())

    // Pre-step: Process payment
    const paymentProcessedSuccessfully = true

    /**
     * Step 2:
     * Send seats update request to SeatUpdatingService
     * Or Reverse the booking by sending a compensation request to CompensationService
     */
    if (paymentProcessedSuccessfully) {
      await sendMessageToQueue(seatUpdatingQueue, { booking })
    } else {
      await sendMessageToQueue(compensationQueue, {
        booking,
        event: 'PaymentServiceFailed',
      })
    }

    // Acknowledge the message
    channel.ack(msg as amqp.Message)
  })
}

async function sendMessageToQueue(queue: string, message: unknown) {
  await channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)))
}

processPayment()

// seat-update-service.ts
import amqp from 'amqplib'

let channel: amqp.Channel

const seatUpdatingQueue = 'seat_update_queue'
const notificationQueue = 'notification_queue'
const compensationQueue = 'compensation_queue'

async function processSeatUpdate() {
  const connection = await amqp.connect('amqp://localhost')

  channel = await connection.createChannel()

  await channel.assertQueue(seatUpdatingQueue)

  channel.consume(seatUpdatingQueue, async (msg) => {
    const { booking } = JSON.parse(msg?.content.toString() || '{}')

    // Add your seat updating logic here
    const seatsUpdatedSuccessfully = true

    /**
     * Step 3:
     * Send seats update request to NotificationService
     * Or Reverse the booking by sending a compensation request to CompensationService
     */
    if (seatsUpdatedSuccessfully) {
      await sendMessageToQueue(notificationQueue, { booking, isSuccess: true })
    } else {
      await sendMessageToQueue(compensationQueue, {
        booking,
        event: 'SeatUpdatingServiceFailed',
      })
    }

    // Acknowledge the message
    channel.ack(msg as amqp.Message)
  })
}

async function sendMessageToQueue(queue: string, message: unknown) {
  await channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)))
}

processSeatUpdate()

// seat-update-service.ts
import amqp from 'amqplib'

let channel: amqp.Channel

const compensationQueue = 'compensation_queue'
const notificationQueue = 'notification_queue'

async function processCompensation() {
  const connection = await amqp.connect('amqp://localhost')

  channel = await connection.createChannel()

  await channel.assertQueue(compensationQueue)

  channel.consume(compensationQueue, async (msg) => {
    const { booking, event } = JSON.parse(msg?.content.toString() || '{}')

    console.log(`Compensating for user ${booking.userId} and event ${event}`)

    /**
     * Step 4: Do compensation
     */
    switch (event) {
      case 'PaymentServiceFailed':
        console.log(
          `Compensating PaymentServiceFailed for user ${booking.userId}`
        )
        // Add your compensation logic here
        break
      case 'SeatUpdatingServiceFailed':
        console.log(
          `Compensating SeatUpdatingServiceFailed for user ${booking.userId}`
        )
        // Add your compensation logic here
        break
    }

    // Send notification to user
    await sendMessageToQueue(notificationQueue, { booking, isSuccess: false })

    // Acknowledge the message
    channel.ack(msg as amqp.Message)
  })
}

async function sendMessageToQueue(queue: string, message: unknown) {
  await channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)))
}

processCompensation()

// seat-update-service.ts
import amqp from 'amqplib'

let channel: amqp.Channel

const notificationQueue = 'notification_queue'

async function processSendNotification() {
  const connection = await amqp.connect('amqp://localhost')

  channel = await connection.createChannel()

  await channel.assertQueue(notificationQueue)

  channel.consume(notificationQueue, async (msg) => {
    const { booking, isSuccess } = JSON.parse(msg?.content.toString() || '{}')

    /**
     * Step 4: Send notification to user
     */
    if (isSuccess) {
      console.log(`Booking SUCCESS sent to user ${booking.userId}`)
    } else {
      console.log(`Booking FAIL send to user ${booking.userId}`)
    }

    // Acknowledge the message
    channel.ack(msg as amqp.Message)
  })
}

processSendNotification()