package server

import (
	"context"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/testtidb"
)

func newTestStoreForEventBus(t *testing.T) *datastore.Store {
	t.Helper()
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	testtidb.ResetDB(t, store.DB())
	// Ensure fs_events table exists.
	if _, err := store.DB().Exec(`CREATE TABLE IF NOT EXISTS fs_events (
		seq        BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
		path       TEXT NOT NULL,
		op         VARCHAR(64) NOT NULL,
		actor      VARCHAR(255),
		ts         BIGINT NOT NULL,
		created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3)
	)`); err != nil {
		t.Fatal(err)
	}
	// Create index, ignoring "duplicate key" error for idempotency (MySQL has no IF NOT EXISTS for indexes).
	if _, err := store.DB().Exec(`CREATE INDEX idx_fs_events_created ON fs_events(created_at)`); err != nil && !strings.Contains(err.Error(), "Duplicate key") {
		t.Fatal(err)
	}
	return store
}

func TestEventBusSubscribeAndNotify(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)

	subID, notify := bus.Subscribe()
	defer bus.Unsubscribe(subID)

	// No signal yet.
	select {
	case <-notify:
		t.Fatal("unexpected signal before publish")
	case <-time.After(50 * time.Millisecond):
	}

	// Insert an event and signal.
	if _, err := store.InsertFSEvent(context.Background(), "/a.txt", "write", "actor1", time.Now().UnixMilli()); err != nil {
		t.Fatal(err)
	}
	bus.Publish()

	// Should receive a signal.
	select {
	case _, open := <-notify:
		if !open {
			t.Fatal("channel closed unexpectedly")
		}
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for signal")
	}
}

func TestEventBusSetMetricOrgIDMovesInFlightGauge(t *testing.T) {
	const tenantID = "eventbus-relabel-tenant"
	bus := NewEventBusWithOrg(tenantID, "", nil)
	subID, _ := bus.Subscribe()
	defer bus.Unsubscribe(subID)

	bus.SetMetricOrgID("org-eventbus-relabel")

	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if strings.Contains(text, `drive9_sse_inflight_connections{tenant_id="`+tenantID+`",tidbcloud_org_id="guest"}`) {
		t.Fatalf("old guest in-flight series was not deleted:\n%s", text)
	}
	if !strings.Contains(text, `drive9_sse_inflight_connections{tenant_id="`+tenantID+`",tidbcloud_org_id="org-eventbus-relabel"} 1.000000`) {
		t.Fatalf("new org in-flight series was not recorded:\n%s", text)
	}
}

func TestEventBusEventsSinceZeroReturnsReset(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)

	if _, err := store.InsertFSEvent(context.Background(), "/a.txt", "write", "", time.Now().UnixMilli()); err != nil {
		t.Fatal(err)
	}

	events, headSeq, ok := bus.EventsSince(context.Background(), 0)
	if ok {
		t.Fatalf("EventsSince(0) should return ok=false for initial sync, got ok=true with %d events", len(events))
	}
	if headSeq == 0 {
		t.Fatal("headSeq should be > 0 after insert")
	}
}

func TestEventBusEventsSinceReplays(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)

	// Insert two events.
	seq1, err := store.InsertFSEvent(context.Background(), "/a.txt", "write", "actor1", time.Now().UnixMilli())
	if err != nil {
		t.Fatal(err)
	}
	seq2, err := store.InsertFSEvent(context.Background(), "/b.txt", "delete", "actor2", time.Now().UnixMilli())
	if err != nil {
		t.Fatal(err)
	}

	// Query events since seq1 (should get event 2).
	events, headSeq, ok := bus.EventsSince(context.Background(), uint64(seq1))
	if !ok {
		t.Fatal("EventsSince should succeed")
	}
	if len(events) != 1 {
		t.Fatalf("expected 1 event, got %d", len(events))
	}
	if events[0].Path != "/b.txt" || events[0].Op != "delete" {
		t.Fatalf("unexpected event: %+v", events[0])
	}
	if headSeq < uint64(seq2) {
		t.Fatalf("headSeq=%d, want >= %d", headSeq, seq2)
	}
}

