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

通过 getUpdates 将群写入 Redis，供创建/编辑渠道时选择 chat_id，避免前端重复传递 token。
2026-08-15 00:40:40 +08:00
parent 0d0cd0c510
commit bd8a9ff96d
14 changed files with 2265 additions and 4 deletions
@@ -0,0 +1,88 @@
+package safew
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"aiaa-notification-service/internal/adapter"
+)
+
+type ChatStore interface {
+	MergeSafewChats(ctx context.Context, token string, chats []adapter.SafewChat) error
+	ListSafewChats(ctx context.Context, token string) ([]adapter.SafewChat, error)
+	GetSafewOffset(ctx context.Context, token string) (int64, error)
+	SetSafewOffset(ctx context.Context, token string, offset int64) error
+	TrySafewPollLock(ctx context.Context, token string, ttl time.Duration) (bool, error)
+	UnlockSafewPoll(ctx context.Context, token string) error
+}
+
+type MemStore struct {
+	mu     sync.Mutex
+	chats  map[string]map[string]adapter.SafewChat
+	offset map[string]int64
+	locks  map[string]bool
+}
+
+func NewMemStore() *MemStore {
+	return &MemStore{
+		chats:  map[string]map[string]adapter.SafewChat{},
+		offset: map[string]int64{},
+		locks:  map[string]bool{},
+	}
+}
+
+func (m *MemStore) MergeSafewChats(_ context.Context, token string, chats []adapter.SafewChat) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.chats[token] == nil {
+		m.chats[token] = map[string]adapter.SafewChat{}
+	}
+	for _, c := range chats {
+		m.chats[token][c.ID] = c
+	}
+	return nil
+}
+
+func (m *MemStore) ListSafewChats(_ context.Context, token string) ([]adapter.SafewChat, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var out []adapter.SafewChat
+	for _, c := range m.chats[token] {
+		out = append(out, c)
+	}
+	if out == nil {
+		out = []adapter.SafewChat{}
+	}
+	return out, nil
+}
+
+func (m *MemStore) GetSafewOffset(_ context.Context, token string) (int64, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.offset[token], nil
+}
+
+func (m *MemStore) SetSafewOffset(_ context.Context, token string, offset int64) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.offset[token] = offset
+	return nil
+}
+
+func (m *MemStore) TrySafewPollLock(_ context.Context, token string, _ time.Duration) (bool, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.locks[token] {
+		return false, nil
+	}
+	m.locks[token] = true
+	return true, nil
+}
+
+func (m *MemStore) UnlockSafewPoll(_ context.Context, token string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	delete(m.locks, token)
+	return nil
+}
@@ -0,0 +1,117 @@
+package safew
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+
+	"aiaa-notification-service/internal/adapter"
+)
+
+type Poller func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error)
+
+type Watcher struct {
+	store  ChatStore
+	poll   Poller
+	bgIdle time.Duration
+
+	mu      sync.Mutex
+	running map[string]context.CancelFunc
+	stopped bool
+}
+
+func NewWatcher(store ChatStore, poll Poller) *Watcher {
+	return &Watcher{
+		store:   store,
+		poll:    poll,
+		bgIdle:  time.Second,
+		running: map[string]context.CancelFunc{},
+	}
+}
+
+func (w *Watcher) Ensure(token string) {
+	if token == "" || w == nil {
+		return
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.stopped {
+		return
+	}
+	if _, ok := w.running[token]; ok {
+		return
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	w.running[token] = cancel
+	go w.loop(ctx, token)
+}
+
+func (w *Watcher) Stop() {
+	w.mu.Lock()
+	w.stopped = true
+	for _, cancel := range w.running {
+		cancel()
+	}
+	w.running = map[string]context.CancelFunc{}
+	w.mu.Unlock()
+}
+
+func (w *Watcher) Refresh(ctx context.Context, token string) error {
+	return w.pollOnce(ctx, token, 0)
+}
+
+func (w *Watcher) List(ctx context.Context, token, q string) ([]adapter.SafewChat, error) {
+	chats, err := w.store.ListSafewChats(ctx, token)
+	if err != nil {
+		return nil, err
+	}
+	return adapter.FilterSafewChats(chats, q), nil
+}
+
+func (w *Watcher) loop(ctx context.Context, token string) {
+	for {
+		if err := w.pollOnce(ctx, token, 30); err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			slog.Warn("safew watcher poll", "error", err)
+		}
+		idle := w.bgIdle
+		if idle <= 0 {
+			idle = time.Second
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(idle):
+		}
+	}
+}
+
+func (w *Watcher) pollOnce(ctx context.Context, token string, timeout int) error {
+	ok, err := w.store.TrySafewPollLock(ctx, token, 35*time.Second)
+	if err != nil {
+		return err
+	}
+	if !ok {
+		return nil
+	}
+	defer func() { _ = w.store.UnlockSafewPoll(ctx, token) }()
+
+	offset, err := w.store.GetSafewOffset(ctx, token)
+	if err != nil {
+		return err
+	}
+	chats, next, err := w.poll(token, offset, timeout)
+	if err != nil {
+		return err
+	}
+	if err := w.store.MergeSafewChats(ctx, token, chats); err != nil {
+		return err
+	}
+	if next != offset {
+		return w.store.SetSafewOffset(ctx, token, next)
+	}
+	return nil
+}
@@ -0,0 +1,76 @@
+package safew
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"aiaa-notification-service/internal/adapter"
+)
+
+func TestRefreshMergesAndAdvancesOffset(t *testing.T) {
+	st := NewMemStore()
+	poll := func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error) {
+		if timeout != 0 {
+			t.Fatalf("timeout=%d", timeout)
+		}
+		if offset != 0 {
+			t.Fatalf("offset=%d", offset)
+		}
+		return []adapter.SafewChat{{ID: "10000778141", Type: "group", Title: "测试AI"}}, 11, nil
+	}
+	w := NewWatcher(st, poll)
+	ctx := context.Background()
+	if err := w.Refresh(ctx, "tok"); err != nil {
+		t.Fatal(err)
+	}
+	list, err := w.List(ctx, "tok", "测试")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(list) != 1 || list[0].ID != "10000778141" {
+		t.Fatalf("%#v", list)
+	}
+	off, _ := st.GetSafewOffset(ctx, "tok")
+	if off != 11 {
+		t.Fatalf("offset=%d", off)
+	}
+}
+
+func TestRefreshAuthError(t *testing.T) {
+	st := NewMemStore()
+	w := NewWatcher(st, func(string, int64, int) ([]adapter.SafewChat, int64, error) {
+		return nil, 0, &adapter.SafewAuthError{Description: "BOT_TOKEN_INVALID"}
+	})
+	err := w.Refresh(context.Background(), "bad")
+	if err == nil {
+		t.Fatal("expected auth error")
+	}
+	if _, ok := err.(*adapter.SafewAuthError); !ok {
+		t.Fatalf("%T", err)
+	}
+}
+
+func TestEnsurePollsInBackground(t *testing.T) {
+	st := NewMemStore()
+	got := make(chan int, 1)
+	w := NewWatcher(st, func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error) {
+		if timeout != 30 {
+			return nil, offset, errors.New("not background")
+		}
+		select {
+		case got <- timeout:
+		default:
+		}
+		return []adapter.SafewChat{{ID: "1", Type: "group", Title: "g"}}, offset + 1, nil
+	})
+	w.bgIdle = 10 * time.Millisecond
+	w.Ensure("tok")
+	select {
+	case <-got:
+	case <-time.After(2 * time.Second):
+		t.Fatal("background poll not called")
+	}
+	w.Stop()
+}