import Fastify from 'fastify' import { Queue } from 'bullmq' import Redis from 'ioredis' import { loadConfig } from './config.js' import { getLogger } from './lib/logger.js' import { handleVerification } from './webhook/verification.js' import { createWebhookHandler } from './webhook/handler.js' import { WhatsAppClient } from './whatsapp/WhatsAppClient.js' import { ClaudeProvider } from './llm/ClaudeProvider.js' import { ConversationManager } from './bot/ConversationManager.js' import { EscalationManager } from './bot/EscalationManager.js' import { MessageRouter } from './bot/MessageRouter.js' import { PayloadClient } from './payload/PayloadClient.js' import { InteractionWriter } from './payload/InteractionWriter.js' import { RulesLoader } from './payload/RulesLoader.js' import { TemplateResolver } from './payload/TemplateResolver.js' import { MessageWorkerManager } from './queue/message-worker.js' import { INCOMING_MESSAGE_QUEUE, STATUS_UPDATE_QUEUE, type IncomingMessageJobData, type StatusUpdateJobData, } from './queue/message-job.js' const config = loadConfig() const log = getLogger('server') // --- Initialize infrastructure --- const redis = new Redis(config.REDIS_URL, { maxRetriesPerRequest: null, // Required by BullMQ enableReadyCheck: false, }) // --- Initialize services --- const whatsappClient = new WhatsAppClient({ phoneNumberId: config.WHATSAPP_PHONE_NUMBER_ID, accessToken: config.WHATSAPP_ACCESS_TOKEN, }) const llmProvider = new ClaudeProvider(config.ANTHROPIC_API_KEY) const conversationManager = new ConversationManager(redis) const escalationManager = new EscalationManager(conversationManager, whatsappClient) const payloadClient = new PayloadClient() const interactionWriter = new InteractionWriter(payloadClient) const rulesLoader = new RulesLoader(payloadClient) const templateResolver = new TemplateResolver(payloadClient) const messageRouter = new MessageRouter( conversationManager, escalationManager, whatsappClient, llmProvider, rulesLoader, templateResolver, interactionWriter, ) // --- BullMQ queues --- const incomingQueue = new Queue(INCOMING_MESSAGE_QUEUE, { connection: redis, defaultJobOptions: { attempts: 3, backoff: { type: 'exponential', delay: 2000 }, removeOnComplete: { count: 1000 }, removeOnFail: { count: 5000 }, }, }) const statusQueue = new Queue(STATUS_UPDATE_QUEUE, { connection: redis, defaultJobOptions: { attempts: 2, removeOnComplete: { count: 500 }, removeOnFail: { count: 1000 }, }, }) // --- Workers --- const workerManager = new MessageWorkerManager({ redis, messageRouter, whatsappClient, }) // --- Fastify server --- const app = Fastify({ logger: false, // Using pino directly trustProxy: true, }) // Custom JSON parser that preserves rawBody for signature validation app.removeContentTypeParser('application/json') app.addContentTypeParser( 'application/json', { parseAs: 'buffer' }, (req, body, done) => { ;(req as typeof req & { rawBody: Buffer }).rawBody = body as Buffer try { const json = JSON.parse((body as Buffer).toString()) done(null, json) } catch (err) { done(err as Error, undefined) } }, ) // --- Routes --- // Meta Webhook verification (GET) app.get('/webhook', handleVerification) // Meta Webhook handler (POST) const webhookHandler = createWebhookHandler({ onMessage: async (message) => { await incomingQueue.add('process', { message, receivedAt: new Date().toISOString(), }) }, onStatusUpdate: async (status) => { await statusQueue.add('update', { status, receivedAt: new Date().toISOString(), }) }, }) app.post('/webhook', webhookHandler) // Health check app.get('/health', async () => { const redisOk = redis.status === 'ready' let payloadOk = false try { await payloadClient.find('users', { limit: '0' }) payloadOk = true } catch { // Payload API down } const status = redisOk && payloadOk ? 'ok' : 'degraded' return { status, timestamp: new Date().toISOString(), services: { redis: redisOk ? 'ok' : 'down', payloadApi: payloadOk ? 'ok' : 'down', }, } }) // --- Lifecycle --- async function start(): Promise { try { // Initialize interaction writer await interactionWriter.init() // Start workers await workerManager.start() // Start HTTP server await app.listen({ port: config.PORT, host: '0.0.0.0' }) log.info( { port: config.PORT, env: config.NODE_ENV }, 'WhatsApp Bot server started', ) } catch (err) { log.fatal({ error: (err as Error).message }, 'Failed to start server') process.exit(1) } } async function shutdown(): Promise { log.info('Shutting down...') await workerManager.stop() await incomingQueue.close() await statusQueue.close() await app.close() redis.disconnect() log.info('Shutdown complete') process.exit(0) } process.on('SIGTERM', shutdown) process.on('SIGINT', shutdown) start()