Campaign Schedule Dispatch (process_campaign)

Overview

A campaign scheduled through the CMS is stored as a row in the campaign table with status scheduled and a start_date.

This feature is the scanner that runs every minute, finds campaigns whose send time has arrived, claims them atomically, and then enqueues the real work into either line_broadcast_rich_message or line_multicast_rich_message depending on cast_type.

The central concern here is preventing duplicate sends. Whether several worker replicas scan at the same time or a message is redelivered, the end recipient must never receive the campaign twice.

Business Flow

1. Tick — a */1 * * * * cron on the cron-scheduler profile (CampaignScheduleScannerService) publishes a single empty message to the process_campaign queue.
   • A Redis SET NX on the key CAMPAIGN_SCHEDULE_TICK:<yyyy-MM-ddTHH:mm> (TTL 55 seconds) ensures only one replica publishes.
   • If Redis is down, the tick still publishes, because the database-level claim already guarantees correctness.
   • This tick previously came from external infrastructure and has since been moved into the repository.
2. Reaper — the main profile receives the tick and first recovers any campaign stuck in sending whose claimed_at is older than 15 minutes, resetting it to scheduled on the assumption that the claiming worker has died.
3. Atomic claim — issue UPDATE campaign SET status='sending', claimed_at=NOW() guarded by status='scheduled' AND start_date <= NOW() AND deleted_date IS NULL with a RETURNING clause. Only rows this tick actually claimed come back, so no other scanner can pick them up.
4. Enqueue — for each claimed campaign, publish a payload containing campaignId, audienceId, organizationId, and lineOaId to the queue matching cast_type: broadcast goes to line_broadcast_rich_message, everything else to line_multicast_rich_message.
5. Roll back on publish failure — the status is reverted to scheduled immediately on a best-effort basis, so the next tick can retry it without waiting the full 15 minutes for the reaper.
6. Heartbeat while sending — the delivery consumer refreshes claimed_at every 5 minutes during the send so the reaper does not reclaim the campaign mid-flight (see Multicast / Targeted Campaign Delivery).
7. Re-check at the destination — the delivery consumer verifies the status once more. If the campaign is no longer sending — for instance it is already sent — the message is discarded quietly as a permanent error without flipping the campaign to failed.

Key Files & Functions

• internal/campaign/campaign.go
  • Service.ProcessCampaign(ctx) — reaper, atomic claim, and enqueue in one pass
  • The campaignStatusScheduled, campaignStatusSending, campaignStatusSent, campaignStatusCancel, campaignStatusDraft, campaignStatusFailed, castTypeBroadcast, and castTypeMulticast constants
• internal/campaign/consumer.go — Consumer.HandleProcessCampaign for the process_campaign queue, plus Consumer.Register()
• internal/cronscheduler/campaign_schedule_scanner.go — CampaignScheduleScannerService.Run(ctx) and campaignTickKeyPrefix
• cmd/worker/integration.go — campaignForLineMessageApi.GetCampaignAndValidate(), the point where the sending status is verified and duplicate delivery is blocked
• Queues: consumes process_campaign on the main profile and publishes to line_broadcast_rich_message or line_multicast_rich_message

Connections to Other Services

• The campaign table — reads and writes status, claimed_at, start_date, cast_type, audience_id, organization_id, and line_oa_id
• Redis — the CAMPAIGN_SCHEDULE_TICK:* keys used to deduplicate ticks across replicas
• RabbitMQ — consumes process_campaign and publishes to the actual delivery queues
• cms-api-go — creates and edits campaigns and sets start_date in the campaign-management domain. When a user chooses to send immediately, cms-api-go flips the status to sending and enqueues the job itself without waiting for the scanner.
• This feature makes no direct LINE API calls; those happen during delivery