// Package performance is the Phase 5 filter→upload correlation engine.
//
// Linker subscribes to eventbus.TopicAutobrrGrab and writes a row to
// filter_performance for every grab. Rows arrive with a known info_hash
// (from autobrr's payload) or unlinked (info_hash NULL). The Scorer
// periodically attempts to link unlinked rows via fuzzy name matching
// against newly-seen torrents AND updates final_uploaded_bytes /
// final_ratio from the latest torrent_snapshots.
package performance

import (
	"context"
	"database/sql"
	"errors"
	"strings"
	"sync"
	"time"

	"github.com/rs/zerolog"

	"github.com/operator/command-center/internal/eventbus"
	"github.com/operator/command-center/internal/webhooks"
)

// Linker writes filter_performance rows in response to bus events.
type Linker struct {
	db        *sql.DB
	bus       *eventbus.Bus
	logger    zerolog.Logger

	mu     sync.Mutex
	stop   context.CancelFunc
	wg     sync.WaitGroup
}

// NewLinker constructs a Linker bound to the given db + bus.
func NewLinker(db *sql.DB, bus *eventbus.Bus, logger zerolog.Logger) *Linker {
	return &Linker{
		db:     db,
		bus:    bus,
		logger: logger.With().Str("component", "performance-linker").Logger(),
	}
}

// Start subscribes to TopicAutobrrGrab and processes events until parent ctx
// is canceled. Returns immediately.
func (l *Linker) Start(parent context.Context) {
	ctx, cancel := context.WithCancel(parent)
	l.mu.Lock()
	l.stop = cancel
	l.mu.Unlock()
	l.wg.Add(1)
	go l.run(ctx)
}

// Stop terminates the worker. Safe to call multiple times.
func (l *Linker) Stop() {
	l.mu.Lock()
	if l.stop != nil {
		l.stop()
	}
	l.mu.Unlock()
	l.wg.Wait()
}

func (l *Linker) run(ctx context.Context) {
	defer l.wg.Done()
	sub := l.bus.Subscribe(eventbus.TopicAutobrrGrab)
	defer sub.Close()

	for {
		select {
		case <-ctx.Done():
			return
		case ev, ok := <-sub.Chan():
			if !ok {
				return
			}
			grab, ok := ev.Payload.(webhooks.AutobrrGrabEvent)
			if !ok {
				l.logger.Warn().Msg("non-AutobrrGrabEvent payload on TopicAutobrrGrab")
				continue
			}
			if err := l.recordGrab(ctx, grab); err != nil {
				l.logger.Error().Err(err).
					Str("filter", grab.FilterID).
					Str("release", grab.ReleaseName).
					Msg("failed to record grab")
			}
		}
	}
}

// RecordGrab writes one filter_performance row. Exported so the reconciler
// (Phase 4 reconcile loop) can replay missed grabs through the same path.
func (l *Linker) RecordGrab(ctx context.Context, grab webhooks.AutobrrGrabEvent) error {
	return l.recordGrab(ctx, grab)
}

func (l *Linker) recordGrab(ctx context.Context, grab webhooks.AutobrrGrabEvent) error {
	if grab.FilterID == "" {
		return errors.New("performance: empty filter_id")
	}
	// The webhook payload's EndpointID maps 1:1 to an automation_tools.id
	// when the operator wires them together (see Phase 5 brief). For Phase 5
	// we accept the convention that the endpoint id matches the tool id.
	toolID := grab.EndpointID
	if toolID == "" {
		toolID = "autobrr"
	}

	// Idempotency: skip if we've already recorded this grab. autobrr's
	// payload includes a stable event_id; fall back to (filter_id +
	// release_name + grabbed_at within 60s) when it's empty.
	var existing int
	if grab.EventID != "" {
		if err := l.db.QueryRowContext(ctx, `
			SELECT COUNT(*) FROM filter_performance
			WHERE filter_external_id = ? AND release_name = ?
			  AND grabbed_at = ?
		`, grab.FilterID, grab.ReleaseName, grab.GrabbedAt.Unix()).Scan(&existing); err == nil && existing > 0 {
			return nil
		}
	}

	_, err := l.db.ExecContext(ctx, `
		INSERT INTO filter_performance
		  (automation_tool_id, filter_external_id, filter_name,
		   info_hash, release_name, grabbed_at)
		VALUES (?, ?, ?, ?, ?, ?)
	`,
		toolID, grab.FilterID, grab.FilterName,
		nullString(grab.InfoHash), nullString(grab.ReleaseName), grab.GrabbedAt.Unix(),
	)
	return err
}

func nullString(s string) any {
	if strings.TrimSpace(s) == "" {
		return nil
	}
	return s
}

// linkWindow is the maximum lag between a grab's grabbed_at and the moment a
// torrent must appear in the `torrents` table for the fuzzy linker to claim
// the row. Beyond this, the row stays unlinked forever (autobrr grabbed
// something the torrent client never actually got).
const linkWindow = 1 * time.Hour