configcenter/internal/watch/hub_test.go

package watch

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"

	"github.com/longpeng/configcenter/internal/domain"
	runtimepkg "github.com/longpeng/configcenter/internal/runtime"
)

func TestHubReconnectsFromNextRevision(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	first := make(chan runtimepkg.WatchResult, 4)
	second := make(chan runtimepkg.WatchResult, 4)
	runtimeStore := &fakeRuntimeStore{
		current: domain.RuntimeConfig{Items: map[string]string{"feature": "one"}, Revision: 3},
		streams: []chan runtimepkg.WatchResult{first, second},
	}
	hub := New(ctx, runtimeStore)
	events, unsubscribe := hub.Subscribe("/config/PROD/orders/application", Scope{ApplicationID: 1, NamespaceID: 2, EnvironmentID: 3})
	defer unsubscribe()

	initial := receiveEvent(t, events)
	if initial.Revision != 3 || initial.Force || initial.Items["feature"] != "one" {
		t.Fatalf("unexpected initial resync: %#v", initial)
	}
	waitForWatchCalls(t, runtimeStore, 1)
	first <- runtimepkg.WatchResult{Event: domain.ConfigEvent{Type: "UPDATED", Items: map[string]string{"feature": "two"}, Revision: 4}}
	if event := receiveEvent(t, events); event.Revision != 4 || event.Force {
		t.Fatalf("unexpected first event: %#v", event)
	}

	runtimeStore.setCurrent(domain.RuntimeConfig{Items: map[string]string{"feature": "four"}, Revision: 6})
	first <- runtimepkg.WatchResult{Err: errors.New("watch connection lost")}

	waitForWatchCalls(t, runtimeStore, 2)
	calls := runtimeStore.watchRevisions()
	if calls[0] != 4 || calls[1] != 5 {
		t.Fatalf("unexpected watch revisions: %v", calls)
	}

	second <- runtimepkg.WatchResult{Event: domain.ConfigEvent{Type: "UPDATED", Items: map[string]string{"feature": "three"}, Revision: 5}}
	second <- runtimepkg.WatchResult{Event: domain.ConfigEvent{Type: "UPDATED", Items: map[string]string{"feature": "four"}, Revision: 6}}
	if event := receiveEvent(t, events); event.Revision != 5 || event.Items["feature"] != "three" {
		t.Fatalf("unexpected resumed event: %#v", event)
	}
	if event := receiveEvent(t, events); event.Revision != 6 || event.Items["feature"] != "four" {
		t.Fatalf("unexpected second resumed event: %#v", event)
	}
}

func TestHubFallsBackToFullSyncOnlyWhenRevisionCompacted(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	first := make(chan runtimepkg.WatchResult, 4)
	second := make(chan runtimepkg.WatchResult, 4)
	runtimeStore := &fakeRuntimeStore{
		current: domain.RuntimeConfig{Items: map[string]string{"feature": "one"}, Revision: 3},
		streams: []chan runtimepkg.WatchResult{first, second},
	}
	hub := New(ctx, runtimeStore)
	events, unsubscribe := hub.Subscribe("/config/PROD/orders/application", Scope{ApplicationID: 1, NamespaceID: 2, EnvironmentID: 3})
	defer unsubscribe()

	initial := receiveEvent(t, events)
	if initial.Type != "FULL_SYNC" || initial.Revision != 3 {
		t.Fatalf("unexpected initial event: %#v", initial)
	}
	waitForWatchCalls(t, runtimeStore, 1)

	runtimeStore.setCurrent(domain.RuntimeConfig{Items: map[string]string{"feature": "latest"}, Revision: 8})
	first <- runtimepkg.WatchResult{Err: errors.Join(runtimepkg.ErrRevisionCompacted, errors.New("compacted")), CompactRevision: 7}
	resync := receiveEvent(t, events)
	if resync.Type != "FULL_SYNC" || resync.Revision != 8 || resync.Items["feature"] != "latest" {
		t.Fatalf("unexpected compacted resync: %#v", resync)
	}

	waitForWatchCalls(t, runtimeStore, 2)
	if calls := runtimeStore.watchRevisions(); calls[1] != 9 {
		t.Fatalf("watch did not resume from snapshot+1: %v", calls)
	}
}

func TestHubRefreshesMatchingScopeWithoutRevisionChange(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	stream := make(chan runtimepkg.WatchResult)
	runtimeStore := &fakeRuntimeStore{
		current: domain.RuntimeConfig{Items: map[string]string{"feature": "base"}, Revision: 7},
		streams: []chan runtimepkg.WatchResult{stream},
	}
	hub := New(ctx, runtimeStore)
	scope := Scope{ApplicationID: 11, NamespaceID: 12, EnvironmentID: 13}
	events, unsubscribe := hub.Subscribe("/config/PROD/catalog/application", scope)
	defer unsubscribe()
	initial := receiveEvent(t, events)
	if initial.Revision != 7 || initial.Force {
		t.Fatalf("unexpected initial event: %#v", initial)
	}
	waitForWatchCalls(t, runtimeStore, 1)

	if err := hub.Refresh(ctx, scope); err != nil {
		t.Fatal(err)
	}
	event := receiveEvent(t, events)
	if event.Revision != 7 || !event.Force || event.Items["feature"] != "base" {
		t.Fatalf("unexpected forced refresh event: %#v", event)
	}
}

