// Package preview reports what a filter would have matched, since a filter is
// silent about what it fails to match: a subscription that will never fire and
// one that was never going to fire look identical in the form.
//
// Two properties carry the weight. The corpus follows the mode (see corpusFor)
// — counting the wrong table is worse than counting nothing. And matching is
// the production matcher: SQL narrows only on exactly-compared indexed columns,
// then every row goes through SubscriptionFilter.Matches with a target from
// subscription.TargetFromAlert/TargetFromEvent, the same call the fan-out and
// the mute check make.
package preview

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/jmoiron/sqlx"

	"github.com/tidbcloud/alertstore/pkg/alert"
	"github.com/tidbcloud/alertstore/pkg/event"
	"github.com/tidbcloud/alertstore/pkg/subscription"
)

// ModeMute previews a mute rule. Not a subscription mode — a rule has none —
// but it is the same filter type asked the same question.
const ModeMute = "mute"

// The two histories a filter can be run over. Reported in the response: which
// one was used decides what the number means.
const (
	CorpusAlert = "alert"
	CorpusEvent = "event"
)

// ScanLimit bounds one preview at the most recent 1000 rows. The worst case is
// not exotic — a new form starts empty, and an empty filter matches everything.
// Scanning newest-first is what makes truncation explicable: recent history
// stays exact and older history becomes a lower bound, per window.
const ScanLimit = 1000

// HistogramDays is the daily breakdown's span and the minimum scan window. The
// week's shape separates a one-off burst from steady noise.
const HistogramDays = 7

// MaxWindow caps how far back a caller may ask; longer is a reporting question,
// not a "what am I about to save" one.
const MaxWindow = 30 * 24 * time.Hour

type Store struct {
	db *sqlx.DB
}

func NewStore(db *sqlx.DB) *Store {
	return &Store{db: db}
}

type Request struct {
	Mode    string
	Filter  subscription.SubscriptionFilter
	Windows []Window
}

// Window carries the caller's own spelling of the span, so the response needs
// no label re-derived from a duration.
type Window struct {
	Label string
	Span  time.Duration
}

type Result struct {
	// Corpus names what was counted — alerts, or raw observations — which the
	// number alone cannot say.
	Corpus  string        `json:"corpus" validate:"required"`
	Windows []WindowCount `json:"windows" validate:"required"`
	Daily   []DayCount    `json:"daily" validate:"required"`
	// TopAlertnames catches what counts cannot: 300 matches where 290 are the
	// intended alertname and 10 are not has a perfectly reasonable total.
	TopAlertnames []NameCount `json:"top_alertnames" validate:"required"`
	Latest        []Sample    `json:"latest" validate:"required"`
	// MatchesEverything marks a filter naming no field, which for a mute rule
	// means the fleet.
	MatchesEverything bool `json:"matches_everything" validate:"required"`
	// Truncated says ScanLimit was reached, so counts reaching past
	// ScannedSince are lower bounds.
	Scanned      int       `json:"scanned" validate:"required"`
	Truncated    bool      `json:"truncated" validate:"required"`
	ScannedSince time.Time `json:"scanned_since" validate:"required"`
}

type WindowCount struct {
	Window string `json:"window" validate:"required"`
	Count  int    `json:"count" validate:"required"`
	// Exact is false when the scan stopped short of this window's start,
	// making Count a floor. Reporting a floor as a total is the failure the
	// flag exists to prevent.
	Exact bool `json:"exact" validate:"required"`
}

type DayCount struct {
	Date  string `json:"date" validate:"required"` // YYYY-MM-DD, UTC
	Count int    `json:"count" validate:"required"`
	Exact bool   `json:"exact" validate:"required"`
}

type NameCount struct {
	Alertname string `json:"alertname" validate:"required"`
	Count     int    `json:"count" validate:"required"`
}

// Sample is one matched row, carrying enough to list and link it without a
// second request.
type Sample struct {
	ID             string    `json:"id" validate:"required"`
	Alertname      string    `json:"alertname" validate:"required"`
	ClusterID      string    `json:"cluster_id,omitempty"`
	Classification string    `json:"classification,omitempty"`
	State          string    `json:"state,omitempty"`
	At             time.Time `json:"at" validate:"required"`
}

// How many matched rows and ranked alertnames come back with the counts.
const (
	SampleCount       = 3
	TopAlertnameCount = 5
)

// corpusFor maps a mode to the history its filter is matched against. The
// alert modes deliver alerts; the event modes deliver one card per observation,
// split by visibility (see eligible).
//
// Mute is the one that looks wrong and is not. Its check sits after the logging
// short-circuit and before the alert is created (webhook.Handler.route), so an
// alert is all a rule can suppress — counting observations would add resolved
// notifications and logging traffic no rule could touch, several times the
// pages actually silenced.
func corpusFor(mode string) (string, error) {
	switch mode {
	case subscription.ModeTopicForward, subscription.ModeCardOnly, ModeMute:
		return CorpusAlert, nil
	case subscription.ModeEvent, subscription.ModeCustomer:
		return CorpusEvent, nil
	default:
		return "", fmt.Errorf("unsupported preview mode %q", mode)
	}
}

