"""推送订阅与分发模块（云端架构实施）。

零第三方依赖，只用 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``，并且持久化进 ``push_digest_events``
  表——worker 循环每小时按项目/订阅汇总发一次摘要
  （``send_digest_due``），不再是“返回值里假装汇总、实际
  永不送达”。失败类事件不进摘要。
- 推送失败后补：webhook/slack 发送端抛异常时，事件写进
  ``push_outbox`` 表并后台重试（``process_outbox_due``，最多
  3 次、默认 60 秒退避）；缺凭证（webpush/apns/fcm）与无
  sender 不是瞬时故障，不进 outbox，仍明确报错。重试是
  at-least-once：同一事件可能重复送达，接收方须去重。
- 发送端可注入：``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 权限）。P1 收紧后：真实发送层对 target 做 SSRF
  校验——loopback、私有网段、链路本地、云元数据地址一律不
  POST；主机名先 DNS 解析再按解析出的 IP 判，解析失败拒绝
  （见 validate_push_target）。本地接收器如需放行，须运维显式
  置 CODEX_PUSH_ALLOW_PRIVATE_TARGETS=1（默认关闭，测试用
  本地接收器也须经此开关，不在代码里默认放行 loopback）。
  密钥、token、证书一律不由本模块读取、写入或返回。
"""

import errno
import http.client
import ipaddress
import json
import os
import socket
import sqlite3
import urllib.error
import urllib.parse
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",
    "validate_push_target",
    "send_digest_due",
    "process_outbox_due",
    "list_pending_digest",
    "list_outbox",
    "DIGEST_INTERVAL_SECONDS",
    "OUTBOX_MAX_ATTEMPTS",
    "OUTBOX_BACKOFF_SECONDS",
]

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"})

# G11/G12 的节奏与重试上限：digest 每小时汇总一次（worker
# 循环按此间隔调用 send_digest_due）；outbox 最多重试 3 次，
# 每次失败后至少等 60 秒再试。at-least-once 语义：webhook/slack
# 的重试可能重复送达同一事件，接收方须按 event/task 去重，
# 本模块不保证 exactly-once。
DIGEST_INTERVAL_SECONDS = 3600
OUTBOX_MAX_ATTEMPTS = 3
OUTBOX_BACKOFF_SECONDS = 60

