package server

import (
	"context"
	"sync"
	"sync/atomic"
	"time"

	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"go.uber.org/zap"
)

// ChangeEvent represents a single filesystem mutation event.
type ChangeEvent struct {
	Seq   uint64 `json:"seq"`             // monotonic per-tenant seq (fs_events.seq AUTO_INCREMENT)
	Path  string `json:"path"`            // affected path
	Op    string `json:"op"`              // "write" | "delete" | "rename" | "mkdir" | "copy" | "upload_complete"
	Actor string `json:"actor,omitempty"` // X-Dat9-Actor header value (per-mount ID)
	Ts    int64  `json:"ts"`              // unix milliseconds
}

const (
	eventBusListenerChanSize = 1 // signal-only channel
)

var (
	// eventPageSize is the max number of events one EventsSince call returns.
	// Callers (SSE Phase-1 replay and Phase-2 poll) must loop on a full page:
	// a short page signals the backlog is drained. A var (not a const) so
	// tests can shrink the page instead of inserting tens of thousands of
	// rows to exercise multi-page drains. Tests that mutate it must restore
	// it via t.Cleanup and must NOT use t.Parallel() — the mutation is
	// process-global.
	eventPageSize = 1000
)

// EventBus is a per-tenant event hub backed by the durable fs_events table.
// Events are stored in the shared tenant database so that they propagate across
// all pods. The local notify channel provides instant push to same-pod SSE
// clients; cross-pod events are discovered via the central tenant_notify_outbox
// table in the meta DB (polled by a single global notifyPoller per pod, not a
// per-bus goroutine).
//
// The store field is an atomic pointer so it can be safely refreshed by
// eventBuses.get (when the pool recreates a backend) while SSE handlers read it.
//
// Design note: prior to the outbox redesign, each EventBus ran its own 1s poll
// goroutine querying the tenant's fs_events table. With ~100k tenants this
// kept every serverless tenant TiDB awake (RCU cost). Now cross-pod discovery is
// centralized: a single notifyPoller reads the lightweight tenant_notify_outbox
// (in the always-provisioned meta DB) and calls Publish() on matching buses.
// That path never touches a tenant TiDB unless the tenant actually has new
// events.
type EventBus struct {
	tenantID       string
	tidbCloudOrgID string
	store          atomic.Pointer[datastore.Store]
	mu             sync.Mutex
	listeners      map[uint64]chan struct{}
	nextID         uint64
	// lastSweepUnixNano throttles the lazy fs_events retention sweep on the
	// write path to at most once per configured interval per tenant, in unix
	// nanoseconds so sub-second intervals (e.g. 500ms for e2e) throttle
	// correctly. The bus is per-tenant, so one atomic per bus gives the
	// per-tenant throttle.
	lastSweepUnixNano atomic.Int64
	// done is closed by Close when the tenant is being drained (delete
	// paths): established SSE streams select on it and return, releasing
	// the pool backend reference their request pinned in auth.
	done      chan struct{}
	closeOnce sync.Once
}

// eventBuses manages per-tenant EventBus instances. For single-tenant mode
// (fallback backend), the empty-string key is used.
type eventBuses struct {
	mu    sync.RWMutex
	buses map[string]*EventBus
}

func newEventBuses() *eventBuses {
	return &eventBuses{
		buses: make(map[string]*EventBus),
	}
}

func (ebs *eventBuses) get(tenantID string, store *datastore.Store) *EventBus {
	return ebs.getWithOrg(tenantID, "", store)
}

func (ebs *eventBuses) getWithOrg(tenantID, tidbCloudOrgID string, store *datastore.Store) *EventBus {
	ebs.mu.Lock()
	defer ebs.mu.Unlock()
	if bus, ok := ebs.buses[tenantID]; ok {
		bus.SetMetricOrgID(tidbCloudOrgID)
		// Refresh the store reference if a non-nil store is provided: the pool
		// may have invalidated and recreated the backend (closing the old store
		// and opening a new one), so the cached bus's store could be stale/closed.
		// Don't overwrite with nil — that would break a bus that already has a
		// valid store (e.g. when tenantEventBus can't resolve a backend but the
		// bus was previously initialized with one).
		if store != nil {
			bus.SetStore(store)
		}
		return bus
	}
	bus := NewEventBusWithOrg(tenantID, tidbCloudOrgID, store)
	ebs.buses[tenantID] = bus
	return bus
}

// getIfExists returns the EventBus for tenantID if one already exists, or nil
// if no bus has been created for that tenant. Unlike get, it does NOT create a
// new bus. Used by the notifyPoller to wake only tenants that have local SSE
// subscribers — skipping idle tenants avoids touching their TiDB.
func (ebs *eventBuses) getIfExists(tenantID string) *EventBus {
	ebs.mu.RLock()
	defer ebs.mu.RUnlock()
	return ebs.buses[tenantID]
}

