Warmbly文档
warmbly.com打开控制台

事件系统

介绍连接 Warmbly 各服务的事件总线主题、消息封装和实时事件分发机制。

Warmbly 各服务通过可插拔的事件总线异步通信。默认自托管构建使用 JSON 编码的 NATS JetStream；云部署可以选择使用 Kafka、Avro 和 Confluent Schema Registry 的构建。控制台的实时事件分发使用独立且更简单的发布/订阅渠道，本页末尾会介绍。

传输与编解码

事件总线是一个抽象层（internal/infrastructure/eventbus/），有两种提供程序，由 EVENTBUS_PROVIDER 选择：

  • nats：NATS JetStream。一个持久流（默认名为 warmbly）以带前缀的主题形式保存所有消息；例如主题 jobs.worker-events 会变为 subject warmbly.jobs.worker-events。消息保留 7 天，消费者组映射到 JetStream 持久消费者，处理失败时最多重新投递 10 次。docker-compose.yml 运行的就是这种配置。
  • kafka：Apache Kafka，受 Go 构建标签 kafka 控制。默认二进制不包含它，也没有 librdkafka 和 CGO，因此默认构建设置 EVENTBUS_PROVIDER=kafka 后会在启动时明确报错。每个读取总线的服务都为它发布第二种镜像，标签相同但带 -kafka 后缀，例如 ghcr.io/warmbly/warmbly/backend:prod-kafka；consumer、worker 和 tracking 也遵循相同规则。自行构建时，Go 需要 GO_TAGS=kafka，跟踪服务需要 CARGO_FEATURES=kafka。

消息编码与传输方式相互独立，由 CODEC_PROVIDER 选择：

共享发送预留，以及操作和诊断授权，属于 SQL/控制平面能力，不是新的队列命令格式。现有 JSON 和 Avro SEND_EMAIL 载荷仍可读取。协议 warmup_send_protocol=2 控制新的执行器检查，协议 1 仅代表此前的来源链能力。先升级所有控制平面和结果消费者，暂停并处理完已接受的旧任务，再升级工作节点。结果未知的发送保留预留，不会变成盲目队列重试。结果核对会在提交前，将退信 Webhook 和通知写入持久化发件箱；消费者通过租约获取行，并使用稳定的下游 ID 投递。外部集成和实时投递不保证恰好一次。参见发送安全与发布。

  • json：普通 JSON，无外部依赖，是自托管服务栈的默认值。
  • avro：Avro 与 Confluent Schema Registry，使用 SCHEMA_REGISTRY_URL 及可选的密钥/秘密，同样受 kafka 构建标签控制。subject 为主题名加 -value，采用 Confluent 默认命名策略，因此 warmup-events 注册为 warmup-events-value。

每个派生字段都带有默认值

Schema 从 Go 结构体（internal/models/event_schema.go）派生，每个字段都会获得与其零值对应的 Avro 默认值。正因如此，新增字段才是安全的：使用新 Schema 的读取器仍可解码此前注册 Schema 写入的数据，并用默认值填补旧数据没有的字段，注册表也因此接受新版本。新增字段如果没有默认值，会在 BACKWARD 兼容性下被拒绝；而发布者在序列化之前注册 Schema，因此拒绝会阻止该主题的每一次发布，不只是影响某个字段。

由此有两点要求。注册的文档必须是序列化后的 Schema，而不是会省略默认值、破坏这一保证的 Schema.String()。注册表保持 BACKWARD，不使用 FORWARD：两者都允许新增字段，但新增事件类型会添加联合分支，BACKWARD 接受，FORWARD 则拒绝。当注册被拒绝，且 subject 的实际兼容级别为 FORWARD 或 FULL，或其传递形式时，发布者会把该 subject 设为 BACKWARD 并重新注册，因此注册表密钥需要修改 subject 兼容性的权限。其他级别及其他拒绝情况保持原样。

这些改动必须经过测试才会进入生产。internal/app/eventschemas 列出每个版本发布的所有 Schema，并在 testdata/ 下保留注册表最后接受的版本。CI 用新 Schema 解码已记录的旧 Schema，这与注册表自身的 BACKWARD 检查一致；注册表会拒绝的改动，CI 也会失败。即使改动兼容，但未更新记录，CI 同样会失败，因此 Schema 改动必然出现在差异中：运行 make schemas 并提交结果。发布时，后端在启动过程中注册所有列出的 Schema，成功后节点集群才会切换到该版本。

