at-container-registry/pkg/hold/pds/events_test.go

package pds

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	atproto "github.com/bluesky-social/indigo/api/atproto"
	"github.com/ipfs/go-cid"
)

// TestNewEventBroadcaster tests event broadcaster creation
func TestNewEventBroadcaster(t *testing.T) {
	holdDID := "did:web:hold.example.com"
	broadcaster := NewEventBroadcaster(holdDID, 100, "")

	if broadcaster.holdDID != holdDID {
		t.Errorf("Expected holdDID=%s, got %s", holdDID, broadcaster.holdDID)
	}

	if broadcaster.eventSeq != 0 {
		t.Errorf("Expected initial eventSeq=0, got %d", broadcaster.eventSeq)
	}

	if broadcaster.maxHistory != 100 {
		t.Errorf("Expected maxHistory=100, got %d", broadcaster.maxHistory)
	}

	if len(broadcaster.subscribers) != 0 {
		t.Errorf("Expected 0 subscribers initially, got %d", len(broadcaster.subscribers))
	}
}

// TestNewEventBroadcaster_DefaultHistory tests default history size
func TestNewEventBroadcaster_DefaultHistory(t *testing.T) {
	// Zero or negative maxHistory should default to 100
	broadcaster := NewEventBroadcaster("did:web:test", 0, "")
	if broadcaster.maxHistory != 100 {
		t.Errorf("Expected default maxHistory=100 for input 0, got %d", broadcaster.maxHistory)
	}

	broadcaster2 := NewEventBroadcaster("did:web:test", -5, "")
	if broadcaster2.maxHistory != 100 {
		t.Errorf("Expected default maxHistory=100 for negative input, got %d", broadcaster2.maxHistory)
	}
}

// TestGetCurrentSeq tests sequence number tracking
func TestGetCurrentSeq(t *testing.T) {
	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "")

	// Initial seq should be 0
	seq := broadcaster.GetCurrentSeq()
	if seq != 0 {
		t.Errorf("Expected initial seq=0, got %d", seq)
	}

	// After broadcasting, seq should increment
	ctx := context.Background()
	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")

	event := &RepoEvent{
		NewRoot:   testCID,
		Rev:       "test-rev-1",
		RepoSlice: []byte("test CAR data"),
		Ops: []RepoOp{
			{
				Kind:       EvtKindCreateRecord,
				Collection: "io.atcr.hold.crew",
				Rkey:       "test123",
			},
		},
	}

	broadcaster.Broadcast(ctx, event)

	seq = broadcaster.GetCurrentSeq()
	if seq != 1 {
		t.Errorf("Expected seq=1 after one broadcast, got %d", seq)
	}

	// Broadcast again
	broadcaster.Broadcast(ctx, event)

	seq = broadcaster.GetCurrentSeq()
	if seq != 2 {
		t.Errorf("Expected seq=2 after two broadcasts, got %d", seq)
	}
}

// TestBroadcast tests event broadcasting
func TestBroadcast(t *testing.T) {
	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "")
	ctx := context.Background()

	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")

	event := &RepoEvent{
		NewRoot:   testCID,
		Rev:       "test-rev-1",
		RepoSlice: []byte("test CAR data"),
		Ops: []RepoOp{
			{
				Kind:       EvtKindCreateRecord,
				Collection: "io.atcr.hold.crew",
				Rkey:       "test123",
				RecCid:     &testCID,
			},
		},
	}

	// Broadcast should not panic without subscribers
	broadcaster.Broadcast(ctx, event)

	// Verify sequence incremented
	if broadcaster.eventSeq != 1 {
		t.Errorf("Expected eventSeq=1, got %d", broadcaster.eventSeq)
	}

	// Verify event added to history
	if len(broadcaster.eventHistory) != 1 {
		t.Errorf("Expected 1 event in history, got %d", len(broadcaster.eventHistory))
	}

	he := broadcaster.eventHistory[0]
	if he.Seq != 1 {
		t.Errorf("Expected history seq=1, got %d", he.Seq)
	}

	if he.Event.Repo != "did:web:hold.example.com" {
		t.Errorf("Expected repo=did:web:hold.example.com, got %s", he.Event.Repo)
	}

	if he.Event.Type != "#commit" {
		t.Errorf("Expected type=#commit, got %s", he.Event.Type)
	}

	if len(he.Event.Ops) != 1 {
		t.Errorf("Expected 1 op, got %d", len(he.Event.Ops))
	}
}

