Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| bd8a9ff96d | |||
| 0d0cd0c510 |
@@ -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
@@ -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)
|
||||||
|
|||||||
@@ -260,6 +260,39 @@ curl -X POST 'http://localhost:8080/api/v1/channels' \
|
|||||||
}'
|
}'
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### 创建 Channel(SafeW)
|
||||||
|
|
||||||
|
```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
|
||||||
|
### 创建 Channel(SafeW)
|
||||||
|
|
||||||
|
```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 返回一批 update,offset 前进。
|
||||||
|
4. **handler**:非 safew channel → 400;缺 token → 400。
|
||||||
|
|
||||||
|
Watcher 用可注入的 poll 函数 + 假 cache,不断言真实 30s 阻塞。
|
||||||
|
|
||||||
|
## Success criteria
|
||||||
|
|
||||||
|
- 两个接口返回已监控群,`id` 为字符串
|
||||||
|
- 默认只有群;`q` 可过滤
|
||||||
|
- 后台 `timeout=30` 持续写入 Redis;列表接口不卡 30 秒
|
||||||
|
- 仓库与 Redis key 中不出现明文 token
|
||||||
@@ -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:
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
Vendored
+94
@@ -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()
|
||||||
|
}
|
||||||
Vendored
+22
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
}
|
||||||
|
|||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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()
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user