// Package snapshot owns the per-client snapshot collectors that turn live
// torrent state into rows in `torrent_snapshots` (SQLite + DuckDB) and into
// events on the eventbus.
package snapshot

import (
	"context"
	"sync"
	"time"

	"github.com/rs/zerolog"

	"github.com/operator/command-center/internal/db"
	"github.com/operator/command-center/internal/eventbus"
	"github.com/operator/command-center/internal/integrations/qbit"
	"github.com/operator/command-center/internal/observability"
)

// TorrentEvent is the payload broadcast on eventbus.TopicTorrents whenever a
// torrent's state changes or a new torrent appears. The frontend uses these
// to invalidate cached torrent lists in near-real-time.
type TorrentEvent struct {
	ClientID  string  `json:"client_id"`
	InfoHash  string  `json:"info_hash"`
	Kind      string  `json:"kind"` // "added" | "updated" | "removed"
	Name      string  `json:"name,omitempty"`
	State     string  `json:"state,omitempty"`
	Ratio     float64 `json:"ratio,omitempty"`
	Timestamp int64   `json:"timestamp"`
}

// Collector runs one polling goroutine per configured TorrentClient. Each
// goroutine calls qbit.Client.Sync, writes a snapshot row per torrent, and
// publishes TorrentEvent rows for state transitions.
type Collector struct {
	writer *db.SnapshotWriter
	bus    *eventbus.Bus
	events *observability.EventRecorder
	logger zerolog.Logger

	mu      sync.Mutex
	cancels map[string]context.CancelFunc
	wg      sync.WaitGroup
}

// NewCollector constructs a stopped collector.
func NewCollector(writer *db.SnapshotWriter, bus *eventbus.Bus, events *observability.EventRecorder, logger zerolog.Logger) *Collector {
	return &Collector{
		writer:  writer,
		bus:     bus,
		events:  events,
		logger:  logger.With().Str("component", "snapshot").Logger(),
		cancels: map[string]context.CancelFunc{},
	}
}

// Job is one client the collector polls.
type Job struct {
	Client      qbit.TorrentClient
	ClientID    string
	IntervalSec int
}

// Replace stops removed workers, leaves existing workers in place, and
// starts new workers. Called on initial boot and on every
// torrent-clients.yaml reload.
func (c *Collector) Replace(parent context.Context, jobs []Job) {
	c.mu.Lock()
	defer c.mu.Unlock()

	wanted := map[string]Job{}
	for _, j := range jobs {
		wanted[j.ClientID] = j
	}
	for id, cancel := range c.cancels {
		if _, keep := wanted[id]; !keep {
			cancel()
			delete(c.cancels, id)
			c.logger.Info().Str("client", id).Msg("stopped snapshot worker")
		}
	}
	for id, job := range wanted {
		if _, already := c.cancels[id]; already {
			continue
		}
		ctx, cancel := context.WithCancel(parent)
		c.cancels[id] = cancel
		c.wg.Add(1)
		go c.runOne(ctx, job)
	}
}

// Stop terminates all workers.
func (c *Collector) Stop() {
	c.mu.Lock()
	for id, cancel := range c.cancels {
		cancel()
		delete(c.cancels, id)
	}
	c.mu.Unlock()
	c.wg.Wait()
}

func (c *Collector) runOne(ctx context.Context, job Job) {
	defer c.wg.Done()
	logger := c.logger.With().Str("client", job.ClientID).Logger()
	logger.Info().Int("interval_sec", job.IntervalSec).Msg("snapshot worker started")

	interval := time.Duration(job.IntervalSec) * time.Second
	if interval < time.Second {
		interval = 30 * time.Second
	}

	prevState := map[string]string{} // hash -> previous state (for transition detection)
	rid := 0

	for {
		// Cheap initial scrape so the dashboard has data immediately.
		md, err := job.Client.Sync(ctx, rid)
		if err != nil {
			logger.Warn().Err(err).Msg("sync failed")
			if c.events != nil {
				c.events.Record(ctx, observability.LevelError, "snapshot",
					"sync failed",
					map[string]any{"client": job.ClientID, "error": err.Error()}, "")
			}
		} else {
			rid = md.RID
			c.process(ctx, job.ClientID, md, prevState, logger)
		}

		timer := time.NewTimer(interval)
		select {
		case <-timer.C:
		case <-ctx.Done():
			timer.Stop()
			logger.Info().Msg("snapshot worker stopped")
			return
		}
	}
}

func (c *Collector) process(ctx context.Context, clientID string, md qbit.MainData, prevState map[string]string, logger zerolog.Logger) {
	now := time.Now()
	seenThisCycle := map[string]struct{}{}

	for hash, t := range md.Torrents {
		seenThisCycle[hash] = struct{}{}
		// Upsert the torrent row (cheap; the unique constraint on info_hash
		// makes this idempotent).
		_ = c.writer.UpsertTorrent(ctx, hash, t.Name, t.Size, t.Category, t.Tags, "")

		// Write the per-tick snapshot. Pointer values let nil mean "not
		// reported" — qBittorrent always reports the numeric fields, but
		// future client types (rTorrent, Transmission) may not.
		up := t.Uploaded
		dn := t.Downloaded
		ratio := t.Ratio
		seed := int64(t.NumSeeds)
		leech := int64(t.NumLeechs)
		us := t.UpSpeed
		ds := t.DlSpeed
		_ = c.writer.WriteTorrentSnapshot(ctx, db.TorrentSnapshotRow{
			InfoHash:         hash,
			ClientID:         clientID,
			Timestamp:        now,
			UploadedBytes:    &up,
			DownloadedBytes:  &dn,
			State:            t.State,
			Ratio:            &ratio,
			Seeders:          &seed,
			Leechers:         &leech,
			UploadSpeedBps:   &us,
			DownloadSpeedBps: &ds,
		})

		// Transition detection.
		kind := "updated"
		if prev, ok := prevState[hash]; !ok {
			kind = "added"
		} else if prev != t.State {
			kind = "updated"
		}
		prevState[hash] = t.State

		if c.bus != nil {
			c.bus.Publish(eventbus.Event{
				Topic: eventbus.TopicTorrents,
				Payload: TorrentEvent{
					ClientID:  clientID,
					InfoHash:  hash,
					Kind:      kind,
					Name:      t.Name,
					State:     t.State,
					Ratio:     t.Ratio,
					Timestamp: now.Unix(),
				},
			})
		}
	}

	// Handle explicit removals (qBittorrent's sync API tells us).
	for _, hash := range md.TorrentsRemoved {
		delete(prevState, hash)
		if c.bus != nil {
			c.bus.Publish(eventbus.Event{
				Topic: eventbus.TopicTorrents,
				Payload: TorrentEvent{
					ClientID:  clientID,
					InfoHash:  hash,
					Kind:      "removed",
					Timestamp: now.Unix(),
				},
			})
		}
	}

	if md.FullUpdate {
		logger.Debug().Int("torrents", len(md.Torrents)).Msg("full sync")
	}
}