Messaging Pipeline: Channel Routing and the Inbound/Outbound Job Chain

Yaya Engine talks to users over WhatsApp, SMS, and Telegram (and delivers push through the same notification layer), but almost none of the business logic in the messaging path knows which of those it is talking to. Three queued jobs carry a message from a webhook to a user and back: ProcessIncomingMessage turns an inbound webhook payload into a Dify reply, SendChannelMessage is the single point where any outbound message actually leaves the system, and ProcessMessageBatch fans a bulk send (an alert, a broadcast) out into per-recipient SendChannelMessage dispatches. All three delegate the "which channel, which credentials, which API" questions to a ChannelManager/ChannelResolver pair keyed on ChannelTypeEnum, which is what lets channel abstraction mean something concrete in this codebase rather than being aspirational.

Inbound: webhook to AI reply

A WhatsApp or Telegram webhook resolves the sending user and dispatches ProcessIncomingMessage with the channel, the identifier, and the raw content. The job first tries to find the user by that channel identifier; if nobody matches, it calls MessagingService::handleUnknownSender() and dispatches SendUnregisteredUserResponse instead of proceeding, so an unregistered sender gets a canned response rather than triggering an AI call on their behalf [@process-incoming]. For a known user, the job syncs a placeholder name from the WhatsApp profile if one hasn't been set, gets or creates the channel account and conversation, and stores the inbound message before doing anything else — so the message is durably recorded even if everything downstream fails [@process-incoming].

Before calling out to Dify, the job checks KillSwitchSettings::isDifyEnabledFor($channel) and returns early, logging dify.disabled_noop, if the AI hand-off is paused for that channel — this is the per-channel admin toggle documented in Dify Kill Switch; it stops only the AI response, not message storage or account/conversation bookkeeping [@process-incoming]. If the switch is on, DifyService::chat() is called with the user as the session key (so a conversation stays continuous across channels), the conversation's dify_conversation_id is set on first use, and the reply is handed to SendChannelMessage::dispatch() [@process-incoming]. A Dify failure is caught rather than left to crash the job: it logs the error, sends an AiProcessingFailedNotification to Slack (throttled — see Notifications), and still dispatches SendChannelMessage with a generic "trouble processing your request" fallback, so a broken AI backend degrades to a polite failure message instead of silence [@process-incoming].

Outbound: the one place a message actually sends

SendChannelMessage is the terminal step for both conversational replies and batch sends, and it is the only place that calls a channel's send() method. It looks up the channel implementation from ChannelManager, sends the message, and records the result: it either updates an existing Message row (when dispatched from a batch, which pre-creates the row) or creates one via MessagingService::storeMessage(), then writes a MessageDeliveryLog row and increments Mixpanel counters [@send-channel-message]. Weather alerts on Telegram get special treatment here: if the message carries an alert_id in its metadata and the channel is Telegram, SendChannelMessage sets options['open_app_button'] = true, which TelegramChannel uses to render an inline button that opens the Mini-App — replacing what used to be a magic link opened in Telegram's in-app browser [@send-channel-message].

Retry behavior is channel-specific by design: WhatsApp sends get up to config('whatsapp.retry.max_attempts', 5) tries with backoff computed by WhatsAppErrorHandler, while every other channel gets 3 tries with a flat 10-second backoff [@send-channel-message]. On failure, WhatsAppErrorHandler::categorize() decides whether the error is retryable; a rate-limit category releases the job back onto the queue with a computed delay instead of throwing, while other retryable failures throw so Laravel's queue retry mechanism picks them up, and a final non-retryable or exhausted-attempts failure calls $this->fail() and updates the batch's failure counter [@send-channel-message]. This job is also where batch bookkeeping closes the loop: incrementBatchSuccess()/incrementBatchFailure() bump the parent MessageBatch counters whenever a message belongs to one, whether the job succeeds, fails, or is later marked failed by Laravel's failed() hook [@send-channel-message].

Bulk sends: ProcessMessageBatch and the ChannelResolver

ProcessMessageBatch is how an alert or broadcast turns a list of recipients into individual SendChannelMessage dispatches. For each recipient it resolves a channel through ChannelResolver::resolve(), which can be forced per-recipient or per-batch (forced_channel in batch metadata, or a per-recipient override) with an explicit allowFallbackFromPreferred flag controlling whether an unreachable forced channel silently falls through to the default preference order or leaves the recipient unreachable [@process-batch]. When no channel is forced, ChannelResolver tries channels in a fixed order — Telegram, then WhatsApp, then SMS, then a bare phone-number fallback for SMS — reasoning that pilot users are Telegram-only, so falling back to a phone-based channel for them would require a phone number they were never asked for [@channel-resolver].

Content assembly branches by channel and message type: WhatsApp keeps template messages as templates, while Telegram and SMS get interpolated free text via TemplateInterpolator and SmsTemplateVariableBuilder, with a hard-coded default alert text as the last-resort fallback if a template body is empty [@process-batch]. Each recipient's Message row is created with firstOrCreate() keyed on (message_batch_id, user_id, conversation_id, direction), so a retried chunk cannot create duplicate outbound messages, and wasRecentlyCreated gates whether SendChannelMessage is dispatched at all for that recipient on a re-run [@process-batch]. The job re-fetches the batch's metadata before merging in channel_distribution updates specifically to avoid clobbering concurrent chunk writes, and uses DB::raw() atomic increments for processed_count and failed_count rather than read-modify-write [@process-batch].

Why batches are chunked into groups of 500

ProcessMessageBatch did not always chunk its recipients. In two real alert dispatches, a single monolithic job carrying 5,000-7,000 recipients ran long enough to exceed the queue worker's 90-second retry_after, causing the queue to re-dispatch the job mid-run and deliver only 17-25% of the intended messages [@chunk-design]. The fix, recorded in Chunk Batch Message Processing, moved chunking upstream into BatchMessagingService::createBatch(), which now splits recipients into groups of 500 and dispatches one ProcessMessageBatch job per chunk against a single shared MessageBatch record, relying on the atomic counters described above to make concurrent chunk completion safe [@chunk-design]. ProcessMessageBatch carries an explicit $timeout = 60, deliberately shorter than the 90-second retry_after, as a visible guard against the same failure mode recurring silently [@process-batch]. The queue connection itself defaults to database [@queue-config].

The channel abstraction underneath it all

ChannelManager is a small registry: register() stores a ChannelInterface implementation keyed by its ChannelTypeEnum, and channel($type) throws UnsupportedChannelException if nothing is registered for that type [@channel-manager]. ChannelInterface itself defines only getType(), send(), sendToIdentifier(), and normalizeIdentifier() [@channel-interface], and ChannelTypeEnum enumerates WhatsApp, Sms, Pwa, and Telegram, with a requiresPhoneNumber() helper that returns true only for WhatsApp and SMS [@channel-type-enum]. Because every job in this pipeline talks to channels only through that interface and through ChannelResolver's reachability logic, adding a new channel is a matter of implementing ChannelInterface and registering it — see Add a Messaging Channel — rather than touching ProcessIncomingMessage, SendChannelMessage, or ProcessMessageBatch. The Telegram Bot Architecture page covers how Telegram in particular plugs into this same job chain alongside its own Nutgram-specific inbound handling, and Debug Message Delivery covers how to trace a message through this pipeline when a send goes missing.