// Interior seq holes (e.g. AUTO_INCREMENT ids burned by rolled-back inserts,
// or other tenants interleaved on a shared-schema table) must NOT trigger a
// reset: no events of this tenant were lost, so delivery continues normally.
func TestEventBusEventsSinceInteriorGapDelivers(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)
	ctx := context.Background()

	seq1, err := store.InsertFSEvent(ctx, "/a.txt", "write", "actor1", time.Now().UnixMilli())
	if err != nil {
		t.Fatal(err)
	}
	seq2, err := store.InsertFSEvent(ctx, "/hole.txt", "write", "actor1", time.Now().UnixMilli())
	if err != nil {
		t.Fatal(err)
	}
	seq3, err := store.InsertFSEvent(ctx, "/b.txt", "write", "actor1", time.Now().UnixMilli())
	if err != nil {
		t.Fatal(err)
	}
	// Burn the middle seq, leaving an interior hole at seq2 while seq1 stays
	// retained (the cursor is not behind the oldest event).
	if _, err := store.DB().ExecContext(ctx, `DELETE FROM fs_events WHERE seq = ?`, seq2); err != nil {
		t.Fatal(err)
	}

	events, headSeq, ok := bus.EventsSince(ctx, uint64(seq1))
	if !ok {
		t.Fatal("EventsSince with an interior hole should deliver, not reset")
	}
	if len(events) != 1 || events[0].Path != "/b.txt" {
		t.Fatalf("events = %+v, want only /b.txt", events)
	}
	if headSeq != uint64(seq3) {
		t.Fatalf("headSeq = %d, want %d", headSeq, seq3)
	}
}

// A gap reaching back past the oldest retained seq means events were pruned
// after the client's cursor — that must still reset.
func TestEventBusEventsSincePrunedGapResets(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)
	ctx := context.Background()

	seq1, err := store.InsertFSEvent(ctx, "/a.txt", "write", "actor1", time.Now().UnixMilli())
	if err != nil {
		t.Fatal(err)
	}
	seq2, err := store.InsertFSEvent(ctx, "/b.txt", "write", "actor1", time.Now().UnixMilli())
	if err != nil {
		t.Fatal(err)
	}
	seq3, err := store.InsertFSEvent(ctx, "/c.txt", "write", "actor1", time.Now().UnixMilli())
	if err != nil {
		t.Fatal(err)
	}
	// Prune everything through seq2, so the client's cursor (seq1) falls
	// behind the oldest retained event (seq3).
	for _, seq := range []int64{seq1, seq2} {
		if _, err := store.DB().ExecContext(ctx, `DELETE FROM fs_events WHERE seq = ?`, seq); err != nil {
			t.Fatal(err)
		}
	}

	events, headSeq, ok := bus.EventsSince(ctx, uint64(seq1))
	if ok {
		t.Fatalf("EventsSince with a pruned gap should reset, got events %+v", events)
	}
	if headSeq != uint64(seq3) {
		t.Fatalf("headSeq = %d, want %d", headSeq, seq3)
	}
}

func TestEventBusSeq(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)

	// Reset AUTO_INCREMENT so seq starts at 1.
	if _, err := store.DB().Exec(`ALTER TABLE fs_events AUTO_INCREMENT = 1`); err != nil {
		t.Fatal(err)
	}

	if bus.Seq(context.Background()) != 0 {
		t.Fatal("initial seq should be 0")
	}

	seq, err := store.InsertFSEvent(context.Background(), "/a.txt", "write", "", time.Now().UnixMilli())
	if err != nil {
		t.Fatal(err)
	}
	if seq != 1 {
		t.Fatalf("insert returned seq=%d, want 1", seq)
	}
	if bus.Seq(context.Background()) != 1 {
		t.Fatalf("seq after 1 insert = %d, want 1", bus.Seq(context.Background()))
	}
}

func TestEventBusSubscribeUnsubscribe(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)

	subID, _ := bus.Subscribe()
	bus.Unsubscribe(subID)

	// After unsubscribe, a new subscribe should work.
	subID2, notify := bus.Subscribe()
	defer bus.Unsubscribe(subID2)

	bus.Publish()
	select {
	case _, open := <-notify:
		if !open {
			t.Fatal("channel closed")
		}
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for signal")
	}
}

func TestEventBusMultipleSubscribers(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)

	var wg sync.WaitGroup
	const n = 5
	for i := 0; i < n; i++ {
		wg.Add(1)
		subID, notify := bus.Subscribe()
		go func(id uint64, ch chan struct{}) {
			defer wg.Done()
			select {
			case _, open := <-ch:
				if open {
					bus.Unsubscribe(id)
				}
			case <-time.After(2 * time.Second):
			}
		}(subID, notify)
	}

	bus.Publish()
	wg.Wait()
}

// TestEventBusStoreRefreshNoRace exercises concurrent SetStore + EventsSince
// to catch data races on the store field. Run with -race to detect violations.
func TestEventBusStoreRefreshNoRace(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)

	done := make(chan struct{})

	// Writer goroutine: repeatedly refresh the store.
	go func() {
		defer close(done)
		for i := 0; i < 100; i++ {
			bus.SetStore(store)
		}
	}()

	// Reader goroutine: repeatedly call EventsSince.
	for i := 0; i < 100; i++ {
		bus.EventsSince(context.Background(), 0)
	}

	<-done
}

