通知与 Webhook
流量余额预警的配置方式,四种通知渠道的要求,以及 Webhook 的实际负载格式。
平台的通知能力比"完整的事件订阅系统"要窄——目前实现的是流量余额预警这一个触发场景,可以投递到四种渠道。把预期对齐了再接。
支持的渠道
| 渠道 | 说明 |
|---|---|
| 发到账号邮箱 | |
| Webhook | POST 到你的 HTTPS 端点 |
| Bark | iOS 推送服务 |
| Telegram | 由绑定流程管理,不通过普通渠道接口创建 |
Webhook 的要求
Webhook 的实际负载
这是完整的请求内容,没有别的字段:
POST /your-endpoint HTTP/1.1
Content-Type: application/json
User-Agent: NextProxy-Notification/1.0
{
"type": "webhook",
"title": "流量余额不足提醒",
"body": "当前剩余流量 12.3 GB,已低于您设置的 20 GB 预警阈值。"
}
怎么在这个前提下用得安全
既然没有签名,用一个不可猜测的 URL 路径作为共享秘密,并配合来源校验:
import crypto from 'node:crypto';
import express from 'express';
const app = express();
app.use(express.json());
// 把 secret 放在路径里,因为负载里没有签名可校验
const WEBHOOK_PATH = `/hooks/nextproxy/${process.env.WEBHOOK_SECRET}`;
app.post(WEBHOOK_PATH, (req, res) => {
// 平台固定发这个 UA,可以作为一层弱校验
if (req.get('user-agent') !== 'NextProxy-Notification/1.0') {
return res.status(404).end();
}
const { title, body } = req.body ?? {};
if (typeof title !== 'string' || typeof body !== 'string') {
return res.status(400).json({ error: 'bad payload' });
}
// 必须先返回 2xx 再做耗时处理——没有重试,超时就丢了
res.status(200).end();
queueAlert({ title, body }).catch((error) => {
console.error('告警入队失败', error);
});
});
async function queueAlert(alert) {
// 转发到你自己的告警系统
console.log('[NextProxy]', alert.title, alert.body);
}
app.listen(3000);
流量余额预警
这是目前唯一实现的预警触发场景,针对动态住宅的流量池余量。
配置
GET
/api/v1/traffic-alert-settingsPUT
/api/v1/traffic-alert-settingscurl -X PUT 'https://api.example.com/api/v1/traffic-alert-settings' \
-H 'Authorization: Bearer at_YOUR_ACCESS_TOKEN' \
-H 'Origin: https://console.example.com' \
-H 'Content-Type: application/json' \
-d '{
"enabled": true,
"thresholdBytes": 21474836480,
"channelIds": ["channel-id-1", "channel-id-2"]
}'
规则
预警阈值,范围 1 MB – 100 TB。剩余流量低于这个值时触发。
关联的通知渠道,最多 20 个。
低于阈值期间,按北京时间自然日每天最多发送一次。
不会因为你一直低于阈值就反复轰炸,但也意味着你不会立刻收到第二次提醒。
触发时同时写入站内通知,所以即使外部渠道投递失败,在控制台里也能看到。
创建通知渠道
POST
/api/v1/notification-channelscurl -X POST 'https://api.example.com/api/v1/notification-channels' \
-H 'Authorization: Bearer at_YOUR_ACCESS_TOKEN' \
-H 'Origin: https://console.example.com' \
-H 'Content-Type: application/json' \
-d '{
"type": "webhook",
"name": "运维告警群",
"target": "https://hooks.example.com/nextproxy/SECRET"
}'
其他端点:
GET
/api/v1/notification-channelsPATCH
/api/v1/notification-channels/:idDELETE
/api/v1/notification-channels/:id站内通知
分三类:system(系统)、renewal(续期)、activity(活动)。
GET
/api/v1/notificationsPOST
/api/v1/notifications/:id/readPOST
/api/v1/notifications/read-all当前没有的能力
把这些列清楚,免得你按不存在的功能设计方案:
| 你可能想要的 | 现状 |
|---|---|
| 金额钱包余额预警 | 没有。只有流量余量预警 |
| 按百分比的用量预警 | 没有。只支持绝对字节阈值 |
| 静态 IP 到期提醒 | 没有对应的定时任务 |
| 通用业务事件 Webhook 订阅 | 没有。Webhook 只作为流量预警的投递渠道 |
| Webhook 签名验证 | 没有签名头或 secret |
| Webhook 失败重试 | 没有重试队列 |
一个自建监控的例子
import time
import requests
def check_traffic_balance(client, warn_gb: float = 20.0) -> None:
"""轮询流量余额,低于阈值就走自己的告警通道。
平台的预警每天只发一次,自建轮询能补上"当天第二次"这个缺口。
"""
summary = client.get("/api/v1/proxy/usage/summary")
remaining_gb = summary["trafficRemainBytes"] / (1024 ** 3)
if remaining_gb < warn_gb:
notify_ops(
f"NextProxy 流量余额 {remaining_gb:.1f} GB,低于 {warn_gb} GB"
)
def check_expiring_resources(client, warn_days: int = 7) -> None:
"""静态资源到期没有平台侧提醒,自己查。"""
deadline = time.time() + warn_days * 86400
for path in ("/api/v1/static-residential", "/api/v1/static-datacenter"):
for resource in client.get(path).get("items", []):
expires_at = resource.get("expiresAt")
if expires_at and _to_epoch(expires_at) < deadline:
notify_ops(f"{resource['ip']} 将于 {expires_at} 到期")