lingniu-vehicle-ingest/vehicle-data-platform/apps/api/internal/platform/alert_test.go

package platform

import (
	"bytes"
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"regexp"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
)

type configuredMetricStore struct {
	*MockStore
	definitions []MetricDefinition
}

func (s *configuredMetricStore) MetricDefinitions(context.Context) ([]MetricDefinition, error) {
	return append([]MetricDefinition(nil), s.definitions...), nil
}

func BenchmarkAlertEvaluationPlan10000x20(b *testing.B) {
	evidence := make([]alertEvaluationEvidence, 10_000)
	for index := range evidence {
		evidence[index] = alertEvaluationEvidence{VIN: "VIN" + strconv.Itoa(index), Protocol: "JT808", SpeedKmh: float64(index % 120), FreshnessSec: 5}
	}
	rules := make([]AlertRule, 20)
	for index := range rules {
		rules[index] = AlertRule{ID: "rule" + strconv.Itoa(index), Enabled: true, Metric: "speed_kmh", Operator: "gt", Threshold: float64(60 + index)}
	}
	now := time.Now()
	b.ResetTimer()
	for iteration := 0; iteration < b.N; iteration++ {
		upserts := make([]alertCandidateUpsert, 0, len(evidence))
		for _, rule := range rules {
			for _, item := range evidence {
				value, ok := alertMetricValue(rule.Metric, item)
				if ok && alertRuleInScope(rule, item) && compareAlertValue(value, rule.Operator, rule.Threshold) {
					upserts = append(upserts, alertCandidateUpsert{RuleID: rule.ID, VIN: item.VIN, Protocol: item.Protocol, FirstMatchedAt: now, LastMatchedAt: now, LatestValue: value})
				}
			}
		}
		for start := 0; start < len(upserts); start += alertCandidateBatchSize {
			end := min(start+alertCandidateBatchSize, len(upserts))
			_, args := buildAlertCandidateUpsert(upserts[start:end])
			if len(args) == 0 {
				b.Fatal("expected batched candidates")
			}
		}
	}
}

func TestAlertEventWorkflowIsVersionedAndAudited(t *testing.T) {
	handler := NewHandler(NewService(NewMockStore()))
	list := httptest.NewRecorder()
	handler.ServeHTTP(list, httptest.NewRequest(http.MethodPost, "/api/v2/alerts/events", bytes.NewBufferString(`{"status":"unprocessed","limit":20}`)))
	if list.Code != http.StatusOK {
		t.Fatalf("list status=%d body=%s", list.Code, list.Body.String())
	}
	var body struct {
		Data Page[AlertEvent] `json:"data"`
	}
	if err := json.Unmarshal(list.Body.Bytes(), &body); err != nil {
		t.Fatal(err)
	}
	if body.Data.Total < 2 {
		t.Fatalf("expected seeded unprocessed events, got %+v", body.Data)
	}
	event := body.Data.Items[0]
	action := func(version int) *httptest.ResponseRecorder {
		rec := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodPost, "/api/v2/alerts/events/"+event.ID+"/actions", bytes.NewBufferString(`{"version":`+itoa(version)+`,"action":"acknowledge","note":"已联系司机","actor":"spoofed-body-user"}`))
		req.Header.Set("X-User-Name", "spoofed-header-user")
		req = req.WithContext(WithPrincipal(req.Context(), Principal{Name: "operator-a", Role: "operator"}))
		handler.ServeHTTP(rec, req)
		return rec
	}
	rec := action(event.Version)
	if rec.Code != http.StatusOK {
		t.Fatalf("action status=%d body=%s", rec.Code, rec.Body.String())
	}
	var updated struct {
		Data AlertEvent `json:"data"`
	}
	_ = json.Unmarshal(rec.Body.Bytes(), &updated)
	if updated.Data.Status != "processing" || updated.Data.Handler != "operator-a" || len(updated.Data.Actions) < 2 {
		t.Fatalf("workflow not persisted: %+v", updated.Data)
	}
	stale := action(event.Version)
	if stale.Code != http.StatusConflict {
		t.Fatalf("stale update should conflict, status=%d body=%s", stale.Code, stale.Body.String())
	}
}