注册表检查覆盖 Schema，不覆盖发布过程。读取器使用每条消息携带的 Schema ID 解码，因此仍运行上一版本的工作节点可以读取新增字段；但从未见过的命令类型会被记录并跳过。后端先于节点集群发布，因此新增工作节点命令类型必须允许在集群追赶期间丢失：只向报告版本已支持该命令的节点发送，或让后端持续重试，直到节点支持为止。

Rust 跟踪服务遵循这两个开关。只有启用 Cargo 特性 kafka 编译时才支持 Kafka，tracking:*-kafka 镜像就是这种构建；两种传输都通过 CODEC_PROVIDER 选择编码。因此，Kafka 部署使用 CODEC_PROVIDER=json 时完全不需要 Schema Registry；选择 avro 却没有 SCHEMA_REGISTRY_URL 时，会在启动时被拒绝，而不会等到发布第一个事件才失败。

两个主题使用同一种编码

消费者解码 jobs.worker-events 和 tracking-events 时使用相同的 CODEC_PROVIDER。因此，跟踪发布者必须读取此设置，不能始终写 Avro：JSON 消费者收到 Avro 时，会丢弃所有打开和点击事件，只留下反序列化警告。

主题

以下为实际主题名。在 NATS 中，每个主题都会加上 subject 前缀，默认是 warmbly.。

主题生产者消费者用途
w.<worker-uuid>后端具有对应 ID 的工作节点发给指定工作节点的命令
jobs.worker-events工作节点消费者服务同步及发送结果、邮箱更新、健康状态
tracking-events跟踪服务消费者服务邮件打开和链接点击
email-events后端无，仅写分析流营销活动发送记录
warmup-events后端无，仅写分析流预热发送记录

每个工作节点只订阅自己的 w.<worker-uuid> 主题；UUID 是工作节点身份，在安装时根据公网 IP 派生。消费者服务读取 jobs.worker-events 时使用消费者组 consumer-group，读取 tracking-events 时使用组 tracking-consumer。

工作节点命令

后端使用 {type, body} 消息封装驱动工作节点，定义见 internal/models/event.go：

{
  "type": "SEND_EMAIL",
  "body": { ... }
}

类型包括：SEND_EMAIL、ADD_EMAIL、REMOVE_EMAIL、EMAIL_VALIDATION、WARMUP_ACTION、MESSAGE_SEEN、MESSAGE_FOLDER 和 MAILBOX_IDENTITY。每种类型都在 internal/models/ 中有对应的类型化消息体结构，例如 models.SendEmail 对应 SEND_EMAIL。

工作节点逐条读取自己的主题，因此耗时操作不会直接运行在该循环中。ADD_EMAIL 在后台连接邮箱；邮箱已在加载时，重复发布会被忽略，针对它的命令则等待加载完成。无法加载的邮箱会产生与同步失败相同的 EMAIL_AUTH_ERROR 或 EMAIL_SERVER_ERROR，从而显示错误，而不是看起来已连接。在 Kafka 中，已处理消息的偏移量会保存并在后台提交，不会每条消息都同步提交。

后端的工作节点协调器会立即为没有工作节点、所在节点停止心跳或已迁移到其他节点的邮箱发送 ADD_EMAIL。其余活跃邮箱每 30 分钟重新发送一次，作为安全兜底；各邮箱分散在该时间间隔内的不同时间点执行。

EMAIL_VALIDATION 通常完全不通过主题发送。后端把相同消息体发布到工作节点的 Redis 渠道 email_validation_request:<worker uuid>，工作节点并行读取该渠道和自己的主题，因此检查无需排在发送队列后。Redis 会报告接收到消息的订阅者数量；没有订阅者时尝试下一个健康工作节点，只有没有任何工作节点监听该渠道时，检查才会进入主题。

MESSAGE_SEEN 将统一收件箱中的已读或未读更改传递给邮箱服务商；每个邮箱一个事件，每个事件最多包含 models.SeenRelayChunk 封邮件。它不返回响应：事件发布之前存储已更新，因此它只修改服务商侧副本。失败只记录日志，不重试，因为下一次同步会报告服务商实际保存的状态。每封邮件携带三类服务商所需的句柄，工作节点选择其客户端适用者：Gmail 和 Graph 的服务商邮件 ID、IMAP 文件夹及 UID，以及不可变的 RFC Message-ID。Graph 会根据后者重新解析实时 ID，因为邮件每次移动都会改变 Graph ID。转发的是再次读取该行时保存的状态，不是请求最初要求的状态，因此发生冲突切换时，服务商会与存储一致，而不是与最后发布消息的人一致。