// Run reports what the filter would have matched over its mode's corpus.
func (s *Store) Run(ctx context.Context, req Request) (*Result, error) {
	corpus, err := corpusFor(req.Mode)
	if err != nil {
		return nil, err
	}
	for _, w := range req.Windows {
		if w.Span <= 0 {
			return nil, fmt.Errorf("window %s is not a positive duration", w.Label)
		}
		if w.Span > MaxWindow {
			return nil, fmt.Errorf("window %s is longer than the %s preview limit", w.Label, MaxWindow)
		}
	}

	now := time.Now()
	since := now.Add(-scanSpan(req.Windows))

	var candidates []candidate
	if corpus == CorpusAlert {
		candidates, err = s.scanAlerts(ctx, req.Filter, since)
	} else {
		candidates, err = s.scanEvents(ctx, req.Filter, since)
	}
	if err != nil {
		return nil, err
	}
	return aggregate(req, corpus, candidates, since, now), nil
}

// scanSpan reads back the longest window asked for, but never less than the
// histogram's span, which is always drawn.
func scanSpan(windows []Window) time.Duration {
	span := HistogramDays * 24 * time.Hour
	for _, w := range windows {
		if w.Span > span {
			span = w.Span
		}
	}
	return span
}

// candidate is one row of either corpus, reduced to what matching needs.
type candidate struct {
	id     string
	at     time.Time
	state  string
	target subscription.Target
	// The alert corpus's discriminator and the event corpus's. Two fields
	// rather than one reused bool: they mean opposite things — an unowned
	// alert reaches nobody, an external observation reaches a different mode.
	owned    bool
	external bool
}

func (s *Store) scanAlerts(ctx context.Context, f subscription.SubscriptionFilter, since time.Time) ([]candidate, error) {
	where, args := pushdown(f)
	q := `SELECT id, created_at, state, attrs FROM alert WHERE created_at >= ?` + where +
		` ORDER BY created_at DESC LIMIT ?`
	args = append([]any{since}, args...)
	args = append(args, ScanLimit)

	var rows []struct {
		ID        string          `db:"id"`
		CreatedAt time.Time       `db:"created_at"`
		State     string          `db:"state"`
		Attrs     json.RawMessage `db:"attrs"`
	}
	if err := s.db.SelectContext(ctx, &rows, q, args...); err != nil {
		return nil, fmt.Errorf("scan alerts for preview: %w", err)
	}

	out := make([]candidate, 0, len(rows))
	for _, r := range rows {
		var attrs alert.AlertAttrs
		// Skipped, not fatal: one bad row should not deny an answer about the
		// other 999, and the matcher skips unreadable filters for the same
		// reason.
		if err := json.Unmarshal(r.Attrs, &attrs); err != nil {
			continue
		}
		out = append(out, candidate{
			id:     r.ID,
			at:     r.CreatedAt,
			state:  r.State,
			target: subscription.TargetFromAlert(&attrs),
			owned:  attrs.Owned,
		})
	}
	return out, nil
}

func (s *Store) scanEvents(ctx context.Context, f subscription.SubscriptionFilter, since time.Time) ([]candidate, error) {
	where, args := pushdown(f)
	q := `SELECT id, received_at, status, observation_type, attrs FROM event WHERE received_at >= ?` + where +
		` ORDER BY received_at DESC LIMIT ?`
	args = append([]any{since}, args...)
	args = append(args, ScanLimit)

	var rows []struct {
		ID              int64           `db:"id"`
		ReceivedAt      time.Time       `db:"received_at"`
		Status          string          `db:"status"`
		ObservationType string          `db:"observation_type"`
		Attrs           json.RawMessage `db:"attrs"`
	}
	if err := s.db.SelectContext(ctx, &rows, q, args...); err != nil {
		return nil, fmt.Errorf("scan events for preview: %w", err)
	}

	out := make([]candidate, 0, len(rows))
	for _, r := range rows {
		var attrs event.EventAttrs
		if err := json.Unmarshal(r.Attrs, &attrs); err != nil {
			continue
		}
		target := subscription.TargetFromEvent(&attrs)
		target.ObservationType = r.ObservationType
		out = append(out, candidate{
			id:       fmt.Sprint(r.ID),
			at:       r.ReceivedAt,
			state:    r.Status,
			target:   target,
			external: attrs.Visibility == alert.VisibilityExternal,
		})
	}
	return out, nil
}