func TestAlertEventsActiveStatusIncludesOnlyOpenWorkflowStates(t *testing.T) {
	service := NewService(NewMockStore())
	page, err := service.AlertEvents(t.Context(), AlertQuery{Status: "active", Limit: 200})
	if err != nil {
		t.Fatal(err)
	}
	if page.Total == 0 {
		t.Fatal("expected seeded active alerts")
	}
	for _, event := range page.Items {
		if event.Status != "unprocessed" && event.Status != "processing" {
			t.Fatalf("active query returned terminal state: %+v", event)
		}
	}
	if where, args := buildAlertWhere(AlertQuery{Status: "active"}); !strings.Contains(where, "status IN ('unprocessed','processing')") || len(args) != 0 {
		t.Fatalf("active SQL filter is not authoritative: where=%s args=%v", where, args)
	}
}

func TestAlertRuleCreationNormalizesBooleanAndChannels(t *testing.T) {
	service := NewService(NewMockStore())
	truth := true
	rule, err := service.SaveAlertRule(t.Context(), AlertRuleInput{Name: "主电源异常", Severity: "major", ValueType: "boolean", Metric: "alarm_active", Operator: "eq", BooleanThreshold: &truth, ScopeModels: []string{"纯电客车", "纯电客车"}, ScopeCompanies: []string{"示范公交"}, NotificationChannels: []string{"sms", "sms"}, Enabled: true})
	if err != nil {
		t.Fatal(err)
	}
	if rule.Threshold != 1 {
		t.Fatalf("boolean threshold should normalize to numeric evaluator value: %+v", rule)
	}
	if len(rule.NotificationChannels) != 2 || rule.NotificationChannels[0] != "in_app" {
		t.Fatalf("channels should include unique in-app truth: %+v", rule.NotificationChannels)
	}
	if len(rule.ScopeModels) != 1 || rule.ScopeModels[0] != "纯电客车" || len(rule.ScopeCompanies) != 1 {
		t.Fatalf("master-data scopes should normalize and round-trip: %+v", rule)
	}
	_, err = service.SaveAlertRule(t.Context(), AlertRuleInput{ID: rule.ID, Version: 99, Name: rule.Name, Severity: rule.Severity, ValueType: rule.ValueType, Metric: rule.Metric, Operator: rule.Operator, BooleanThreshold: &truth})
	clientErr, ok := asClientError(err)
	if !ok || clientErr.Code != "ALERT_RULE_VERSION_CONFLICT" {
		t.Fatalf("expected optimistic conflict, got %v", err)
	}
}

func TestAlertNotificationsReadUpdatesUnreadSummary(t *testing.T) {
	service := NewService(NewMockStore())
	before, err := service.AlertSummary(t.Context(), AlertQuery{})
	if err != nil {
		t.Fatal(err)
	}
	page, err := service.AlertNotifications(t.Context(), AlertNotificationQuery{UnreadOnly: true, Limit: 20})
	if err != nil {
		t.Fatal(err)
	}
	if page.Total != before.UnreadNotifications || page.Total == 0 {
		t.Fatalf("unread truth differs: summary=%+v page=%+v", before, page)
	}
	count, err := service.MarkAlertNotificationsRead(t.Context(), AlertNotificationReadRequest{IDs: []int64{page.Items[0].ID}, Actor: "operator-a"})
	if err != nil || count != 1 {
		t.Fatalf("mark read count=%d err=%v", count, err)
	}
	after, _ := service.AlertSummary(t.Context(), AlertQuery{})
	if after.UnreadNotifications != before.UnreadNotifications-1 {
		t.Fatalf("unread should recalculate: before=%d after=%d", before.UnreadNotifications, after.UnreadNotifications)
	}
}