// closeBus closes and removes the EventBus for tenantID, terminating any
// established SSE streams for the tenant. Tenant delete paths call this
// before draining the pool backend: an SSE connection pins its backend
// reference for the whole connection lifetime (auth Acquire + deferred
// release) and never re-checks tenant status, so without an explicit close
// it would hold the reference until the client disconnects and force every
// delete to wait out the full pool drain timeout. Removal (not just close)
// lets a later request build a fresh bus if the delete is rolled back.
func (ebs *eventBuses) closeBus(ctx context.Context, tenantID string) {
	if ebs == nil {
		return
	}
	ebs.mu.Lock()
	bus, ok := ebs.buses[tenantID]
	if ok {
		delete(ebs.buses, tenantID)
	}
	ebs.mu.Unlock()
	if ok {
		logger.Info(ctx, "event_bus_closed_for_tenant_drain",
			zap.String("tenant_id", tenantID))
		bus.Close()
	}
}

// activeTenantIDs returns the tenant IDs for which this pod has EventBus
// instances (and thus potentially SSE subscribers). Used by the pod registry
// to periodically report this pod's subscription set to the central DB so
// writers can route push notifications to the right pods.
func (ebs *eventBuses) activeTenantIDs() []string {
	ebs.mu.RLock()
	defer ebs.mu.RUnlock()
	ids := make([]string, 0, len(ebs.buses))
	for id, bus := range ebs.buses {
		// Only report buses that currently have listeners. A bus without
		// listeners means all SSE connections for that tenant disconnected;
		// reporting it would cause peers to push notifications to a pod that
		// has no one to deliver them to.
		if bus.HasListeners() {
			ids = append(ids, id)
		}
	}
	return ids
}

// NewEventBus creates a new EventBus backed by the given tenant store.
func NewEventBus(tenantID string, store *datastore.Store) *EventBus {
	return NewEventBusWithOrg(tenantID, "", store)
}

// NewEventBusWithOrg creates a new EventBus with explicit tenant metric org scope.
func NewEventBusWithOrg(tenantID, tidbCloudOrgID string, store *datastore.Store) *EventBus {
	eb := &EventBus{
		tenantID:       tenantID,
		tidbCloudOrgID: normalizeTenantMetricTiDBCloudOrgID(tidbCloudOrgID),
		listeners:      make(map[uint64]chan struct{}),
		done:           make(chan struct{}),
	}
	eb.store.Store(store)
	return eb
}

// TenantID returns the tenant identifier for this bus.
func (eb *EventBus) TenantID() string {
	return eb.tenantID
}

// TiDBCloudOrgID returns the org label value used by tenant-scoped SSE metrics.
func (eb *EventBus) TiDBCloudOrgID() string {
	eb.mu.Lock()
	defer eb.mu.Unlock()
	return normalizeTenantMetricTiDBCloudOrgID(eb.tidbCloudOrgID)
}

// SetStore atomically updates the backing store. Called by eventBuses.get
// when the pool invalidates and recreates a backend (closing the old store
// and opening a new one).
func (eb *EventBus) SetStore(store *datastore.Store) {
	eb.store.Store(store)
}

// SetMetricOrgID updates the org label for this bus without touching its store.
func (eb *EventBus) SetMetricOrgID(tidbCloudOrgID string) {
	nextOrgID := normalizeTenantMetricTiDBCloudOrgID(tidbCloudOrgID)
	eb.mu.Lock()
	prevOrgID := normalizeTenantMetricTiDBCloudOrgID(eb.tidbCloudOrgID)
	if prevOrgID == nextOrgID {
		eb.tidbCloudOrgID = nextOrgID
		eb.mu.Unlock()
		return
	}
	listeners := len(eb.listeners)
	eb.tidbCloudOrgID = nextOrgID
	eb.mu.Unlock()
	if listeners > 0 {
		metrics.RecordSSEInFlightWithOrg(eb.tenantID, prevOrgID, 0)
		metrics.RecordSSEInFlightWithOrg(eb.tenantID, nextOrgID, float64(listeners))
	}
}

// HasListeners returns true if this bus currently has at least one subscriber.
// Used by the notify poller to skip tenants with no local SSE connections
// (avoiding unnecessary wakeups).
func (eb *EventBus) HasListeners() bool {
	eb.mu.Lock()
	defer eb.mu.Unlock()
	return len(eb.listeners) > 0
}

// Publish signals all local subscribers that a new event is available.
// The actual event row is inserted into fs_events by the caller (publishEvent)
// before calling Publish; this method only wakes same-pod SSE clients for instant
// delivery. Cross-pod clients discover new rows via the central notifyPoller.
func (eb *EventBus) Publish() {
	eb.mu.Lock()
	for _, ch := range eb.listeners {
		select {
		case ch <- struct{}{}:
		default:
		}
	}
	eb.mu.Unlock()
}

