# wa-agent A WhatsApp Cloud API agent framework for Cloudflare Workers. Extracted from a production bot serving daily devotional content to thousands of users on WhatsApp. **Cloudflare-native by design.** Uses D1 for durable state, R2 for media caching, Workers for compute, and the cron trigger for scheduled outreach. No Durable Objects required, no external queue service, no managed database — just bindings. ```js import { Agent, mountWebhook } from '@emerleite/wa-agent' import { Hono } from 'hono' const app = new Hono() const agent = new Agent({ whatsapp: { endpoint: env.META_WA_ENDPOINT, token: env.META_WA_TOKEN, verifyToken: env.META_WH_TOKEN, appSecret: env.META_APP_SECRET }, db: env.DB, }) agent.command(['help', 'h'], async ({ reply }) => reply.text('Hi!')) agent.onText(async ({ text, reply }) => reply.text(`You said: ${text}`)) mountWebhook(agent, app) export default { fetch: app.fetch, scheduled: agent.scheduled.bind(agent), } ``` ## Why this exists Building a WhatsApp bot on Cloudflare Workers means solving the same problems every time: - **Webhook signature verification** against `X-Hub-Signature-256` (HMAC-SHA256 over the raw body — easy to get wrong) - **Burst coalescing**: users send three messages in two seconds; you want one LLM turn, not three - **Meta's 24h/72h messaging window**: outside it you can only send pre-approved templates; tracking when each user is reachable is a constant footgun - **Opt-in / opt-out** plumbing for proactive broadcasts (devotionals, reminders, tips) - **Long answers** that exceed WhatsApp's interactive-message cap, forcing a summarize-and-expand-button pattern - **AI thread persistence**: OpenAI Assistants need a `thread_id` per user, which has to live somewhere - **Daily drip content**: 21-day plans, habit prompts, weekly progress recaps - **Audio in / audio out**: Whisper transcription for inbound voice notes, Azure TTS for outbound narration `wa-agent` packages all of those into composable primitives. Each one works independently — you can use just `D1CoalesceQueue` if that's all you need. ## Architecture at a glance ``` Meta webhook │ ▼ (verifyMetaSignature) Hono /webhook (POST) │ ▼ D1CoalesceQueue.enqueue() ─┐ │ per-user 3s debounce in D1 waitUntil(setTimeout(3s) → ─┘ buffer rapid-fire messages Agent.drain()) │ ▼ one batch per user, text combined extractInbound() │ ▼ (lifecycle: onFirstContact / onMessage) ┌─────────┴──────────┐ │ │ ▼ ▼ ButtonRouter CommandRouter ──▶ onText fallback │ ▼ reply.ai(text) ──▶ OpenAIAssistant │ ▼ if answer > 1024 chars Summarizer + expand button ``` Cron jobs run alongside webhook traffic: ``` scheduled() event │ ├─▶ queue.processAll() (drain anything stuck) ├─▶ queue.cleanup() (drop done rows > 7d) └─▶ matched cron handler: Broadcast(channel='devotional') ReEngagement.ask() SequentialPlan.usersForDelivery() → send + markDelivered SlotDelivery.pickForUser() → send + recordImpression ``` ## Quickstart Scaffold a runnable bot in one command: ```bash npx @emerleite/wa-agent init my-bot cd my-bot npm install cp .dev.vars.example .dev.vars # then fill in Meta secrets npm run db:create # → paste the printed database_id into wrangler.toml npm run db:migrate # framework migrations, local D1 npm run dev # wrangler dev on http://localhost:8787 ``` For local development without a real Meta account, boot the fake Meta server in a second terminal and point the Worker at it: ```bash npm run mock:meta # fake graph.facebook.com on :4000 echo 'META_GRAPH_BASE_URL=http://localhost:4000' >> .dev.vars ``` Templates: ```bash npx @emerleite/wa-agent init my-bot # echo-bot (default) npx @emerleite/wa-agent init my-tools --template=tool-agent # AgentLoop + Zod tools npx @emerleite/wa-agent init my-support --template=support-bot # pipeline (intent → policy → LLM) npx @emerleite/wa-agent init my-bsp --template=multi-tenant-bot # BSP-style, many numbers npx @emerleite/wa-agent init my-all --template=full-bot # every primitive (reference) ``` ## Install (manual) ```bash npm install @emerleite/wa-agent hono openai ``` `hono` and `openai` are optional peers — pull them only if you use them. Apply the migrations to your D1 database: ```bash wrangler d1 migrations apply \ --migrations-dir node_modules/@emerleite/wa-agent/migrations ``` The migrations are layered so you only enable the pieces you need: | File | What it adds | Required for | |------|---|---| | `001_core.sql` | `messages`, `sessions` | every bot | | `002_users.sql` | `leads`, `message_windows` | every bot | | `003_queue.sql` | `message_queue` | webhook queue | | `004_broadcast.sql` | `broadcast_log`, `engagement_answers` | `Broadcast`, `ReEngagement` | | `005_plans.sql` | `plans`, `plan_days`, `user_plans`, `user_plan_progress` | `SequentialPlan` | | `006_slots.sql` | `ads`, `ad_impressions` | `SlotDelivery` | | `007_usage.sql` | `feature_usage` | `UsageCounter` | | `008_preferences.sql` | `user_preferences` | `PreferenceStore` | ## Configuration Required environment / wrangler vars: ```toml # wrangler.toml [[d1_databases]] binding = "DB" database_name = "..." database_id = "..." ``` ```bash # Secrets (wrangler secret put) META_WA_ENDPOINT # https://graph.facebook.com/v22.0/ META_WA_TOKEN # permanent system-user token META_WH_TOKEN # webhook verify token META_APP_SECRET # app secret (for signature verification) ``` For AI features, also: ```bash AZURE_OPENAI_ENDPOINT AZURE_OPENAI_API_KEY AZURE_API_VERSION AZURE_MODEL_DEPLOYMENT ASSISTANT_ID ``` ## Core concepts ### Agent The composer. Holds your config, registers handlers, exposes a `scheduled()` for the cron trigger and webhook helpers for the fetch handler. ```js const agent = new Agent({ whatsapp: { endpoint, token, verifyToken, appSecret }, db: env.DB, ai: new OpenAIAssistant({ client, assistantId }), // optional summarizer: new Summarizer({ client }), // optional summarizeOver: 1024, // chars queue: { debounceSeconds: 3, maxAttempts: 3 }, // optional D1 queue tuning contextHook: async (ctx) => ({ tier: await fetchTier(ctx.user.whatsapp) }), }) ``` ### Commands and buttons ```js agent.command('help', async ({ reply }) => reply.text(helpText)) agent.command(['plan', 'plans'], async ({ user, reply }) => { … }) agent.button('opt-in', async ({ user, reply, leads }) => { await leads.optIn(user.whatsapp) await reply.text('Subscribed. ✓') }) // Button ids are usually prefix-encoded so we can attach state. // plan_done_42_3 → user finished day 3 of plan 42 agent.buttonPrefix('plan_done_', async ({ user, suffix, reply }) => { const [planId, day] = suffix.split('_').map(Number) const r = await new SequentialPlan({ db: agent.db }).markDone(user.whatsapp, planId, day) await reply.text(r.completed ? 'Plan complete!' : `Day ${day} done.`) }) ``` The `reply` object is pre-bound to the current user — `reply.text(...)`, `reply.buttons(...)`, `reply.cta(...)`, `reply.image(...)`, `reply.audio(...)`, `reply.template(...)`, `reply.markRead()`. `reply.ai(text)` is the high-leverage shortcut: it runs the configured assistant with the persisted thread id, sends the answer (summarized if too long), persists the new thread id, and offers an "expand" button when summarization happened. Most bots just need: ```js agent.onText(async ({ text, reply, session }) => { await reply.markRead() await reply.ai(text, { threadId: session?.threadId }) }) ``` ### D1CoalesceQueue (the interesting bit) Why coalesce? A user types: ``` hi i have a question about romans 8 ``` Three separate webhooks fire over ~2 seconds. You want one LLM turn that sees all three lines, not three turns where the model says "Hello!" then "Tell me more" then "Sure, what's your question?". Cloudflare Queues don't debounce per-key. Durable Objects work but add a binding, a class definition, migrations, and cost. This package uses D1 alone: 1. `enqueue()` writes a row with `scheduled_at = now + 3s`, then *pushes every pending row for that user forward by 3s*. So a burst all settles to the same fire time. 2. Your fetch handler `waitUntil`s a 3-second sleep, then calls `agent.drain()`. 3. `drain()` claims all pending rows for the user in one batch, hands them to the agent. The agent combines the text bodies and runs one turn. State lives entirely in D1. Workers can come and go; recovery is automatic via `recoverStale()` (runs on every `scheduled()` invocation). Failed batches retry up to 3 times, then go to `status='failed'`. You can use `D1CoalesceQueue` standalone — it doesn't depend on anything else in the package. ### MessageWindow ```js const window = new MessageWindow({ db: env.DB }) await window.start(whatsapp, 'paid') // open or renew const { inWindow, type } = await window.status(whatsapp) const open = await window.listOpen() // all reachable users right now ``` The two window types are Meta's 24h paid and 72h free conversation categories (we use 23h30m / 71h30m to be safe). Every inbound message renews the window automatically inside `Agent.handleBatch()`. ### LeadStore A minimal CRM-style table with opt_in, funnel_state, and ctwa_clid. The Agent calls `leads.upsert()` on every message, so you never lose a lead if your AI call later throws. Subclass or replace if your app already has a user table. ### Broadcast Send the same payload to many users with a "received today" log so re-running the cron is safe. ```js const devotional = new Broadcast({ client: agent.client, db: env.DB, channel: 'devotional' }) await devotional.run({ send: async ({ whatsapp }) => agent.client.sendText(whatsapp, todaysContent), }) ``` Default audience is "opt-in users with an open window who haven't been broadcast-to on this channel today" — override `audienceQuery` if you need different rules (paid only, specific funnel state, etc). ### ReEngagement Daily yes/no question with a weekly progress view. Bíblia Fala uses this for "did you pray yesterday?" — the bot asks at noon, the user taps Yes/No, and the answer is logged. A separate handler renders ✅/❌ for the past 7 days. ```js const reading = new ReEngagement({ client: agent.client, db: env.DB, topicId: 1, question: { body: 'Did you read your devotional yesterday?', yesLabel: 'Yes', noLabel: 'No' }, template: { name: 'engagement_reading', language: 'pt_BR' }, // sent when user is OUT of window }) // Cron: ask everyone in the audience await new Broadcast({ client: agent.client, db: env.DB, channel: 'engagement_reading' }) .run({ send: ({ whatsapp }) => reading.ask(whatsapp) }) // Webhook: persist the answer + show progress agent.buttonPrefix('engagement_1_', async ({ user, buttonId, reply }) => { await reading.recordAnswer(user.whatsapp, buttonId) const week = await reading.weekProgress(user.whatsapp) await reply.text(renderProgress(week)) }) ``` ### SequentialPlan Multi-day drip plans with done/skip buttons and 24h auto-advance. ```js const plans = new SequentialPlan({ db: env.DB }) await plans.enroll(whatsapp, planId) const today = await plans.usersForDelivery() // for the cron const r = await plans.markDone(whatsapp, planId, day) // r === { completed: false, nextDay: 4 } | { completed: true, day: 21 } ``` `autoAdvanceStale({ staleHours: 24 })` skips users who got their day delivered but never tapped — keeps them moving through the plan. ### SlotDelivery One item per slot per user per day, weighted random with recent-impression skipping. Works for ads, content tips, daily verses, etc. ```js const slot = new SlotDelivery({ db: agent.db }) const users = await slot.usersForSlot('morning') for (const u of users) { const item = await slot.pickForUser(u.whatsapp) await agent.client.sendCtaUrl(u.whatsapp, { body: item.body, displayText: item.cta_text, url: item.cta_url }) await slot.recordImpression(u.whatsapp, item.id, 'morning') } ``` ### TierProvider + AccessGate Pluggable subscription lookup and feature gating. Most apps store paid status outside the bot (Stripe, your billing service). `TierProvider` is the interface; `HttpTierProvider` is a ready-made HTTP-backed implementation with a 60-second in-memory cache. ```js import { HttpTierProvider, AccessGate, Upsell } from '@emerleite/wa-agent' const tierProvider = new HttpTierProvider({ baseUrl: 'https://billing.example.com/subscriptions', token: env.BILLING_API_TOKEN, // urlFor: (wa) => `${baseUrl}/${wa}/tier` ← override if your URL shape differs }) const gate = new AccessGate({ tierProvider, log: agent.log, // counts user's prior messages allowedTiers: ['premium', 'lifetime'], freeMessageLimit: 10, // first 10 messages free, then paywall }) agent = new Agent({ /* … */, tierProvider, gate }) const upsell = new Upsell({ client: agent.client, leads: agent.leads, pitch: ({ name }) => `Hi ${name}, want to keep chatting? …`, ctaText: 'Subscribe', ctaUrl: async (wa) => `https://checkout.example.com/${wa}`, // resolver per user video: { url: 'https://media.example.com/pitch.mp4', caption: 'See how it works' }, }) agent.onText(async ({ user, text, reply, gate, session }) => { const access = await gate.check(user.whatsapp) if (!access.allowed) { await upsell.send(user.whatsapp, { name: user.name }) return } await reply.markRead() await reply.ai(text, { threadId: session?.threadId }) }) ``` `AccessGate.check()` returns `{ allowed, tier, reason: 'tier'|'trial'|'denied', remaining }` — useful if you want to nudge users approaching the trial limit ("3 free messages left"). ### OnboardingFlow First-contact composition. Pass it to the `Agent` and it fires automatically when a user messages for the first time. ```js import { OnboardingFlow } from '@emerleite/wa-agent' const onboarding = new OnboardingFlow({ client, contact: { name: { formatted_name: 'My Bot', first_name: 'My Bot' }, phones: [{ phone: '+15551234567', type: 'Main', wa_id: '15551234567' }], }, welcomeBody: ({ name }) => `Hi ${name || 'there'}! Tap below to start chatting.`, optInButtonId: 'opt-in', optInButtonTitle: 'Let’s go', helpText: 'Type "help" any time to see what I can do.', }) agent = new Agent({ /* … */, onboarding }) ``` The `opt-in` button is wired by the agent by default — when tapped, it calls `leads.optIn()` and replies with a confirmation. Override with `agent.button('opt-in', yourHandler)`. ### Upsell Multi-step pitch composition: optional video → pause → pitch text + CTA-URL button. Updates the user's `funnel_state` to `CHECKOUT` (or whatever you pass) so dashboard funnels light up. `pitch` and `ctaUrl` accept either strings or functions of `({ whatsapp, ...vars })` — useful for per-user payment links. **Throttled mode** — sending the full video pitch every time a free user hits the same paywall annoys them and doesn't lift conversion. Configure a shorter `reminder`, then call `sendSmart()` to let the framework pick: ```js const upsell = new Upsell({ client, leads, pitch: 'full pitch with video pitch text…', ctaText: 'Subscribe', ctaUrl: async (wa) => `https://checkout.example.com/${wa}`, video: { url: 'https://media.example.com/pitch.mp4' }, reminder: { pitch: 'Already saw the demo — tap below to subscribe', // ctaText / ctaUrl optional; falls back to the main config if omitted }, }) await upsell.send(whatsapp) // full pitch + transition lead → CHECKOUT await upsell.sendReminder(whatsapp) // CTA only, no video, no funnel change await upsell.sendSmart(whatsapp) // picks based on lead.funnel_state ``` `sendSmart()` uses the lead's funnel state as the "already pitched" marker — if the lead is already in the configured `funnelState` (default `'CHECKOUT'`), it sends the reminder; otherwise it fires the full pitch. Falls back to the full pitch if `reminder`/`leads` aren't configured or the lookup throws. ### HybridSearch Generic BM25 + LIKE + Reciprocal Rank Fusion search over any FTS5-backed table. ```js import { HybridSearch, buildSearchSchema } from '@emerleite/wa-agent' // Once at deploy: build the FTS5 mirror of your content table for (const sql of buildSearchSchema({ contentTable: 'docs', ftsColumns: ['title', 'body'], tokenize: 'trigram remove_diacritics 1', })) await env.DB.prepare(sql).run() // Per request: const search = new HybridSearch({ db: env.DB, contentTable: 'docs', searchColumns: ['title', 'body'], filters: { language: 'en' }, // applied to every query }) const hits = await search.search('mercy', { limit: 5 }) ``` Why hybrid: BM25 alone misses on short queries (FTS5 trigram needs ≥3 chars); LIKE alone has no ranking. RRF fuses both into one score-weighted list with `bm25Weight` / `keywordWeight` knobs. ### Dashboard HTMX-based admin dashboard with default cards over the framework's tables and pluggable custom cards. ```js import { Hono } from 'hono' import { Dashboard, defaultCards } from '@emerleite/wa-agent' const app = new Hono() const dash = new Dashboard({ title: 'My Bot', cards: [ ...defaultCards.all(), // summary, queue, messages-chart, funnel { id: 'sales', refreshSeconds: 60, render: async ({ db }) => { const r = await db.prepare(`SELECT COUNT(*) as c FROM orders WHERE created_at >= date('now')`).first() return `

Sales today

${r.c}
` }, }, ], auth: { username: 'admin', password: env.DASHBOARD_PASSWORD }, }) await dash.mount(app, '/dashboard') ``` Each card is `{ id, refreshSeconds?, render({ db, env, query }) }` returning an HTML string. The shell auto-loads HTMX and Chart.js so cards can return `` + inline `