// TestAddToHistory_RingBuffer tests ring buffer behavior
func TestAddToHistory_RingBuffer(t *testing.T) {
	// Create broadcaster with small history
	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 3, "")
	ctx := context.Background()

	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")

	// Broadcast 5 events (exceeds maxHistory of 3)
	for i := 0; i < 5; i++ {
		event := &RepoEvent{
			NewRoot:   testCID,
			Rev:       "test-rev",
			RepoSlice: []byte("test CAR data"),
			Ops:       []RepoOp{},
		}
		broadcaster.Broadcast(ctx, event)
	}

	// Should only keep last 3 events
	if len(broadcaster.eventHistory) != 3 {
		t.Errorf("Expected 3 events in history (ring buffer), got %d", len(broadcaster.eventHistory))
	}

	// Verify we kept the most recent events (seq 3, 4, 5)
	expectedSeqs := []int64{3, 4, 5}
	for i, expected := range expectedSeqs {
		if broadcaster.eventHistory[i].Seq != expected {
			t.Errorf("Expected history[%d].Seq=%d, got %d", i, expected, broadcaster.eventHistory[i].Seq)
		}
	}

	// Final sequence should be 5
	if broadcaster.eventSeq != 5 {
		t.Errorf("Expected eventSeq=5, got %d", broadcaster.eventSeq)
	}
}

// TestConvertToCommitEvent tests event conversion
func TestConvertToCommitEvent(t *testing.T) {
	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "")

	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")
	since := "prev-rev"

	event := &RepoEvent{
		NewRoot:   testCID,
		Rev:       "test-rev-123",
		Since:     &since,
		RepoSlice: []byte("test CAR data"),
		Ops: []RepoOp{
			{
				Kind:       EvtKindCreateRecord,
				Collection: "io.atcr.hold.crew",
				Rkey:       "member1",
				RecCid:     &testCID,
			},
			{
				Kind:       EvtKindUpdateRecord,
				Collection: "io.atcr.hold.captain",
				Rkey:       "self",
				RecCid:     &testCID,
			},
			{
				Kind:       EvtKindDeleteRecord,
				Collection: "io.atcr.hold.crew",
				Rkey:       "oldmember",
				RecCid:     nil, // Deletes don't have CIDs
			},
		},
	}

	commitEvent := broadcaster.convertToCommitEvent(event, 42)

	// Verify basic fields
	if commitEvent.Seq != 42 {
		t.Errorf("Expected seq=42, got %d", commitEvent.Seq)
	}

	if commitEvent.Repo != "did:web:hold.example.com" {
		t.Errorf("Expected repo=did:web:hold.example.com, got %s", commitEvent.Repo)
	}

	if commitEvent.Commit != testCID.String() {
		t.Errorf("Expected commit=%s, got %s", testCID.String(), commitEvent.Commit)
	}

	if commitEvent.Rev != "test-rev-123" {
		t.Errorf("Expected rev=test-rev-123, got %s", commitEvent.Rev)
	}

	if commitEvent.Since == nil || *commitEvent.Since != since {
		t.Errorf("Expected since=%s, got %v", since, commitEvent.Since)
	}

	if string(commitEvent.Blocks) != "test CAR data" {
		t.Errorf("Expected blocks='test CAR data', got %s", string(commitEvent.Blocks))
	}

	if commitEvent.Type != "#commit" {
		t.Errorf("Expected type=#commit, got %s", commitEvent.Type)
	}

	// Verify time is set
	if commitEvent.Time == "" {
		t.Error("Expected non-empty time")
	}

	// Parse time to verify it's valid RFC3339
	_, err := time.Parse(time.RFC3339, commitEvent.Time)
	if err != nil {
		t.Errorf("Expected valid RFC3339 time, got error: %v", err)
	}

	// Verify ops conversion
	if len(commitEvent.Ops) != 3 {
		t.Fatalf("Expected 3 ops, got %d", len(commitEvent.Ops))
	}

	// Check create op
	createOp := commitEvent.Ops[0]
	if createOp.Action != "create" {
		t.Errorf("Expected action=create, got %s", createOp.Action)
	}
	if createOp.Path != "io.atcr.hold.crew/member1" {
		t.Errorf("Expected path=io.atcr.hold.crew/member1, got %s", createOp.Path)
	}
	if createOp.Cid == nil {
		t.Error("Expected non-nil CID for create op")
	}

	// Check update op
	updateOp := commitEvent.Ops[1]
	if updateOp.Action != "update" {
		t.Errorf("Expected action=update, got %s", updateOp.Action)
	}
	if updateOp.Path != "io.atcr.hold.captain/self" {
		t.Errorf("Expected path=io.atcr.hold.captain/self, got %s", updateOp.Path)
	}

	// Check delete op
	deleteOp := commitEvent.Ops[2]
	if deleteOp.Action != "delete" {
		t.Errorf("Expected action=delete, got %s", deleteOp.Action)
	}
	if deleteOp.Path != "io.atcr.hold.crew/oldmember" {
		t.Errorf("Expected path=io.atcr.hold.crew/oldmember, got %s", deleteOp.Path)
	}
	if deleteOp.Cid != nil {
		t.Error("Expected nil CID for delete op")
	}
}

