1 files changed, 247 insertions, 0 deletions
diff --git a/internal/telegram/poller_test.go b/internal/telegram/poller_test.go
new file mode 100644
index 0000000..4d4ccb4
--- /dev/null
+++ b/internal/telegram/poller_test.go
@@ -0,0 +1,247 @@
+package telegram
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+type recordingHandler struct {
+	mu       sync.Mutex
+	handled  []int64
+	block    chan struct{}
+	started  chan struct{}
+	finished chan struct{}
+}
+
+func (h *recordingHandler) Handle(ctx context.Context, update Update) error {
+	if h.started != nil {
+		select {
+		case h.started <- struct{}{}:
+		default:
+		}
+	}
+	if h.block != nil {
+		<-h.block
+	}
+	h.mu.Lock()
+	h.handled = append(h.handled, update.UpdateID)
+	h.mu.Unlock()
+	if h.finished != nil {
+		select {
+		case h.finished <- struct{}{}:
+		default:
+		}
+	}
+	return nil
+}
+
+func (h *recordingHandler) seen() []int64 {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return append([]int64(nil), h.handled...)
+}
+
+type updateServer struct {
+	mu      sync.Mutex
+	batches [][]Update
+	offsets []int64
+}
+
+func (s *updateServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if !strings.HasSuffix(r.URL.Path, "/getUpdates") {
+		writeJSON(w, 200, `{"ok":true,"result":true}`)
+		return
+	}
+
+	var req GetUpdatesRequest
+	json.NewDecoder(r.Body).Decode(&req)
+
+	s.mu.Lock()
+	s.offsets = append(s.offsets, req.Offset)
+	var batch []Update
+	if len(s.batches) > 0 {
+		batch = s.batches[0]
+		s.batches = s.batches[1:]
+	}
+	s.mu.Unlock()
+
+	if len(batch) == 0 {
+		select {
+		case <-r.Context().Done():
+		case <-time.After(50 * time.Millisecond):
+		}
+	}
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]any{"ok": true, "result": batch})
+}
+
+func (s *updateServer) seenOffsets() []int64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]int64(nil), s.offsets...)
+}
+
+func message(id int64) Update {
+	return Update{
+		UpdateID: id,
+		Message:  &Message{MessageID: id, Chat: Chat{ID: 1, Type: "private"}, Text: "01001000"},
+	}
+}
+
+func newPoller(t *testing.T, source *updateServer, handler Handler, opts ...PollerOption) *Poller {
+	t.Helper()
+	server := httptest.NewServer(source)
+	t.Cleanup(server.Close)
+
+	client, err := New(testToken,
+		WithBaseURL(server.URL),
+		WithTimeout(2*time.Second),
+		WithBackoff(time.Millisecond, 5*time.Millisecond),
+	)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+
+	base := []PollerOption{
+		WithLogger(slog.New(slog.NewJSONHandler(io.Discard, nil))),
+		WithPollTimeout(0),
+		WithShutdownGrace(2 * time.Second),
+		WithErrorBackoff(time.Millisecond, 5*time.Millisecond),
+	}
+	return NewPoller(client, handler, append(base, opts...)...)
+}
+
+func TestPollerAdvancesTheOffset(t *testing.T) {
+	source := &updateServer{batches: [][]Update{
+		{message(4), message(5)},
+		{message(9)},
+	}}
+	handler := &recordingHandler{finished: make(chan struct{}, 8)}
+	poller := newPoller(t, source, handler)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- poller.Run(ctx) }()
+
+	deadline := time.After(5 * time.Second)
+	for len(handler.seen()) < 3 {
+		select {
+		case <-handler.finished:
+		case <-deadline:
+			t.Fatalf("only handled %v", handler.seen())
+		}
+	}
+	cancel()
+	<-done
+
+	offsets := source.seenOffsets()
+	if offsets[0] != 0 {
+		t.Fatalf("first poll asked for %d, want 0", offsets[0])
+	}
+	if offsets[1] != 6 {
+		t.Fatalf("second poll asked for %d, want 6", offsets[1])
+	}
+	if last := offsets[len(offsets)-1]; last != 10 {
+		t.Fatalf("last poll asked for %d, want 10", last)
+	}
+	if poller.Offset() != 10 {
+		t.Fatalf("offset = %d, want 10", poller.Offset())
+	}
+}
+
+func TestPollerFinishesTheUpdateInFlightAndConfirmsIt(t *testing.T) {
+	source := &updateServer{batches: [][]Update{{message(3)}}}
+	handler := &recordingHandler{
+		block:    make(chan struct{}),
+		started:  make(chan struct{}, 1),
+		finished: make(chan struct{}, 1),
+	}
+	poller := newPoller(t, source, handler)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- poller.Run(ctx) }()
+
+	select {
+	case <-handler.started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the update was never dispatched")
+	}
+	cancel()
+
+	select {
+	case <-done:
+		t.Fatal("Run returned while an update was still being handled")
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	close(handler.block)
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Run did not return after the update finished")
+	}
+
+	if seen := handler.seen(); len(seen) != 1 || seen[0] != 3 {
+		t.Fatalf("handled %v, want [3]", seen)
+	}
+	offsets := source.seenOffsets()
+	if last := offsets[len(offsets)-1]; last != 4 {
+		t.Fatalf("shutdown confirmed offset %d, want 4", last)
+	}
+}
+
+func TestPollerSurvivesAFailedPoll(t *testing.T) {
+	var calls int
+	var mu sync.Mutex
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		mu.Lock()
+		calls++
+		current := calls
+		mu.Unlock()
+		if current <= 2 {
+			writeJSON(w, 500, `{"ok":false,"error_code":500,"description":"boom"}`)
+			return
+		}
+		writeJSON(w, 200, `{"ok":true,"result":[{"update_id":1,"message":{"message_id":1,"chat":{"id":1,"type":"private"},"text":"01001000"}}]}`)
+	}))
+	t.Cleanup(server.Close)
+
+	client, err := New(testToken,
+		WithBaseURL(server.URL),
+		WithTimeout(time.Second),
+		WithMaxAttempts(1),
+		WithBackoff(time.Millisecond, 2*time.Millisecond),
+	)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+
+	handler := &recordingHandler{finished: make(chan struct{}, 4)}
+	poller := NewPoller(client, handler,
+		WithLogger(slog.New(slog.NewJSONHandler(io.Discard, nil))),
+		WithPollTimeout(0),
+		WithShutdownGrace(time.Second),
+		WithErrorBackoff(time.Millisecond, 5*time.Millisecond),
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- poller.Run(ctx) }()
+
+	select {
+	case <-handler.finished:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the poller gave up after transient failures")
+	}
+	cancel()
+	<-done
+}