func TestAlertEvaluatorComparisonsAndScope(t *testing.T) {
	item := alertEvaluationEvidence{VIN: "VIN1", Protocol: "JT808", OEM: "宇通", Model: "纯电客车", Company: "示范公交", SpeedKmh: 96, AlarmFlag: 1}
	rule := AlertRule{Metric: "speed_kmh", Operator: "gt", Threshold: 80, ScopeProtocols: []string{"JT808"}, ScopeVINs: []string{"VIN1"}}
	value, ok := alertMetricValue(rule.Metric, item)
	if !ok || !compareAlertValue(value, rule.Operator, rule.Threshold) || !alertRuleInScope(rule, item) {
		t.Fatalf("numeric rule should match current evidence")
	}
	if compareAlertValue(75, "gt", 80) || !compareAlertValue(75, "lte", 75) {
		t.Fatal("comparison boundary incorrect")
	}
	rule.ScopeVINs = []string{"OTHER"}
	if alertRuleInScope(rule, item) {
		t.Fatal("VIN scope must prevent cross-vehicle evaluation")
	}
	rule.ScopeVINs = nil
	rule.ScopeOEMs = []string{"其他厂家"}
	if alertRuleInScope(rule, item) {
		t.Fatal("OEM scope must prevent cross-manufacturer evaluation")
	}
	rule.ScopeOEMs = nil
	rule.ScopeModels = []string{"氢燃料重卡"}
	if alertRuleInScope(rule, item) {
		t.Fatal("model scope must use authoritative vehicle profile")
	}
	rule.ScopeModels = []string{"纯电客车"}
	rule.ScopeCompanies = []string{"其他企业"}
	if alertRuleInScope(rule, item) {
		t.Fatal("company scope must use authoritative vehicle profile")
	}
	if !compareAlertRuleValue(80, AlertRule{Operator: "between", Threshold: 70, ThresholdHigh: 90}) || compareAlertRuleValue(95, AlertRule{Operator: "between", Threshold: 70, ThresholdHigh: 90}) {
		t.Fatal("between comparison boundary incorrect")
	}
	if !compareAlertRuleValue(95, AlertRule{Operator: "outside", Threshold: 70, ThresholdHigh: 90}) || compareAlertRuleValue(80, AlertRule{Operator: "outside", Threshold: 70, ThresholdHigh: 90}) {
		t.Fatal("outside comparison boundary incorrect")
	}
}

func TestAlertRuleSQLMatchesMasterDataScopeArguments(t *testing.T) {
	if placeholders := strings.Count(alertRuleInsertSQL, "?"); placeholders != 23 {
		t.Fatalf("alert insert placeholders=%d query=%s", placeholders, alertRuleInsertSQL)
	}
	if placeholders := strings.Count(alertRuleUpdateSQL, "?"); placeholders != 23 {
		t.Fatalf("alert update placeholders=%d query=%s", placeholders, alertRuleUpdateSQL)
	}
}

func TestAlertRuleScopeBoundsProtectEvaluator(t *testing.T) {
	values := make([]string, 501)
	for index := range values {
		values[index] = "company-" + strconv.Itoa(index)
	}
	err := validateAlertRuleScopes(AlertRuleInput{ScopeCompanies: values})
	clientErr, ok := asClientError(err)
	if !ok || clientErr.Code != "ALERT_RULE_SCOPE_TOO_LARGE" {
		t.Fatalf("large scope should be rejected, got %v", err)
	}
}

func TestAlertCandidateUpsertBatchesRows(t *testing.T) {
	now := time.Now()
	query, args := buildAlertCandidateUpsert([]alertCandidateUpsert{
		{RuleID: "r1", VIN: "v1", Protocol: "JT808", FirstMatchedAt: now, LastMatchedAt: now, LatestValue: 91, SourceEventID: "e1"},
		{RuleID: "r1", VIN: "v2", Protocol: "GB32960", FirstMatchedAt: now, LastMatchedAt: now, LatestValue: 92, SourceEventID: "e2"},
	})
	if strings.Count(query, "(?,?,?,?,?,?,?)") != 2 || len(args) != 14 || !strings.Contains(query, "ON DUPLICATE KEY UPDATE") {
		t.Fatalf("unexpected batch query=%s args=%#v", query, args)
	}
}