// eligible reports whether a row is one this mode could have delivered: the
// corpus is the table, this is the part of it the mode reaches.
func eligible(mode string, c candidate) bool {
	switch mode {
	case subscription.ModeTopicForward, subscription.ModeCardOnly:
		// Shadow alerts stop before the fan-out (delivery.Worker), so counting
		// them would over-report.
		return c.owned
	case subscription.ModeCustomer:
		return c.external
	case subscription.ModeEvent:
		return !c.external
	case ModeMute:
		// Checked before ownership is decided, so it reaches every alert.
		return true
	}
	return false
}

// pushdown narrows the scan in SQL, and only where SQL and the matcher agree.
// Deliberately partial: these five are indexed generated columns compared
// exactly, so `IN` means what Matches means. The rest stays in Go — biz_region,
// tenant_kind and customer_type are matched case-insensitively and would depend
// on collation here, while component and labels have no column at all.
// Narrowing further would be faster and would eventually disagree.
func pushdown(f subscription.SubscriptionFilter) (string, []any) {
	var sb strings.Builder
	var args []any
	add := func(column string, m *subscription.FieldMatch) {
		// Only the plain equal operator is an `IN`. Negation and the regex pair
		// have no pushdown that stays a subset of Matches, so those fields stay
		// wholly in Go rather than narrowing here.
		if m == nil || m.Op != subscription.OpEq || len(m.Values) == 0 {
			return
		}
		sb.WriteString(" AND " + column + " IN (?" + strings.Repeat(",?", len(m.Values)-1) + ")")
		for _, v := range m.Values {
			args = append(args, v)
		}
	}
	add("region", f.Region)
	add("classification", f.Classification)
	add("alertname", f.Alertname)
	add("cluster_id", f.ClusterID)
	add("tenant_id", f.TenantID)
	add("observation_type", f.ObservationType)
	return sb.String(), args
}

// aggregate turns the scanned rows into the answer. Pure and separate from the
// scan, so the counting rules are testable without a database.
func aggregate(req Request, corpus string, candidates []candidate, since, now time.Time) *Result {
	truncated := len(candidates) >= ScanLimit
	res := &Result{
		Corpus:            corpus,
		MatchesEverything: isEmpty(req.Filter),
		Scanned:           len(candidates),
		Truncated:         truncated,
		ScannedSince:      since,
	}

	// A truncated scan holds the newest rows, so a span starting after the
	// oldest row it reached is still complete.
	oldest := since
	if truncated && len(candidates) > 0 {
		oldest = candidates[len(candidates)-1].at
	}
	exact := func(start time.Time) bool { return !truncated || !start.Before(oldest) }

	counts := make([]int, len(req.Windows))
	starts := make([]time.Time, len(req.Windows))
	for i, w := range req.Windows {
		starts[i] = now.Add(-w.Span)
	}

	days := make([]DayCount, HistogramDays)
	today := now.UTC().Truncate(24 * time.Hour)
	dayIndex := make(map[string]int, HistogramDays)
	for i := range days {
		date := today.AddDate(0, 0, i-(HistogramDays-1))
		days[i] = DayCount{Date: date.Format("2006-01-02"), Exact: exact(date)}
		dayIndex[days[i].Date] = i
	}

	byName := map[string]int{}
	for _, c := range candidates {
		if !eligible(req.Mode, c) || !req.Filter.Matches(c.target) {
			continue
		}
		for i := range req.Windows {
			if !c.at.Before(starts[i]) {
				counts[i]++
			}
		}
		if i, ok := dayIndex[c.at.UTC().Format("2006-01-02")]; ok {
			days[i].Count++
		}
		byName[c.target.Alertname]++
		// Newest-first, so the first matches seen are the most recent.
		if len(res.Latest) < SampleCount {
			res.Latest = append(res.Latest, Sample{
				ID:             c.id,
				Alertname:      c.target.Alertname,
				ClusterID:      c.target.ClusterID,
				Classification: c.target.Classification,
				State:          c.state,
				At:             c.at,
			})
		}
	}

	res.Windows = make([]WindowCount, len(req.Windows))
	for i, w := range req.Windows {
		res.Windows[i] = WindowCount{Window: w.Label, Count: counts[i], Exact: exact(starts[i])}
	}
	res.Daily = days
	res.TopAlertnames = topNames(byName)
	return res
}

// topNames ranks by count, then name, so equal counts do not reorder between
// two previews of one filter.
func topNames(byName map[string]int) []NameCount {
	out := make([]NameCount, 0, len(byName))
	for name, count := range byName {
		out = append(out, NameCount{Alertname: name, Count: count})
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Count != out[j].Count {
			return out[i].Count > out[j].Count
		}
		return out[i].Alertname < out[j].Alertname
	})
	if len(out) > TopAlertnameCount {
		out = out[:TopAlertnameCount]
	}
	return out
}

// isEmpty reports a filter that names nothing and so matches everything — said
// out loud rather than left to be inferred from a large count.
func isEmpty(f subscription.SubscriptionFilter) bool {
	return f.IsEmpty()
}