func TestHubRetriesInitialSnapshotBeforeWatching(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	runtimeStore := &fakeRuntimeStore{
		current:     domain.RuntimeConfig{Items: map[string]string{"feature": "ready"}, Revision: 9},
		streams:     []chan runtimepkg.WatchResult{make(chan runtimepkg.WatchResult)},
		getFailures: 1,
	}
	hub := New(ctx, runtimeStore)
	events, unsubscribe := hub.Subscribe("/config/PROD/billing/application", Scope{ApplicationID: 21, NamespaceID: 22, EnvironmentID: 23})
	defer unsubscribe()

	event := receiveEvent(t, events)
	if event.Type != "FULL_SYNC" || event.Revision != 9 || event.Items["feature"] != "ready" {
		t.Fatalf("unexpected snapshot after retry: %#v", event)
	}
	waitForWatchCalls(t, runtimeStore, 1)
	if calls := runtimeStore.watchRevisions(); len(calls) != 1 || calls[0] != 10 {
		t.Fatalf("watch started without a safe baseline: %v", calls)
	}
}

func TestHubRetainsNewestSnapshotForSlowSubscriber(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	runtimeStore := &fakeRuntimeStore{
		current: domain.RuntimeConfig{Items: map[string]string{}, Revision: 1},
		streams: []chan runtimepkg.WatchResult{make(chan runtimepkg.WatchResult)},
	}
	hub := New(ctx, runtimeStore)
	const key = "/config/PROD/search/application"
	events, unsubscribe := hub.Subscribe(key, Scope{ApplicationID: 31, NamespaceID: 32, EnvironmentID: 33})
	defer unsubscribe()
	_ = receiveEvent(t, events)

	hub.mu.Lock()
	target := hub.keys[key]
	hub.mu.Unlock()
	for revision := int64(2); revision <= 40; revision++ {
		hub.broadcast(key, target, domain.ConfigEvent{Type: "UPDATED", Items: map[string]string{}, Revision: revision})
	}
	latest := int64(0)
	for {
		select {
		case event := <-events:
			latest = event.Revision
		default:
			if latest != 40 {
				t.Fatalf("newest snapshot was dropped: latest revision=%d", latest)
			}
			return
		}
	}
}

type fakeRuntimeStore struct {
	mu          sync.Mutex
	current     domain.RuntimeConfig
	streams     []chan runtimepkg.WatchResult
	revisions   []int64
	getFailures int
}

func (s *fakeRuntimeStore) Close() error { return nil }

func (s *fakeRuntimeStore) Ping(context.Context) error { return nil }

func (s *fakeRuntimeStore) Put(context.Context, string, []byte, domain.Release) (int64, error) {
	return 0, nil
}

func (s *fakeRuntimeStore) Get(context.Context, string) (domain.RuntimeConfig, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.getFailures > 0 {
		s.getFailures--
		return domain.RuntimeConfig{}, errors.New("runtime unavailable")
	}
	items := make(map[string]string, len(s.current.Items))
	for key, value := range s.current.Items {
		items[key] = value
	}
	current := s.current
	current.Items = items
	return current, nil
}

func (s *fakeRuntimeStore) setCurrent(current domain.RuntimeConfig) {
	s.mu.Lock()
	s.current = current
	s.mu.Unlock()
}

func (s *fakeRuntimeStore) Watch(_ context.Context, _ string, revision int64) <-chan runtimepkg.WatchResult {
	s.mu.Lock()
	defer s.mu.Unlock()
	index := len(s.revisions)
	s.revisions = append(s.revisions, revision)
	if index < len(s.streams) {
		return s.streams[index]
	}
	return make(chan runtimepkg.WatchResult)
}

func (s *fakeRuntimeStore) watchRevisions() []int64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return append([]int64(nil), s.revisions...)
}

func waitForWatchCalls(t *testing.T, store *fakeRuntimeStore, count int) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if len(store.watchRevisions()) >= count {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("timed out waiting for %d watch calls; got %v", count, store.watchRevisions())
}

func receiveEvent(t *testing.T, events <-chan domain.ConfigEvent) domain.ConfigEvent {
	t.Helper()
	select {
	case event, open := <-events:
		if !open {
			t.Fatal("event channel closed")
		}
		return event
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for event")
		return domain.ConfigEvent{}
	}
}