Echo — 私有通知分发网关 (架构研读与使用笔记)

⚠️ 个人自用项目，不对外开源。
本文档主要用于：1. 剖析整洁六边形架构与源码学习路径；2. 记录日常调测与上游接入用法。

---

第一部分：架构剖析与学习指南

1. 这个项目的架构到底长什么样？

Echo 的核心定位是 纯粹的消息投递网关（Pure Delivery Gateway）。
它遵循了经典的 六边形架构（Hexagonal Architecture / Ports & Adapters） 与 整洁架构（Clean Architecture） 思想：核心领域层完全不依赖外部框架与具体通道，业务数据与计算全部留在外部上游。

依赖倒置拓扑（Dependency Rule）

所有依赖关系严格单向朝内，核心层 (core) 拥有绝对的主权：

flowchart TD
    subgraph Inbound ["输入端适配器 (Entrypoints)"]
        HTTP["HTTP API (/v1/messages)"]
        CLI["CLI 命令行 (echo-push)"]
    end

    subgraph Core ["【领域核心层 core】(绝对内向依赖)"]
        Gateway["MessageGateway\n(应用统一门面)"]
        Dispatcher["MessageDispatcher\n(路由校验、并发分发、故障隔离)"]
        Ports["Provider 抽象契约\n(core/ports.py)"]
        Models["不可变数据模型\n(Message / DispatchReport)"]

        Gateway --> Dispatcher
        Dispatcher --> Ports
        Dispatcher -.-> Models
    end

    subgraph Outbound ["输出端适配器 (Providers)"]
        Bark["BarkProvider\n(bark.py)"]
        TG["TelegramProvider\n(telegram.py)"]
    end

    subgraph Wiring ["装配与配置 (Bootstrap & Config)"]
        Config["配置加载器\n(config/loader.py)"]
        Bootstrap["组合根 Composition Root\n(bootstrap.py)"]
    end

    HTTP -->|调用| Gateway
    CLI -->|调用| Gateway
    Bark -->|实现接口| Ports
    TG -->|实现接口| Ports
    Config -->|读取环境与TOML| Bootstrap
    Bootstrap -->|实例化并注入| Gateway
    Bootstrap -.->|构建| Bark
    Bootstrap -.->|构建| TG

请求执行生命周期时序图

一次完整的从消息传入到多目标并发隔离投递的全过程：

sequenceDiagram
    autonumber
    actor Caller as 上游服务 / CLI
    participant Entry as Entrypoint (HTTP / CLI)
    participant Gateway as MessageGateway
    participant Dispatcher as MessageDispatcher
    participant Pool as ThreadPoolExecutor
    participant Target as Provider (Bark / TG)

    Caller->>Entry: 提交请求 (Message + RouteSelection)
    Entry->>Entry: 校验传输协议 / Bearer Token 鉴权
    Entry->>Gateway: 调用 .send(message, selection)
    Gateway->>Dispatcher: 委派 .dispatch(message, selection)
    Dispatcher->>Dispatcher: 防御性校验 (未知通道/未选目标立即拦截)
    par 并发投递各目标设备 (故障隔离)
        Dispatcher->>Pool: 提交任务 pool.submit(send)
        Pool->>Target: 发起 HTTP POST 请求
        Target-->>Pool: 返回 DeliveryResult (耗时与状态)
    end
    Pool-->>Dispatcher: 汇总所有目标结果
    Dispatcher-->>Gateway: 封装为不可变 DispatchReport
    Gateway-->>Entry: 返回 DispatchReport
    Entry-->>Caller: 响应 JSON (全成功 200 / 部分失败 207)

• 绝对单向依赖：core 内部绝不 import providers、entrypoints 或外部网络库。
• 组合根设计（Composition Root）：整个工程只有 bootstrap.py 知道所有具体类的存在，负责把配置、具体 Provider 和 Dispatcher 装配成 MessageGateway。
• 零外部运行时依赖：100% 纯 Python 3.11+ 标准库编写，不用 requests、FastAPI 或 Celery，冷启动毫秒级。

---

2. 怎么高效研读这个项目的源码？（推荐学习路线）

按照以下 5 个阶段 顺序看代码，能最快掌握其设计精髓：

第 1 步：读数据模型与协议契约（理解系统骨架）

