smile-nestjs-sqs - สร้างขึ้นเพื่อให้ใช้งาน NestJS SQS ในโปรดักชันได้อย่างปลอดภัยยิ่งขึ้น

สวัสดีครับ
ระหว่างที่พัฒนาบริการด้วยชุดผสมของ nestjs + sqs ผมได้ใช้งานแพ็กเกจ sqs อื่น ๆ ใน ecosystem ของ nestjs มาอยู่เรื่อย ๆ
แต่มีจุดที่ใช้งานจริงในโปรดักชันแล้วไม่สะดวกอยู่หลายอย่าง และแม้จะเคยส่ง PR ไปให้บางแพ็กเกจแล้ว แต่ก็มีทั้งกรณีที่ไม่ได้อัปเดตต่อแล้ว หรือบางที่ก็ไม่ได้มองว่านี่เป็นปัญหา ผมเลยสร้างโมดูลนี้ขึ้นมาเพื่อแก้ปัญหาเหล่านี้ด้วยตัวเอง ตอนนี้ก็ได้เปลี่ยนจากโมดูล sqs เดิมที่ใช้อยู่มาใช้ตัวนี้ และใช้งานได้อย่างเสถียรแล้ว
เลยอยากแชร์ให้กับคนที่กำลังมองหาโมดูล sqs สำหรับ nestjs หรือคนที่เคยกังวลคล้าย ๆ กันกับผม เผื่อว่าจะเป็นประโยชน์ครับ

แพ็กเกจนี้มีจุดเด่นดังต่อไปนี้

เป็นมิตรกับ nestjs

ใน nestjs สามารถสร้างและ import โมดูลแบบไดนามิก พร้อม inject handler ที่ติดตั้ง decorator ไว้แล้ว ทำให้พัฒนาได้ง่าย
และมี signature ที่คล้ายกับสิ่งที่พบในแพ็กเกจ sqs อื่น ๆ ส่วนใหญ่

@Module({  
  imports: [  
    ConfigModule.forRoot(),  
    SqsModule.registerAsync({  
      inject: [ConfigService],  
      useFactory: (config: ConfigService) => {  
        const region = config.getOrThrow<string>('AWS_REGION');  
        const queueUrl = config.getOrThrow<string>('ORDERS_QUEUE_URL');  
        const defaultSqsClient = new SQSClient({ region });  
  
        return {  
          defaultSqsClient,  
          consumers: [{ name: 'orders', queueUrl }],  
          producers: [{ name: 'orders', queueUrl }],  
        };  
      },  
    }),  
  ],  
})  
export class AppModule {}  
  
@Injectable()  
export class OrderQueueHandler {  
  @SqsMessageHandler('orders')  
  public async onMessage(message: Message): Promise<Message> {  
    // return message to ack/delete  
    return message;  
  }  
  
  @SqsConsumerEventHandler('orders', 'processing_error')  
  public onProcessingError(error: Error, message: Message) {  
    // report error  
  }  
}  

ใช้แพ็กเกจ bbc เวอร์ชันล่าสุด

แพ็กเกจนี้อ้างอิงจากสองไลบรารีด้านล่าง ซึ่งมียอดดาวน์โหลดเกิน 1.3 ล้านครั้งต่อสัปดาห์ และเป็นไลบรารี sqs ที่มีชื่อเสียงที่สุดในฝั่ง nodejs

https://github.com/bbc/sqs-producer
https://github.com/bbc/sqs-consumer

sqs-consumer มี breaking change ที่สำคัญตั้งแต่ v14
https://github.com/bbc/sqs-consumer/discussions/584

ในเวอร์ชัน 13 และต่ำกว่า ต่อให้ sqs handler รีเทิร์น void ก็ยังถือว่าเป็น ack behaviour
แต่ตั้งแต่เวอร์ชัน 14 เป็นต้นไป จะถือว่า message ที่ sqs handler รีเทิร์นกลับมาอย่างชัดเจนเท่านั้นคือ ack
การเปลี่ยนแปลงนี้มีความชัดเจนและคาดการณ์ได้มากขึ้น และเป็นการเปลี่ยนที่ช่วยไม่ให้ผู้ใช้สับสนในการ implement เรื่อง ack
แต่แพ็กเกจ npm ส่วนใหญ่ที่เรียกตัวเองว่าเป็น nestjs sqs module ในตอนนี้ ยังใช้ sqs-consumer เวอร์ชัน 13 หรือต่ำกว่าอยู่

การตรวจสอบตั้งแต่ตอนบูต

ช่วยจับการตั้งค่าที่ผิดพลาดได้ทันทีตอนแอปเริ่มทำงาน ทำให้ใช้งานได้ปลอดภัยยิ่งขึ้น

• ตรวจจับชื่อ consumer/producer ที่ซ้ำกัน
• หากใช้ decorator กับ consumer ที่ไม่มีอยู่จริง จะเกิด error ทันที
• ตรวจจับการพิมพ์ชื่อ event ผิด
• ตรวจจับความไม่ตรงกันของชนิดพารามิเตอร์ระหว่าง handler แบบ batch/single
• ตรวจจับความไม่ตรงกันของชนิดค่าที่รีเทิร์นของ batch/single

graceful shutdown

bbc/sqs-consumer มีตัวเลือก pollingCompleteWaitTimeMs สำหรับ graceful shutdown
และใน nestjs ก็มี lifecycle ของ nestjs เอง
ในขั้น onModuleDestroy ของ nestjs ภายในจะรอจนกว่า consumer แต่ละตัวจะ emit stopped event ออกมาก่อนจึงค่อยปิดการทำงาน
เพราะ event นี้จะเกิดขึ้นก็ต่อเมื่อข้อความที่ consumer กำลังประมวลผลอยู่เสร็จสิ้นแล้ว จึงทำให้การประมวลผลข้อความเสร็จสมบูรณ์และปิดโปรเซสได้อย่างปลอดภัย
เนื่องจากเชื่อมต่อกับ lifecycle ของ nestjs โดยอัตโนมัติ
เพียงเปิด app.enableShutdownHooks() ไว้ และ
ตั้งค่า pollingCompleteWaitTimeMs, shutdownTimeoutMs ให้เหมาะสม ก็จะช่วยให้ใช้งานในสภาพแวดล้อม k8s ได้ปลอดภัยยิ่งขึ้น แม้ในกรณีที่ถูกปิดแบบกะทันหัน

ระหว่างที่ใช้ sqs ในโปรดักชัน ผมมักรู้สึกว่าเรื่องที่คิดไว้ว่าเมื่อไรจะอัปเดตสักทีกลับไม่ได้รับการอัปเดตต่อเนื่อง ทำให้มีอุปสรรคในการสร้างวัฒนธรรมการพัฒนาที่แข็งแรง เลยกลายมาเป็นผลลัพธ์ชิ้นนี้เพื่อแก้ปัญหา
ถ้าใครอยู่ในสภาพแวดล้อมคล้ายกันและกำลังมีโจทย์แบบเดียวกัน ลองนำไปใช้ดูแล้วส่งฟีดแบ็กมาได้ จะขอบคุณมากครับ