// Subscribe registers a new listener. Returns a unique ID and a signal channel.
// The channel receives a signal whenever new events are published locally or
// discovered by the central notifyPoller.
func (eb *EventBus) Subscribe() (uint64, chan struct{}) {
	eb.mu.Lock()
	defer eb.mu.Unlock()

	id := eb.nextID
	eb.nextID++
	ch := make(chan struct{}, eventBusListenerChanSize)
	eb.listeners[id] = ch
	metrics.RecordSSEInFlightWithOrg(eb.tenantID, eb.tidbCloudOrgID, float64(len(eb.listeners)))
	return id, ch
}

// Unsubscribe removes a listener and closes its channel.
func (eb *EventBus) Unsubscribe(id uint64) {
	eb.mu.Lock()
	if ch, ok := eb.listeners[id]; ok {
		delete(eb.listeners, id)
		close(ch)
	}
	metrics.RecordSSEInFlightWithOrg(eb.tenantID, eb.tidbCloudOrgID, float64(len(eb.listeners)))
	eb.mu.Unlock()
}

// Close signals established SSE streams on this bus to terminate. It is
// idempotent. Tenant delete paths close the bus before draining the pool
// backend so a long-lived /v1/events connection cannot pin the backend
// reference past the drain timeout. The bus is removed from eventBuses by
// the caller, so a later request (e.g. after a delete rollback) builds a
// fresh bus. A zero-value bus (done == nil) is a no-op: its Done channel
// never fires.
func (eb *EventBus) Close() {
	eb.closeOnce.Do(func() {
		if eb.done != nil {
			close(eb.done)
		}
	})
}

// Done returns a channel that is closed when Close is called. A zero-value
// bus returns nil, which blocks forever in a select — the neutral "never
// closed" behavior. Buses must be built via NewEventBusWithOrg.
func (eb *EventBus) Done() <-chan struct{} {
	return eb.done
}

// Seq returns the current maximum fs_events seq, or 0 if no events exist.
func (eb *EventBus) Seq(ctx context.Context) uint64 {
	return uint64(clampNonNegative(latestSeqSafeWithMetric(ctx, eb.store.Load(), eb.tenantID)))
}

// EventsSince returns all events with seq > since and the current head seq.
// ok=false signals the caller to send a reset (cursor is stale, table missing,
// or initial sync). The reset reasons mirror the old ring buffer semantics:
//   - since == 0: initial_sync
//   - since < oldestSeq: events pruned between client cursor and retained window
//   - since > headSeq: server_restart (client cursor ahead of head)
//
// Interior seq holes (other tenants interleaved on a shared-schema table, or
// AUTO_INCREMENT ids burned by rolled-back inserts) are NOT a reset
// condition: no events of this tenant were lost, so delivery continues.
//
// DB query errors are masked (ok=true, empty events, headSeq=since): see
// EventsSinceE for the rationale and for the error-distinguishing variant.
func (eb *EventBus) EventsSince(ctx context.Context, since uint64) (events []ChangeEvent, headSeq uint64, ok bool) {
	events, headSeq, ok, _ = eb.EventsSinceE(ctx, since)
	return events, headSeq, ok
}

