fix:(sks): Add queue between kafka and lambda

[benjaminpaige]

Before	After
Email processing pipeline lacks delay queue between Kafka and Lambda functions.	Introduced an SQS delay queue to improve fault tolerance and added monitoring for delays and errors.

🔧 Changes

• Added an SQS delay queue between Kafka and Lambda for improved message handling.
• Updated processEmails.ts to handle both Kafka and SQS events:
  • Forward Kafka events to the delay queue.
  • Process delayed SQS messages for email generation and sending.
• Enhanced error handling in processEmails.ts:
  • Implemented error handling for failed SQS SendMessageCommand.
  • Added debug logging for easier troubleshooting.
• Updated email.ts stack:
  • Configured a new delay queue with a 2-minute delivery delay and required IAM permissions.
  • Added CloudWatch alarms for Lambda errors, throttling, and DLQ message count.
  • Improved alarm configurations and added SNS notifications for critical events.
• Documented key functions and stack components with JSDoc comments for clarity.

🚀 How to Test

1. Deploy the updated stack:
   • Ensure the environment variables (e.g., DELAY_QUEUE_URL) are set correctly in the Lambda configurations.
2. Test Kafka event processing:
   • Publish Kafka messages and verify that they are forwarded to the SQS delay queue.
3. Test SQS message processing:
   • Check if delayed messages from the queue are processed by the Lambda and trigger the email sending flow.
4. Simulate failure scenarios:
   • Introduce errors in the processing pipeline to validate error handling, DLQ behavior, and alarm triggers.
5. Review CloudWatch logs and alarms:
   • Confirm that appropriate metrics are being tracked and alarms are configured as expected.

📄 Additional Notes

• The processEmails.ts handler now logs detailed debug information for both Kafka and SQS events, aiding in identifying potential issues.
• New alarms monitor Lambda execution errors, throttling, and DLQ activity, ensuring visibility into critical operational metrics.
• This update is backward-compatible with the existing email processing logic.

[github-actions]

Coverage Report

Status	Category	Percentage	Covered / Total
🔵	Lines	0%	0 / 4515
🔵	Statements	0%	0 / 4752
🔵	Functions	0%	0 / 1370
🔵	Branches	0%	0 / 2639

File Coverage

File	Stmts	Branches	Functions	Lines	Uncovered Lines
Changed Files
lib/lambda/delayedEmailProcessor.ts	0%	0%	0%	0%	28-299
lib/lambda/kafkaToSqs.ts	0%	0%	0%	0%	5-34

Generated in workflow #1733 for commit aaa8fcb by the Vitest Coverage Report Action

[asharonbaltazar]

Great work. What wizardry 🧙🏻‍♂️

[asharonbaltazar]

You can make a guard function that narrows the sqsRecord.body to a KafkaRecord

const isKafkaRecord = (parsedBody: unknown): parsedBody is KafkaRecord => /* logic pertaining to kafkaRecord, i.e. typeof parsedBody === 'object' && parsedBody?.value */ 

[asharonbaltazar]

But that's a nit