MESSAGE_FOLDER 对启用了 email_accounts.relay_folder_moves 的邮箱，将统一收件箱中的归档、删除或移至收件箱操作传给服务商。每个邮箱和目标位置一个事件，每个事件最多 models.FolderRelayChunk 封邮件，携带与 MESSAGE_SEEN 相同的句柄，另加行 ID。Gmail 通过一次系统标签 batchModify 批量整理。Outlook 将每封邮件移到其标准文件夹，并在移动前后把邮件映射从旧 Graph ID 更新为新 ID，因为增量同步会报告旧 ID 已删除，否则会误删该行。IMAP 移动前通过 Message-ID 查找每封邮件，因为先前归档的响应仍在途中时，存储中的 UID 可能已经过期，例如归档后立刻撤销；随后对每个文件夹的邮件用一次 MOVE 移动。归档改变了行的位置，或 provider_folder 与其文件夹不一致时，邮件才会被转发；Outlook 和 IMAP 的已发送副本保持原位，草稿和预热回执绝不移动。与 MESSAGE_SEEN 一样，它在响应后发布且不重试，因此旧版工作节点会丢弃它，邮箱副本会保持原位，直到下一次整理。

MAILBOX_IDENTITY 请求持有邮箱的工作节点，从服务商读取其可用发件地址和一个签名。响应发送到以请求 process_id 命名的 Redis 渠道，不走 worker-events 主题，与 EMAIL_VALIDATION 使用相同往返方式，因为调用方是正在等待结果的 HTTP 请求。控制平面仅在一处自行执行此调用：OAuth 握手期间，此时令牌来自客户刚完成的同意流程，邮箱还没有工作节点。

工作节点结果

工作节点通过 jobs.worker-events 发布相同封装格式的结果：

{
  "type": "NEW_EMAIL",
  "body": { ... }
}

类型包括：NEW_EMAIL、INBOUND_BOUNCE、REMOVE_EMAIL、FLAGS_ADD、FLAGS_REMOVE、UPDATE_EMAIL、UPDATE_FOLDER、UPDATE_MAILBOX、DELETE_MAILBOX、TOKEN_UPDATE、HISTORY_ID_UPDATE、GRAPH_DELTA_UPDATE、SYNC_STATE、EMAIL_SENT、EMAIL_FAILED、EMAIL_AUTH_ERROR、EMAIL_DISABLED、EMAIL_RATE_LIMITED、EMAIL_SERVER_ERROR 和 WORKER_HEALTH。

消费者为每个类型注册一个处理器，见 internal/app/consumer/events.go；未注册的类型会记录日志并确认，不再重新投递。

每个 SEND_EMAIL 都得到且只得到一个任务级结果：EMAIL_SENT 表示服务商接受了邮件，消费者将实际传输的 Message-ID 记录到任务；EMAIL_FAILED 则携带包含错误代码和消息的 SendEmailResult。控制平面在把发送交给工作节点时，就将营销活动步骤标为已发送，因此需要由 EMAIL_FAILED 撤销：消费者将任务标为失败，清除步骤的 sent_at，并在 campaign_contact_progress 中记录尝试次数，退还营销活动每日计数，将失败写入活动日志，并重新打开在该发送尚未完成时已经结束的营销活动。达到 CampaignSendMaxAttempts（5）后，潜在客户会被标为失败并退出路由；RECIPIENT_REJECTED 结果表示在 RCPT 阶段被拒绝，会跳过重试，直接作为退信处理。账户级情况 EMAIL_AUTH_ERROR、EMAIL_DISABLED、EMAIL_RATE_LIMITED 和 EMAIL_SERVER_ERROR 会在任务级结果之外额外产生，绝不代替任务结果；它们携带 EmailErrorEvent 并作用于邮箱。工作节点尚未持有邮箱时，会允许发送重投几次再报告失败；后端也不会向未发送心跳的节点发布发送命令。

