Files

T

ryan bd8a9ff96d feat: 增加 SafeW 已监控群列表接口

通过 getUpdates 将群写入 Redis，供创建/编辑渠道时选择 chat_id，避免前端重复传递 token。

2026-08-15 00:40:40 +08:00

31 KiB

Raw Permalink Blame History

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:

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.

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:

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:

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:

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:

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:

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

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:

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:

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:

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):

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:

	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)

	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:

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"}'

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

31 KiB Raw Permalink Blame History Unescape Escape

SafeW Chat List Implementation Plan

Global Constraints

File Structure

Task 1: Parse updates and filter chats

Task 2: PollGroupChats (httptest)

Task 3: Redis chat store + ChatStore interface

Task 4: Watcher + Refresh + List

Task 5: HTTP handlers, routes, docs

31 KiB

Raw Permalink Blame History