# 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 表（已存在则不动），以及
    G11 的 push_digest_events 与 G12 的 push_outbox 表。

    老库同样走这里懒建表：dispatch/list 入口都先调它，
    所以旧部署升级后第一次分发即建好新表，不用单独迁移。
    """
    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.execute(
            "CREATE TABLE IF NOT EXISTS push_digest_events("
            "id TEXT PRIMARY KEY, "
            "project_id TEXT NOT NULL, "
            "subscription_id TEXT NOT NULL, "
            "user_id TEXT NOT NULL, "
            "channel TEXT NOT NULL, "
            "target TEXT NOT NULL, "
            "event_type TEXT NOT NULL, "
            "payload_json TEXT NOT NULL, "
            "created_at REAL, "
            "sent INTEGER DEFAULT 0, "
            "sent_at REAL)"
        )
        conn.execute(
            "CREATE TABLE IF NOT EXISTS push_outbox("
            "id TEXT PRIMARY KEY, "
            "project_id TEXT NOT NULL, "
            "subscription_id TEXT NOT NULL, "
            "user_id TEXT NOT NULL, "
            "channel TEXT NOT NULL, "
            "target TEXT NOT NULL, "
            "event_type TEXT NOT NULL, "
            "payload_json TEXT NOT NULL, "
            "attempts INTEGER DEFAULT 0, "
            "next_attempt_at REAL, "
            "created_at REAL, "
            "last_error TEXT)"
        )
        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, resolver=None):
    """POST JSON 到 url：返回 (ok, detail)，2xx 为 ok。

    建连时解析一次并判段、钉 IP 直连（见 _checked_opener）：
    Host 头与 TLS SNI 仍为原域名，只把拨号地址换成判过的
    IP——预检解析与建连解析不一致时，建连侧的判定为准，
    绝不会拨向未验证的 IP。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 _checked_opener(resolver).open(
                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 _pick_connect_ip(host, resolver):
    """建连时解析一次并判段，返回要拨的 IP（只抛 OSError）。

    字面量 IP 不经过 DNS（只复核段）；域名解析一次：
    allow-private 时取首个，否则取首个全局单播 IP；无可用
    IP 时抛 OSError。调用方拨的就是这里判过的 IP，之后
    不再二次解析——校验与建连之间的 DNS 重绑定窗口就此
    闭合（旧版 validate 后 urlopen 再解析一次的 TOCTOU）。
    """
    if _allow_private_targets():
        try:
            ipaddress.ip_address(host)
            return host
        except ValueError:
            pass
        ips = (resolver or _default_resolver)(host)
        if not ips:
            raise OSError("DNS resolution failed for target host")
        return ips[0]
    try:
        ipaddress.ip_address(host)
    except ValueError:
        pass
    else:
        if _ip_is_blocked(host):
            raise OSError(
                "target resolves to a non-public address: %s" % host)
        return host
    ips = (resolver or _default_resolver)(host)
    if not ips:
        raise OSError("DNS resolution failed for target host")
    for ip_text in ips:
        if not _ip_is_blocked(ip_text):
            return ip_text
    raise OSError("target host resolves to a non-public address")


def _checked_connection_class(base, resolver):
    """建连时解析+判段的连接类：Host 头与 TLS SNI 仍为原域名
    （self.host 不动），只把拨号地址换成判过的 IP。

    resolver 由调用方注入（测试 mock DNS 用）；为 None 时
    用默认解析。代理隧道场景（_tunnel_host 已置）仍走基类：
    隧道里解析权在代理方，钉 IP 不适用，预检判定为准。
    """
    class CheckedConnection(base):
        def connect(self):
            if getattr(self, "_tunnel_host", None):
                super().connect()
                return
            picked = _pick_connect_ip(self.host, resolver)
            self.sock = socket.create_connection(
                (picked, self.port), self.timeout, self.source_address)
            try:
                self.sock.setsockopt(
                    socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            except OSError as exc:
                if exc.errno != errno.ENOPROTOOPT:
                    raise
            if isinstance(self, http.client.HTTPSConnection):
                self.sock = self._context.wrap_socket(
                    self.sock, server_hostname=self.host)
    return CheckedConnection


class _CheckedHTTPHandler(urllib.request.HTTPHandler):
    def __init__(self, resolver=None):
        super().__init__()
        self._resolver = resolver

    def http_open(self, req):
        return self.do_open(
            _checked_connection_class(http.client.HTTPConnection,
                                      self._resolver), req)


class _CheckedHTTPSHandler(urllib.request.HTTPSHandler):
    def __init__(self, resolver=None, debuglevel=0, context=None,
                 check_hostname=None):
        super().__init__(debuglevel=debuglevel, context=context,
                         check_hostname=check_hostname)
        self._resolver = resolver

    def https_open(self, req):
        return self.do_open(
            _checked_connection_class(http.client.HTTPSConnection,
                                      self._resolver), req,
            context=self._context, check_hostname=self._check_hostname)


def _checked_opener(resolver=None):
    """发送用 opener：建连时解析+判段的连接类 + 重定向/错误处理。

    刻意直接构造 OpenerDirector、不用 build_opener：后者会
    自动塞入 ProxyHandler，本机 egress 代理会把请求改写成
    代理端口、钉 IP 失效（实测拨号跑到代理端口）。钉 IP 的
    前提是直连；代理环境下发送会直连失败并记入 errors
    （loud，不静默成功）。重定向仍跟随，但每一跳都重新
    走建连判定（新目标同样解析+判段）。
    """
    opener = urllib.request.OpenerDirector()
    opener.add_handler(_CheckedHTTPHandler(resolver))
    opener.add_handler(_CheckedHTTPSHandler(resolver))
    opener.add_handler(urllib.request.HTTPRedirectHandler())
    opener.add_handler(urllib.request.HTTPErrorProcessor())
    return opener


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


# 云元数据主机名：即使解析出的 IP 恰好不在私有段，也直接挡。
_BLOCKED_HOSTNAMES = frozenset({"metadata.google.internal"})


def _allow_private_targets():
    """运维显式开关：仅 CODEX_PUSH_ALLOW_PRIVATE_TARGETS=1 时放行
    内网目标（本地接收器场景）。默认关闭；开关只影响 IP 段判定，
    不影响 scheme 校验。"""
    return os.environ.get(
        "CODEX_PUSH_ALLOW_PRIVATE_TARGETS", "").strip() == "1"


def _default_resolver(hostname):
    """默认 DNS 解析：返回 IP 字符串列表；解析失败抛异常，
    由 validate_push_target 转成“拒绝”。"""
    infos = socket.getaddrinfo(hostname, None)
    out = []
    for info in infos:
        sockaddr = info[4] if len(info) > 4 else None
        if sockaddr and sockaddr[0]:
            out.append(str(sockaddr[0]))
    return out


def _ip_is_blocked(ip_text):
    """非全局单播 IP 一律算内网/特殊地址：loopback、私有网段、
    链路本地（169.254/16，含云元数据 169.254.169.254）、保留段、
    组播、未指定地址都在 is_global=False 里。"""
    try:
        return not ipaddress.ip_address(ip_text).is_global
    except ValueError:
        return True


def validate_push_target(target, resolver=None, check_dns=True):
    """SSRF 校验：通过返回 None，否则返回原因字符串（不抛）。

    顺序：scheme（http/https）-> 主机名存在 -> 元数据主机名 ->
    IP 字面量直接判段 -> （check_dns 时）DNS 解析后逐 IP 判段，
    解析失败或解析出空结果都拒绝。本预检只是 fail-fast：
    真正起强制作用的是发送时建连类的解析+判段（解析一次、
    只拨判过的 IP），预检解析与建连解析不一致时以建连侧
    的判定为准，不会拨向未验证的 IP（旧版 validate 后
    urlopen 再解析一次的 TOCTOU 已闭合）。
    check_dns=False 时只做字面量判定：供直接调 send_* 但不想
    做 DNS 的调用方用；dispatch 的真实发送路径一律
    check_dns=True，显式 sender 路径不走本预检（见 dispatch）。
    """
    if not isinstance(target, str) or not (
            target.startswith("http://") or target.startswith("https://")):
        return "target must be an http(s) URL"
    try:
        host = urllib.parse.urlparse(target).hostname
    except ValueError:
        return "target host is invalid"
    if not host:
        return "target host required"
    lowered = host.lower()
    if lowered in _BLOCKED_HOSTNAMES or lowered.endswith(
            ".metadata.google.internal"):
        return "target host is a cloud metadata address"
    if _allow_private_targets():
        return None
    try:
        ipaddress.ip_address(host)
        is_literal = True
    except ValueError:
        is_literal = False
    if is_literal:
        if _ip_is_blocked(host):
            return "target resolves to a non-public address: %s" % host
        return None
    if not check_dns:
        return None
    resolve = resolver or _default_resolver
    try:
        ips = resolve(host)
    except Exception:
        return "DNS resolution failed for target host"
    if not ips:
        return "DNS resolution failed for target host"
    for ip_text in ips:
        if _ip_is_blocked(ip_text):
            return ("target host resolves to a non-public address: %s"
                    % ip_text)
    return None


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")
    blocked = validate_push_target(
        target, resolver=message.get("_resolver"))
    if blocked is not None:
        raise RuntimeError(
            "webhook delivery failed: target blocked: %s" % blocked)
    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"],
    }, resolver=message.get("_resolver"))
    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")
    blocked = validate_push_target(
        target, resolver=message.get("_resolver"))
    if blocked is not None:
        raise RuntimeError(
            "slack delivery failed: target blocked: %s" % blocked)
    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},
                            resolver=message.get("_resolver"))
    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 _payload_json(payload):
    try:
        return json.dumps(payload, ensure_ascii=False)
    except (TypeError, ValueError):
        return json.dumps({"value": str(payload)}, ensure_ascii=False)


