feat: enhance Safew error handling and polling mechanism

- Added error_msg field to safewAPIResponse for better error descriptions. - Updated safewErrorDescription function to include error_msg in the output. - Introduced a new test to verify that error messages are included in Safew error descriptions. - Improved the polling mechanism in the Watcher to handle conflicts and ensure no overlapping polls occur.
2026-08-15 01:07:01 +08:00
parent bd8a9ff96d
commit e369f398d8
4 changed files with 201 additions and 28 deletions
@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"strings"
+	"time"
 )

 const safewAPIBase = "https://api.safew.bot"
@@ -29,6 +30,7 @@ type safewMessage struct {
 type safewAPIResponse struct {
 	OK          bool   `json:"ok"`
 	Description string `json:"description"`
+	ErrorMsg    string `json:"error_msg"`
 }

 func (s *SafeWSender) Type() string { return "safew" }
@@ -112,17 +114,22 @@ func (e *SafewAuthError) Error() string {
 	return e.Description
 }

+var safewLongPollClient = &http.Client{Timeout: 45 * time.Second}
+
 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{
+	payload := map[string]any{
 		"timeout": timeout,
-		"offset":  offset,
 		"limit":   100,
-	})
-	resp, err := http.Post(s.methodURL(token, "getUpdates"), "application/json", bytes.NewReader(reqBody))
+	}
+	if offset > 0 {
+		payload["offset"] = offset
+	}
+	reqBody, _ := json.Marshal(payload)
+	resp, err := safewLongPollClient.Post(s.methodURL(token, "getUpdates"), "application/json", bytes.NewReader(reqBody))
 	if err != nil {
 		return nil, offset, fmt.Errorf("safew getUpdates: %w", err)
 	}
@@ -183,7 +190,14 @@ func safewErrorDescription(body []byte) string {
 	if err := json.Unmarshal(body, &api); err != nil {
 		return strings.TrimSpace(string(body))
 	}
+	switch {
+	case api.Description != "" && api.ErrorMsg != "":
+		return api.Description + ": " + api.ErrorMsg
+	case api.ErrorMsg != "":
+		return api.ErrorMsg
+	default:
 		return api.Description
+	}
 }

 func escapeMarkdownV2(s string) string {
@@ -99,6 +99,13 @@ func TestPollGroupChatsSuccess(t *testing.T) {
 	}
 }

+func TestSafewErrorDescriptionIncludesErrorMsg(t *testing.T) {
+	got := safewErrorDescription([]byte(`{"ok":false,"description":"BAD_REQUEST","error_msg":"Conflict: terminated by other getUpdates request"}`))
+	if !strings.Contains(got, "Conflict") {
+		t.Fatalf("got %q, want error_msg", got)
+	}
+}
+
 func TestPollGroupChatsUnauthorized(t *testing.T) {
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusUnauthorized)
@@ -2,7 +2,9 @@ package safew

 import (
 	"context"
+	"errors"
 	"log/slog"
+	"strings"
 	"sync"
 	"time"

@@ -11,13 +13,44 @@ import (

 type Poller func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error)

+const pollLockTTL = 60 * time.Second
+
+type runner struct {
+	cancel    context.CancelFunc
+	firstDone chan struct{}
+	mu        sync.Mutex
+	firstErr  error
+}
+
+func (r *runner) setFirst(err error) {
+	r.mu.Lock()
+	r.firstErr = err
+	r.mu.Unlock()
+	select {
+	case <-r.firstDone:
+	default:
+		close(r.firstDone)
+	}
+}
+
+func (r *runner) waitFirst(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-r.firstDone:
+		r.mu.Lock()
+		defer r.mu.Unlock()
+		return r.firstErr
+	}
+}
+
 type Watcher struct {
 	store  ChatStore
 	poll   Poller
 	bgIdle time.Duration

 	mu      sync.Mutex
-	running map[string]context.CancelFunc
+	running map[string]*runner
 	stopped bool
 }