// TestConvertToCommitEvent_NoSince tests event without since field
func TestConvertToCommitEvent_NoSince(t *testing.T) {
	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "")

	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")

	event := &RepoEvent{
		NewRoot:   testCID,
		Rev:       "test-rev-123",
		Since:     nil, // No since
		RepoSlice: []byte("test CAR data"),
		Ops:       []RepoOp{},
	}

	commitEvent := broadcaster.convertToCommitEvent(event, 1)

	if commitEvent.Since != nil {
		t.Errorf("Expected nil since, got %v", commitEvent.Since)
	}
}

// TestSetRepoEventHandler tests handler registration
func TestSetRepoEventHandler(t *testing.T) {
	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "")

	handler := broadcaster.SetRepoEventHandler()
	if handler == nil {
		t.Fatal("Expected non-nil handler")
	}

	// Call handler
	ctx := context.Background()
	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")

	event := &RepoEvent{
		NewRoot:   testCID,
		Rev:       "test-rev",
		RepoSlice: []byte("test CAR data"),
		Ops:       []RepoOp{},
	}

	handler(ctx, event)

	// Verify event was broadcast
	if broadcaster.eventSeq != 1 {
		t.Errorf("Expected eventSeq=1 after handler call, got %d", broadcaster.eventSeq)
	}

	if len(broadcaster.eventHistory) != 1 {
		t.Errorf("Expected 1 event in history after handler call, got %d", len(broadcaster.eventHistory))
	}
}

// TestEncodeCBOR tests CBOR encoding (currently JSON)
func TestEncodeCBOR(t *testing.T) {
	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")

	event := &RepoCommitEvent{
		Seq:    1,
		Repo:   "did:web:hold.example.com",
		Commit: testCID.String(),
		Rev:    "test-rev",
		Blocks: []byte("test data"),
		Ops:    []*atproto.SyncSubscribeRepos_RepoOp{},
		Time:   time.Now().Format(time.RFC3339),
		Type:   "#commit",
	}

	encoded, err := encodeCBOR(event)
	if err != nil {
		t.Fatalf("Failed to encode CBOR: %v", err)
	}

	if len(encoded) == 0 {
		t.Error("Expected non-empty encoded data")
	}

	// Current implementation uses JSON, so verify it's valid JSON
	// In future, this would be proper CBOR validation
	var decoded RepoCommitEvent
	if err := json.Unmarshal(encoded, &decoded); err != nil {
		t.Errorf("Failed to decode JSON: %v", err)
	}

	if decoded.Seq != 1 {
		t.Errorf("Expected decoded seq=1, got %d", decoded.Seq)
	}
}