工作节点为 SEND_EMAIL 提供 120 秒，足以覆盖 SMTP 客户端与慢服务器之间最长 90 秒的会话，以及此前的派发检查和正文获取；其他命令仍为 30 秒。命令订阅的时限为 195 秒，也覆盖结果记录，避免第一次投递仍在完成时，总线再次分发发送命令。两种总线都按邮箱有序通道运行命令，由 WORKER_COMMAND_CONCURRENCY 控制，因此卡住的邮件服务器只会阻塞自己邮箱的命令。

UPDATE_FOLDER 报告服务商当前保存邮件的标准文件夹，不包含邮件其他内容。Gmail 历史记录添加或移除 INBOX、TRASH、SPAM、SENT 或 DRAFT 时会发送它，并根据邮件标签重新计算文件夹；每六小时及每次邮箱加载时的核对也会发送。工作节点列出 Gmail 收件箱，并通过后端 GET /api/v1/internal/sync/provider-folder-messages 查询系统认为处于收件箱、归档、垃圾邮件或垃圾箱的行，再报告 Gmail 实际放在其他位置的邮件。消费者像处理 UPDATE_EMAIL 一样，以服务商上次报告的文件夹为基准，因此在 Warmbly 中已整理的邮件会保持原位，除非服务商自身移动了它。工作节点也会发送设置了 relayed 的该事件，覆盖每封被 MESSAGE_FOLDER 放到目标位置的邮件，同时带上移动改变的句柄：新的 Graph ID 放在 provider_id，或 IMAP 的 folder_path、mailbox 和 uid。转发产生的事件只写 provider_folder 和句柄，不修改文件夹，因此随后同步报告相同位置时不会被理解为移动；在此期间已撤销的整理仍保持撤销。

SYNC_STATE 是工作节点转发的邮箱同步状态，包含历史回填进度和游标、合理使用限流以及最后同步时间。它携带完整状态而非增量，因此事件丢失可由下一次修复。消费者写入 email_sync_state，后端在下次加载的 ADD_EMAIL 中返回，使替换后的工作节点能够继续导入，而不是从头开始。

跟踪事件

Rust 跟踪服务在像素加载或跟踪链接被点击时生成事件，消费者将其计入营销活动统计和送达能力数据：

{
  "event_type": "EMAIL_OPENED",
  "task_id": "uuid",
  "original_url": "https://example.com/page",
  "link_id": "uuid",
  "timestamp": "2026-01-29T12:00:00Z",
  "user_agent": "Mozilla/5.0...",
  "ip_hash": "sha256...",
  "client_ip": "203.0.113.0",
  "scanner": "proofpoint",
  "scanner_probable": true
}

event_type 为 EMAIL_OPENED 或 EMAIL_CLICKED；original_url 和 link_id 仅在点击时设置，后者是点击票据，指向链接存储的目标和锚文本。IP 仅以哈希保存，不保留原始值，client_ip 只携带来源网段。scanner 表示边缘识别出的邮件过滤网络；scanner_probable 表示该网络也可能承载真人访问，邮件安全服务商的浏览器隔离正会产生这种情况。这些字段出现之前写入的事件均不含它们；缺失的 scanner_probable 按 false 读取，采用保守判断。结构体为 events.TrackingEvent，定义在 internal/events/schemas.go 中，并在 tracking/src/events.rs 中有对应定义。

消费者先分类，再计数。如果用户代理是邮件隐私代理或缺失，或事件在步骤派发后的 tracking.machine_window_open_seconds 内到达，打开会被判为自动行为；它以 opened_machine 记录，后续真人打开可以升级该记录。点击在以下情况下判为自动行为：缺少用户代理；在同一派发后的 tracking.machine_window_click_seconds 内到达，默认 30 秒，与打开窗口独立设置，同样接受 1 至 900 秒，默认较短是因为误判点击影响的是自动化操作，而不只是指标；或者同一来源在 TrackingClickBurstSeconds 内点击了同一邮件的另一链接。每次点击都在 email_link_clicks 中按链接记录原因，只有真人点击才写入 clicked_at、触发即时操作或发出 Webhook。突发点击必须从第二次点击起才能识别，因此真人点击的时间戳和日志行会立即写入，但其效果，包括证据、即时操作、Webhook 和实时事件，会延迟到突发窗口结束再加一秒后，按届时的分类执行。如果期间识别出突发行为，先前点击会重新标记；没有真人点击剩余时，时间戳会撤销。消费者在该窗口内重启只会丢失这些延迟效果，路由仍会在下一个步骤边界走已点击分支。