// TestEventBusEventsSinceQueryErrorReturnsCaughtUp verifies that when the
// store query fails (e.g. DB closed, table missing), EventsSince returns
// ok=true with empty events (caught up) instead of a reset — preventing
// continuous full-cache invalidation on every poll. This is the silent-
// failure branch flagged in D1.
func TestEventBusEventsSinceQueryErrorReturnsCaughtUp(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)

	// Insert one event so since=1 is valid.
	if _, err := store.InsertFSEvent(context.Background(), "/a.txt", "write", "", time.Now().UnixMilli()); err != nil {
		t.Fatal(err)
	}

	// Close the store's DB to simulate query failure.
	if err := store.Close(); err != nil {
		t.Fatal(err)
	}

	events, headSeq, ok := bus.EventsSince(context.Background(), 1)
	if !ok {
		t.Fatal("EventsSince should return ok=true on query error (caught up), not reset")
	}
	if len(events) != 0 {
		t.Fatalf("expected 0 events on query error, got %d", len(events))
	}
	if headSeq != 1 {
		t.Fatalf("headSeq should be unchanged (=since) on query error, got %d", headSeq)
	}
}

// TestEventBusUnsubscribeOneOfManyReturnsImmediately verifies that unsubscribing
// while other subscribers remain connected returns immediately (B1 regression test).
// Before the original fix, Unsubscribe unconditionally waited on the poll goroutine,
// blocking until all other subscribers left. (The per-bus poll goroutine has since
// been removed entirely — cross-pod discovery is now centralized in the tenantOutboxPoller
// — but the fast-return property is still asserted here.)
func TestEventBusUnsubscribeOneOfManyReturnsImmediately(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)

	// Two subscribers.
	id1, _ := bus.Subscribe()
	id2, _ := bus.Subscribe()
	defer bus.Unsubscribe(id2)

	// Unsubscribe id1 while id2 is still subscribed — must return within 1s.
	done := make(chan struct{})
	go func() {
		bus.Unsubscribe(id1)
		close(done)
	}()

	select {
	case <-done:
		// Success: returned immediately.
	case <-time.After(time.Second):
		t.Fatal("Unsubscribe blocked while another subscriber remained connected")
	}
}

// TestEventsSinceEDistinguishesQueryError verifies that EventsSinceE surfaces
// DB query errors while the EventsSince wrapper keeps the tolerated
// caught-up fallback (ok=true, empty events, headSeq=since).
func TestEventsSinceEDistinguishesQueryError(t *testing.T) {
	store := newTestStoreForEventBus(t)
	bus := NewEventBus("test-tenant", store)

	// Insert one event so since=1 is a valid cursor (not initial sync).
	if _, err := store.InsertFSEvent(context.Background(), "/a.txt", "write", "", time.Now().UnixMilli()); err != nil {
		t.Fatal(err)
	}

	// Close the store's DB to simulate query failure.
	if err := store.Close(); err != nil {
		t.Fatal(err)
	}

	events, headSeq, ok, err := bus.EventsSinceE(context.Background(), 1)
	if err == nil {
		t.Fatal("EventsSinceE should return the query error")
	}
	if !ok || len(events) != 0 || headSeq != 1 {
		t.Fatalf("EventsSinceE fallback = (%v, %d, %v), want ([], 1, true) alongside the error", events, headSeq, ok)
	}

	events, headSeq, ok = bus.EventsSince(context.Background(), 1)
	if !ok || len(events) != 0 || headSeq != 1 {
		t.Fatalf("EventsSince tolerated fallback = (%v, %d, %v), want ([], 1, true)", events, headSeq, ok)
	}
}

func TestEventBusCloseIdempotent(t *testing.T) {
	bus := NewEventBus("t-close", nil)
	select {
	case <-bus.Done():
		t.Fatal("Done closed before Close")
	default:
	}
	bus.Close()
	bus.Close()
	select {
	case <-bus.Done():
	default:
		t.Fatal("Done not closed after Close")
	}
}

func TestEventBusesCloseBusRemovesAndCloses(t *testing.T) {
	ebs := newEventBuses()
	bus := ebs.get("t-close", nil)
	ebs.closeBus(context.Background(), "t-close")
	select {
	case <-bus.Done():
	default:
		t.Fatal("bus not closed by closeBus")
	}
	if got := ebs.getIfExists("t-close"); got != nil {
		t.Fatal("bus still present after closeBus")
	}
	// A later request (e.g. after a delete rollback) builds a fresh, open bus.
	fresh := ebs.get("t-close", nil)
	if fresh == bus {
		t.Fatal("expected a fresh bus after closeBus")
	}
	select {
	case <-fresh.Done():
		t.Fatal("fresh bus is closed")
	default:
	}
	// Closing an unknown tenant is a no-op.
	ebs.closeBus(context.Background(), "t-unknown")
	// A nil registry is a no-op (hand-constructed servers).
	var nilBuses *eventBuses
	nilBuses.closeBus(context.Background(), "t-close")
}