package capacity

import (
	"strings"
	"testing"
)

func TestEvaluateReportsOKWhenCriticalBacklogsAreZero(t *testing.T) {
	report := Evaluate(map[string]string{
		"gateway": `vehicle_gateway_active_connections{protocol="JT808"} 50000
vehicle_gateway_connection_rejections_total{protocol="JT808",reason="max_connections"} 0
vehicle_async_sink_queue_depth{sink="nats"} 0`,
		"fast-writer": `vehicle_fast_writer_nats_consumer_ack_pending{consumer="vehicle-fast-writer",stream="VEHICLE_INGEST"} 0
vehicle_fast_writer_nats_consumer_pending{consumer="vehicle-fast-writer",stream="VEHICLE_INGEST"} 0
vehicle_fast_writer_batch_pending_messages 0`,
		"history": `vehicle_history_batch_pending_messages 0
vehicle_history_batch_pending_rows 0
vehicle_history_kafka_lag{topic="vehicle.raw.go.jt808.v1",partition="0"} 0`,
	})

	if report.Status != StatusOK {
		t.Fatalf("status = %s, want ok: %#v", report.Status, report)
	}
	if report.Totals.ActiveConnections != 50000 {
		t.Fatalf("active connections = %d, want 50000", report.Totals.ActiveConnections)
	}
	if len(report.Findings) != 0 {
		t.Fatalf("findings = %#v, want none", report.Findings)
	}
}

func TestEvaluateReportsDegradedForPendingAndRejects(t *testing.T) {
	report := Evaluate(map[string]string{
		"gateway": `vehicle_gateway_active_connections{protocol="JT808"} 120000
vehicle_gateway_connection_rejections_total{protocol="JT808",reason="max_connections"} 3
vehicle_async_sink_queue_depth{sink="nats"} 25000`,
		"bridge": `vehicle_bridge_nats_consumer_ack_pending{consumer="vehicle-kafka-bridge",stream="VEHICLE_INGEST"} 101
vehicle_bridge_nats_consumer_pending{consumer="vehicle-kafka-bridge",stream="VEHICLE_INGEST"} 12001
vehicle_bridge_batch_pending_messages 1001`,
		"fast-writer": `vehicle_fast_writer_nats_consumer_ack_pending{consumer="vehicle-fast-writer",stream="VEHICLE_INGEST"} 11
vehicle_fast_writer_nats_consumer_pending{consumer="vehicle-fast-writer",stream="VEHICLE_INGEST"} 12001
vehicle_fast_writer_batch_pending_messages 9`,
		"history": `vehicle_history_batch_pending_messages 6
vehicle_history_batch_pending_rows 7
vehicle_history_kafka_lag{topic="vehicle.raw.go.jt808.v1",partition="0"} 42`,
	})

	if report.Status != StatusDegraded {
		t.Fatalf("status = %s, want degraded: %#v", report.Status, report)
	}
	joined := strings.Join(report.Findings, "\n")
	for _, want := range []string{
		"gateway connection rejections",
		"async sink queue depth",
		"bridge ack pending",
		"bridge consumer pending",
		"bridge batch pending",
		"fast writer ack pending",
		"fast writer consumer pending",
		"history batch pending",
		"kafka lag",
	} {
		if !strings.Contains(joined, want) {
			t.Fatalf("finding missing %q in:\n%s", want, joined)
		}
	}
	if report.Totals.BridgeAckPending != 101 {
		t.Fatalf("bridge ack pending = %v, want 101", report.Totals.BridgeAckPending)
	}
	if report.Totals.BridgeConsumerPending != 12001 {
		t.Fatalf("bridge consumer pending = %v, want 12001", report.Totals.BridgeConsumerPending)
	}
	if report.Totals.FastWriterAckPending != 11 {
		t.Fatalf("fast writer ack pending = %v, want 11", report.Totals.FastWriterAckPending)
	}
	if report.Totals.FastWriterBatchPending != 9 {
		t.Fatalf("fast writer batch pending = %v, want 9", report.Totals.FastWriterBatchPending)
	}
	if report.Totals.HistoryBatchPending != 6 {
		t.Fatalf("history batch pending = %v, want 6", report.Totals.HistoryBatchPending)
	}
	if report.Totals.HistoryRowsPending != 7 {
		t.Fatalf("history rows pending = %v, want 7", report.Totals.HistoryRowsPending)
	}
}