app/services/webhook_store.py

"""Webhook 通知服务。

事件类型：
- chat.completed   聊天成功
- chat.failed      聊天失败
- file.uploaded    文件上传
- config.updated   配置更新
- session.created  会话创建
- session.deleted  会话删除

配置存 SQLite webhook_config 表；触发时异步 httpx POST，
失败不影响主流程。事件用 events 字段（逗号分隔）过滤。
"""
from __future__ import annotations

import asyncio
import json
import time
from typing import Any, Optional

from ..database import get_conn
from ..http_client import get_http_client


# ===== 配置 CRUD =====

def _now_ts() -> int:
    return int(time.time())


def list_webhooks() -> list[dict]:
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT id, name, url, events, enabled, created_at FROM webhook_config "
            "ORDER BY id ASC"
        ).fetchall()
    out = []
    for r in rows:
        out.append({
            "id": int(r["id"]),
            "name": r["name"],
            "url": r["url"],
            "events": [e for e in (r["events"] or "").split(",") if e],
            "enabled": bool(r["enabled"]),
            "created_at": int(r["created_at"]),
        })
    return out


def create_webhook(*, name: str, url: str, events: list[str], enabled: bool = True) -> dict:
    events_str = ",".join([e for e in (events or []) if e])
    now = _now_ts()
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO webhook_config(name, url, events, enabled, created_at) "
            "VALUES(?,?,?,?,?)",
            (name, url, events_str, 1 if enabled else 0, now),
        )
        wid = int(cur.lastrowid)
    return {"id": wid, "name": name, "url": url, "events": events or [], "enabled": enabled, "created_at": now}


def update_webhook(
    webhook_id: int,
    *,
    name: Optional[str] = None,
    url: Optional[str] = None,
    events: Optional[list[str]] = None,
    enabled: Optional[bool] = None,
) -> Optional[dict]:
    sets: list[str] = []
    args: list = []
    if name is not None:
        sets.append("name = ?")
        args.append(name)
    if url is not None:
        sets.append("url = ?")
        args.append(url)
    if events is not None:
        sets.append("events = ?")
        args.append(",".join([e for e in events if e]))
    if enabled is not None:
        sets.append("enabled = ?")
        args.append(1 if enabled else 0)
    if not sets:
        return get_webhook(webhook_id)
    args.append(webhook_id)
    with get_conn() as conn:
        cur = conn.execute(
            f"UPDATE webhook_config SET {', '.join(sets)} WHERE id = ?",
            tuple(args),
        )
        if cur.rowcount == 0:
            return None
    return get_webhook(webhook_id)


def get_webhook(webhook_id: int) -> Optional[dict]:
    with get_conn() as conn:
        row = conn.execute(
            "SELECT id, name, url, events, enabled, created_at FROM webhook_config WHERE id = ?",
            (webhook_id,),
        ).fetchone()
    if not row:
        return None
    return {
        "id": int(row["id"]),
        "name": row["name"],
        "url": row["url"],
        "events": [e for e in (row["events"] or "").split(",") if e],
        "enabled": bool(row["enabled"]),
        "created_at": int(row["created_at"]),
    }


def delete_webhook(webhook_id: int) -> bool:
    with get_conn() as conn:
        cur = conn.execute("DELETE FROM webhook_config WHERE id = ?", (webhook_id,))
        return cur.rowcount > 0


# ===== 事件触发 =====

def _matching_webhooks(event: str) -> list[dict]:
    """返回订阅了该事件且启用的 webhook 列表。"""
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT id, name, url, events FROM webhook_config WHERE enabled = 1"
        ).fetchall()
    out = []
    for r in rows:
        events = [e for e in (r["events"] or "").split(",") if e]
        if event in events or "*" in events:
            out.append({"id": int(r["id"]), "name": r["name"], "url": r["url"]})
    return out


async def notify(event: str, payload: Any) -> None:
    """异步通知所有订阅了 event 的 webhook。失败静默。"""
    targets = _matching_webhooks(event)
    if not targets:
        return
    body = json.dumps(
        {"event": event, "ts": _now_ts(), "data": payload},
        ensure_ascii=False,
    ).encode("utf-8")
    client = get_http_client()
    for t in targets:
        try:
            resp = await client.post(
                t["url"],
                content=body,
                headers={
                    "Content-Type": "application/json",
                    "X-XTC-Event": event,
                    "X-XTC-Webhook-Id": str(t["id"]),
                    "User-Agent": "xtc-backend-hf-webhook/1.0",
                },
                timeout=10.0,
            )
            if resp.status_code >= 400:
                print(f"[webhook] {t['name']} ({t['url']}) returned {resp.status_code}")
        except Exception as e:
            print(f"[webhook] {t['name']} ({t['url']}) failed: {e}")


def notify_fire_and_forget(event: str, payload: Any) -> None:
    """同步上下文用：把通知丢到事件循环后台跑，不阻塞主流程。"""
    try:
        loop = asyncio.get_event_loop()
    except RuntimeError:
        return
    loop.create_task(notify(event, payload))