host
fb3a311da8
此前订阅与分发框架都在，但分发端点传 sender=None， webhook/slack 永远报无 sender、从未真发过。本轮： - push：send_webhook/send_slack（标准库 urllib POST， 单次 10 秒超时，不重试，失败记 errors）；sender=None 时用内置端，显式 sender 优先；webpush/apns/fcm 仍报 缺凭证（原因写明：webpush 需浏览器端三字段订阅与标准库 没有的 P-256 ECDH + AES-GCM；apns 需 HTTP/2；fcm 需 Firebase 服务账号，三者均无现成接收方，不造假发送） - worker：run_one 终态（completed/failed，缺 key 与仓库 失败两条早退路径也在内）后 best-effort 调分发，通知 失败只记日志、绝不影响任务状态；取消（interrupted） 不在事件集合里，不通知 - 测试：真实发送端 6 个（本地回环服务器）、worker 终态 通知 2 个，旧无 sender 断言按新语义改写
473 lines
18 KiB
Python
473 lines
18 KiB
Python
"""推送订阅与分发模块（云端架构实施）。
|
||
|
||
零第三方依赖，只用 Python 标准库。本模块只做两件事：
|
||
|
||
1. 推送订阅的增删查（SQLite 表 ``push_subscriptions``）。
|
||
2. 状态变化事件的分发（只发状态变化，不发进行中状态）。
|
||
|
||
边界（不要超出本模块宣称的范围）：
|
||
|
||
- 合法事件类型与 ``webhooks.ALLOWED_NOTIFICATION_EVENTS`` 一致：
|
||
``completed``、``failed``、``goal-unmet``、``merge-failed``、
|
||
``review-requested``。其他事件类型拒绝或忽略，并写明原因，
|
||
绝不静默丢弃。
|
||
- 偏好过滤：``silent``（静默）不发；失败类事件（``failed``、
|
||
``goal-unmet``、``merge-failed``）即使偏好是 ``digest``（摘要）
|
||
也即时发；非失败类（``completed``、``review-requested``）在
|
||
``digest`` 偏好下本次不立即发，返回结果里明确列入
|
||
``skipped_for_digest``（待摘要列表），不是静默丢弃。
|
||
- 发送端可注入：``webhook`` / ``slack`` 走可注入的 ``sender``。
|
||
``sender`` 为 ``None`` 时用内置真实发送端（见 :func:`send_webhook` /
|
||
:func:`send_slack`，标准库 urllib POST，单次 10 秒超时，不
|
||
重试，失败记入 errors）；显式传入的 sender 优先于内置端。
|
||
``sender`` 可以是可调用对象（收到一个 dict，见
|
||
:func:`dispatch_event`），也可以是按通道分的 dict
|
||
（``{"webhook": callable, "slack": callable}``），或带 ``send``
|
||
方法的对象。显式 sender 里没有某通道时，该通道仍报
|
||
“无 sender”，不静默成功。
|
||
- ``webpush`` / ``apns`` / ``fcm`` 只有接口，没有真实发送端：
|
||
一律明确报错“缺凭证（VAPID / APNs 证书 / FCM key）”，不编造
|
||
真实推送、不生成或提交任何私钥到仓库、不向真实推送服务发
|
||
任何请求。原因不是“没空写”：webpush 需要浏览器端订阅
|
||
（endpoint + p256dh + auth 三字段，现有订阅模型只有一个
|
||
target 文本字段，装不下）与标准库没有的 P-256 ECDH +
|
||
AES-GCM；apns 需要标准库没有的 HTTP/2；fcm 需要用户
|
||
另建 Firebase 项目与服务账号——三者都没有现成接收方，
|
||
本模块不为此造一套半吊子加密或假发送。真实凭证与真实
|
||
发送端是后续待接线的工作，见模块末尾的缺口说明。
|
||
- 本模块不跑真实模型回合。webhook/slack 的真实发送只 POST
|
||
到订阅者自己填的 target URL（创建订阅已要求项目
|
||
create_task 权限；能建订阅的成员本来就能经任务执行代码，
|
||
因此不对 target 做内网屏蔽——屏蔽既拦不住提权，也会误伤
|
||
本地接收器）。密钥、token、证书一律不由本模块读取、写入
|
||
或返回。
|
||
"""
|
||
|
||
import json
|
||
import sqlite3
|
||
import urllib.error
|
||
import urllib.request
|
||
from cloud.db import connect as _db_connect
|
||
from cloud.db import PgConnection as _PgConnection
|
||
import time
|
||
import uuid
|
||
from contextlib import contextmanager
|
||
|
||
__all__ = [
|
||
"CHANNELS",
|
||
"PREFERENCES",
|
||
"ALLOWED_PUSH_EVENTS",
|
||
"FAILURE_EVENTS",
|
||
"MISSING_CREDENTIAL_REASONS",
|
||
"DEFAULT_SENDERS",
|
||
"init_push_db",
|
||
"create_subscription",
|
||
"delete_subscription",
|
||
"list_subscriptions",
|
||
"send_webhook",
|
||
"send_slack",
|
||
"dispatch_event",
|
||
]
|
||
|
||
CHANNELS = ("webhook", "slack", "webpush", "apns", "fcm")
|
||
PREFERENCES = ("instant", "digest", "silent")
|
||
|
||
# 只发状态变化的事件集合，与 webhooks.ALLOWED_NOTIFICATION_EVENTS 一致。
|
||
ALLOWED_PUSH_EVENTS = frozenset(
|
||
{"completed", "failed", "goal-unmet", "merge-failed", "review-requested"}
|
||
)
|
||
|
||
# 失败类事件：即使偏好是 digest 也即时发。
|
||
FAILURE_EVENTS = frozenset({"failed", "goal-unmet", "merge-failed"})
|
||
|
||
# webpush/apns/fcm 缺凭证时的明确报错文案（不含任何真实凭证）。
|
||
MISSING_CREDENTIAL_REASONS = {
|
||
"webpush": "missing credential: VAPID keys not configured",
|
||
"apns": "missing credential: APNs certificate not configured",
|
||
"fcm": "missing credential: FCM key not configured",
|
||
}
|
||
|
||
|
||
def _is_connection(db):
|
||
return isinstance(db, (sqlite3.Connection, _PgConnection))
|
||
|
||
|
||
@contextmanager
|
||
def _connect(db):
|
||
"""按 db 类型取连接：Connection 直接用（不关闭）；路径则新建并关闭。"""
|
||
if _is_connection(db):
|
||
yield db
|
||
elif isinstance(db, (str, bytes)) or hasattr(db, "__fspath__"):
|
||
conn = _db_connect(str(db))
|
||
try:
|
||
yield conn
|
||
finally:
|
||
conn.close()
|
||
else:
|
||
raise TypeError("db must be a sqlite3 connection or a db path")
|
||
|
||
|
||
def init_push_db(db):
|
||
"""创建 push_subscriptions 表（已存在则不动）。"""
|
||
with _connect(db) as conn:
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS push_subscriptions("
|
||
"id TEXT PRIMARY KEY, "
|
||
"project_id TEXT NOT NULL, "
|
||
"user_id TEXT NOT NULL, "
|
||
"channel TEXT NOT NULL, "
|
||
"target TEXT NOT NULL, "
|
||
"preference TEXT NOT NULL, "
|
||
"created_at REAL, "
|
||
"UNIQUE(project_id, user_id, channel, target))"
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def _row_to_dict(row):
|
||
return {
|
||
"id": row[0],
|
||
"project_id": row[1],
|
||
"user_id": row[2],
|
||
"channel": row[3],
|
||
"target": row[4],
|
||
"preference": row[5],
|
||
"created_at": row[6],
|
||
}
|
||
|
||
|
||
def _validate_subscription_fields(project_id, user_id, channel, target,
|
||
preference):
|
||
if not project_id or not isinstance(project_id, str):
|
||
raise ValueError("project_id required")
|
||
if not user_id or not isinstance(user_id, str):
|
||
raise ValueError("user_id required")
|
||
if channel not in CHANNELS:
|
||
raise ValueError("channel must be one of %s" % (CHANNELS,))
|
||
if not target or not isinstance(target, str):
|
||
raise ValueError("target required")
|
||
if preference not in PREFERENCES:
|
||
raise ValueError("preference must be one of %s" % (PREFERENCES,))
|
||
|
||
|
||
def create_subscription(db, project_id, user_id, channel, target,
|
||
preference="instant"):
|
||
"""创建一条推送订阅，返回订阅 dict。
|
||
|
||
同一 (project_id, user_id, channel, target) 已存在时返回已有订阅，
|
||
不建重复行。channel 取值 webhook/slack/webpush/apns/fcm；
|
||
preference 取值 instant/digest/silent；非法取值抛 ValueError。
|
||
"""
|
||
_validate_subscription_fields(
|
||
project_id, user_id, channel, target, preference)
|
||
init_push_db(db)
|
||
with _connect(db) as conn:
|
||
existing = conn.execute(
|
||
"SELECT id, project_id, user_id, channel, target, preference,"
|
||
" created_at FROM push_subscriptions"
|
||
" WHERE project_id = ? AND user_id = ? AND channel = ?"
|
||
" AND target = ?",
|
||
(project_id, user_id, channel, target),
|
||
).fetchone()
|
||
if existing is not None:
|
||
return _row_to_dict(existing)
|
||
sub_id = uuid.uuid4().hex
|
||
now = time.time()
|
||
conn.execute(
|
||
"INSERT INTO push_subscriptions(id, project_id, user_id,"
|
||
" channel, target, preference, created_at)"
|
||
" VALUES(?, ?, ?, ?, ?, ?, ?)",
|
||
(sub_id, project_id, user_id, channel, target, preference, now),
|
||
)
|
||
conn.commit()
|
||
row = conn.execute(
|
||
"SELECT id, project_id, user_id, channel, target, preference,"
|
||
" created_at FROM push_subscriptions WHERE id = ?",
|
||
(sub_id,),
|
||
).fetchone()
|
||
return _row_to_dict(row)
|
||
|
||
|
||
def delete_subscription(db, subscription_id=None, *, project_id=None,
|
||
user_id=None, channel=None, target=None):
|
||
"""删除订阅，返回 True（删到至少一条）或 False（无匹配）。
|
||
|
||
优先按 subscription_id 删除；未传 id 时按给出的
|
||
project_id/user_id/channel/target 过滤删除（至少要给一个过滤条件，
|
||
否则不删任何行并返回 False，避免误删整表）。
|
||
"""
|
||
init_push_db(db)
|
||
with _connect(db) as conn:
|
||
if subscription_id is not None:
|
||
cur = conn.execute(
|
||
"DELETE FROM push_subscriptions WHERE id = ?",
|
||
(subscription_id,),
|
||
)
|
||
else:
|
||
conditions = []
|
||
params = []
|
||
for column, value in (
|
||
("project_id", project_id),
|
||
("user_id", user_id),
|
||
("channel", channel),
|
||
("target", target),
|
||
):
|
||
if value is not None:
|
||
conditions.append("%s = ?" % column)
|
||
params.append(value)
|
||
if not conditions:
|
||
return False
|
||
cur = conn.execute(
|
||
"DELETE FROM push_subscriptions WHERE "
|
||
+ " AND ".join(conditions),
|
||
params,
|
||
)
|
||
conn.commit()
|
||
return cur.rowcount > 0
|
||
|
||
|
||
def list_subscriptions(db, project_id, user_id=None):
|
||
"""列出某项目的订阅；传 user_id 时只列该用户的。
|
||
|
||
返回按 created_at、id 排序的订阅 dict 列表。只查本项目，
|
||
跨项目订阅绝不会混入（见单测）。
|
||
"""
|
||
if not project_id or not isinstance(project_id, str):
|
||
raise ValueError("project_id required")
|
||
init_push_db(db)
|
||
with _connect(db) as conn:
|
||
if user_id is None:
|
||
rows = conn.execute(
|
||
"SELECT id, project_id, user_id, channel, target,"
|
||
" preference, created_at FROM push_subscriptions"
|
||
" WHERE project_id = ? ORDER BY created_at, id",
|
||
(project_id,),
|
||
).fetchall()
|
||
else:
|
||
rows = conn.execute(
|
||
"SELECT id, project_id, user_id, channel, target,"
|
||
" preference, created_at FROM push_subscriptions"
|
||
" WHERE project_id = ? AND user_id = ?"
|
||
" ORDER BY created_at, id",
|
||
(project_id, user_id),
|
||
).fetchall()
|
||
return [_row_to_dict(row) for row in rows]
|
||
|
||
|
||
def _resolve_channel_sender(sender, channel):
|
||
"""从注入的 sender 里取出某通道的发送函数，取不到返回 None。"""
|
||
if sender is None:
|
||
return None
|
||
if isinstance(sender, dict):
|
||
candidate = sender.get(channel)
|
||
if candidate is None:
|
||
return None
|
||
if callable(candidate):
|
||
return candidate
|
||
send = getattr(candidate, "send", None)
|
||
return send if callable(send) else None
|
||
if callable(sender):
|
||
return sender
|
||
send = getattr(sender, "send", None)
|
||
return send if callable(send) else None
|
||
|
||
|
||
# 真实发送端：单次投递超时（秒），不重试——失败记入 errors，
|
||
# 下一次事件自然重发，不在这里循环撞墙。
|
||
_PUSH_SEND_TIMEOUT = 10
|
||
|
||
|
||
def _post_json(url, payload):
|
||
"""POST JSON 到 url：返回 (ok, detail)，2xx 为 ok。
|
||
|
||
HTTP 状态错误与网络异常一律返回 (False, reason)，不抛
|
||
（调用方据此记 errors）；只有调用方传了非法 URL 时
|
||
Request 构造本身可能抛 ValueError，由发送端先校验 scheme
|
||
挡住，不走到这里。
|
||
"""
|
||
data = json.dumps(payload).encode("utf-8")
|
||
request = urllib.request.Request(
|
||
url, data=data, method="POST",
|
||
headers={"Content-Type": "application/json",
|
||
"User-Agent": "codex-app-server"})
|
||
try:
|
||
with urllib.request.urlopen(
|
||
request, timeout=_PUSH_SEND_TIMEOUT) as response:
|
||
response.read()
|
||
return True, "HTTP %s" % response.status
|
||
except urllib.error.HTTPError as e:
|
||
return False, "HTTP %s" % e.code
|
||
except Exception as exc: # noqa: BLE001 -- 网络异常转 reason，不抛
|
||
return False, "%s: %s" % (type(exc).__name__, exc)
|
||
|
||
|
||
def _require_http_url(target, channel):
|
||
"""发送端 target 校验：只接受 http(s) URL，其他一律 ValueError
|
||
（dispatch 记入 errors，不静默成功、不乱 POST）。"""
|
||
if not isinstance(target, str) or not (
|
||
target.startswith("http://") or target.startswith("https://")):
|
||
raise ValueError(
|
||
"%s target must be an http(s) URL" % channel)
|
||
return target
|
||
|
||
|
||
def send_webhook(message):
|
||
"""webhook 通道真实发送：POST JSON 到订阅的 target URL。
|
||
|
||
body 固定为 {event_type, project_id, task_id, payload}，
|
||
task_id 从 payload 里取（没有则为 None）。非 2xx 抛
|
||
RuntimeError（dispatch 记入 errors，不算 sent）。
|
||
"""
|
||
target = _require_http_url(message["target"], "webhook")
|
||
payload = message["payload"] if isinstance(
|
||
message["payload"], dict) else {}
|
||
ok, detail = _post_json(target, {
|
||
"event_type": message["event_type"],
|
||
"project_id": message["project_id"],
|
||
"task_id": payload.get("task_id"),
|
||
"payload": message["payload"],
|
||
})
|
||
if not ok:
|
||
raise RuntimeError("webhook delivery failed: %s" % detail)
|
||
|
||
|
||
def send_slack(message):
|
||
"""slack 通道真实发送：POST Slack incoming webhook 格式
|
||
（{"text": ...}）到 target URL（Slack webhook 地址本身
|
||
就是 http(s) URL，与 webhook 通道同一投递路径）。
|
||
|
||
文本一行讲清事件：任务 id + 事件类型 + 项目，失败类
|
||
事件附 error 摘要（截断 500 字）。非 2xx 抛 RuntimeError。
|
||
"""
|
||
target = _require_http_url(message["target"], "slack")
|
||
payload = message["payload"] if isinstance(
|
||
message["payload"], dict) else {}
|
||
text = "[codex] task %s %s (project %s)" % (
|
||
payload.get("task_id", "?"), message["event_type"],
|
||
message["project_id"])
|
||
if payload.get("error"):
|
||
text += "\nerror: %s" % str(payload["error"])[:500]
|
||
ok, detail = _post_json(target, {"text": text})
|
||
if not ok:
|
||
raise RuntimeError("slack delivery failed: %s" % detail)
|
||
|
||
|
||
# sender=None 时的内置真实发送端：只覆盖 webhook/slack；
|
||
# webpush/apns/fcm 不在此列，仍走缺凭证报错。
|
||
DEFAULT_SENDERS = {"webhook": send_webhook, "slack": send_slack}
|
||
|
||
|
||
def dispatch_event(db, event_type, project_id, payload, sender=None):
|
||
"""分发一次状态变化事件，只发本项目的订阅。
|
||
|
||
参数：
|
||
db: sqlite3 连接或数据库路径。
|
||
event_type: 合法类型见 ALLOWED_PUSH_EVENTS；其他类型拒绝，
|
||
返回 ``{"sent": [], ..., "reason": "event not allowed"}``
|
||
且不调用任何 sender。
|
||
project_id: 只查该项目的订阅，跨项目不混。
|
||
payload: 原事件载荷（dict 或其他），原样放入发送给 sender
|
||
的消息 dict 的 ``payload`` 字段。
|
||
sender: 可注入发送端（见模块 docstring）；为 None 时用
|
||
内置真实发送端（webhook/slack 真 POST，webpush/apns/fcm
|
||
仍报缺凭证）；显式传入的 sender 优先于内置端，显式
|
||
sender 里没有某通道时该通道报“无 sender”，不静默成功。
|
||
|
||
返回 dict，键固定为::
|
||
|
||
{
|
||
"event_type": ...,
|
||
"project_id": ...,
|
||
"sent": [{"subscription_id", "user_id", "channel", "target"}],
|
||
"skipped_for_digest": [订阅 dict（待摘要，本次未立即发）],
|
||
"skipped_silent": [订阅 dict（静默偏好，本次未发）],
|
||
"errors": [{"subscription_id", "user_id", "channel", "target",
|
||
"reason": ...}],
|
||
}
|
||
|
||
非法事件类型时额外含 ``"reason": "event not allowed"``。
|
||
sender 抛异常时记入 errors，不中断其他订阅的分发。
|
||
"""
|
||
result = {
|
||
"event_type": event_type,
|
||
"project_id": project_id,
|
||
"sent": [],
|
||
"skipped_for_digest": [],
|
||
"skipped_silent": [],
|
||
"errors": [],
|
||
}
|
||
if event_type not in ALLOWED_PUSH_EVENTS:
|
||
result["reason"] = "event not allowed"
|
||
result["errors"] = [{
|
||
"subscription_id": None,
|
||
"user_id": None,
|
||
"channel": None,
|
||
"target": None,
|
||
"reason": "event not allowed: %r not in %s"
|
||
% (event_type, sorted(ALLOWED_PUSH_EVENTS)),
|
||
}]
|
||
return result
|
||
if not project_id or not isinstance(project_id, str):
|
||
result["reason"] = "project_id required"
|
||
result["errors"] = [{
|
||
"subscription_id": None,
|
||
"user_id": None,
|
||
"channel": None,
|
||
"target": None,
|
||
"reason": "project_id required",
|
||
}]
|
||
return result
|
||
|
||
if sender is None:
|
||
# 内置真实发送端：webhook/slack 真 POST；显式 sender
|
||
# 传入时不走这里（注入优先），webpush/apns/fcm 永远
|
||
# 不走 sender，仍在循环里报缺凭证。
|
||
sender = DEFAULT_SENDERS
|
||
subscriptions = list_subscriptions(db, project_id)
|
||
is_failure = event_type in FAILURE_EVENTS
|
||
for sub in subscriptions:
|
||
base = {
|
||
"subscription_id": sub["id"],
|
||
"user_id": sub["user_id"],
|
||
"channel": sub["channel"],
|
||
"target": sub["target"],
|
||
}
|
||
# 偏好过滤：silent 永远不发；digest 只对非失败类延迟。
|
||
if sub["preference"] == "silent":
|
||
result["skipped_silent"].append(sub)
|
||
continue
|
||
if sub["preference"] == "digest" and not is_failure:
|
||
result["skipped_for_digest"].append(sub)
|
||
continue
|
||
# webpush/apns/fcm：只有接口，缺凭证，明确报错，不走 sender、
|
||
# 不联网、不编造推送。
|
||
if sub["channel"] in MISSING_CREDENTIAL_REASONS:
|
||
result["errors"].append(
|
||
dict(base, reason=MISSING_CREDENTIAL_REASONS[sub["channel"]]))
|
||
continue
|
||
# webhook/slack：sender=None 时已换成内置真实发送端；
|
||
# 显式 sender 里没有该通道时明确报错，不静默成功。
|
||
channel_sender = _resolve_channel_sender(sender, sub["channel"])
|
||
if channel_sender is None:
|
||
result["errors"].append(
|
||
dict(base, reason="no sender for channel %s"
|
||
% sub["channel"]))
|
||
continue
|
||
message = {
|
||
"event_type": event_type,
|
||
"project_id": project_id,
|
||
"payload": payload,
|
||
"channel": sub["channel"],
|
||
"target": sub["target"],
|
||
"user_id": sub["user_id"],
|
||
"subscription_id": sub["id"],
|
||
}
|
||
try:
|
||
channel_sender(message)
|
||
except Exception as exc: # noqa: BLE001 -- 分发失败记 errors，不中断
|
||
result["errors"].append(
|
||
dict(base, reason="sender failed: %s" % exc))
|
||
continue
|
||
result["sent"].append(base)
|
||
return result
|