// TestSubscribe_CursorZeroBackfill tests that cursor=0 replays all events
func TestSubscribe_CursorZeroBackfill(t *testing.T) {
	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100, "")
	ctx := context.Background()

	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")

	// Broadcast 5 events before subscribing
	for i := 1; i <= 5; i++ {
		event := &RepoEvent{
			NewRoot:   testCID,
			Rev:       "test-rev",
			RepoSlice: []byte("test CAR data"),
			Ops:       []RepoOp{},
		}
		broadcaster.Broadcast(ctx, event)
	}

	// Verify we have 5 events in history
	if broadcaster.eventSeq != 5 {
		t.Fatalf("Expected eventSeq=5, got %d", broadcaster.eventSeq)
	}

	// Create mock websocket connection (we won't actually use it)
	// We just need to verify backfillSubscriber is called
	// For this test, we'll check the history directly
	if len(broadcaster.eventHistory) != 5 {
		t.Errorf("Expected 5 events in history, got %d", len(broadcaster.eventHistory))
	}

	// Verify all events have sequential sequence numbers
	for i, he := range broadcaster.eventHistory {
		expectedSeq := int64(i + 1)
		if he.Seq != expectedSeq {
			t.Errorf("Expected history[%d].Seq=%d, got %d", i, expectedSeq, he.Seq)
		}
	}

	// Test backfillSubscriber directly with cursor=0
	// Create a subscriber manually (conn not needed for backfill test)
	sub := &Subscriber{
		conn:   nil,                              // Not used in backfillSubscriber
		send:   make(chan *RepoCommitEvent, 100), // Large buffer for testing
		cursor: 0,
	}

	// Run backfill in a goroutine
	go broadcaster.backfillSubscriber(sub, 0)

	// Wait for events to be sent
	time.Sleep(100 * time.Millisecond)

	// Should receive all 5 events
	receivedCount := len(sub.send)
	if receivedCount != 5 {
		t.Errorf("Expected to receive 5 events with cursor=0, got %d", receivedCount)
	}

	// Verify events are in order
	for i := 1; i <= 5; i++ {
		select {
		case event := <-sub.send:
			if event.Seq != int64(i) {
				t.Errorf("Expected event seq=%d, got %d", i, event.Seq)
			}
		default:
			t.Errorf("Expected event %d but channel was empty", i)
		}
	}
}

// TestSubscribe_MidCursorBackfill tests that cursor=N only gets events after N
func TestSubscribe_MidCursorBackfill(t *testing.T) {
	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100, "")
	ctx := context.Background()

	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")

	// Broadcast 10 events before subscribing
	for i := 1; i <= 10; i++ {
		event := &RepoEvent{
			NewRoot:   testCID,
			Rev:       "test-rev",
			RepoSlice: []byte("test CAR data"),
			Ops:       []RepoOp{},
		}
		broadcaster.Broadcast(ctx, event)
	}

	// Test backfillSubscriber with cursor=5 (conn not needed for backfill test)
	sub := &Subscriber{
		conn:   nil,                              // Not used in backfillSubscriber
		send:   make(chan *RepoCommitEvent, 100), // Large buffer for testing
		cursor: 5,
	}

	// Run backfill
	go broadcaster.backfillSubscriber(sub, 5)

	// Wait for events to be sent
	time.Sleep(100 * time.Millisecond)

	// Should receive events 6-10 (5 events after cursor=5)
	receivedCount := len(sub.send)
	if receivedCount != 5 {
		t.Errorf("Expected to receive 5 events with cursor=5, got %d", receivedCount)
	}

	// Verify events start at seq=6
	for i := 6; i <= 10; i++ {
		select {
		case event := <-sub.send:
			if event.Seq != int64(i) {
				t.Errorf("Expected event seq=%d, got %d", i, event.Seq)
			}
		default:
			t.Errorf("Expected event %d but channel was empty", i)
		}
	}
}

// TestSubscribe_NegativeCursorNoBackfill tests that negative cursor means no backfill
func TestSubscribe_NegativeCursorNoBackfill(t *testing.T) {
	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100, "")
	ctx := context.Background()

	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")

	// Broadcast 5 events before subscribing
	for i := 1; i <= 5; i++ {
		event := &RepoEvent{
			NewRoot:   testCID,
			Rev:       "test-rev",
			RepoSlice: []byte("test CAR data"),
			Ops:       []RepoOp{},
		}
		broadcaster.Broadcast(ctx, event)
	}

	// Create subscriber with cursor=-1 (no backfill, conn not needed)
	sub := &Subscriber{
		conn:   nil, // Not used in this test
		send:   make(chan *RepoCommitEvent, 100),
		cursor: -1,
	}

	// Subscribe should not trigger backfill
	broadcaster.mu.Lock()
	currentSeq := broadcaster.eventSeq
	broadcaster.mu.Unlock()

	// Check the condition: cursor >= 0 && cursor < currentSeq
	// For cursor=-1, this should be false
	shouldBackfill := -1 >= 0 && -1 < currentSeq
	if shouldBackfill {
		t.Error("Expected shouldBackfill=false for cursor=-1, but condition evaluated to true")
	}

	// Verify no events in send channel (no backfill happened)
	if len(sub.send) != 0 {
		t.Errorf("Expected 0 events with cursor=-1 (no backfill), got %d", len(sub.send))
	}
}