@@ -26,39 +59,49 @@ func NewWatcher(store ChatStore, poll Poller) *Watcher {
 		store:   store,
 		poll:    poll,
 		bgIdle:  time.Second,
-		running: map[string]context.CancelFunc{},
+		running: map[string]*runner{},
 	}
 }

 func (w *Watcher) Ensure(token string) {
+	w.ensure(token)
+}
+
+func (w *Watcher) ensure(token string) *runner {
 	if token == "" || w == nil {
-		return
+		return nil
 	}
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	if w.stopped {
-		return
+		return nil
 	}
-	if _, ok := w.running[token]; ok {
-		return
+	if r, ok := w.running[token]; ok {
+		return r
 	}
 	ctx, cancel := context.WithCancel(context.Background())
-	w.running[token] = cancel
-	go w.loop(ctx, token)
+	r := &runner{cancel: cancel, firstDone: make(chan struct{})}
+	w.running[token] = r
+	go w.loop(ctx, token, r)
+	return r
 }

 func (w *Watcher) Stop() {
 	w.mu.Lock()
 	w.stopped = true
-	for _, cancel := range w.running {
-		cancel()
+	for _, r := range w.running {
+		r.cancel()
 	}
-	w.running = map[string]context.CancelFunc{}
+	w.running = map[string]*runner{}
 	w.mu.Unlock()
 }

 func (w *Watcher) Refresh(ctx context.Context, token string) error {
-	return w.pollOnce(ctx, token, 0)
+	r := w.ensure(token)
+	if r == nil {
+		return nil
+	}
+	return r.waitFirst(ctx)
 }

 func (w *Watcher) List(ctx context.Context, token, q string) ([]adapter.SafewChat, error) {
@@ -69,28 +112,66 @@ func (w *Watcher) List(ctx context.Context, token, q string) ([]adapter.SafewCha
 	return adapter.FilterSafewChats(chats, q), nil
 }

-func (w *Watcher) loop(ctx context.Context, token string) {
+func (w *Watcher) loop(ctx context.Context, token string, r *runner) {
+	for {
+		err := w.pollOnce(ctx, token, 0)
+		if ctx.Err() != nil {
+			r.setFirst(ctx.Err())
+			return
+		}
+		if isSafewAuth(err) {
+			r.setFirst(err)
+			return
+		}
+		if isGetUpdatesConflict(err) {
+			if !w.sleep(ctx, conflictBackoff(w.bgIdle)) {
+				r.setFirst(ctx.Err())
+				return
+			}
+			continue
+		}
+		r.setFirst(err)
+		break
+	}
+
 	for {
 		if err := w.pollOnce(ctx, token, 30); err != nil {
 			if ctx.Err() != nil {
 				return
 			}
+			if isSafewAuth(err) {
+				slog.Warn("safew watcher poll", "error", err)
+				return
+			}
+			if isGetUpdatesConflict(err) {
+				slog.Debug("safew watcher poll conflict", "error", err)
+				if !w.sleep(ctx, conflictBackoff(w.bgIdle)) {
+					return
+				}
+				continue
+			}
 			slog.Warn("safew watcher poll", "error", err)
 		}
-		idle := w.bgIdle
-		if idle <= 0 {
-			idle = time.Second
-		}
-		select {
-		case <-ctx.Done():
+		if !w.sleep(ctx, w.bgIdle) {
 			return
-		case <-time.After(idle):
 		}
 	}
 }

+func (w *Watcher) sleep(ctx context.Context, d time.Duration) bool {
+	if d <= 0 {
+		d = time.Second
+	}
+	select {
+	case <-ctx.Done():
+		return false
+	case <-time.After(d):
+		return true
+	}
+}
+
 func (w *Watcher) pollOnce(ctx context.Context, token string, timeout int) error {
-	ok, err := w.store.TrySafewPollLock(ctx, token, 35*time.Second)
+	ok, err := w.store.TrySafewPollLock(ctx, token, pollLockTTL)
 	if err != nil {
 		return err
 	}
@@ -115,3 +196,27 @@ func (w *Watcher) pollOnce(ctx context.Context, token string, timeout int) error
 	}
 	return nil
 }
+
+func isSafewAuth(err error) bool {
+	var e *adapter.SafewAuthError
+	return errors.As(err, &e)
+}
+
+func isGetUpdatesConflict(err error) bool {
+	if err == nil {
+		return false
+	}
+	s := strings.ToLower(err.Error())
+	return strings.Contains(s, "conflict") && strings.Contains(s, "getupdates")
+}
+
+func conflictBackoff(idle time.Duration) time.Duration {
+	if idle <= 0 {
+		return 3 * time.Second
+	}
+	d := idle * 3
+	if d < 3*time.Second {
+		return 3 * time.Second
+	}
+	return d
+}
@@ -3,6 +3,7 @@ package safew
 import (
 	"context"
 	"errors"
+	"sync/atomic"
 	"testing"
 	"time"

@@ -13,7 +14,7 @@ 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)
+			return nil, offset, nil
 		}
 		if offset != 0 {
 			t.Fatalf("offset=%d", offset)
@@ -36,6 +37,7 @@ func TestRefreshMergesAndAdvancesOffset(t *testing.T) {
 	if off != 11 {
 		t.Fatalf("offset=%d", off)
 	}
+	w.Stop()
 }

 func TestRefreshAuthError(t *testing.T) {
@@ -50,6 +52,44 @@ func TestRefreshAuthError(t *testing.T) {
 	if _, ok := err.(*adapter.SafewAuthError); !ok {
 		t.Fatalf("%T", err)
 	}
+	w.Stop()
+}
+
+type alwaysLockStore struct{ *MemStore }
+
+func (a *alwaysLockStore) TrySafewPollLock(context.Context, string, time.Duration) (bool, error) {
+	return true, nil
+}
+
+func (a *alwaysLockStore) UnlockSafewPoll(context.Context, string) error { return nil }
+
+func TestRefreshAndEnsureDoNotOverlapPolls(t *testing.T) {
+	st := &alwaysLockStore{MemStore: NewMemStore()}
+	var inflight int32
+	var overlap int32
+	poll := func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error) {
+		n := atomic.AddInt32(&inflight, 1)
+		if n > 1 {
+			atomic.AddInt32(&overlap, 1)
+		}
+		time.Sleep(40 * time.Millisecond)
+		atomic.AddInt32(&inflight, -1)
+		return []adapter.SafewChat{{ID: "1", Type: "group", Title: "g"}}, offset + 1, nil
+	}
+	w := NewWatcher(st, poll)
+	w.bgIdle = 20 * time.Millisecond
+	ctx := context.Background()
+	errCh := make(chan error, 1)
+	go func() { errCh <- w.Refresh(ctx, "tok") }()
+	w.Ensure("tok")
+	if err := <-errCh; err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(80 * time.Millisecond)
+	w.Stop()
+	if atomic.LoadInt32(&overlap) > 0 {
+		t.Fatalf("overlapping getUpdates: %d", overlap)
+	}
 }

 func TestEnsurePollsInBackground(t *testing.T) {
@@ -57,7 +97,7 @@ func TestEnsurePollsInBackground(t *testing.T) {
 	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")
+			return nil, offset, nil
 		}
 		select {
 		case got <- timeout:
@@ -74,3 +114,10 @@ func TestEnsurePollsInBackground(t *testing.T) {
 	}
 	w.Stop()
 }
+
+func TestPollConflictIsRetryable(t *testing.T) {
+	err := errors.New("safew getUpdates status 400: BAD_REQUEST: Conflict: terminated by other getUpdates request; make sure that only one bot instance is running")
+	if !isGetUpdatesConflict(err) {
+		t.Fatal("expected conflict")
+	}
+}