def _enqueue_digest(conn, sub, event_type, payload, now):
    conn.execute(
        "INSERT INTO push_digest_events(id, project_id, subscription_id,"
        " user_id, channel, target, event_type, payload_json,"
        " created_at, sent) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, 0)",
        (uuid.uuid4().hex, sub["project_id"], sub["id"],
         sub["user_id"], sub["channel"], sub["target"], event_type,
         _payload_json(payload), now))
    conn.commit()


def _enqueue_outbox(conn, sub, event_type, payload, reason, now,
                    backoff_seconds=OUTBOX_BACKOFF_SECONDS):
    conn.execute(
        "INSERT INTO push_outbox(id, project_id, subscription_id,"
        " user_id, channel, target, event_type, payload_json,"
        " attempts, next_attempt_at, created_at, last_error)"
        " VALUES(?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?)",
        (uuid.uuid4().hex, sub["project_id"], sub["id"],
         sub["user_id"], sub["channel"], sub["target"], event_type,
         _payload_json(payload), now + backoff_seconds, now, reason))
    conn.commit()


def _send_via(sender, sub, event_type, project_id, payload):
    """按订阅通道解析发送端并发送一次；无 sender 时抛 ValueError。

    调用方区分：ValueError（无 sender/缺凭证在外层已处理，
    这里只对无 sender）不进 outbox；发送端抛的其他异常才
    是瞬时投递失败，进 outbox 的依据。
    """
    channel_sender = _resolve_channel_sender(sender, sub["channel"])
    if channel_sender is None:
        raise ValueError("no sender for channel %s" % sub["channel"])
    channel_sender({
        "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"],
    })


