diff --git a/README.md b/README.md index 60e1977..6327840 100644 --- a/README.md +++ b/README.md @@ -409,6 +409,41 @@ SMTP 使用全局 `config.yaml` 的 `smtp` 段;支持 587 STARTTLS / 465 TLS `chat_id` 可为数字或字符串(含 `@username`)。消息以 MarkdownV2 发送:标题加粗,标题和正文均自动转义。 +#### `POST /api/v1/channels/safew/chats` — 列出已监控的 SafeW 群 + +创建渠道时选群。Admin Key。用 `getUpdates` 攒到 Redis 里的群;bot 尚未在群里收到过更新时列表为空。 + +```json +{ + "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` — 列表 Query:`page`、`page_size`。**200:** `{ "data": Channel[], "total", "page" }` diff --git a/cmd/server/main.go b/cmd/server/main.go index 2fa45c4..ec72faa 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -15,6 +15,7 @@ import ( "aiaa-notification-service/internal/config" "aiaa-notification-service/internal/engine" "aiaa-notification-service/internal/handler" + "aiaa-notification-service/internal/safew" "aiaa-notification-service/internal/store" "github.com/gin-gonic/gin" @@ -125,7 +126,21 @@ func main() { notifyH := handler.NewNotifyHandler(st, redisCache, matcher, renderer, router) sourceH := handler.NewSourceHandler(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) msgLogH := handler.NewMessageLogHandler(st) @@ -168,6 +183,8 @@ func main() { admin.DELETE("/templates/:id", templateH.Delete) // Channels + admin.POST("/channels/safew/chats", channelH.ListSafewChats) + admin.GET("/channels/:id/chats", channelH.ListChannelSafewChats) admin.POST("/channels", channelH.Create) admin.GET("/channels", channelH.List) admin.GET("/channels/:id", channelH.Get) diff --git a/docs/httpie/curls.md b/docs/httpie/curls.md index 2fc0518..8da680f 100644 --- a/docs/httpie/curls.md +++ b/docs/httpie/curls.md @@ -277,6 +277,22 @@ curl -X POST 'http://localhost:8080/api/v1/channels' \ }' ``` +### 列出 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":"","q":"测试AI"}' +``` + +### 列出 SafeW 群(已有渠道) + +```bash +curl -X GET 'http://localhost:8080/api/v1/channels/1/chats?q=' \ + -H 'Authorization: Bearer admin-sk-change-me' +``` + ### 列出 Channel ```bash diff --git a/docs/superpowers/plans/2026-08-15-safew-chats-list.md b/docs/superpowers/plans/2026-08-15-safew-chats-list.md new file mode 100644 index 0000000..71b0871 --- /dev/null +++ b/docs/superpowers/plans/2026-08-15-safew-chats-list.md @@ -0,0 +1,1191 @@ +# SafeW Chat List 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 two admin APIs that return SafeW groups already seen via `getUpdates`, persisted in Redis, with a per-token background long-poller (`timeout=30`). + +**Architecture:** Parse group chats from `getUpdates`. Redis Hash keyed by `sha256(token)` stores chats and offset. List handlers drain with `timeout=0` then return Redis. A lazy in-process Watcher long-polls `timeout=30`. Same token is serialized with a Redis lock. + +**Tech Stack:** Go 1.22+, gin, go-redis, stdlib `net/http/httptest`. No new dependencies. No real bot tokens in repo. + +## Global Constraints + +- Endpoints: `POST /api/v1/channels/safew/chats`, `GET /api/v1/channels/:id/chats?q=` +- Admin Bearer auth (existing middleware) +- `id` is always a JSON string; never decode chat id as `float64` +- Default types: `group` and `supergroup` only +- Redis keys: `safew:chats:{sha256hex}`, `safew:offset:{sha256hex}`, `safew:poll:{sha256hex}` — token never in the key +- List path uses `getUpdates` `timeout=0`; Watcher uses `timeout=30` +- Empty list is `200` `{"data":[],"total":0}` +- Invalid token: HTTP `401` with SafeW `description` +- Do not commit unless the user explicitly asks +- Do not write the user's bot token into any file + +--- + +## File Structure + +| File | Responsibility | +|------|----------------| +| `internal/adapter/safew.go` | `SafewChat`, parse updates, `PollGroupChats`, `FilterSafewChats`, `SafewAuthError` | +| `internal/adapter/safew_chats_test.go` | Parse / filter / poll unit tests | +| `internal/cache/safew_chats.go` | Token hash, Redis merge/list/offset/lock | +| `internal/safew/store.go` | `ChatStore` interface + memory fake for tests | +| `internal/safew/watcher.go` | Per-token goroutine, `Ensure` / `Stop` / `Refresh` / `List` | +| `internal/safew/watcher_test.go` | Watcher + Refresh with fake store + fake poller | +| `internal/handler/channel.go` | Two HTTP handlers | +| `internal/handler/channel_chats_test.go` | POST missing token / fake list; GET type/token checks | +| `cmd/server/main.go` | Routes + Watcher wiring + shutdown Stop | +| `README.md` / `docs/httpie/curls.md` | API docs | + +--- + +### Task 1: Parse updates and filter chats + +**Files:** +- Modify: `internal/adapter/safew.go` +- Create: `internal/adapter/safew_chats_test.go` + +**Interfaces:** +- Produces: + - `type SafewChat struct { ID string; Type string; Title string; Username *string }` with JSON tags `id,type,title,username` + - `func GroupsFromUpdates(raw []byte) ([]SafewChat, int64, error)` — `raw` is the full getUpdates JSON body (`{"ok":true,"result":[...]}`); returns unique groups and max `update_id` (0 if none) + - `func FilterSafewChats(chats []SafewChat, q string) []SafewChat` + +- [ ] **Step 1: Write the failing tests** + +Create `internal/adapter/safew_chats_test.go`: + +```go +package adapter + +import ( + "encoding/json" + "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) + if !json.Valid(raw) || string(raw)[0] != '{' { + t.Fatal(string(raw)) + } + if got := string(mustJSONField(t, raw, "id")); got != `"10000778141"` && false { + t.Fatal(got) + } + 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 mustJSONField(t *testing.T, raw []byte, key string) string { + t.Helper() + var m map[string]json.RawMessage + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + return string(m[key]) +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `go test ./internal/adapter/ -run 'TestGroupsFromUpdates|TestFilterSafewChats' -count=1` + +Expected: FAIL `undefined: GroupsFromUpdates` + +- [ ] **Step 3: Implement parse + filter** + +Append to `internal/adapter/safew.go` (keep existing send code). Add imports: `strconv`. + +```go +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 +} +``` + +Unused `strconv` — do **not** add that import. + +- [ ] **Step 4: Run tests and make sure they pass** + +Run: `go test ./internal/adapter/ -count=1` + +Expected: `PASS` (existing send tests + new tests) + +--- + +### Task 2: PollGroupChats (httptest) + +**Files:** +- Modify: `internal/adapter/safew.go` +- Modify: `internal/adapter/safew_chats_test.go` + +**Interfaces:** +- Consumes: `GroupsFromUpdates` +- Produces: + - `type SafewAuthError struct { Description string }` with `Error() string` + - `func (s *SafeWSender) PollGroupChats(token string, offset int64, timeout int) (chats []SafewChat, nextOffset int64, err error)` + - `nextOffset` is `max(update_id)+1` when any updates; otherwise unchanged `offset` + +- [ ] **Step 1: Write the failing poll tests** + +Append to `internal/adapter/safew_chats_test.go`: + +```go +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) + } +} +``` + +Add imports in that file: `io`, `net/http`, `net/http/httptest`, `strings`. + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `go test ./internal/adapter/ -run TestPollGroupChats -count=1` + +Expected: FAIL `undefined: PollGroupChats` + +- [ ] **Step 3: Implement poll** + +In `internal/adapter/safew.go`, change `endpoint` to go through `methodURL`: + +```go +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 +} +``` + +Need `net/http` import in `safew.go` (already present). + +- [ ] **Step 4: Run adapter tests** + +Run: `go test ./internal/adapter/ -count=1` + +Expected: `PASS` + +--- + +### Task 3: Redis chat store + ChatStore interface + +**Files:** +- Create: `internal/cache/safew_chats.go` +- Create: `internal/cache/safew_chats_test.go` (hash only) +- Create: `internal/safew/store.go` + +**Interfaces:** +- Produces: + - `func TokenHash(token string) string` in package `cache` — sha256 hex of token + - `cache.Cache` methods: `MergeSafewChats`, `ListSafewChats`, `GetSafewOffset`, `SetSafewOffset`, `TrySafewPollLock`, `UnlockSafewPoll` + - `type safew.ChatStore interface` wrapping those operations by **token string** (hash inside Redis impl) + +- [ ] **Step 1: Write hash test** + +`internal/cache/safew_chats_test.go`: + +```go +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") + } +} +``` + +- [ ] **Step 2: Run to verify fail** + +Run: `go test ./internal/cache/ -run TestTokenHash -count=1` + +Expected: FAIL `undefined: TokenHash` + +- [ ] **Step 3: Implement hash + Redis + interface** + +`internal/cache/safew_chats.go`: + +```go +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() +} +``` + +`internal/safew/store.go`: + +```go +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 +} +``` + +MemStore keys by raw token (tests only). Redis impl hashes. That is OK: production always uses `cache.Cache`. + +- [ ] **Step 4: Run tests** + +Run: `go test ./internal/cache/ ./internal/safew/ -count=1` + +Expected: cache hash test PASS; safew package may have no tests yet (`?` or PASS) + +--- + +### Task 4: Watcher + Refresh + List + +**Files:** +- Create: `internal/safew/watcher.go` +- Create: `internal/safew/watcher_test.go` + +**Interfaces:** +- Consumes: `ChatStore`, `PollGroupChats` via `type Poller func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error)` +- Produces: + - `type Watcher struct` + - `func NewWatcher(store ChatStore, poll Poller) *Watcher` + - `func (w *Watcher) Ensure(token string)` + - `func (w *Watcher) Stop()` + - `func (w *Watcher) Refresh(ctx context.Context, token string) error` — timeout=0 drain + - `func (w *Watcher) List(ctx context.Context, token, q string) ([]adapter.SafewChat, error)` + +- [ ] **Step 1: Write failing watcher tests** + +```go +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() +} +``` + +- [ ] **Step 2: Run to verify fail** + +Run: `go test ./internal/safew/ -count=1` + +Expected: FAIL `undefined: NewWatcher` + +- [ ] **Step 3: Implement Watcher** + +`internal/safew/watcher.go`: + +```go +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 +} +``` + +- [ ] **Step 4: Run safew tests** + +Run: `go test ./internal/safew/ -count=1` + +Expected: `PASS` + +If `TestEnsurePollsInBackground` is flaky, increase wait or call `loop` once via exported test hook — do not lengthen beyond 2s. + +--- + +### Task 5: HTTP handlers, routes, docs + +**Files:** +- Modify: `internal/handler/channel.go` +- Create: `internal/handler/channel_chats_test.go` +- Modify: `cmd/server/main.go` +- Modify: `README.md` +- Modify: `docs/httpie/curls.md` + +**Interfaces:** +- Consumes: `*safew.Watcher` (`Ensure`, `Refresh`, `List`) +- Produces: the two admin endpoints + +- [ ] **Step 1: Write handler tests** + +`internal/handler/channel_chats_test.go`: + +```go +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") + } +} +``` + +`MergeSafewChats(nil, ...)` — MemStore ignores ctx, OK. If compile requires context, use `context.Background()`. + +- [ ] **Step 2: Run to verify fail** + +Run: `go test ./internal/handler/ -count=1` + +Expected: FAIL `undefined: ListSafewChats` or `unknown field chats` + +- [ ] **Step 3: Implement handlers and wire main** + +Update `ChannelHandler`: + +```go +type ChannelHandler struct { + store *store.Store + cache *cache.Cache + chats *safew.Watcher +} + +func NewChannelHandler(s *store.Store, c *cache.Cache, w *safew.Watcher) *ChannelHandler { + return &ChannelHandler{store: s, cache: c, chats: w} +} +``` + +Add handlers (same file): + +```go +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 +} +``` + +Add imports in `channel.go`: `fmt`, `log/slog`, `strings`, `adapter`, `safew`. + +**Every** existing `NewChannelHandler(st, redisCache)` call must become `NewChannelHandler(st, redisCache, watcher)`. In `cmd/server/main.go`: + +After Redis is connected: + +```go + 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) +``` + +Register routes **before** `GET /channels/:id`: + +```go + admin.POST("/channels/safew/chats", channelH.ListSafewChats) + admin.GET("/channels/:id/chats", channelH.ListChannelSafewChats) + admin.POST("/channels", channelH.Create) + admin.GET("/channels", channelH.List) + admin.GET("/channels/:id", channelH.Get) +``` + +On shutdown, `safewWatcher.Stop()` via defer is enough. + +Grep for `NewChannelHandler` and update all call sites. + +README: after SafeW config block, before `#### GET /api/v1/channels`, insert the two endpoints with the JSON examples from the spec. + +`docs/httpie/curls.md`: after 创建 Channel(SafeW), add: + +```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":"","q":"测试AI"}' + +curl -X GET 'http://localhost:8080/api/v1/channels/1/chats?q=' \ + -H 'Authorization: Bearer admin-sk-change-me' +``` + +- [ ] **Step 4: Run all internal tests** + +Run: `go test ./internal/... -count=1` + +Expected: `PASS` + +Also: `go build ./cmd/server` + +Expected: success diff --git a/docs/superpowers/specs/2026-08-15-safew-chats-list-design.md b/docs/superpowers/specs/2026-08-15-safew-chats-list-design.md new file mode 100644 index 0000000..5319e52 --- /dev/null +++ b/docs/superpowers/specs/2026-08-15-safew-chats-list-design.md @@ -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": "", + "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":""}`。 + +## 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":,"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 diff --git a/internal/adapter/safew.go b/internal/adapter/safew.go index c0d3416..4661b21 100644 --- a/internal/adapter/safew.go +++ b/internal/adapter/safew.go @@ -90,11 +90,73 @@ func (s *SafeWSender) Send(title, content string, config json.RawMessage) error } 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 + "/sendMessage" + 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) { @@ -136,3 +198,121 @@ func escapeMarkdownV2(s string) string { } 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 +} diff --git a/internal/adapter/safew_chats_test.go b/internal/adapter/safew_chats_test.go new file mode 100644 index 0000000..84e60ad --- /dev/null +++ b/internal/adapter/safew_chats_test.go @@ -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) + } +} diff --git a/internal/cache/safew_chats.go b/internal/cache/safew_chats.go new file mode 100644 index 0000000..de49e90 --- /dev/null +++ b/internal/cache/safew_chats.go @@ -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() +} diff --git a/internal/cache/safew_chats_test.go b/internal/cache/safew_chats_test.go new file mode 100644 index 0000000..d4114c3 --- /dev/null +++ b/internal/cache/safew_chats_test.go @@ -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") + } +} diff --git a/internal/handler/channel.go b/internal/handler/channel.go index 3b26c56..230f488 100644 --- a/internal/handler/channel.go +++ b/internal/handler/channel.go @@ -2,11 +2,16 @@ package handler import ( "encoding/json" + "fmt" + "log/slog" "net/http" "strconv" + "strings" + "aiaa-notification-service/internal/adapter" "aiaa-notification-service/internal/cache" "aiaa-notification-service/internal/model" + "aiaa-notification-service/internal/safew" "aiaa-notification-service/internal/store" "github.com/gin-gonic/gin" @@ -15,10 +20,11 @@ import ( type ChannelHandler struct { store *store.Store cache *cache.Cache + chats *safew.Watcher } -func NewChannelHandler(s *store.Store, c *cache.Cache) *ChannelHandler { - return &ChannelHandler{store: s, cache: c} +func NewChannelHandler(s *store.Store, c *cache.Cache, w *safew.Watcher) *ChannelHandler { + return &ChannelHandler{store: s, cache: c, chats: w} } type createChannelReq struct { @@ -96,3 +102,80 @@ func (h *ChannelHandler) Delete(c *gin.Context) { } 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 +} diff --git a/internal/handler/channel_chats_test.go b/internal/handler/channel_chats_test.go new file mode 100644 index 0000000..14b91bf --- /dev/null +++ b/internal/handler/channel_chats_test.go @@ -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") + } +} diff --git a/internal/safew/store.go b/internal/safew/store.go new file mode 100644 index 0000000..a5205fb --- /dev/null +++ b/internal/safew/store.go @@ -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 +} diff --git a/internal/safew/watcher.go b/internal/safew/watcher.go new file mode 100644 index 0000000..237ec04 --- /dev/null +++ b/internal/safew/watcher.go @@ -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 +} diff --git a/internal/safew/watcher_test.go b/internal/safew/watcher_test.go new file mode 100644 index 0000000..37fc8c1 --- /dev/null +++ b/internal/safew/watcher_test.go @@ -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() +}