• 📄 src/echo_service/core/models.py：
  • 学习 Message、RouteSelection、DeliveryResult、DispatchReport。
  • 亮点：全面使用 @dataclass(frozen=True, slots=True) 实现不可变值对象，杜绝运行时状态篡改。
• 📄 src/echo_service/core/ports.py：
  • 看 Provider 抽象基类。整个网关对下游推送通道的抽象极其克制，仅要求实现 name、target_ids 和 send()。

第 2 步：读并发与分发核心（核心精髓）

• 📄 src/echo_service/core/dispatcher.py：
  • 学习重点 1（严格防御性校验）：在发送前比对配置，若请求中包含未知 provider 或未选 provider 的 target，立刻拦截抛出 SelectionError。
  • 学习重点 2（并发与故障隔离）：看 ThreadPoolExecutor 的使用。关键在于 future.result() 的异常捕获——即使某个 Provider 崩溃抛出异常，也绝不影响其他 Provider 和 Target 的投递，所有错误会被清洗包装进 DeliveryResult 中。
• 📄 src/echo_service/core/gateway.py：
  • 学习门面模式（Facade），收敛对外暴露的接口，只提供 providers() 和 send()。

第 3 步：看外部适配器与轻量设施

• 📄 src/echo_service/infrastructure/http.py：
  • 学习如何仅用 Python 标准库 urllib.request 实现一个线程安全、带超时、支持 JSON 自动反序列化的极简 HTTP 客户端。
• 📄 src/echo_service/providers/：
  • 看 bark.py 和 telegram.py 如何把抽象的 Message 映射为各自平台的专用参数（如 Bark 分组、Telegram 线程 ID）。

第 4 步：看依赖装配（组合根）

• 📄 src/echo_service/bootstrap.py：
  • 学习 build_gateway(config)：从纯配置解析出具体的 Provider 列表，注入给 Dispatcher，最后暴露 Gateway。所有脏活（依赖构建）全部收敛在此处。

第 5 步：看接入层与配置校验

• 📄 src/echo_service/entrypoints/http.py：
  • 原生 ThreadingHTTPServer 实现。学习 Bearer 令牌的恒定时间比较（hmac.compare_digest 防时序攻击）以及 HTTP 207 Multi-Status 的状态码设计。
• 📄 src/echo_service/config/loader.py：
  • 学习基于正则的高效环境变量展开（${ENV}），以及启动时的严格白名单校验（显式拒绝旧版非网关字段）。

---

第二部分：怎么使用？（日常运维与调用）

1. 本地初始化与配置

步骤一：准备环境与密钥

# 1. 创建虚拟环境
python3 -m venv .venv
source .venv/bin/activate
pip install -e .

# 2. 从模板复制配置文件
cp config/echo.example.toml config/echo.toml
cp .env.example .env

# 3. 编辑 .env，填写真实密钥
# ECHO_API_KEY=my-secret-key
# BARK_DEVICE_KEY_IPHONE=xxxx
# TELEGRAM_BOT_TOKEN=xxxx
# TELEGRAM_CHAT_ID=xxxx

步骤二：配置文件说明 (config/echo.toml)

配置只负责声明服务网络与推送渠道，绝不包含业务计算规则：

[server]
host = "127.0.0.1"
port = 8787
api_key = "${ECHO_API_KEY}"
max_body_bytes = 1048576

[runtime]
max_workers = 4  # 推送并发线程数

[providers.bark]
base_url = "https://api.day.app"
timeout_seconds = 10

[[providers.bark.targets]]
id = "my-iphone"
device_key = "${BARK_DEVICE_KEY_IPHONE}"

[providers.telegram]
api_url = "https://api.telegram.org"
timeout_seconds = 10

[[providers.telegram.targets]]
id = "personal"
bot_token = "${TELEGRAM_BOT_TOKEN}"
chat_id = "${TELEGRAM_CHAT_ID}"

---

2. 服务的启动与常驻

开发与本地调试

# 导入 .env 并启动前置配置语法检查
set -a; source .env; set +a
echo-push --config config/echo.toml check-config

# 前台启动 HTTP 服务
echo-push --config config/echo.toml serve

生产服务器部署 (Docker Compose)

容器被配置为只读根文件系统、非 root 运行、内存限制 16MB：

# 启动常驻
docker compose up -d --build

# 查看运行状态与日志
docker compose ps
docker compose logs -f

# 探针检测 (无需鉴权)
curl http://127.0.0.1:8787/health