def list_pending_digest(db, project_id=None):
    """未发送的摘要事件（按 created_at 排序），供检查与单测。"""
    init_push_db(db)
    with _connect(db) as conn:
        if project_id:
            rows = conn.execute(
                "SELECT id, project_id, subscription_id, user_id,"
                " channel, target, event_type, payload_json, created_at"
                " FROM push_digest_events WHERE sent = 0"
                " AND project_id = ? ORDER BY created_at, id",
                (project_id,)).fetchall()
        else:
            rows = conn.execute(
                "SELECT id, project_id, subscription_id, user_id,"
                " channel, target, event_type, payload_json, created_at"
                " FROM push_digest_events WHERE sent = 0"
                " ORDER BY created_at, id").fetchall()
    out = []
    for row in rows:
        try:
            payload = json.loads(row[7])
        except (TypeError, ValueError):
            payload = {}
        out.append({
            "id": row[0], "project_id": row[1],
            "subscription_id": row[2], "user_id": row[3],
            "channel": row[4], "target": row[5],
            "event_type": row[6], "payload": payload,
            "created_at": row[8],
        })
    return out


def list_outbox(db, project_id=None):
    """待重试的 outbox 行（按 next_attempt_at 排序）。"""
    init_push_db(db)
    with _connect(db) as conn:
        if project_id:
            rows = conn.execute(
                "SELECT id, project_id, subscription_id, user_id,"
                " channel, target, event_type, payload_json, attempts,"
                " next_attempt_at, created_at, last_error"
                " FROM push_outbox WHERE project_id = ?"
                " ORDER BY next_attempt_at, id",
                (project_id,)).fetchall()
        else:
            rows = conn.execute(
                "SELECT id, project_id, subscription_id, user_id,"
                " channel, target, event_type, payload_json, attempts,"
                " next_attempt_at, created_at, last_error"
                " FROM push_outbox ORDER BY next_attempt_at, id").fetchall()
    out = []
    for row in rows:
        try:
            payload = json.loads(row[7])
        except (TypeError, ValueError):
            payload = {}
        out.append({
            "id": row[0], "project_id": row[1],
            "subscription_id": row[2], "user_id": row[3],
            "channel": row[4], "target": row[5],
            "event_type": row[6], "payload": payload,
            "attempts": row[8], "next_attempt_at": row[9],
            "created_at": row[10], "last_error": row[11],
        })
    return out


def send_digest_due(db, sender=None, now=None):
    """汇总发送所有未发送的摘要事件：按（项目、订阅）分组，
    每组只发一条摘要消息，成功后整组标记已发送。

    摘要消息的 event_type 为 "digest"，payload 为
    {"digest": True, "events": [{"event_type", "payload"}...],
    "count": n}。某组发送失败时整组保持未发送（下一次
    send_digest_due 再试），不丢事件、不部分标记。返回
    {"sent": [组摘要], "errors": [...]}。
    """
    now = time.time() if now is None else now
    init_push_db(db)
    effective = DEFAULT_SENDERS if sender is None else sender
    pending = list_pending_digest(db)
    groups = {}
    for event in pending:
        key = (event["project_id"], event["subscription_id"],
               event["user_id"], event["channel"], event["target"])
        groups.setdefault(key, []).append(event)
    result = {"sent": [], "errors": []}
    for (project_id, sub_id, user_id, channel, target), events in groups.items():
        message = {
            "event_type": "digest",
            "project_id": project_id,
            "payload": {
                "digest": True,
                "events": [{"event_type": e["event_type"],
                            "payload": e["payload"]} for e in events],
                "count": len(events),
            },
            "channel": channel,
            "target": target,
            "user_id": user_id,
            "subscription_id": sub_id,
        }
        sub = {"id": sub_id, "user_id": user_id,
               "channel": channel, "target": target,
               "project_id": project_id}
        try:
            _send_via(effective, sub, "digest", project_id,
                      message["payload"])
        except Exception as exc:  # noqa: BLE001 -- 组失败整组保留
            result["errors"].append({
                "subscription_id": sub_id, "user_id": user_id,
                "channel": channel, "target": target,
                "reason": "digest send failed: %s" % exc})
            continue
        with _connect(db) as conn:
            for event in events:
                conn.execute(
                    "UPDATE push_digest_events SET sent = 1, sent_at = ?"
                    " WHERE id = ?", (now, event["id"]))
            conn.commit()
        result["sent"].append({
            "subscription_id": sub_id, "user_id": user_id,
            "channel": channel, "target": target,
            "count": len(events)})
    return result


