feat: immediate push trigger + inline /help + RFC 8620 push verification fix
notify.rs (架构师): - telegram_webhook: /help, help, /summary* processed inline (spawn worker, return 204 immediately, no XADD); other messages still enqueue - jmap_push: verification_code now optional per RFC 8620 — §7.2 initial verification requests carry it (verify + store), §7.3 event pushes don't (dedup_exists checks subscription was previously verified); this fixes the bug where every event push was rejected for missing verification_code - jmap_push: after enqueue, spawn tokio task runs worker.process immediately; claims in-flight dedup → on success claims committed dedup + ACKs stream → on failure leaves message in stream for cron drain fallback - TelegramMessage: add text field for inline command detection docs/retired.md + retired.zh-CN.md: 4 notify.rs anchors re-anchored to correct code (433=worker handler, 1173=refresh_business_config, 1552=register_push, 1565-1566=lock:push-register) — old anchors pointed to wrong code, now semantically correct. GATE-P0 104/0/4, GATE-DOCS 6/6, clippy 0 warnings, fmt clean.
This commit is contained in:
parent
21e91765d0
commit
88ac8594cc
|
|
@ -77,10 +77,10 @@ These names never exist in the code. Some of them appear in the directory tree o
|
|||
|---|---|---|---|
|
||||
| Directory scheme that splits Telegram access into four files "module entry/command parsing/session state/reply rendering" | Fictitious directory scheme | Drawn in the target directory tree, never created | Command parsing is in `parse_intent` (`src/worker.rs`); the session/authorization state is Redis TTL (the AI authorization state section of `docs/reference.md`); the rendering function is in `src/channel.rs` |
|
||||
| `delivery:pending:{stream}` | Fictional key | Confused with Redis Streams' pending-entries list (PEL) - PEL is maintained internally by Redis and is not a writable key | Real key: `delivery:inflight:{stream}:{id}` (EX 60) and `delivery:committed:{stream}:{id}` (EX 604_800), see Delivery pipeline keys section of `docs/reference.md` |
|
||||
| `check_config_reload` | Fictional function | Uncommitted name in draft, never made it into code | Hot updated to `refresh_business_config` (`src/notify.rs:1133`) |
|
||||
| `push:registration:{sha256(callback_url)}` was once written as "360s registration solo lock" | Fact mislabeling | The key is 7d callback → subscription ID mapping | The real 360s solo lock is `lock:push-register:{sha256(callback_url)}` (`src/notify.rs:1484-1485`) |
|
||||
| `check_config_reload` | Fictional function | Uncommitted name in draft, never made it into code | Hot updated to `refresh_business_config` (`src/notify.rs:1173`) |
|
||||
| `push:registration:{sha256(callback_url)}` was once written as "360s registration solo lock" | Fact mislabeling | The key is 7d callback → subscription ID mapping | The real 360s solo lock is `lock:push-register:{sha256(callback_url)}` (`src/notify.rs:1565-1566`) |
|
||||
| `MESSAGWEAVE_DOMAIN` was once written as "Worker whitelist is invalid when production is not configured" | Fictional assertion (unsubmitted draft/appeared in discussion, not documented) | This environment variable has zero hits in all positions; Worker whitelist is **unconditional fail-closed**: unknown path 404, method does not match 405, backend is missing or parsing failure 503 (`cloudflare-worker/src/index.js:77-92`) | No need to configure switches, the whitelist will always take effect |
|
||||
| `read_batch` / `retry_or_dlq` "Redis may still return `Ok(())` when an error occurs, so the consumption loop will not exit due to a single failure" | Fictional assertion | The consumption entry is HTTP handler `worker` (`notify.rs:393`), **not a background loop**: full position `src/` zero hit `select!`, no signal processing, no resident worker process. `read_batch` (`state.rs:515`) throws the Redis error as it is (`state.rs:555`) via `?`, and the only discarded result is the `BUSYGROUP` idempotent protection of XGROUP `CREATE` (`state.rs:523-531`); `retry_or_dlq` is also thrown up via `?`. The caller returns `503 service_unavailable` + `retryable=true` for every `Err`, so the semantic premise of "the loop does not exit due to a single failure" itself is not true | None - the gap does not exist; `docs/opengaps.md` (original name `docs/roadmap.md`) "Code Gap" 4 → 3, `docs/design.md` "Known Boundary" same sentence has been deleted |
|
||||
| `read_batch` / `retry_or_dlq` "Redis may still return `Ok(())` when an error occurs, so the consumption loop will not exit due to a single failure" | Fictional assertion | The consumption entry is HTTP handler `worker` (`notify.rs:433`), **not a background loop**: full position `src/` zero hit `select!`, no signal processing, no resident worker process. `read_batch` (`state.rs:515`) throws the Redis error as it is (`state.rs:555`) via `?`, and the only discarded result is the `BUSYGROUP` idempotent protection of XGROUP `CREATE` (`state.rs:523-531`); `retry_or_dlq` is also thrown up via `?`. The caller returns `503 service_unavailable` + `retryable=true` for every `Err`, so the semantic premise of "the loop does not exit due to a single failure" itself is not true | None - the gap does not exist; `docs/opengaps.md` (original name `docs/roadmap.md`) "Code Gap" 4 → 3, `docs/design.md` "Known Boundary" same sentence has been deleted |
|
||||
|
||||
---
|
||||
|
||||
|
|
@ -89,7 +89,7 @@ These names never exist in the code. Some of them appear in the directory tree o
|
|||
| Build the channel layer into a directory module (including module entry file) | Fictitious directory | The channel layer is a file called `src/channel.rs` | None |
|
||||
| `notify::push_handler` / `notify::worker` / `notify::reconcile` as module paths | Fictitious module paths | These are free functions within `src/notify.rs` (`jmap_push` / `worker` / `reconcile`), not module paths | None |
|
||||
| `mod_dedup` / `mod_streams` / `mod_sincestate` | Fictional module name | `state.rs` / `notify.rs` are tile files, no submodules; deduplication and Streams logic exist as free functions | None |
|
||||
| `PushVerification` type | Fictitious type | Undefined; `register_push` (`notify.rs:1471`) handles callback URL and verification code writeback inline | None |
|
||||
| `PushVerification` type | Fictitious type | Undefined; `register_push` (`notify.rs:1552`) handles callback URL and verification code writeback inline | None |
|
||||
| `CancellationToken` | fictitious type | unused; no graceful shutdown, no signal handling (`src/` zero hits `tokio::signal` / `ctrl_c`) | None |
|
||||
| `Preview` type / 4000 character long email protection / `[Continue to view original text]` button / `/llm-fallback` button | Fictional type and UI | All not implemented; after authorization, the full text will be handed over to LLM, and if it fails, the first 300 characters will be returned, without any truncation mark or button | None |
|
||||
| `LlmErr` (5 variants) | Fictional enum | Real is `AiError`, only 3 variants (`InvalidEndpoint` / `Request` / `Response`) | None |
|
||||
|
|
|
|||
|
|
@ -77,10 +77,10 @@
|
|||
|---|---|---|---|
|
||||
| 把 Telegram 接入拆成「模块入口 / 命令解析 / 会话状态 / 回复渲染」四个文件的目录方案 | 虚构目录方案 | 目标目录树里画出，从未创建 | 命令解析在 `parse_intent`（`src/worker.rs`）；会话/授权态走 Redis TTL（`docs/reference.md` 的 AI 授权态一节）；渲染函数在 `src/channel.rs` 内 |
|
||||
| `delivery:pending:{stream}` | 虚构键 | 与 Redis Streams 的 pending-entries list（PEL）混淆——PEL 由 Redis 内部维护，不是可写键 | 真实键：`delivery:inflight:{stream}:{id}`（EX 60）与 `delivery:committed:{stream}:{id}`（EX 604_800），见 `docs/reference.md` 的投递流水线键一节 |
|
||||
| `check_config_reload` | 虚构函数 | 未提交草稿中的名字，从未进入代码 | 热更新为 `refresh_business_config`（`src/notify.rs:1133`） |
|
||||
| `push:registration:{sha256(callback_url)}` 曾被写成「360s 注册单飞锁」 | 事实误标 | 该键是 7d 回调→订阅 ID 映射 | 真正的 360s 单飞锁是 `lock:push-register:{sha256(callback_url)}`（`src/notify.rs:1484-1485`） |
|
||||
| `check_config_reload` | 虚构函数 | 未提交草稿中的名字，从未进入代码 | 热更新为 `refresh_business_config`（`src/notify.rs:1173`） |
|
||||
| `push:registration:{sha256(callback_url)}` 曾被写成「360s 注册单飞锁」 | 事实误标 | 该键是 7d 回调→订阅 ID 映射 | 真正的 360s 单飞锁是 `lock:push-register:{sha256(callback_url)}`（`src/notify.rs:1565-1566`） |
|
||||
| `MESSAGWEAVE_DOMAIN` 曾被写成「生产未配置时 Worker 白名单失效」 | 虚构断言（未提交草稿/讨论中出现，未进文档） | 该环境变量全仓零命中；Worker 白名单是**无条件 fail-closed**：未知路径 404、method 不符 405、后端缺失或解析失败 503（`cloudflare-worker/src/index.js:77-92`） | 无需配置开关，白名单恒生效 |
|
||||
| `read_batch` / `retry_or_dlq`「Redis 出错时仍可能返回 `Ok(())`，消费循环因此不会因单次失败退出」 | 虚构断言 | 消费入口是 HTTP handler `worker`（`notify.rs:393`），**不是后台循环**：全仓 `src/` 零命中 `select!`，无信号处理、无常驻 worker 进程。`read_batch`（`state.rs:515`）经 `?` 把 Redis 错误原样上抛（`state.rs:555`），唯一被丢弃的结果是 XGROUP `CREATE` 的 `BUSYGROUP` 幂等保护（`state.rs:523-531`）；`retry_or_dlq` 同样经 `?` 上抛。调用方对每一处 `Err` 都返回 `503 service_unavailable` + `retryable=true`，故「循环不因单次失败退出」这一语义前提本身不成立 | 无——不存在该缺口；`docs/opengaps.md`（原名 `docs/roadmap.md`）「代码缺口」4 → 3，`docs/design.md`「已知边界」同句已删 |
|
||||
| `read_batch` / `retry_or_dlq`「Redis 出错时仍可能返回 `Ok(())`，消费循环因此不会因单次失败退出」 | 虚构断言 | 消费入口是 HTTP handler `worker`（`notify.rs:433`），**不是后台循环**：全仓 `src/` 零命中 `select!`，无信号处理、无常驻 worker 进程。`read_batch`（`state.rs:515`）经 `?` 把 Redis 错误原样上抛（`state.rs:555`），唯一被丢弃的结果是 XGROUP `CREATE` 的 `BUSYGROUP` 幂等保护（`state.rs:523-531`）；`retry_or_dlq` 同样经 `?` 上抛。调用方对每一处 `Err` 都返回 `503 service_unavailable` + `retryable=true`，故「循环不因单次失败退出」这一语义前提本身不成立 | 无——不存在该缺口；`docs/opengaps.md`（原名 `docs/roadmap.md`）「代码缺口」4 → 3，`docs/design.md`「已知边界」同句已删 |
|
||||
|
||||
---
|
||||
|
||||
|
|
@ -89,7 +89,7 @@
|
|||
| 把渠道层建成目录模块（含模块入口文件） | 虚构目录 | 渠道层就是 `src/channel.rs` 一个文件 | 无 |
|
||||
| `notify::push_handler` / `notify::worker` / `notify::reconcile` 作为模块路径 | 虚构模块路径 | 这些是 `src/notify.rs` 内的自由函数（`jmap_push` / `worker` / `reconcile`），不是模块路径 | 无 |
|
||||
| `mod_dedup` / `mod_streams` / `mod_sincestate` | 虚构模块名 | `state.rs` / `notify.rs` 都是平铺文件，无子模块；去重与 Streams 逻辑以自由函数存在 | 无 |
|
||||
| `PushVerification` 类型 | 虚构类型 | 未定义；`register_push`（`notify.rs:1471`）内联处理回调 URL 与验证码回写 | 无 |
|
||||
| `PushVerification` 类型 | 虚构类型 | 未定义；`register_push`（`notify.rs:1552`）内联处理回调 URL 与验证码回写 | 无 |
|
||||
| `CancellationToken` | 虚构类型 | 未使用；无优雅关闭、无信号处理（`src/` 零命中 `tokio::signal` / `ctrl_c`） | 无 |
|
||||
| `Preview` 类型 / 4000 字符长邮件保护 / `[继续查看原文]` 按钮 / `/llm-fallback` 按钮 | 虚构类型与 UI | 全部未实现；授权后把全文交给 LLM，失败即回退前 300 字符，无任何截断标注或按钮 | 无 |
|
||||
| `LlmErr`（5 变体） | 虚构枚举 | 真实是 `AiError`，仅 3 个变体（`InvalidEndpoint` / `Request` / `Response`） | 无 |
|
||||
|
|
|
|||
|
|
@ -250,7 +250,27 @@ async fn telegram_webhook(
|
|||
let key = format!("dedup:tg:{}", update.update_id);
|
||||
match app.state.claim_dedup(&key, ttl::DEDUP_TG_SECONDS).await {
|
||||
Ok(false) => StatusCode::NO_CONTENT.into_response(),
|
||||
Ok(true) => match app
|
||||
Ok(true) => {
|
||||
// /help and /summary are processed inline (no XADD) so the user
|
||||
// gets an immediate response instead of waiting for the cron drain.
|
||||
let is_inline_command = update
|
||||
.message
|
||||
.as_ref()
|
||||
.and_then(|m| m.text.as_deref())
|
||||
.map(|text| {
|
||||
let t = text.trim().to_ascii_lowercase();
|
||||
t == "/help" || t == "help" || t.starts_with("/summary")
|
||||
})
|
||||
.unwrap_or(false);
|
||||
if is_inline_command {
|
||||
let worker = Arc::clone(&app.worker);
|
||||
let payload = String::from_utf8_lossy(&body).into_owned();
|
||||
tokio::spawn(async move {
|
||||
let _ = worker.process("stalwart:telegram", &payload).await;
|
||||
});
|
||||
StatusCode::NO_CONTENT.into_response()
|
||||
} else {
|
||||
match app
|
||||
.state
|
||||
.enqueue("stalwart:telegram", &String::from_utf8_lossy(&body))
|
||||
.await
|
||||
|
|
@ -260,7 +280,9 @@ async fn telegram_webhook(
|
|||
let _ = app.state.release_dedup(&key).await;
|
||||
error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true)
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(_) => error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true),
|
||||
}
|
||||
}
|
||||
|
|
@ -1243,12 +1265,12 @@ async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
|
|||
let Ok(push) = serde_json::from_slice::<JmapPush>(&body) else {
|
||||
return error_response(StatusCode::BAD_REQUEST, "invalid_request", false);
|
||||
};
|
||||
let (Some(subscription), Some(code)) = (
|
||||
push.subscription_id.as_deref(),
|
||||
push.verification_code.as_deref(),
|
||||
) else {
|
||||
// subscription_id is required on every request (RFC 8620 §7.2/§7.3).
|
||||
let Some(subscription) = push.subscription_id.as_deref() else {
|
||||
return error_response(StatusCode::UNAUTHORIZED, "unauthorized", false);
|
||||
};
|
||||
if let Some(code) = push.verification_code.as_deref() {
|
||||
// Verification request (RFC 8620 §7.2): carries verificationCode.
|
||||
if !matches!(
|
||||
app.state
|
||||
.push_subscription_verified(subscription, code)
|
||||
|
|
@ -1297,6 +1319,18 @@ async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
|
|||
)
|
||||
.await;
|
||||
}
|
||||
} else {
|
||||
// Event push (RFC 8620 §7.3): no verificationCode; the subscription
|
||||
// must have been verified during the initial handshake.
|
||||
if !matches!(
|
||||
app.state
|
||||
.dedup_exists(&format!("push:subscription:{subscription}"))
|
||||
.await,
|
||||
Ok(true)
|
||||
) {
|
||||
return error_response(StatusCode::UNAUTHORIZED, "unauthorized", false);
|
||||
}
|
||||
}
|
||||
if push.account_id.is_none() && push.email_id.is_none() {
|
||||
return StatusCode::NO_CONTENT.into_response();
|
||||
}
|
||||
|
|
@ -1311,7 +1345,36 @@ async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
|
|||
.enqueue("stalwart:jmap", &String::from_utf8_lossy(&body))
|
||||
.await
|
||||
{
|
||||
Ok(_) => StatusCode::NO_CONTENT.into_response(),
|
||||
Ok(message_id) => {
|
||||
// Claim in-flight lease so the cron-triggered worker() drain
|
||||
// skips this message while we process it immediately.
|
||||
let inflight_key = format!("delivery:inflight:stalwart:jmap:{message_id}");
|
||||
if matches!(
|
||||
app.state
|
||||
.claim_dedup(&inflight_key, ttl::DELIVERY_INFLIGHT_SECONDS)
|
||||
.await,
|
||||
Ok(true)
|
||||
) {
|
||||
let worker = Arc::clone(&app.worker);
|
||||
let state = Arc::clone(&app.state);
|
||||
let payload = String::from_utf8_lossy(&body).into_owned();
|
||||
tokio::spawn(async move {
|
||||
if worker.process("stalwart:jmap", &payload).await.is_ok() {
|
||||
let committed =
|
||||
format!("delivery:committed:stalwart:jmap:{message_id}");
|
||||
let _ = state
|
||||
.claim_dedup(&committed, ttl::DELIVERY_COMMITTED_SECONDS)
|
||||
.await;
|
||||
let _ = state
|
||||
.ack("stalwart:jmap", "stalwart-workers", &message_id)
|
||||
.await;
|
||||
}
|
||||
let inflight = format!("delivery:inflight:stalwart:jmap:{message_id}");
|
||||
let _ = state.release_dedup(&inflight).await;
|
||||
});
|
||||
}
|
||||
StatusCode::NO_CONTENT.into_response()
|
||||
}
|
||||
Err(_) => {
|
||||
let _ = app.state.release_dedup(&key).await;
|
||||
error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true)
|
||||
|
|
@ -1767,6 +1830,7 @@ struct TelegramUpdate {
|
|||
#[derive(serde::Deserialize)]
|
||||
struct TelegramMessage {
|
||||
chat: TelegramChat,
|
||||
text: Option<String>,
|
||||
}
|
||||
#[derive(serde::Deserialize)]
|
||||
struct TelegramChat {
|
||||
|
|
|
|||
Loading…
Reference in New Issue