// EventsSinceE is EventsSince but distinguishes DB query errors: when the
// ListFSEventsSince query fails it returns the same tolerated fallback
// (ok=true, empty events, headSeq=since — the FUSE TTL/HEAD revalidation
// provides correctness without SSE, and sending a reset on every failed poll
// would cause continuous full-cache invalidation) PLUS the error itself.
// Callers that cannot tolerate a false "caught up" marker (the SSE Phase-1
// backlog drain) use this variant to stop instead of continuing blind.
func (eb *EventBus) EventsSinceE(ctx context.Context, since uint64) (events []ChangeEvent, headSeq uint64, ok bool, err error) {
	store := eb.store.Load()
	if store == nil {
		return nil, 0, false, nil
	}
	if since == 0 {
		// Initial sync: send reset so client starts fresh.
		headSeq = uint64(clampNonNegative(latestSeqSafeWithMetric(ctx, store, eb.tenantID)))
		return nil, headSeq, false, nil
	}
	queryStart := time.Now()
	rows, qErr := store.ListFSEventsSince(ctx, int64(since), eventPageSize)
	metrics.RecordEventBusQuery(eb.tenantID, "events_since", queryResult(qErr), time.Since(queryStart))
	if qErr != nil {
		// Table missing or query failed: don't send reset on every poll —
		// that would cause continuous full-cache invalidation. Instead, return
		// ok=true with empty events (caught up). The FUSE client's TTL/HEAD
		// revalidation provides correctness without SSE. Only since=0 (initial
		// connection) sends a reset. Log the error and emit a metric counter so
		// operators can detect and alert on persistent table-missing or DB
		// connectivity issues. The error is returned so callers that must not
		// claim "caught up" (Phase-1 drain) can distinguish this branch.
		logger.Warn(ctx, "event_bus_query_failed",
			zap.String("tenant_id", eb.tenantID),
			zap.Uint64("since", since),
			zap.Float64("query_ms", float64(time.Since(queryStart).Microseconds())/1000),
			zap.Error(qErr))
		metrics.RecordTenantOperationWithOrg(eb.tenantID, eb.TiDBCloudOrgID(), "event_bus", "query", metrics.ResultForError(qErr), 0)
		headSeq = since // keep the client's cursor unchanged
		return nil, headSeq, true, qErr
	}
	events = make([]ChangeEvent, 0, len(rows))
	for _, r := range rows {
		events = append(events, ChangeEvent{
			Seq:   r.Seq,
			Path:  r.Path,
			Op:    r.Op,
			Actor: r.Actor,
			Ts:    r.Ts,
		})
	}
	if len(events) > 0 {
		// Got rows: a gap between the client's cursor and the first returned
		// event means one of two things. Either events were pruned at the left
		// edge of the retained window (the client missed real events → reset),
		// or the hole is interior: other tenants' interleaved seqs on a
		// shared-schema fs_events table, or AUTO_INCREMENT ids burned by
		// rolled-back inserts (no events lost → deliver normally). Pruning
		// only happens at the left edge, so compare against the oldest
		// retained seq to tell them apart.
		firstSeq := events[0].Seq
		if since+1 < firstSeq {
			oldest := oldestSeqSafeWithMetric(ctx, store, eb.tenantID)
			if int64(since)+1 < oldest {
				headSeq = events[len(events)-1].Seq
				return nil, headSeq, false, nil // pruned gap → reset
			}
		}
		headSeq = events[len(events)-1].Seq
		return events, headSeq, true, nil
	}
	// No rows after since. Check the actual table state to determine whether
	// the client's cursor is stale (events pruned) or simply caught up.
	latestStart := time.Now()
	latest := latestSeqSafeWithMetric(ctx, store, eb.tenantID)
	latestQueryMs := float64(time.Since(latestStart).Microseconds()) / 1000
	oldestStart := time.Now()
	oldest := oldestSeqSafeWithMetric(ctx, store, eb.tenantID)
	oldestQueryMs := float64(time.Since(oldestStart).Microseconds()) / 1000
	if latest == 0 {
		// Table is empty (all events pruned): cursor is stale → reset.
		return nil, 0, false, nil
	}
	headSeq = uint64(latest)
	if int64(since) > latest {
		// Client cursor is ahead of head (server_restart): reset.
		return nil, headSeq, false, nil
	}
	if int64(since)+1 < oldest {
		// Client cursor is behind the oldest retained event: events between
		// since+1 and oldest-1 were pruned, so the client has a gap → reset
		// to avoid silently missing those events.
		// Log the gap with table bounds so operators can detect table growth
		// without direct DB access (oldest-seq - since delta indicates how
		// much was pruned).
		logger.Info(ctx, "event_bus_cursor_gap",
			zap.String("tenant_id", eb.tenantID),
			zap.Uint64("since", since),
			zap.Int64("oldest", oldest),
			zap.Int64("latest", latest),
			zap.Float64("latest_query_ms", latestQueryMs),
			zap.Float64("oldest_query_ms", oldestQueryMs))
		return nil, headSeq, false, nil
	}
	// Client is caught up: no new events after since, cursor within range.
	return nil, headSeq, true, nil
}

// latestSeqSafeWithMetric is latestSeqSafe plus a per-query duration metric.
func latestSeqSafeWithMetric(ctx context.Context, store *datastore.Store, tenantID string) int64 {
	if store == nil {
		return 0
	}
	start := time.Now()
	seq, err := store.LatestFSEventSeq(ctx)
	metrics.RecordEventBusQuery(tenantID, "latest", queryResult(err), time.Since(start))
	if err != nil {
		return 0
	}
	return seq
}

// oldestSeqSafeWithMetric is oldestSeqSafe plus a per-query duration metric.
func oldestSeqSafeWithMetric(ctx context.Context, store *datastore.Store, tenantID string) int64 {
	if store == nil {
		return 0
	}
	start := time.Now()
	seq, err := store.OldestFSEventSeq(ctx)
	metrics.RecordEventBusQuery(tenantID, "oldest", queryResult(err), time.Since(start))
	if err != nil {
		return 0
	}
	return seq
}

// queryResult maps a query error to a metric result label.
func queryResult(err error) string {
	return metrics.ResultForError(err)
}

// clampNonNegative returns v if v >= 0, else 0.
func clampNonNegative(v int64) int64 {
	if v < 0 {
		return 0
	}
	return v
}