Compare commits

..

2 Commits

Author SHA1 Message Date
ryan bd8a9ff96d feat: 增加 SafeW 已监控群列表接口
通过 getUpdates 将群写入 Redis,供创建/编辑渠道时选择 chat_id,避免前端重复传递 token。
2026-08-15 00:40:40 +08:00
ryan 0d0cd0c510 feat: 新增 SafeW 出站通知渠道
通过 Bot API sendMessage 投递渲染后的通知,MarkdownV2 自动转义标题与正文。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-14 00:58:44 +08:00
18 changed files with 3267 additions and 6 deletions
+49 -3
View File
@@ -1,6 +1,6 @@
# AIAA Notification Service # AIAA Notification Service
纯 Go 实现的多渠道通知服务。上游系统通过 Webhook 推送消息,服务按来源解析、匹配规则、条件过滤、渲染模板后,异步分发到钉钉 / 企业微信 / 邮件 / Bark。 纯 Go 实现的多渠道通知服务。上游系统通过 Webhook 推送消息,服务按来源解析、匹配规则、条件过滤、渲染模板后,异步分发到钉钉 / 企业微信 / 邮件 / Bark / SafeW
## 核心概念 ## 核心概念
@@ -14,7 +14,7 @@ rule N──M channel (via rule_channel,可独立开关)
|------|------| |------|------|
| **Source** | 来源系统,持有独立 `api_key`,定义 body 解析方式(json / regex / text | | **Source** | 来源系统,持有独立 `api_key`,定义 body 解析方式(json / regex / text |
| **Template** | Go `text/template` 模板 | | **Template** | Go `text/template` 模板 |
| **Channel** | 发送渠道配置(钉钉 / 企微 / 邮件 / Bark | | **Channel** | 发送渠道配置(钉钉 / 企微 / 邮件 / Bark / SafeW |
| **Rule** | 绑定 `source + event → template + channels`,可选条件过滤 | | **Rule** | 绑定 `source + event → template + channels`,可选条件过滤 |
| **MessageLog** | 发送记录,便于排查 | | **MessageLog** | 发送记录,便于排查 |
@@ -351,7 +351,7 @@ Query`page`、`page_size`(默认同 sources)。**200** `{ "data": Temp
| 字段 | 类型 | 必填 | 说明 | | 字段 | 类型 | 必填 | 说明 |
|------|------|------|------| |------|------|------|------|
| `name` | string | 是 | 唯一名称(规则里用此名引用) | | `name` | string | 是 | 唯一名称(规则里用此名引用) |
| `type` | string | 是 | `dingtalk` / `wecom` / `email` / `bark` | | `type` | string | 是 | `dingtalk` / `wecom` / `email` / `bark` / `safew` |
| `config` | object | 是 | 见下方各渠道配置 | | `config` | object | 是 | 见下方各渠道配置 |
| `status` | int | 否 | 默认 `1` | | `status` | int | 否 | 默认 `1` |
@@ -398,6 +398,52 @@ SMTP 使用全局 `config.yaml` 的 `smtp` 段;支持 587 STARTTLS / 465 TLS
} }
``` ```
**SafeW `safew`**
```json
{
"token": "<bot token>",
"chat_id": "123456789"
}
```
`chat_id` 可为数字或字符串(含 `@username`)。消息以 MarkdownV2 发送:标题加粗,标题和正文均自动转义。
#### `POST /api/v1/channels/safew/chats` — 列出已监控的 SafeW 群
创建渠道时选群。Admin Key。用 `getUpdates` 攒到 Redis 里的群;bot 尚未在群里收到过更新时列表为空。
```json
{
"token": "<bot token>",
"q": "测试AI"
}
```
`q` 可选,按群名 / username / chat_id 子串过滤(不区分大小写)。
**200**
```json
{
"data": [
{
"id": "10000778141",
"type": "group",
"title": "测试AI",
"username": null
}
],
"total": 1
}
```
`id` 为字符串。Token 无效 → **401**
#### `GET /api/v1/channels/:id/chats?q=` — 用已存 token 列群
编辑已有 safew 渠道。非 safew / 无 token → **400**;渠道不存在 → **404**。响应同上。
#### `GET /api/v1/channels` — 列表 #### `GET /api/v1/channels` — 列表
Query`page``page_size`。**200** `{ "data": Channel[], "total", "page" }` Query`page``page_size`。**200** `{ "data": Channel[], "total", "page" }`
+18 -1
View File
@@ -15,6 +15,7 @@ import (
"aiaa-notification-service/internal/config" "aiaa-notification-service/internal/config"
"aiaa-notification-service/internal/engine" "aiaa-notification-service/internal/engine"
"aiaa-notification-service/internal/handler" "aiaa-notification-service/internal/handler"
"aiaa-notification-service/internal/safew"
"aiaa-notification-service/internal/store" "aiaa-notification-service/internal/store"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
@@ -125,7 +126,21 @@ func main() {
notifyH := handler.NewNotifyHandler(st, redisCache, matcher, renderer, router) notifyH := handler.NewNotifyHandler(st, redisCache, matcher, renderer, router)
sourceH := handler.NewSourceHandler(st, redisCache) sourceH := handler.NewSourceHandler(st, redisCache)
templateH := handler.NewTemplateHandler(st, redisCache) templateH := handler.NewTemplateHandler(st, redisCache)
channelH := handler.NewChannelHandler(st, redisCache)
poller := func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error) {
return (&adapter.SafeWSender{}).PollGroupChats(token, offset, timeout)
}
var watcherStore safew.ChatStore
if redisCache != nil {
watcherStore = redisCache
} else {
watcherStore = safew.NewMemStore()
slog.Warn("safew chats using in-memory store")
}
safewWatcher := safew.NewWatcher(watcherStore, poller)
defer safewWatcher.Stop()
channelH := handler.NewChannelHandler(st, redisCache, safewWatcher)
ruleH := handler.NewRuleHandler(st, redisCache) ruleH := handler.NewRuleHandler(st, redisCache)
msgLogH := handler.NewMessageLogHandler(st) msgLogH := handler.NewMessageLogHandler(st)
@@ -168,6 +183,8 @@ func main() {
admin.DELETE("/templates/:id", templateH.Delete) admin.DELETE("/templates/:id", templateH.Delete)
// Channels // Channels
admin.POST("/channels/safew/chats", channelH.ListSafewChats)
admin.GET("/channels/:id/chats", channelH.ListChannelSafewChats)
admin.POST("/channels", channelH.Create) admin.POST("/channels", channelH.Create)
admin.GET("/channels", channelH.List) admin.GET("/channels", channelH.List)
admin.GET("/channels/:id", channelH.Get) admin.GET("/channels/:id", channelH.Get)
+33
View File
@@ -260,6 +260,39 @@ curl -X POST 'http://localhost:8080/api/v1/channels' \
}' }'
``` ```
### 创建 ChannelSafeW
```bash
curl -X POST 'http://localhost:8080/api/v1/channels' \
-H 'Authorization: Bearer admin-sk-change-me' \
-H 'Content-Type: application/json' \
-d '{
"name": "safew-ops",
"type": "safew",
"config": {
"token": "<bot token>",
"chat_id": "123456789"
},
"status": 1
}'
```
### 列出 SafeW 群(创建渠道时)
```bash
curl -X POST 'http://localhost:8080/api/v1/channels/safew/chats' \
-H 'Authorization: Bearer admin-sk-change-me' \
-H 'Content-Type: application/json' \
-d '{"token":"<bot token>","q":"测试AI"}'
```
### 列出 SafeW 群(已有渠道)
```bash
curl -X GET 'http://localhost:8080/api/v1/channels/1/chats?q=' \
-H 'Authorization: Bearer admin-sk-change-me'
```
### 列出 Channel ### 列出 Channel
```bash ```bash
@@ -0,0 +1,538 @@
# SafeW Channel Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Add outbound channel type `safew` that sends rendered notifications via SafeW Bot API `sendMessage`.
**Architecture:** New `SafeWSender` implementing `adapter.ChannelSender`. Per-channel `token` + `chat_id` in `channel.config`. POST `https://api.safew.bot/bot{token}/sendMessage` with MarkdownV2 text (title and content auto-escaped). Unexported `apiBase` is a test seam only — not part of channel config.
**Tech Stack:** Go 1.22+, stdlib `net/http`, `encoding/json`, `testing`, `httptest`. No new dependencies.
## Global Constraints
- Channel type string: `safew`
- Config JSON keys: `token`, `chat_id` (string or number; normalize to string)
- API: `POST https://api.safew.bot/bot{token}/sendMessage`
- Body: `{chat_id, text, parse_mode: "MarkdownV2"}`
- `text = "*" + escapeMarkdownV2(title) + "*\n" + escapeMarkdownV2(content)`
- MarkdownV2 specials: `_ * [ ] ( ) ~ \` > # + - = | { } . !` — prefix each with `\`
- HTTP 2xx with `ok: false` is failure; include `description` in the error
- Do not add global config, rate limiter, inbound webhook, or e2e changes
- Do not commit unless the user explicitly asks
---
## File Structure
| File | Responsibility |
|------|----------------|
| `internal/adapter/safew.go` | `SafeWSender`, MarkdownV2 escape, chat_id parse, sendMessage HTTP |
| `internal/adapter/safew_test.go` | Escape unit tests + httptest send contract tests |
| `internal/adapter/adapter.go` | Register `"safew"` in `NewSender` |
| `README.md` | Channel type list + safew config example |
| `docs/httpie/curls.md` | Create-channel curl |
---
### Task 1: MarkdownV2 escape helper
**Files:**
- Create: `internal/adapter/safew.go` (helper only in this task)
- Test: `internal/adapter/safew_test.go`
**Interfaces:**
- Consumes: nothing
- Produces: `func escapeMarkdownV2(s string) string` in package `adapter`
- [ ] **Step 1: Write the failing tests**
Create `internal/adapter/safew_test.go`:
```go
package adapter
import "testing"
func TestEscapeMarkdownV2(t *testing.T) {
tests := []struct {
name string
in string
want string
}{
{name: "underscore and dot", in: "hello_world.", want: `hello\_world\.`},
{name: "empty", in: "", want: ""},
{name: "no specials", in: "hello", want: "hello"},
{name: "backslash first", in: `a\b`, want: `a\\b`},
{
name: "all specials",
in: "_*[]()~`>#+-=|{}.!\\",
want: "\\_\\*\\[\\]\\(\\)\\~\\`\\>\\#\\+\\-\\=\\|\\{\\}\\.\\!\\\\",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := escapeMarkdownV2(tt.in)
if got != tt.want {
t.Fatalf("escapeMarkdownV2(%q) = %q, want %q", tt.in, got, tt.want)
}
})
}
}
```
- [ ] **Step 2: Run test to verify it fails**
Run: `go test ./internal/adapter/ -run TestEscapeMarkdownV2 -count=1`
Expected: FAIL with `undefined: escapeMarkdownV2`
- [ ] **Step 3: Write minimal implementation**
Create `internal/adapter/safew.go`:
```go
package adapter
import "strings"
func escapeMarkdownV2(s string) string {
var b strings.Builder
b.Grow(len(s))
for _, r := range s {
switch r {
case '_', '*', '[', ']', '(', ')', '~', '`', '>', '#', '+', '-', '=', '|', '{', '}', '.', '!', '\\':
b.WriteByte('\\')
}
b.WriteRune(r)
}
return b.String()
}
```
- [ ] **Step 4: Run tests and make sure they pass**
Run: `go test ./internal/adapter/ -run TestEscapeMarkdownV2 -count=1`
Expected: `PASS`
---
### Task 2: SafeWSender + factory registration
**Files:**
- Modify: `internal/adapter/safew.go` (full sender)
- Modify: `internal/adapter/safew_test.go` (httptest cases)
- Modify: `internal/adapter/adapter.go` (`case "safew"`)
**Interfaces:**
- Consumes: `escapeMarkdownV2(s string) string`
- Produces:
- `type SafeWSender struct { apiBase string }` — empty `apiBase` means `https://api.safew.bot`
- `func (s *SafeWSender) Type() string` returns `"safew"`
- `func (s *SafeWSender) Send(title, content string, config json.RawMessage) error`
- `NewSender("safew", smtpCfg, limiter)` returns `&SafeWSender{}`
- [ ] **Step 1: Write the failing send tests**
In `internal/adapter/safew_test.go`, expand imports to:
```go
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"aiaa-notification-service/internal/config"
)
```
Keep `TestEscapeMarkdownV2` unchanged. Append:
```go
func TestNewSenderSafew(t *testing.T) {
s, err := NewSender("safew", &config.SMTPConfig{}, nil)
if err != nil {
t.Fatalf("NewSender(safew): %v", err)
}
if s.Type() != "safew" {
t.Fatalf("Type() = %q, want safew", s.Type())
}
}
func TestSafeWSenderSendSuccess(t *testing.T) {
var gotPath string
var gotBody map[string]any
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotPath = r.URL.Path
raw, _ := io.ReadAll(r.Body)
if err := json.Unmarshal(raw, &gotBody); err != nil {
t.Errorf("unmarshal request: %v", err)
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"ok":true,"result":{}}`))
}))
defer srv.Close()
sender := &SafeWSender{apiBase: srv.URL}
cfg, _ := json.Marshal(map[string]any{"token": "tok-1", "chat_id": "123456789"})
if err := sender.Send("hello_world.", "price=1.5", cfg); err != nil {
t.Fatalf("Send: %v", err)
}
if gotPath != "/bottok-1/sendMessage" {
t.Fatalf("path = %q, want /bottok-1/sendMessage", gotPath)
}
if gotBody["chat_id"] != "123456789" {
t.Fatalf("chat_id = %#v, want \"123456789\"", gotBody["chat_id"])
}
if gotBody["parse_mode"] != "MarkdownV2" {
t.Fatalf("parse_mode = %#v, want MarkdownV2", gotBody["parse_mode"])
}
wantText := "*" + escapeMarkdownV2("hello_world.") + "*\n" + escapeMarkdownV2("price=1.5")
if gotBody["text"] != wantText {
t.Fatalf("text = %#v, want %#v", gotBody["text"], wantText)
}
}
func TestSafeWSenderSendNumericChatID(t *testing.T) {
var gotBody map[string]any
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
raw, _ := io.ReadAll(r.Body)
_ = json.Unmarshal(raw, &gotBody)
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"ok":true}`))
}))
defer srv.Close()
sender := &SafeWSender{apiBase: srv.URL}
cfg := json.RawMessage(`{"token":"tok","chat_id":123456789}`)
if err := sender.Send("t", "c", cfg); err != nil {
t.Fatalf("Send: %v", err)
}
if gotBody["chat_id"] != "123456789" {
t.Fatalf("chat_id = %#v, want string \"123456789\"", gotBody["chat_id"])
}
}
func TestSafeWSenderSendOKFalse(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"ok":false,"description":"chat not found"}`))
}))
defer srv.Close()
sender := &SafeWSender{apiBase: srv.URL}
cfg, _ := json.Marshal(map[string]string{"token": "tok", "chat_id": "1"})
err := sender.Send("t", "c", cfg)
if err == nil {
t.Fatal("expected error")
}
if !strings.Contains(err.Error(), "chat not found") {
t.Fatalf("error %q should contain description", err)
}
}
func TestSafeWSenderSendHTTPError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusUnauthorized)
_, _ = w.Write([]byte(`{"ok":false,"description":"Unauthorized"}`))
}))
defer srv.Close()
sender := &SafeWSender{apiBase: srv.URL}
cfg, _ := json.Marshal(map[string]string{"token": "bad", "chat_id": "1"})
err := sender.Send("t", "c", cfg)
if err == nil {
t.Fatal("expected error")
}
if !strings.Contains(err.Error(), "Unauthorized") {
t.Fatalf("error %q should contain description", err)
}
}
func TestSafeWSenderSendMissingFields(t *testing.T) {
hits := 0
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
hits++
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"ok":true}`))
}))
defer srv.Close()
sender := &SafeWSender{apiBase: srv.URL}
cases := []json.RawMessage{
json.RawMessage(`{"chat_id":"1"}`),
json.RawMessage(`{"token":"tok"}`),
json.RawMessage(`{"token":" ","chat_id":"1"}`),
json.RawMessage(`{"token":"tok","chat_id":""}`),
json.RawMessage(`{"token":"tok","chat_id":null}`),
}
for _, cfg := range cases {
if err := sender.Send("t", "c", cfg); err == nil {
t.Fatalf("expected error for config %s", cfg)
}
}
if hits != 0 {
t.Fatalf("unexpected HTTP calls: %d", hits)
}
if err := sender.Send("t", "c", json.RawMessage(`{`)); err == nil {
t.Fatal("expected error for invalid json")
}
if hits != 0 {
t.Fatalf("unexpected HTTP calls after invalid json: %d", hits)
}
}
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `go test ./internal/adapter/ -count=1`
Expected: FAIL with `undefined: SafeWSender` and/or `unknown channel type: safew`
- [ ] **Step 3: Implement sender and register it**
Replace `internal/adapter/safew.go` with:
```go
package adapter
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
)
const safewAPIBase = "https://api.safew.bot"
type SafeWSender struct {
apiBase string
}
type safewConfig struct {
Token string `json:"token"`
ChatID json.RawMessage `json:"chat_id"`
}
type safewMessage struct {
ChatID string `json:"chat_id"`
Text string `json:"text"`
ParseMode string `json:"parse_mode"`
}
type safewAPIResponse struct {
OK bool `json:"ok"`
Description string `json:"description"`
}
func (s *SafeWSender) Type() string { return "safew" }
func (s *SafeWSender) Send(title, content string, config json.RawMessage) error {
var cfg safewConfig
if err := json.Unmarshal(config, &cfg); err != nil {
return fmt.Errorf("parse safew config: %w", err)
}
token := strings.TrimSpace(cfg.Token)
if token == "" {
return fmt.Errorf("safew: token is required")
}
chatID, err := parseSafewChatID(cfg.ChatID)
if err != nil {
return err
}
payload := safewMessage{
ChatID: chatID,
Text: "*" + escapeMarkdownV2(title) + "*\n" + escapeMarkdownV2(content),
ParseMode: "MarkdownV2",
}
body, err := json.Marshal(payload)
if err != nil {
return fmt.Errorf("safew marshal: %w", err)
}
resp, err := http.Post(s.endpoint(token), "application/json", bytes.NewReader(body))
if err != nil {
return fmt.Errorf("safew send: %w", err)
}
defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body)
if err != nil {
return fmt.Errorf("safew read: %w", err)
}
if resp.StatusCode >= 400 {
desc := safewErrorDescription(respBody)
if desc != "" {
return fmt.Errorf("safew returned status %d: %s", resp.StatusCode, desc)
}
return fmt.Errorf("safew returned status %d", resp.StatusCode)
}
var api safewAPIResponse
if err := json.Unmarshal(respBody, &api); err != nil {
return fmt.Errorf("safew decode: %w", err)
}
if !api.OK {
desc := api.Description
if desc == "" {
desc = "ok=false"
}
return fmt.Errorf("safew: %s", desc)
}
return nil
}
func (s *SafeWSender) endpoint(token string) string {
base := s.apiBase
if base == "" {
base = safewAPIBase
}
return strings.TrimRight(base, "/") + "/bot" + token + "/sendMessage"
}
func parseSafewChatID(raw json.RawMessage) (string, error) {
raw = bytes.TrimSpace(raw)
if len(raw) == 0 || string(raw) == "null" {
return "", fmt.Errorf("safew: chat_id is required")
}
if raw[0] == '"' {
var id string
if err := json.Unmarshal(raw, &id); err != nil {
return "", fmt.Errorf("safew: invalid chat_id: %w", err)
}
id = strings.TrimSpace(id)
if id == "" {
return "", fmt.Errorf("safew: chat_id is required")
}
return id, nil
}
return string(raw), nil
}
func safewErrorDescription(body []byte) string {
var api safewAPIResponse
if err := json.Unmarshal(body, &api); err != nil {
return strings.TrimSpace(string(body))
}
return api.Description
}
func escapeMarkdownV2(s string) string {
var b strings.Builder
b.Grow(len(s))
for _, r := range s {
switch r {
case '_', '*', '[', ']', '(', ')', '~', '`', '>', '#', '+', '-', '=', '|', '{', '}', '.', '!', '\\':
b.WriteByte('\\')
}
b.WriteRune(r)
}
return b.String()
}
```
In `internal/adapter/adapter.go`, add the factory case after `"bark"`:
```go
case "bark":
return &BarkSender{}, nil
case "safew":
return &SafeWSender{}, nil
case "email":
return NewEmailSender(*smtpCfg), nil
```
- [ ] **Step 4: Run tests and make sure they pass**
Run: `go test ./internal/adapter/ -count=1`
Expected: `PASS` (existing dingtalk limiter tests plus new safew tests)
---
### Task 3: Docs
**Files:**
- Modify: `README.md`
- Modify: `docs/httpie/curls.md`
**Interfaces:**
- Consumes: channel type `safew`, config `{token, chat_id}`
- Produces: documented create example for operators
- [ ] **Step 1: Update README channel list**
In `README.md` line 3, change:
`异步分发到钉钉 / 企业微信 / 邮件 / Bark。`
to:
`异步分发到钉钉 / 企业微信 / 邮件 / Bark / SafeW。`
Line 17, change:
`| **Channel** | 发送渠道配置(钉钉 / 企微 / 邮件 / Bark |`
to:
`| **Channel** | 发送渠道配置(钉钉 / 企微 / 邮件 / Bark / SafeW |`
Line 354, change:
`| `type` | string | 是 | `dingtalk` / `wecom` / `email` / `bark` |`
to:
`| `type` | string | 是 | `dingtalk` / `wecom` / `email` / `bark` / `safew` |`
After the Bark config block (after the closing ` ``` ` of the bark url example, before `#### GET /api/v1/channels`), insert:
```markdown
**SafeW `safew`**
```json
{
"token": "<bot token>",
"chat_id": "123456789"
}
```
`chat_id` 可为数字或字符串(含 `@username`)。消息以 MarkdownV2 发送:标题加粗,标题和正文均自动转义。
```
- [ ] **Step 2: Add curl example**
In `docs/httpie/curls.md`, after the Bark create example (after its closing ` ``` `, before `### 列出 Channel`), insert:
```markdown
### 创建 ChannelSafeW
```bash
curl -X POST 'http://localhost:8080/api/v1/channels' \
-H 'Authorization: Bearer admin-sk-change-me' \
-H 'Content-Type: application/json' \
-d '{
"name": "safew-ops",
"type": "safew",
"config": {
"token": "<bot token>",
"chat_id": "123456789"
},
"status": 1
}'
```
```
- [ ] **Step 3: Re-run adapter tests**
Run: `go test ./internal/adapter/ -count=1`
Expected: `PASS`
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,122 @@
# SafeW 渠道设计
**Date:** 2026-08-14
**Status:** Approved
## Goal
新增出站渠道 `safew`:规则命中并渲染模板后,通过 SafeW Bot API `sendMessage` 把通知发到指定会话。
## Non-goals
- 接收 SafeW webhook / 入站更新
- 全局 bot token`config.yaml` 不加 safew 段)
- 自定义 API base URL
- 钉钉式按分钟限流排队
- 远程 e2e 真实发送(无现成测试 bot 凭据)
- 管理 API 对 channel `type` 做白名单校验(现有渠道也没有)
## Architecture
沿用现有扩展点:实现 `adapter.ChannelSender`,在 `adapter.NewSender` 注册 `"safew"`
不改表结构。`channel.type` 为字符串,`channel.config` 为 JSON。
发送仍走:
```
POST /api/v1/notify
→ 匹配规则、渲染模板
→ Router 按 ch.Type 取 sender
→ 异步 Send(title, content, config)
→ 失败重试 1s / 5s / 30s
→ 写入 message_log
```
`title` 沿用现有值:`{source.name}: {event}``content` 为模板渲染结果。
## Channel config
创建 channel 时:
```json
{
"name": "safew-ops",
"type": "safew",
"config": {
"token": "<bot token>",
"chat_id": "123456789"
},
"status": 1
}
```
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `token` | string | 是 | SafeW Bot Token,每个 channel 独立 |
| `chat_id` | string 或 number | 是 | 目标会话。JSON 里数字 ID(`123`)或字符串(`"123"` / `"@name"`)都接受,sender 归一成字符串再发给 SafeW |
`token` 或空 `chat_id`:立即返回 error,不发 HTTP。`token` / `chat_id` 读写前 `strings.TrimSpace`
## Send contract
- URL`POST https://api.safew.bot/bot{token}/sendMessage`
- Header`Content-Type: application/json`
- Body
```json
{
"chat_id": "<chat_id>",
"text": "*<escaped title>*\n<escaped content>",
"parse_mode": "MarkdownV2"
}
```
### MarkdownV2 转义
标题和正文都自动转义后再组装。模板里的 markdown 会变成字面量,避免特殊字符导致 400。
转义字符:`_ * [ ] ( ) ~ \` > # + - = | { } . !`
规则:每个字符前加 `\`。
组装:`text = "*" + escape(title) + "*\n" + escape(content)`
## Error handling
| 情况 | 行为 |
|------|------|
| config JSON 非法、缺 `token` / `chat_id` | 返回 error(router 仍会按现有策略重试,每次同样失败) |
| 网络错误、HTTP status ≥ 400 | 返回 error,触发重试 |
| HTTP 2xx 但 body `{"ok":false}` | 视为失败;error 带上 SafeW 的 `description` |
| HTTP 2xx 且 `ok: true` | 成功 |
不单独处理 429。不引入新的限流器。
## Components
| 文件 | 改动 |
|------|------|
| `internal/adapter/safew.go` | 新增 `SafeWSender` |
| `internal/adapter/safew_test.go` | httptest mock 单元测试 |
| `internal/adapter/adapter.go` | `NewSender` 增加 `case "safew"` |
| `README.md` | 渠道列表与创建示例 |
| `docs/httpie/curls.md` | 创建 SafeW channel 的 curl |
## Tests
`internal/adapter/safew_test.go``httptest.Server` 模拟 SafeW):
1. 转义:`hello_world.``hello\_world\.`;标题包在 `*...*`
2. 成功请求:URL 含 `/bot{token}/sendMessage`body 含 `chat_id``parse_mode=MarkdownV2`、转义后的 `text`
3. `ok: false` 或 HTTP 4xx → error,且含 `description`(若有)
4.`token` / `chat_id` → 失败且不发 HTTP
5. `chat_id` 为 JSON number`123456789`)时仍能发出,请求里是字符串 `"123456789"`
不修改 `test/e2e/`。手工验证用 `docs/httpie/curls.md` 里的创建示例。
## Success criteria
- 可创建 `type=safew` 的 channel,并绑到规则
- 命中规则后异步调用 SafeW `sendMessage`
- 含 MarkdownV2 特殊字符的标题/正文能发出(已转义)
- SafeW 业务失败(`ok: false`)记入失败日志,而不是当成成功
@@ -0,0 +1,150 @@
# SafeW 群列表接口设计
**Date:** 2026-08-15
**Status:** Approved
## Goal
为 safew 渠道提供两个管理接口,返回该 bot token **已经监控到的群**。SafeW 没有 `getChats`;用 `getUpdates` 收集群聊,写入 Redis,列表接口读 Redis。
## Non-goals
- 列出 private / channel
- 在请求路径上 `timeout=30` 阻塞 UI
- 把 bot token 明文写入 Redis key 或提交到仓库
- 远程 e2e 真实 SafeW 轮询
- 新增 MySQL 表
## Endpoints
Admin Bearer 鉴权,与现有 `/api/v1/channels` 相同。
### `POST /api/v1/channels/safew/chats`
创建渠道时选群。Body
```json
{
"token": "<bot token>",
"q": "测试AI"
}
```
| 字段 | 必填 | 说明 |
|------|------|------|
| `token` | 是 | SafeW Bot Token |
| `q` | 否 | 按群名 / username / chat_id 子串过滤(不区分大小写) |
路由必须注册在 `GET /channels/:id` 之前,避免 `safew` 被当成 id。
### `GET /api/v1/channels/:id/chats?q=`
编辑已有渠道。用库里 `channel.config.token`,前端不再传 token。
| 情况 | 响应 |
|------|------|
| channel 不存在 | 404 `{"error":"channel not found"}` |
| `type != safew` | 400 `{"error":"channel is not safew"}` |
| config 无 token | 400 `{"error":"safew token is required"}` |
### 成功响应(两个入口相同)
```json
{
"data": [
{
"id": "10000778141",
"type": "group",
"title": "测试AI",
"username": null
}
],
"total": 1
}
```
- `id` 永远是 JSON 字符串(SafeW 的数字 id 可能超过 JS 安全整数)
- `username` 没有则为 `null`
- 默认只含 `group` / `supergroup`
- 空列表:`200``{"data":[],"total":0}`(bot 尚未在任何群收到更新时正常)
Token 无效:`401``{"error":"<SafeW description>"}`
## Architecture
```
POST/GET chats
→ 解析 token
→ Watcher.Ensure(token) // 懒启动该 token 的后台长轮询
→ Poll getUpdates timeout=0 // 抽干积压,不阻塞
→ 合并群到 Redis
→ 按 q 过滤,返回 data/total
Watcher (per token, in-process)
loop:
getUpdates timeout=30
合并群到 Redis
更新 offset
```
同一 token 对 SafeW `getUpdates` 必须串行(Redis 锁),避免并发把 `offset` 冲掉。
## SafeW getUpdates
- URL`POST https://api.safew.bot/bot{token}/getUpdates`
- Body`{"timeout":<0|30>,"offset":<last+1>,"limit":100}`
- 处理完一批后:`offset = max(update_id)+1` 写入 Redis
- 从每条 update 的 `message` / `edited_message` / `my_chat_member` / `chat_member` / `channel_post` 等字段里取 `chat`;只保留 `type``group``supergroup`
- `chat.id``json.Number` / raw 转成十进制字符串,禁止 `float64`
群只有 bot **收到过该群的更新**(加群、有人说话等)才会出现。
## Redis
Key 用 `sha256(token)` 的 hex,不把 token 放进 key。
| Key | 类型 | 内容 | TTL |
|-----|------|------|-----|
| `safew:chats:{hash}` | Hash | field=`chat_id`value=`{"id","type","title","username"}` | 无 |
| `safew:offset:{hash}` | String | 已确认的最大 `update_id` | 无 |
| `safew:poll:{hash}` | String | `getUpdates` 互斥锁,短 TTL(约 35s) | 有 |
进程重启后群记录仍在;Watcher 从存着的 offset 继续。
## Components
| 文件 | 职责 |
|------|------|
| `internal/adapter/safew.go` | `getUpdates` HTTP、解析 chat、id 转字符串 |
| `internal/cache/safew_chats.go` | chats / offset / lock |
| `internal/safew/watcher.go` | 每 token 一条 goroutine`timeout=30` |
| `internal/handler/channel.go` | 两个 HTTP handler |
| `cmd/server/main.go` | 注册路由、注入 Watcher、shutdown 时 Stop |
| `README.md` / `docs/httpie/curls.md` | 接口文档 |
## Error handling
| 情况 | 行为 |
|------|------|
| SafeW 401 / token 无效 | 列表接口 401;Watcher 打日志,该轮结束,下次再试 |
| 网络错误、`getUpdates` 5xx | 列表仍返回 Redis 已有群;Watcher sleep 后重试 |
| 抢不到 poll 锁 | 跳过本次拉取,直接返回 Redis |
| JSON 非法 / 缺 token | 400 |
## Tests
不引入真实 token。
1. **解析**update 含 group + private → 只留下 group`id` 为字符串 `"10000778141"`
2. **过滤**`q=测试` 命中 title`q=10000778141` 命中 id。
3. **getUpdates mock**httptest 返回一批 updateoffset 前进。
4. **handler**:非 safew channel → 400;缺 token → 400。
Watcher 用可注入的 poll 函数 + 假 cache,不断言真实 30s 阻塞。
## Success criteria
- 两个接口返回已监控群,`id` 为字符串
- 默认只有群;`q` 可过滤
- 后台 `timeout=30` 持续写入 Redis;列表接口不卡 30 秒
- 仓库与 Redis key 中不出现明文 token
+2
View File
@@ -21,6 +21,8 @@ func NewSender(channelType string, smtpCfg *config.SMTPConfig, dingtalkLimiter *
return &WeComSender{}, nil return &WeComSender{}, nil
case "bark": case "bark":
return &BarkSender{}, nil return &BarkSender{}, nil
case "safew":
return &SafeWSender{}, nil
case "email": case "email":
return NewEmailSender(*smtpCfg), nil return NewEmailSender(*smtpCfg), nil
default: default:
+318
View File
@@ -0,0 +1,318 @@
package adapter
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
)
const safewAPIBase = "https://api.safew.bot"
type SafeWSender struct {
apiBase string
}
type safewConfig struct {
Token string `json:"token"`
ChatID json.RawMessage `json:"chat_id"`
}
type safewMessage struct {
ChatID string `json:"chat_id"`
Text string `json:"text"`
ParseMode string `json:"parse_mode"`
}
type safewAPIResponse struct {
OK bool `json:"ok"`
Description string `json:"description"`
}
func (s *SafeWSender) Type() string { return "safew" }
func (s *SafeWSender) Send(title, content string, config json.RawMessage) error {
var cfg safewConfig
if err := json.Unmarshal(config, &cfg); err != nil {
return fmt.Errorf("parse safew config: %w", err)
}
token := strings.TrimSpace(cfg.Token)
if token == "" {
return fmt.Errorf("safew: token is required")
}
chatID, err := parseSafewChatID(cfg.ChatID)
if err != nil {
return err
}
payload := safewMessage{
ChatID: chatID,
Text: "*" + escapeMarkdownV2(title) + "*\n" + escapeMarkdownV2(content),
ParseMode: "MarkdownV2",
}
body, err := json.Marshal(payload)
if err != nil {
return fmt.Errorf("safew marshal: %w", err)
}
resp, err := http.Post(s.endpoint(token), "application/json", bytes.NewReader(body))
if err != nil {
return fmt.Errorf("safew send: %w", err)
}
defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body)
if err != nil {
return fmt.Errorf("safew read: %w", err)
}
if resp.StatusCode >= 400 {
desc := safewErrorDescription(respBody)
if desc != "" {
return fmt.Errorf("safew returned status %d: %s", resp.StatusCode, desc)
}
return fmt.Errorf("safew returned status %d", resp.StatusCode)
}
var api safewAPIResponse
if err := json.Unmarshal(respBody, &api); err != nil {
return fmt.Errorf("safew decode: %w", err)
}
if !api.OK {
desc := api.Description
if desc == "" {
desc = "ok=false"
}
return fmt.Errorf("safew: %s", desc)
}
return nil
}
func (s *SafeWSender) endpoint(token string) string {
return s.methodURL(token, "sendMessage")
}
func (s *SafeWSender) methodURL(token, method string) string {
base := s.apiBase
if base == "" {
base = safewAPIBase
}
return strings.TrimRight(base, "/") + "/bot" + token + "/" + method
}
type SafewAuthError struct {
Description string
}
func (e *SafewAuthError) Error() string {
if e.Description == "" {
return "safew unauthorized"
}
return e.Description
}
func (s *SafeWSender) PollGroupChats(token string, offset int64, timeout int) ([]SafewChat, int64, error) {
token = strings.TrimSpace(token)
if token == "" {
return nil, offset, fmt.Errorf("safew: token is required")
}
reqBody, _ := json.Marshal(map[string]any{
"timeout": timeout,
"offset": offset,
"limit": 100,
})
resp, err := http.Post(s.methodURL(token, "getUpdates"), "application/json", bytes.NewReader(reqBody))
if err != nil {
return nil, offset, fmt.Errorf("safew getUpdates: %w", err)
}
defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body)
if err != nil {
return nil, offset, fmt.Errorf("safew getUpdates read: %w", err)
}
if resp.StatusCode == http.StatusUnauthorized {
return nil, offset, &SafewAuthError{Description: safewErrorDescription(respBody)}
}
if resp.StatusCode >= 400 {
desc := safewErrorDescription(respBody)
if desc != "" {
return nil, offset, fmt.Errorf("safew getUpdates status %d: %s", resp.StatusCode, desc)
}
return nil, offset, fmt.Errorf("safew getUpdates status %d", resp.StatusCode)
}
var api safewAPIResponse
if err := json.Unmarshal(respBody, &api); err == nil && !api.OK {
if resp.StatusCode == 401 || strings.Contains(strings.ToLower(api.Description), "token") {
return nil, offset, &SafewAuthError{Description: api.Description}
}
return nil, offset, fmt.Errorf("safew: %s", api.Description)
}
chats, maxID, err := GroupsFromUpdates(respBody)
if err != nil {
return nil, offset, err
}
next := offset
if maxID > 0 {
next = maxID + 1
}
return chats, next, nil
}
func parseSafewChatID(raw json.RawMessage) (string, error) {
raw = bytes.TrimSpace(raw)
if len(raw) == 0 || string(raw) == "null" {
return "", fmt.Errorf("safew: chat_id is required")
}
if raw[0] == '"' {
var id string
if err := json.Unmarshal(raw, &id); err != nil {
return "", fmt.Errorf("safew: invalid chat_id: %w", err)
}
id = strings.TrimSpace(id)
if id == "" {
return "", fmt.Errorf("safew: chat_id is required")
}
return id, nil
}
return string(raw), nil
}
func safewErrorDescription(body []byte) string {
var api safewAPIResponse
if err := json.Unmarshal(body, &api); err != nil {
return strings.TrimSpace(string(body))
}
return api.Description
}
func escapeMarkdownV2(s string) string {
var b strings.Builder
b.Grow(len(s))
for _, r := range s {
switch r {
case '_', '*', '[', ']', '(', ')', '~', '`', '>', '#', '+', '-', '=', '|', '{', '}', '.', '!', '\\':
b.WriteByte('\\')
}
b.WriteRune(r)
}
return b.String()
}
type SafewChat struct {
ID string `json:"id"`
Type string `json:"type"`
Title string `json:"title"`
Username *string `json:"username"`
}
func GroupsFromUpdates(body []byte) ([]SafewChat, int64, error) {
dec := json.NewDecoder(bytes.NewReader(body))
dec.UseNumber()
var wrap struct {
OK bool `json:"ok"`
Result []json.RawMessage `json:"result"`
}
if err := dec.Decode(&wrap); err != nil {
return nil, 0, fmt.Errorf("safew updates decode: %w", err)
}
seen := map[string]SafewChat{}
var maxID int64
for _, item := range wrap.Result {
id, chat, err := parseUpdateItem(item)
if err != nil {
return nil, 0, err
}
if id > maxID {
maxID = id
}
if chat == nil {
continue
}
if chat.Type != "group" && chat.Type != "supergroup" {
continue
}
seen[chat.ID] = *chat
}
out := make([]SafewChat, 0, len(seen))
for _, c := range seen {
out = append(out, c)
}
return out, maxID, nil
}
func parseUpdateItem(item json.RawMessage) (int64, *SafewChat, error) {
dec := json.NewDecoder(bytes.NewReader(item))
dec.UseNumber()
var u map[string]json.RawMessage
if err := dec.Decode(&u); err != nil {
return 0, nil, err
}
var updateID int64
if raw, ok := u["update_id"]; ok {
var n json.Number
if err := json.Unmarshal(raw, &n); err == nil {
updateID, _ = n.Int64()
}
}
for _, key := range []string{"message", "edited_message", "channel_post", "edited_channel_post", "my_chat_member", "chat_member"} {
raw, ok := u[key]
if !ok {
continue
}
chat := chatFromNested(raw)
if chat != nil {
return updateID, chat, nil
}
}
return updateID, nil, nil
}
func chatFromNested(raw json.RawMessage) *SafewChat {
dec := json.NewDecoder(bytes.NewReader(raw))
dec.UseNumber()
var obj map[string]json.RawMessage
if err := dec.Decode(&obj); err != nil {
return nil
}
chatRaw, ok := obj["chat"]
if !ok {
return nil
}
dec = json.NewDecoder(bytes.NewReader(chatRaw))
dec.UseNumber()
var c struct {
ID json.Number `json:"id"`
Type string `json:"type"`
Title string `json:"title"`
Username *string `json:"username"`
}
if err := dec.Decode(&c); err != nil {
return nil
}
id := strings.TrimSpace(c.ID.String())
if id == "" {
return nil
}
return &SafewChat{ID: id, Type: c.Type, Title: c.Title, Username: c.Username}
}
func FilterSafewChats(chats []SafewChat, q string) []SafewChat {
q = strings.TrimSpace(strings.ToLower(q))
if q == "" {
return chats
}
var out []SafewChat
for _, c := range chats {
uname := ""
if c.Username != nil {
uname = *c.Username
}
if strings.Contains(strings.ToLower(c.Title), q) ||
strings.Contains(strings.ToLower(uname), q) ||
strings.Contains(strings.ToLower(c.ID), q) {
out = append(out, c)
}
}
return out
}
+120
View File
@@ -0,0 +1,120 @@
package adapter
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
func TestGroupsFromUpdatesKeepsGroupsDropsPrivate(t *testing.T) {
body := []byte(`{
"ok": true,
"result": [
{"update_id": 100000001, "message": {"chat": {"id": 10000778141, "type": "group", "title": "测试AI"}}},
{"update_id": 100000002, "message": {"chat": {"id": 11, "type": "private", "first_name": "u"}}},
{"update_id": 100000003, "my_chat_member": {"chat": {"id": 22, "type": "supergroup", "title": "SG", "username": "sg_name"}}}
]
}`)
chats, maxID, err := GroupsFromUpdates(body)
if err != nil {
t.Fatal(err)
}
if maxID != 100000003 {
t.Fatalf("maxID=%d", maxID)
}
if len(chats) != 2 {
t.Fatalf("len=%d want 2: %#v", len(chats), chats)
}
byID := map[string]SafewChat{}
for _, c := range chats {
byID[c.ID] = c
}
g := byID["10000778141"]
if g.Type != "group" || g.Title != "测试AI" {
t.Fatalf("group: %#v", g)
}
raw, _ := json.Marshal(g)
var m map[string]any
_ = json.Unmarshal(raw, &m)
if _, ok := m["id"].(string); !ok {
t.Fatalf("id JSON type = %T, want string", m["id"])
}
sg := byID["22"]
if sg.Username == nil || *sg.Username != "sg_name" {
t.Fatalf("username: %#v", sg)
}
}
func TestFilterSafewChats(t *testing.T) {
chats := []SafewChat{
{ID: "10000778141", Type: "group", Title: "测试AI"},
{ID: "99", Type: "group", Title: "ops"},
}
got := FilterSafewChats(chats, "测试")
if len(got) != 1 || got[0].ID != "10000778141" {
t.Fatalf("%#v", got)
}
got = FilterSafewChats(chats, "10000778141")
if len(got) != 1 || got[0].Title != "测试AI" {
t.Fatalf("%#v", got)
}
got = FilterSafewChats(chats, "")
if len(got) != 2 {
t.Fatalf("empty q should keep all, got %d", len(got))
}
}
func TestPollGroupChatsSuccess(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/bottok/getUpdates" {
t.Errorf("path=%s", r.URL.Path)
}
raw, _ := io.ReadAll(r.Body)
var body map[string]any
_ = json.Unmarshal(raw, &body)
if body["timeout"] != float64(0) {
t.Errorf("timeout=%v", body["timeout"])
}
if body["offset"] != float64(5) {
t.Errorf("offset=%v", body["offset"])
}
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"ok":true,"result":[{"update_id":10,"message":{"chat":{"id":10000778141,"type":"group","title":"测试AI"}}}]}`))
}))
defer srv.Close()
s := &SafeWSender{apiBase: srv.URL}
chats, next, err := s.PollGroupChats("tok", 5, 0)
if err != nil {
t.Fatal(err)
}
if next != 11 {
t.Fatalf("next=%d want 11", next)
}
if len(chats) != 1 || chats[0].ID != "10000778141" {
t.Fatalf("%#v", chats)
}
}
func TestPollGroupChatsUnauthorized(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusUnauthorized)
_, _ = w.Write([]byte(`{"ok":false,"error_code":401,"description":"BOT_TOKEN_INVALID"}`))
}))
defer srv.Close()
s := &SafeWSender{apiBase: srv.URL}
_, _, err := s.PollGroupChats("bad", 0, 0)
if err == nil {
t.Fatal("expected error")
}
ae, ok := err.(*SafewAuthError)
if !ok {
t.Fatalf("type %T %v", err, err)
}
if !strings.Contains(ae.Description, "BOT_TOKEN_INVALID") {
t.Fatalf("%q", ae.Description)
}
}
+172
View File
@@ -0,0 +1,172 @@
package adapter
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"aiaa-notification-service/internal/config"
)
func TestEscapeMarkdownV2(t *testing.T) {
tests := []struct {
name string
in string
want string
}{
{name: "underscore and dot", in: "hello_world.", want: `hello\_world\.`},
{name: "empty", in: "", want: ""},
{name: "no specials", in: "hello", want: "hello"},
{name: "backslash first", in: `a\b`, want: `a\\b`},
{
name: "all specials",
in: "_*[]()~`>#+-=|{}.!\\",
want: "\\_\\*\\[\\]\\(\\)\\~\\`\\>\\#\\+\\-\\=\\|\\{\\}\\.\\!\\\\",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := escapeMarkdownV2(tt.in)
if got != tt.want {
t.Fatalf("escapeMarkdownV2(%q) = %q, want %q", tt.in, got, tt.want)
}
})
}
}
func TestNewSenderSafew(t *testing.T) {
s, err := NewSender("safew", &config.SMTPConfig{}, nil)
if err != nil {
t.Fatalf("NewSender(safew): %v", err)
}
if s.Type() != "safew" {
t.Fatalf("Type() = %q, want safew", s.Type())
}
}
func TestSafeWSenderSendSuccess(t *testing.T) {
var gotPath string
var gotBody map[string]any
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotPath = r.URL.Path
raw, _ := io.ReadAll(r.Body)
if err := json.Unmarshal(raw, &gotBody); err != nil {
t.Errorf("unmarshal request: %v", err)
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"ok":true,"result":{}}`))
}))
defer srv.Close()
sender := &SafeWSender{apiBase: srv.URL}
cfg, _ := json.Marshal(map[string]any{"token": "tok-1", "chat_id": "123456789"})
if err := sender.Send("hello_world.", "price=1.5", cfg); err != nil {
t.Fatalf("Send: %v", err)
}
if gotPath != "/bottok-1/sendMessage" {
t.Fatalf("path = %q, want /bottok-1/sendMessage", gotPath)
}
if gotBody["chat_id"] != "123456789" {
t.Fatalf("chat_id = %#v, want \"123456789\"", gotBody["chat_id"])
}
if gotBody["parse_mode"] != "MarkdownV2" {
t.Fatalf("parse_mode = %#v, want MarkdownV2", gotBody["parse_mode"])
}
wantText := "*" + escapeMarkdownV2("hello_world.") + "*\n" + escapeMarkdownV2("price=1.5")
if gotBody["text"] != wantText {
t.Fatalf("text = %#v, want %#v", gotBody["text"], wantText)
}
}
func TestSafeWSenderSendNumericChatID(t *testing.T) {
var gotBody map[string]any
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
raw, _ := io.ReadAll(r.Body)
_ = json.Unmarshal(raw, &gotBody)
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"ok":true}`))
}))
defer srv.Close()
sender := &SafeWSender{apiBase: srv.URL}
cfg := json.RawMessage(`{"token":"tok","chat_id":123456789}`)
if err := sender.Send("t", "c", cfg); err != nil {
t.Fatalf("Send: %v", err)
}
if gotBody["chat_id"] != "123456789" {
t.Fatalf("chat_id = %#v, want string \"123456789\"", gotBody["chat_id"])
}
}
func TestSafeWSenderSendOKFalse(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"ok":false,"description":"chat not found"}`))
}))
defer srv.Close()
sender := &SafeWSender{apiBase: srv.URL}
cfg, _ := json.Marshal(map[string]string{"token": "tok", "chat_id": "1"})
err := sender.Send("t", "c", cfg)
if err == nil {
t.Fatal("expected error")
}
if !strings.Contains(err.Error(), "chat not found") {
t.Fatalf("error %q should contain description", err)
}
}
func TestSafeWSenderSendHTTPError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusUnauthorized)
_, _ = w.Write([]byte(`{"ok":false,"description":"Unauthorized"}`))
}))
defer srv.Close()
sender := &SafeWSender{apiBase: srv.URL}
cfg, _ := json.Marshal(map[string]string{"token": "bad", "chat_id": "1"})
err := sender.Send("t", "c", cfg)
if err == nil {
t.Fatal("expected error")
}
if !strings.Contains(err.Error(), "Unauthorized") {
t.Fatalf("error %q should contain description", err)
}
}
func TestSafeWSenderSendMissingFields(t *testing.T) {
hits := 0
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
hits++
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"ok":true}`))
}))
defer srv.Close()
sender := &SafeWSender{apiBase: srv.URL}
cases := []json.RawMessage{
json.RawMessage(`{"chat_id":"1"}`),
json.RawMessage(`{"token":"tok"}`),
json.RawMessage(`{"token":" ","chat_id":"1"}`),
json.RawMessage(`{"token":"tok","chat_id":""}`),
json.RawMessage(`{"token":"tok","chat_id":null}`),
}
for _, cfg := range cases {
if err := sender.Send("t", "c", cfg); err == nil {
t.Fatalf("expected error for config %s", cfg)
}
}
if hits != 0 {
t.Fatalf("unexpected HTTP calls: %d", hits)
}
if err := sender.Send("t", "c", json.RawMessage(`{`)); err == nil {
t.Fatal("expected error for invalid json")
}
if hits != 0 {
t.Fatalf("unexpected HTTP calls after invalid json: %d", hits)
}
}
+94
View File
@@ -0,0 +1,94 @@
package cache
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"time"
"aiaa-notification-service/internal/adapter"
"github.com/redis/go-redis/v9"
)
func TokenHash(token string) string {
sum := sha256.Sum256([]byte(token))
return hex.EncodeToString(sum[:])
}
func safewChatsKey(hash string) string { return "safew:chats:" + hash }
func safewOffsetKey(hash string) string { return "safew:offset:" + hash }
func safewPollKey(hash string) string { return "safew:poll:" + hash }
func (c *Cache) MergeSafewChats(ctx context.Context, token string, chats []adapter.SafewChat) error {
if c == nil || c.rdb == nil {
return fmt.Errorf("redis unavailable")
}
if len(chats) == 0 {
return nil
}
hash := TokenHash(token)
vals := make([]any, 0, len(chats)*2)
for _, ch := range chats {
b, err := json.Marshal(ch)
if err != nil {
return err
}
vals = append(vals, ch.ID, b)
}
return c.rdb.HSet(ctx, safewChatsKey(hash), vals...).Err()
}
func (c *Cache) ListSafewChats(ctx context.Context, token string) ([]adapter.SafewChat, error) {
if c == nil || c.rdb == nil {
return nil, fmt.Errorf("redis unavailable")
}
m, err := c.rdb.HGetAll(ctx, safewChatsKey(TokenHash(token))).Result()
if err != nil {
return nil, err
}
out := make([]adapter.SafewChat, 0, len(m))
for _, raw := range m {
var ch adapter.SafewChat
if err := json.Unmarshal([]byte(raw), &ch); err != nil {
return nil, err
}
out = append(out, ch)
}
return out, nil
}
func (c *Cache) GetSafewOffset(ctx context.Context, token string) (int64, error) {
if c == nil || c.rdb == nil {
return 0, fmt.Errorf("redis unavailable")
}
n, err := c.rdb.Get(ctx, safewOffsetKey(TokenHash(token))).Int64()
if err == redis.Nil {
return 0, nil
}
return n, err
}
func (c *Cache) SetSafewOffset(ctx context.Context, token string, offset int64) error {
if c == nil || c.rdb == nil {
return fmt.Errorf("redis unavailable")
}
return c.rdb.Set(ctx, safewOffsetKey(TokenHash(token)), offset, 0).Err()
}
func (c *Cache) TrySafewPollLock(ctx context.Context, token string, ttl time.Duration) (bool, error) {
if c == nil || c.rdb == nil {
return false, fmt.Errorf("redis unavailable")
}
ok, err := c.rdb.SetNX(ctx, safewPollKey(TokenHash(token)), "1", ttl).Result()
return ok, err
}
func (c *Cache) UnlockSafewPoll(ctx context.Context, token string) error {
if c == nil || c.rdb == nil {
return nil
}
return c.rdb.Del(ctx, safewPollKey(TokenHash(token))).Err()
}
+22
View File
@@ -0,0 +1,22 @@
package cache
import (
"encoding/hex"
"testing"
)
func TestTokenHashStableAndNotPlainToken(t *testing.T) {
h := TokenHash("secret-token")
if h == "secret-token" || h == "" {
t.Fatalf("hash=%q", h)
}
if _, err := hex.DecodeString(h); err != nil {
t.Fatalf("not hex: %v", err)
}
if TokenHash("secret-token") != h {
t.Fatal("not stable")
}
if TokenHash("other") == h {
t.Fatal("collision")
}
}
+85 -2
View File
@@ -2,11 +2,16 @@ package handler
import ( import (
"encoding/json" "encoding/json"
"fmt"
"log/slog"
"net/http" "net/http"
"strconv" "strconv"
"strings"
"aiaa-notification-service/internal/adapter"
"aiaa-notification-service/internal/cache" "aiaa-notification-service/internal/cache"
"aiaa-notification-service/internal/model" "aiaa-notification-service/internal/model"
"aiaa-notification-service/internal/safew"
"aiaa-notification-service/internal/store" "aiaa-notification-service/internal/store"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
@@ -15,10 +20,11 @@ import (
type ChannelHandler struct { type ChannelHandler struct {
store *store.Store store *store.Store
cache *cache.Cache cache *cache.Cache
chats *safew.Watcher
} }
func NewChannelHandler(s *store.Store, c *cache.Cache) *ChannelHandler { func NewChannelHandler(s *store.Store, c *cache.Cache, w *safew.Watcher) *ChannelHandler {
return &ChannelHandler{store: s, cache: c} return &ChannelHandler{store: s, cache: c, chats: w}
} }
type createChannelReq struct { type createChannelReq struct {
@@ -96,3 +102,80 @@ func (h *ChannelHandler) Delete(c *gin.Context) {
} }
c.JSON(http.StatusOK, gin.H{"ok": true}) c.JSON(http.StatusOK, gin.H{"ok": true})
} }
type listSafewChatsReq struct {
Token string `json:"token"`
Q string `json:"q"`
}
func (h *ChannelHandler) ListSafewChats(c *gin.Context) {
var req listSafewChatsReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
h.respondSafewChats(c, strings.TrimSpace(req.Token), req.Q)
}
func (h *ChannelHandler) ListChannelSafewChats(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
ch, err := h.store.GetChannel(c.Request.Context(), id)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"})
return
}
token, err := safewTokenFromChannel(ch)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
h.respondSafewChats(c, token, c.Query("q"))
}
func (h *ChannelHandler) respondSafewChats(c *gin.Context, token, q string) {
if token == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "safew token is required"})
return
}
if h.chats == nil {
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "safew chat list unavailable"})
return
}
h.chats.Ensure(token)
if err := h.chats.Refresh(c.Request.Context(), token); err != nil {
if _, ok := err.(*adapter.SafewAuthError); ok {
c.JSON(http.StatusUnauthorized, gin.H{"error": err.Error()})
return
}
slog.Warn("safew refresh", "error", err)
}
list, err := h.chats.List(c.Request.Context(), token, q)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
if list == nil {
list = []adapter.SafewChat{}
}
c.JSON(http.StatusOK, gin.H{"data": list, "total": len(list)})
}
func safewTokenFromChannel(ch *model.Channel) (string, error) {
if ch.Type != "safew" {
return "", fmt.Errorf("channel is not safew")
}
if ch.Config == nil {
return "", fmt.Errorf("safew token is required")
}
var cfg struct {
Token string `json:"token"`
}
if err := json.Unmarshal(*ch.Config, &cfg); err != nil {
return "", fmt.Errorf("safew token is required")
}
token := strings.TrimSpace(cfg.Token)
if token == "" {
return "", fmt.Errorf("safew token is required")
}
return token, nil
}
+72
View File
@@ -0,0 +1,72 @@
package handler
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"aiaa-notification-service/internal/adapter"
"aiaa-notification-service/internal/model"
"aiaa-notification-service/internal/safew"
"github.com/gin-gonic/gin"
)
func TestListSafewChatsPOSTMissingToken(t *testing.T) {
gin.SetMode(gin.TestMode)
h := &ChannelHandler{chats: safew.NewWatcher(safew.NewMemStore(), func(string, int64, int) ([]adapter.SafewChat, int64, error) {
return nil, 0, nil
})}
r := gin.New()
r.POST("/api/v1/channels/safew/chats", h.ListSafewChats)
req := httptest.NewRequest(http.MethodPost, "/api/v1/channels/safew/chats", bytes.NewReader([]byte(`{}`)))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusBadRequest {
t.Fatalf("code=%d body=%s", w.Code, w.Body.String())
}
}
func TestListSafewChatsPOSTOk(t *testing.T) {
gin.SetMode(gin.TestMode)
st := safew.NewMemStore()
_ = st.MergeSafewChats(nil, "tok", []adapter.SafewChat{{ID: "10000778141", Type: "group", Title: "测试AI"}})
h := &ChannelHandler{chats: safew.NewWatcher(st, func(string, int64, int) ([]adapter.SafewChat, int64, error) {
return nil, 0, nil
})}
r := gin.New()
r.POST("/api/v1/channels/safew/chats", h.ListSafewChats)
req := httptest.NewRequest(http.MethodPost, "/api/v1/channels/safew/chats", bytes.NewReader([]byte(`{"token":"tok","q":"测试"}`)))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("code=%d body=%s", w.Code, w.Body.String())
}
var resp struct {
Data []adapter.SafewChat `json:"data"`
Total int `json:"total"`
}
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if resp.Total != 1 || resp.Data[0].ID != "10000778141" {
t.Fatalf("%#v", resp)
}
}
func TestSafewTokenFromChannel(t *testing.T) {
raw := json.RawMessage(`{"token":"abc","chat_id":"1"}`)
ch := &model.Channel{Type: "safew", Config: &raw}
tok, err := safewTokenFromChannel(ch)
if err != nil || tok != "abc" {
t.Fatalf("%q %v", tok, err)
}
ch.Type = "bark"
if _, err := safewTokenFromChannel(ch); err == nil {
t.Fatal("expected not safew")
}
}
+88
View File
@@ -0,0 +1,88 @@
package safew
import (
"context"
"sync"
"time"
"aiaa-notification-service/internal/adapter"
)
type ChatStore interface {
MergeSafewChats(ctx context.Context, token string, chats []adapter.SafewChat) error
ListSafewChats(ctx context.Context, token string) ([]adapter.SafewChat, error)
GetSafewOffset(ctx context.Context, token string) (int64, error)
SetSafewOffset(ctx context.Context, token string, offset int64) error
TrySafewPollLock(ctx context.Context, token string, ttl time.Duration) (bool, error)
UnlockSafewPoll(ctx context.Context, token string) error
}
type MemStore struct {
mu sync.Mutex
chats map[string]map[string]adapter.SafewChat
offset map[string]int64
locks map[string]bool
}
func NewMemStore() *MemStore {
return &MemStore{
chats: map[string]map[string]adapter.SafewChat{},
offset: map[string]int64{},
locks: map[string]bool{},
}
}
func (m *MemStore) MergeSafewChats(_ context.Context, token string, chats []adapter.SafewChat) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.chats[token] == nil {
m.chats[token] = map[string]adapter.SafewChat{}
}
for _, c := range chats {
m.chats[token][c.ID] = c
}
return nil
}
func (m *MemStore) ListSafewChats(_ context.Context, token string) ([]adapter.SafewChat, error) {
m.mu.Lock()
defer m.mu.Unlock()
var out []adapter.SafewChat
for _, c := range m.chats[token] {
out = append(out, c)
}
if out == nil {
out = []adapter.SafewChat{}
}
return out, nil
}
func (m *MemStore) GetSafewOffset(_ context.Context, token string) (int64, error) {
m.mu.Lock()
defer m.mu.Unlock()
return m.offset[token], nil
}
func (m *MemStore) SetSafewOffset(_ context.Context, token string, offset int64) error {
m.mu.Lock()
defer m.mu.Unlock()
m.offset[token] = offset
return nil
}
func (m *MemStore) TrySafewPollLock(_ context.Context, token string, _ time.Duration) (bool, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.locks[token] {
return false, nil
}
m.locks[token] = true
return true, nil
}
func (m *MemStore) UnlockSafewPoll(_ context.Context, token string) error {
m.mu.Lock()
defer m.mu.Unlock()
delete(m.locks, token)
return nil
}
+117
View File
@@ -0,0 +1,117 @@
package safew
import (
"context"
"log/slog"
"sync"
"time"
"aiaa-notification-service/internal/adapter"
)
type Poller func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error)
type Watcher struct {
store ChatStore
poll Poller
bgIdle time.Duration
mu sync.Mutex
running map[string]context.CancelFunc
stopped bool
}
func NewWatcher(store ChatStore, poll Poller) *Watcher {
return &Watcher{
store: store,
poll: poll,
bgIdle: time.Second,
running: map[string]context.CancelFunc{},
}
}
func (w *Watcher) Ensure(token string) {
if token == "" || w == nil {
return
}
w.mu.Lock()
defer w.mu.Unlock()
if w.stopped {
return
}
if _, ok := w.running[token]; ok {
return
}
ctx, cancel := context.WithCancel(context.Background())
w.running[token] = cancel
go w.loop(ctx, token)
}
func (w *Watcher) Stop() {
w.mu.Lock()
w.stopped = true
for _, cancel := range w.running {
cancel()
}
w.running = map[string]context.CancelFunc{}
w.mu.Unlock()
}
func (w *Watcher) Refresh(ctx context.Context, token string) error {
return w.pollOnce(ctx, token, 0)
}
func (w *Watcher) List(ctx context.Context, token, q string) ([]adapter.SafewChat, error) {
chats, err := w.store.ListSafewChats(ctx, token)
if err != nil {
return nil, err
}
return adapter.FilterSafewChats(chats, q), nil
}
func (w *Watcher) loop(ctx context.Context, token string) {
for {
if err := w.pollOnce(ctx, token, 30); err != nil {
if ctx.Err() != nil {
return
}
slog.Warn("safew watcher poll", "error", err)
}
idle := w.bgIdle
if idle <= 0 {
idle = time.Second
}
select {
case <-ctx.Done():
return
case <-time.After(idle):
}
}
}
func (w *Watcher) pollOnce(ctx context.Context, token string, timeout int) error {
ok, err := w.store.TrySafewPollLock(ctx, token, 35*time.Second)
if err != nil {
return err
}
if !ok {
return nil
}
defer func() { _ = w.store.UnlockSafewPoll(ctx, token) }()
offset, err := w.store.GetSafewOffset(ctx, token)
if err != nil {
return err
}
chats, next, err := w.poll(token, offset, timeout)
if err != nil {
return err
}
if err := w.store.MergeSafewChats(ctx, token, chats); err != nil {
return err
}
if next != offset {
return w.store.SetSafewOffset(ctx, token, next)
}
return nil
}
+76
View File
@@ -0,0 +1,76 @@
package safew
import (
"context"
"errors"
"testing"
"time"
"aiaa-notification-service/internal/adapter"
)
func TestRefreshMergesAndAdvancesOffset(t *testing.T) {
st := NewMemStore()
poll := func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error) {
if timeout != 0 {
t.Fatalf("timeout=%d", timeout)
}
if offset != 0 {
t.Fatalf("offset=%d", offset)
}
return []adapter.SafewChat{{ID: "10000778141", Type: "group", Title: "测试AI"}}, 11, nil
}
w := NewWatcher(st, poll)
ctx := context.Background()
if err := w.Refresh(ctx, "tok"); err != nil {
t.Fatal(err)
}
list, err := w.List(ctx, "tok", "测试")
if err != nil {
t.Fatal(err)
}
if len(list) != 1 || list[0].ID != "10000778141" {
t.Fatalf("%#v", list)
}
off, _ := st.GetSafewOffset(ctx, "tok")
if off != 11 {
t.Fatalf("offset=%d", off)
}
}
func TestRefreshAuthError(t *testing.T) {
st := NewMemStore()
w := NewWatcher(st, func(string, int64, int) ([]adapter.SafewChat, int64, error) {
return nil, 0, &adapter.SafewAuthError{Description: "BOT_TOKEN_INVALID"}
})
err := w.Refresh(context.Background(), "bad")
if err == nil {
t.Fatal("expected auth error")
}
if _, ok := err.(*adapter.SafewAuthError); !ok {
t.Fatalf("%T", err)
}
}
func TestEnsurePollsInBackground(t *testing.T) {
st := NewMemStore()
got := make(chan int, 1)
w := NewWatcher(st, func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error) {
if timeout != 30 {
return nil, offset, errors.New("not background")
}
select {
case got <- timeout:
default:
}
return []adapter.SafewChat{{ID: "1", Type: "group", Title: "g"}}, offset + 1, nil
})
w.bgIdle = 10 * time.Millisecond
w.Ensure("tok")
select {
case <-got:
case <-time.After(2 * time.Second):
t.Fatal("background poll not called")
}
w.Stop()
}