---

3. 如何调用与推送消息？

场景 A：服务器本地脚本调用 (CLI)

适合在服务器定时备份、系统异常捕获脚本中直接调用：

# 1. 查看当前已加载的通道与设备 ID
echo-push --config config/echo.toml list

# 2. 全员广播推送 (不带 target 时推送到全部启用目标)
echo-push --config config/echo.toml send \
  --title "服务器告警" \
  --body "磁盘空间已低于 10%"

# 3. 精准单点推送 (只推 iPhone)
echo-push --config config/echo.toml send \
  --body "验证码: 839201" \
  --provider bark \
  --target bark:my-iphone

配置邮箱发送（可选）

邮箱沿用现有 Provider 接口，不需要修改 HTTP/CLI 请求结构。将以下配置加入 config/echo.toml，并在 .env 中填写模板列出的邮箱变量，然后按原有方式导入环境变量：

[providers.email]
host = "smtp.example.com" # 替换为邮箱服务商提供的 SMTP 主机
security = "starttls"
port = 587
sender = "${SMTP_SENDER}"
username = "${SMTP_USERNAME}"
password = "${SMTP_PASSWORD}"
timeout_seconds = 10

[[providers.email.targets]]
id = "personal-mail"
address = "${EMAIL_RECIPIENT}"

starttls 默认使用 587 端口；需要隐式 TLS 时设置 security = "ssl"，默认端口为 465，也可显式指定服务商要求的端口。两种模式都会验证服务器证书，STARTTLS 失败时 立即终止发送。凭据优先使用服务商提供的 SMTP 授权码或应用密码；无需认证的加密 中继可同时省略 username 和 password。发件和收件地址使用单个 ASCII 邮箱地址， 不带显示名称。每个收件人使用独立目标配置，可通过 enabled = false 禁用通道或目标。

echo-push --config config/echo.toml check-config
echo-push --config config/echo.toml send \
  --title "服务器告警" --body "备份已完成" \
  --provider email --target email:personal-mail

HTTP 请求使用 "providers": ["email"] 和 "targets": {"email": ["personal-mail"]}，也可与 Bark、Telegram 一起分发。 邮件主题取 message.title（省略时为 Echo notification），正文为 UTF-8 纯文本， 附带 message.url；目前不处理 HTML、附件或额外邮件选项。 成功表示 SMTP 服务器已接受邮件，后续退信或收件箱投递不在网关职责范围内。 启用邮箱后，不指定 provider 的广播也会向邮箱目标发送。

场景 B：上游业务系统集成 (HTTP REST API)

在其他业务服务（如 Bills 记账服务、Git Hook 触发器、监控系统）中，构造标准的 JSON 请求：

• 请求端点：POST http://127.0.0.1:8787/v1/messages
• 鉴权头：Authorization: Bearer <ECHO_API_KEY>

curl -X POST http://127.0.0.1:8787/v1/messages \
  -H "Authorization: Bearer $ECHO_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "message": {
      "title": "Bills 预算预警",
      "body": "本月餐饮消费已达 82%",
      "url": "https://bills.my-domain.com",
      "tags": ["budget", "finance"],
      "options": {
        "bark": {
          "group": "财务提醒",
          "level": "timeSensitive"
        },
        "telegram": {
          "disable_notification": false
        }
      }
    },
    "providers": ["bark", "telegram"],
    "targets": {
      "bark": ["my-iphone"],
      "telegram": ["personal"]
    }
  }'

状态码与响应语义

• HTTP 200 OK：全部指定的目标均投递成功。
• HTTP 207 Multi-Status：部分目标成功，部分失败（例如 iPhone 成功但 Telegram 超时）。响应体中清晰记录每个 target 的错误原因与耗时：

  {
    "ok": false,
    "attempted": 2,
    "succeeded": 1,
    "failed": 1,
    "results": [
      { "provider": "bark", "target": "my-iphone", "ok": true, "duration_ms": 120 },
      { "provider": "telegram", "target": "personal", "ok": false, "error": "Telegram request failed: TimeoutError", "duration_ms": 10002 }
    ]
  }
  

---

4. 代码回归与质量检验

每次修改代码后，运行以下命令确保通过严格的自检：

# 静态字节码编译检查
make check

# 运行全套单元测试（模拟外部发送；HTTP 接口测试监听本地临时端口）
make test