func TestAlertStateLoadersPreserveMySQLDatetimeMilliseconds(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(t.Context(), nil)
	if err != nil {
		t.Fatal(err)
	}
	first := time.Date(2026, 7, 14, 10, 0, 0, 123000000, time.Local)
	last := first.Add(20*time.Second + 333*time.Millisecond)
	candidateSQL := `SELECT c.rule_id,c.vin,c.protocol,c.first_matched_at,c.last_matched_at,c.source_event_id FROM vehicle_alert_candidate c JOIN vehicle_alert_rule r ON r.id=c.rule_id AND r.enabled=1`
	mock.ExpectQuery(regexp.QuoteMeta(candidateSQL)).WillReturnRows(sqlmock.NewRows([]string{"rule_id", "vin", "protocol", "first_matched_at", "last_matched_at", "source_event_id"}).AddRow("r1", "v1", "JT808", first, last, "event-2"))
	candidates, err := loadAlertCandidates(t.Context(), tx)
	if err != nil {
		t.Fatal(err)
	}
	candidate := candidates["r1|v1|JT808"]
	if !candidate.FirstMatchedAt.Equal(first) || !candidate.LastMatchedAt.Equal(last) || candidate.SourceEventID != "event-2" {
		t.Fatalf("candidate datetime precision lost: %+v", candidate)
	}

	stateSQL := `SELECT s.rule_id,s.vin,s.protocol,s.observed_value,s.last_observed_at FROM vehicle_alert_rule_state s JOIN vehicle_alert_rule r ON r.id=s.rule_id AND r.enabled=1`
	mock.ExpectQuery(regexp.QuoteMeta(stateSQL)).WillReturnRows(sqlmock.NewRows([]string{"rule_id", "vin", "protocol", "observed_value", "last_observed_at"}).AddRow("r2", "v2", "GB32960", 1.0, last))
	states, err := loadAlertRuleStates(t.Context(), tx)
	if err != nil {
		t.Fatal(err)
	}
	state := states["r2|v2|GB32960"]
	if state.LastValue != 1 || !state.LastObservedAt.Equal(last) {
		t.Fatalf("rule state datetime precision lost: %+v", state)
	}

	lastTriggeredSQL := `SELECT e.fingerprint,MAX(e.triggered_at) FROM vehicle_alert_event e JOIN vehicle_alert_rule r ON r.id=e.rule_id AND r.enabled=1 GROUP BY e.fingerprint`
	mock.ExpectQuery(regexp.QuoteMeta(lastTriggeredSQL)).WillReturnRows(sqlmock.NewRows([]string{"fingerprint", "triggered_at"}).AddRow("r1|v1|JT808", last))
	lastTriggered, err := loadAlertLastTriggered(t.Context(), tx)
	if err != nil {
		t.Fatal(err)
	}
	if !lastTriggered["r1|v1|JT808"].Equal(last) {
		t.Fatalf("repeat timestamp precision lost: %s", lastTriggered["r1|v1|JT808"])
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestAlertRuleDisableIsAtomicAndClearsEvaluatorState(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	store := NewProductionStore(db, nil, "")
	mock.ExpectQuery(regexp.QuoteMeta(`SELECT COUNT(*) FROM vehicle_alert_rule`)).WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(1))
	mock.ExpectBegin()
	mock.ExpectExec(regexp.QuoteMeta(`UPDATE vehicle_alert_rule SET enabled=?,version=version+1,updated_by=? WHERE id=? AND version=?`)).WithArgs(false, "admin-a", "rule-1", 1).WillReturnResult(sqlmock.NewResult(0, 1))
	columns := []string{"id", "name", "description", "severity", "value_type", "metric", "operator", "threshold_value", "threshold_high", "boolean_threshold", "duration_sec", "recovery_operator", "recovery_threshold", "repeat_interval_sec", "scope_protocols_json", "scope_vins_json", "scope_oems_json", "scope_models_json", "scope_companies_json", "notification_channels_json", "enabled", "version", "created_by", "updated_by", "created_at", "updated_at"}
	mock.ExpectQuery(regexp.QuoteMeta(alertRuleSelect + `WHERE id=?`)).WithArgs("rule-1").WillReturnRows(sqlmock.NewRows(columns).AddRow("rule-1", "超速", "", "minor", "numeric", "speed_kmh", "gte", 0.0, 0.0, nil, 20, "", 0.0, 3600, `["JT808"]`, `["VIN1"]`, `[]`, `[]`, `[]`, `["in_app"]`, false, 2, "admin-a", "admin-a", "2026-07-14T07:00:00.000000Z", "2026-07-14T07:01:00.000000Z"))
	mock.ExpectExec(regexp.QuoteMeta(`DELETE FROM vehicle_alert_candidate WHERE rule_id=?`)).WithArgs("rule-1").WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(regexp.QuoteMeta(`DELETE FROM vehicle_alert_rule_state WHERE rule_id=?`)).WithArgs("rule-1").WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO vehicle_alert_rule_audit(rule_id,rule_version,actor,action,snapshot_json) VALUES(?,?,?,?,?)`)).WithArgs("rule-1", 2, "admin-a", "disable", sqlmock.AnyArg()).WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectCommit()
	rule, err := store.SetAlertRuleEnabled(t.Context(), "rule-1", AlertRuleEnabledUpdate{Version: 1, Enabled: false, Actor: "admin-a"})
	if err != nil {
		t.Fatal(err)
	}
	if rule.Enabled || rule.Version != 2 {
		t.Fatalf("unexpected disabled rule: %+v", rule)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestAlertEvaluationLocksRuleAgainstConcurrentDisable(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(t.Context(), nil)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta(`SELECT enabled FROM vehicle_alert_rule WHERE id=? FOR UPDATE`)).WithArgs("rule-1").WillReturnRows(sqlmock.NewRows([]string{"enabled"}).AddRow(true))
	enabled, err := lockAlertRuleForEvaluation(t.Context(), tx, "rule-1")
	if err != nil || !enabled {
		t.Fatalf("rule lock enabled=%t err=%v", enabled, err)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestAlertRuleStateUpsertBatchesChangedValues(t *testing.T) {
	now := time.Now()
	query, args := buildAlertRuleStateUpsert([]alertRuleStateUpsert{{RuleID: "r1", VIN: "v1", Protocol: "JT808", LastValue: 1, ObservedAt: now}, {RuleID: "r1", VIN: "v2", Protocol: "GB32960", LastValue: 0, ObservedAt: now}})
	if strings.Count(query, "(?,?,?,?,?)") != 2 || len(args) != 10 || !strings.Contains(query, "ON DUPLICATE KEY UPDATE") {
		t.Fatalf("unexpected state batch query=%s args=%#v", query, args)
	}
}

func TestAlertEventInsertContractMatchesArguments(t *testing.T) {
	rule := AlertRule{ID: "rule-1", Name: "超速", Version: 3, Severity: "major", Metric: "speed_kmh", Operator: "between", Threshold: 60, ThresholdHigh: 90, DurationSec: 30}
	item := alertEvaluationEvidence{VIN: "VIN1", Plate: "川A10001", Protocol: "JT808", SourceEventID: "event-1", EventAt: "2026-07-14 10:00:00.000", ReceivedAt: "2026-07-14 10:00:01.000", Location: "104.1,30.6", Longitude: 104.1, Latitude: 30.6}
	query, args := buildAlertEventInsert("alert-1", "rule-1|VIN1|JT808", rule, item, 75)
	if placeholders := strings.Count(query, "?"); placeholders != len(args) {
		t.Fatalf("event insert placeholders=%d args=%d query=%s", placeholders, len(args), query)
	}
	if len(args) != 22 || !strings.Contains(query, "'unprocessed'") || !strings.Contains(query, "CURRENT_TIMESTAMP(3)") {
		t.Fatalf("unexpected event insert contract args=%#v query=%s", args, query)
	}
}

func TestAlertRuleValidationCoversRangesAndStateChange(t *testing.T) {
	base := AlertRuleInput{Name: "区间", Severity: "major", ValueType: "numeric", Metric: "soc_percent", Operator: "between", Threshold: 20, ThresholdHigh: 80}
	if err := validateAlertRule(base); err != nil {
		t.Fatalf("valid range rejected: %v", err)
	}
	base.ThresholdHigh = 10
	if err := validateAlertRule(base); err == nil {
		t.Fatal("inverted range should be rejected")
	}
	changed := AlertRuleInput{Name: "状态变化", Severity: "major", ValueType: "boolean", Metric: "alarm_active", Operator: "changed"}
	if err := validateAlertRule(changed); err != nil {
		t.Fatalf("boolean state change rejected: %v", err)
	}
	unknown := AlertRuleInput{Name: "未知指标", Severity: "major", ValueType: "numeric", Metric: "client_supplied_sql", Operator: "gt"}
	if err := validateAlertRule(unknown); err == nil {
		t.Fatal("metric outside the server catalog should be rejected")
	}
	mismatched := AlertRuleInput{Name: "类型错误", Severity: "major", ValueType: "boolean", Metric: "speed_kmh", Operator: "eq", BooleanThreshold: new(bool)}
	if err := validateAlertRule(mismatched); err == nil {
		t.Fatal("metric catalog type mismatch should be rejected")
	}
}

func TestMetricCatalogAndAlertValidationShareStoreConfiguration(t *testing.T) {
	definitions := metricDefinitions()
	for index := range definitions {
		if definitions[index].Key == "speed_kmh" {
			definitions[index].Label = "数据库配置速度"
			definitions[index].Alertable = false
		}
	}
	store := &configuredMetricStore{MockStore: NewMockStore(), definitions: definitions}
	service := NewService(store)
	catalog, err := service.MetricCatalog(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if catalog.Metrics[0].Label != "数据库配置速度" {
		t.Fatalf("catalog did not use store definitions: %+v", catalog.Metrics[0])
	}
	_, err = service.SaveAlertRule(t.Context(), AlertRuleInput{Name: "应被禁用", Severity: "major", ValueType: "numeric", Metric: "speed_kmh", Operator: "gt"})
	clientErr, ok := asClientError(err)
	if !ok || clientErr.Code != "ALERT_RULE_METRIC_INVALID" {
		t.Fatalf("rule validation must share configured catalog, got %v", err)
	}
}

func TestAlertRepeatIntervalBoundary(t *testing.T) {
	now := time.Now()
	if alertRepeatAllowed(now, now.Add(-59*time.Second), 60) {
		t.Fatal("repeat should remain suppressed inside interval")
	}
	if !alertRepeatAllowed(now, now.Add(-60*time.Second), 60) || !alertRepeatAllowed(now, time.Time{}, 60) {
		t.Fatal("repeat should open at boundary or without history")
	}
}

func TestAlertCandidateDurationOnlyAdvancesWithNewEventTime(t *testing.T) {
	first := time.Date(2026, 7, 14, 10, 0, 0, 0, time.Local)
	candidate, decision := advanceAlertCandidate(alertCandidateState{}, false, first, "event-1", false)
	if decision != alertObservationAdvanced || !candidate.FirstMatchedAt.Equal(first) || !candidate.LastMatchedAt.Equal(first) {
		t.Fatalf("first observation did not create candidate: decision=%v candidate=%+v", decision, candidate)
	}

	unchanged, decision := advanceAlertCandidate(candidate, true, first.Add(30*time.Second), "event-1", false)
	if decision != alertObservationDuplicate || !unchanged.LastMatchedAt.Equal(first) {
		t.Fatalf("same source event must not invent duration: decision=%v candidate=%+v", decision, unchanged)
	}

	advanced, decision := advanceAlertCandidate(candidate, true, first.Add(30*time.Second), "event-2", false)
	if decision != alertObservationAdvanced || advanced.LastMatchedAt.Sub(advanced.FirstMatchedAt) != 30*time.Second {
		t.Fatalf("newer event should advance event-time duration: decision=%v candidate=%+v", decision, advanced)
	}

	late, decision := advanceAlertCandidate(advanced, true, first.Add(20*time.Second), "event-late", false)
	if decision != alertObservationLate || late.SourceEventID != "event-2" {
		t.Fatalf("late event should not regress candidate: decision=%v candidate=%+v", decision, late)
	}
}

func TestAlertCandidateContinuityGapResetsEventTimeWindow(t *testing.T) {
	first := time.Date(2026, 7, 14, 10, 0, 0, 0, time.Local)
	current := alertCandidateState{FirstMatchedAt: first, LastMatchedAt: first.Add(30 * time.Second), SourceEventID: "event-2"}
	nextAt := current.LastMatchedAt.Add(alertCandidateContinuityWindow + time.Second)
	next, decision := advanceAlertCandidate(current, true, nextAt, "event-3", false)
	if decision != alertObservationAdvanced || !next.FirstMatchedAt.Equal(nextAt) || !next.LastMatchedAt.Equal(nextAt) {
		t.Fatalf("continuity gap should start a new window: decision=%v candidate=%+v", decision, next)
	}
}

func TestFreshnessCandidateIsExplicitlyClockDriven(t *testing.T) {
	first := time.Date(2026, 7, 14, 10, 0, 0, 0, time.Local)
	item := alertEvaluationEvidence{SourceEventID: "same-event", EventAt: "2026-07-14 09:55:00.000000"}
	observedAt, clockDriven, ok := alertObservationTime("freshness_sec", item, first)
	if !ok || !clockDriven || !observedAt.Equal(first) {
		t.Fatalf("freshness observation must use evaluator clock: at=%s clock=%t ok=%t", observedAt, clockDriven, ok)
	}
	candidate := alertCandidateState{FirstMatchedAt: first, LastMatchedAt: first, SourceEventID: item.SourceEventID}
	next, decision := advanceAlertCandidate(candidate, true, first.Add(10*time.Second), item.SourceEventID, true)
	if decision != alertObservationAdvanced || next.LastMatchedAt.Sub(next.FirstMatchedAt) != 10*time.Second {
		t.Fatalf("freshness should advance while source event is unchanged: decision=%v candidate=%+v", decision, next)
	}
}

func TestAlertObservationTimePrefersEventTimeAndFallsBackToReceipt(t *testing.T) {
	now := time.Date(2026, 7, 14, 12, 0, 0, 0, time.Local)
	eventItem := alertEvaluationEvidence{EventAt: "2026-07-14 10:00:30.123000", ReceivedAt: "2026-07-14 10:00:31.000000"}
	observedAt, clockDriven, ok := alertObservationTime("speed_kmh", eventItem, now)
	if !ok || clockDriven || observedAt.Second() != 30 || observedAt.Nanosecond() != 123000000 {
		t.Fatalf("event time not selected: at=%s clock=%t ok=%t", observedAt, clockDriven, ok)
	}
	receiptItem := alertEvaluationEvidence{ReceivedAt: "2026-07-14 10:00:31.000000"}
	observedAt, _, ok = alertObservationTime("speed_kmh", receiptItem, now)
	if !ok || observedAt.Second() != 31 {
		t.Fatalf("receipt fallback not selected: at=%s ok=%t", observedAt, ok)
	}
}

func TestSnapshotEvaluatorKeepsOnlyFreshnessRulesWhenStreamIsActive(t *testing.T) {
	rules := []AlertRule{{ID: "speed", Metric: "speed_kmh"}, {ID: "fresh", Metric: "freshness_sec"}, {ID: "delay", Metric: "data_delay_sec"}}
	if got := snapshotAlertRules(rules, "shadow"); len(got) != 3 {
		t.Fatalf("shadow mode filtered rules: %+v", got)
	}
	got := snapshotAlertRules(rules, "active")
	if len(got) != 1 || got[0].ID != "fresh" {
		t.Fatalf("active snapshot ownership is not freshness-only: %+v", got)
	}
}

func itoa(value int) string {
	if value == 0 {
		return "0"
	}
	digits := []byte{}
	for value > 0 {
		digits = append([]byte{byte('0' + value%10)}, digits...)
		value /= 10
	}
	return string(digits)
}