scanner 会覆盖上述判断：来自已识别邮件过滤网络的请求，无论用户代理如何声明、到达多晚，都属于自动行为，因为安全网关会用普通浏览器遍历邮件。scanner_probable 对此作进一步限定。可能承载真人的来源会使用 tracking.machine_window_probable_seconds 与该事件类型窗口中较宽的那个；窗口之外，则与未识别来源完全相同地判断。这使服务商 ASN 可以默认启用：投递时的扫描判为自动行为，一小时后通过浏览器隔离点击的收件人仍能计入。真实收件人若恰在该窗口内点击，也会被判为自动行为，这是扩大窗口的代价。

这一标签还有一个不由消费者决定的效果。跟踪服务不会向任何已识别网络提供点击身份识别票据，包括可能承载真人的网络，因为边缘没有派发时间，无法区分投递时扫描和隔离浏览器中的真人点击。因此，即使消费者后来将某个可疑来源点击计为真人，它仍然没有后续网站访问归因。

网站页面浏览不经过此主题。跟踪服务会将每个接受的访问直接转发到后端内部 API POST /api/v1/internal/page-hits，因为用户代理和 IP 在后端才会转换为设备及位置，IP 不应在传输途中存入持久流。

分析事件

后端将发送记录写入两个只写流，供下游分析：email-events 携带 EMAIL_SENT 记录，包括任务、账户、营销活动、联系人、邮件 ID 和收件人；warmup-events 携带 WARMUP_EMAIL_SENT 记录，包括发件与目标账户，以及是否为回复。目前仓库中没有服务消费它们，保留这些流是为了让外部管道可以接入。

实时事件分发

控制台实时功能不通过事件总线。后端和消费者在专用发布/订阅传输层 internal/infrastructure/pubsub/ 上发布 JSON 事件，Elixir 实时服务订阅并推送到 WebSocket 客户端：

  • 默认使用 Redis 发布/订阅渠道 realtime:events，除了共享 Redis 外无需其他配置。
  • PUBSUB_ENABLED=true 时，改用 Google Cloud Pub/Sub，需要 GCP_PROJECT_ID；后端在启动时创建固定的一组主题和对应的 -sub 订阅。

事件嵌入 BaseEvent，包含 event_type、user_id 和 timestamp，具体载荷再带路由字段 org_id、campaign_id、email_account_id 和 operation_id。Elixir 广播器只根据这些消息体字段，将事件路由到 user:、org:、campaign:、account: 和 bulk: 渠道主题。事件类型列表和权限控制见实时 API 参考。

配置

# Transport
EVENTBUS_PROVIDER=nats            # nats | kafka (kafka needs the kafka build tag)
NATS_URL=nats://nats:4222
NATS_STREAM_NAME=warmbly          # optional, default warmbly
NATS_SUBJECT_PREFIX=warmbly       # optional, default warmbly
EVENTBUS_HANDLER_TIMEOUT=30s      # optional, per-message handler timeout (worker commands run under 195s)

# Codec
CODEC_PROVIDER=json               # json | avro (avro needs the kafka build tag)

# Kafka path only
KAFKA_BOOTSTRAP_SERVERS=broker1:9092,broker2:9092
KAFKA_SASL_USERNAME=...
KAFKA_SASL_PASSWORD=...
SCHEMA_REGISTRY_URL=https://schema.example.com
SCHEMA_REGISTRY_KEY=...
SCHEMA_REGISTRY_SECRET=...
KAFKA_TRACKING_TOPIC=tracking-events   # optional, default tracking-events

# Realtime fanout
PUBSUB_ENABLED=false              # true switches Redis pub/sub to Google Cloud Pub/Sub

代码参考

  • 总线抽象及提供程序：internal/infrastructure/eventbus/
  • 编码选择：internal/infrastructure/codec/
  • 主题名称：internal/infrastructure/kafka/topics.go、internal/events/schemas.go
  • 消息封装：internal/models/event.go
  • 消费者分发：internal/app/consumer/events.go
  • 跟踪生产者：tracking/src/nats.rs、tracking/src/kafka.rs
  • 实时分发：internal/infrastructure/pubsub/、realtime/lib/realtime/redis/event_subscriber.ex

架构

介绍 Warmbly 控制平面与执行平面的协作方式，以及加密模型和工作节点模型。

本页目录

传输与编解码主题工作节点命令工作节点结果跟踪事件分析事件实时事件分发配置代码参考