def process_outbox_due(db, sender=None, now=None,
                       max_attempts=OUTBOX_MAX_ATTEMPTS,
                       backoff_seconds=OUTBOX_BACKOFF_SECONDS):
    """重试到期的 outbox 行：成功删行；失败 attempts+1 并把
    下次时间推后 backoff_seconds；达到 max_attempts 后删行
    并进 failed（丢弃，不再重试），调用方负责记日志。

    返回 {"sent": [...], "failed": [...]}，每项含 subscription
    与 attempts。未到期与已达上限外的行不动。
    """
    now = time.time() if now is None else now
    init_push_db(db)
    effective = DEFAULT_SENDERS if sender is None else sender
    result = {"sent": [], "failed": []}
    for row in list_outbox(db):
        if row["attempts"] >= max_attempts:
            continue
        if row["next_attempt_at"] is not None and \
                row["next_attempt_at"] > now:
            continue
        sub = {"id": row["subscription_id"], "user_id": row["user_id"],
               "channel": row["channel"], "target": row["target"],
               "project_id": row["project_id"]}
        try:
            _send_via(effective, sub, row["event_type"],
                      row["project_id"], row["payload"])
        except Exception as exc:  # noqa: BLE001 -- 失败记 attempts
            attempts = (row["attempts"] or 0) + 1
            with _connect(db) as conn:
                if attempts >= max_attempts:
                    conn.execute(
                        "DELETE FROM push_outbox WHERE id = ?",
                        (row["id"],))
                else:
                    conn.execute(
                        "UPDATE push_outbox SET attempts = ?,"
                        " next_attempt_at = ?, last_error = ?"
                        " WHERE id = ?",
                        (attempts, now + backoff_seconds,
                         "sender failed: %s" % exc, row["id"]))
                conn.commit()
            if attempts >= max_attempts:
                result["failed"].append(dict(
                    sub, attempts=attempts,
                    reason="sender failed: %s" % exc))
            continue
        with _connect(db) as conn:
            conn.execute("DELETE FROM push_outbox WHERE id = ?",
                         (row["id"],))
            conn.commit()
        result["sent"].append(dict(sub, attempts=row["attempts"] or 0))
    return result


def dispatch_event(db, event_type, project_id, payload, sender=None,
                   resolver=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": [],
        "outbox": [],
    }
    init_push_db(db)
    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

    real_send = sender is None
    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:
            # G11：不再只在返回值里“假装汇总”——持久化后由
            # worker 每小时汇总发送；持久化失败不静默：记入
            # errors 并仍列入 skipped，调用方日志可见。
            try:
                with _connect(db) as conn:
                    _enqueue_digest(conn, sub, event_type, payload,
                                    time.time())
            except Exception as exc:  # noqa: BLE001
                result["errors"].append(
                    dict(base, reason="digest persist failed: %s" % exc))
            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
        # SSRF 预检（真实发送路径，发送前挡，不 POST）：含 DNS
        # 的全量校验。显式 sender 路径不预检：注入的 sender 是
        # 调用方自控的投递（测试 seam），不是服务端发起的 HTTP，
        # 挡它不增加生产安全性；生产服务端一律走 sender=None
        # 的真实发送，必经本预检与 send_webhook/send_slack 的
        # 二次校验。原因文案含 "delivery failed"，与旧调用方
        # 对投递失败的断言兼容。
        if real_send:
            blocked = validate_push_target(
                sub["target"], resolver=resolver, check_dns=True)
            if blocked is not None:
                result["errors"].append(
                    dict(base, reason="%s delivery failed: target blocked:"
                         " %s" % (sub["channel"], blocked)))
                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))
            # G12：webhook/slack 的瞬时投递失败进 outbox 后台
            # 重试；缺凭证与无 sender 在上游分支已处理，不到这里。
            try:
                with _connect(db) as conn:
                    _enqueue_outbox(conn, sub, event_type, payload,
                                    "sender failed: %s" % exc,
                                    time.time())
                result["outbox"].append(base)
            except Exception as out_exc:  # noqa: BLE001
                result["errors"].append(
                    dict(base, reason="outbox enqueue failed: %s" % out_exc))
            continue
        result["sent"].append(base)
    return result