package platform

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"time"
)

const alertStreamMaxFutureSkew = 10 * time.Minute

type AlertStreamRecord struct {
	Topic         string
	Partition     int
	Offset        int64
	HighWatermark int64
	Protocol      string
	VIN           string
	Plate         string
	SourceEventID string
	EventAt       time.Time
	ReceivedAt    time.Time
	Fields        map[string]json.RawMessage
	Valid         bool
	Late          bool
	ErrorCode     string
}

type AlertStreamBatchResult struct {
	Fetched               int
	Processed             int
	Valid                 int
	Invalid               int
	Late                  int
	ReplaySkipped         int
	Partitions            int
	RulesEvaluated        int
	CandidatesAdvanced    int
	DuplicateObservations int
	LateObservations      int
	Opened                int
	Recovered             int
}

type alertStreamEnvelope struct {
	EventID       string                     `json:"event_id"`
	EventKind     string                     `json:"event_kind"`
	SourceEventID string                     `json:"source_event_id"`
	FieldMapping  string                     `json:"field_mapping"`
	Protocol      string                     `json:"protocol"`
	VIN           string                     `json:"vin"`
	Plate         string                     `json:"plate"`
	EventTimeMS   int64                      `json:"event_time_ms"`
	ReceivedAtMS  int64                      `json:"received_at_ms"`
	Fields        map[string]json.RawMessage `json:"fields"`
}

func DecodeAlertStreamRecord(topic string, partition int, offset, highWatermark int64, payload []byte, lateness time.Duration) AlertStreamRecord {
	record := AlertStreamRecord{Topic: strings.TrimSpace(topic), Partition: partition, Offset: offset, HighWatermark: highWatermark}
	var envelope alertStreamEnvelope
	if err := json.Unmarshal(payload, &envelope); err != nil {
		record.ErrorCode = "invalid_json"
		return record
	}
	record.Protocol = strings.TrimSpace(envelope.Protocol)
	record.VIN = strings.TrimSpace(envelope.VIN)
	record.Plate = strings.TrimSpace(envelope.Plate)
	record.SourceEventID = firstNonEmpty(strings.TrimSpace(envelope.SourceEventID), strings.TrimSpace(envelope.EventID))
	record.Fields = envelope.Fields
	if expected, known := alertStreamTopicProtocol(record.Topic); known && !strings.EqualFold(expected, record.Protocol) {
		record.ErrorCode = "protocol_topic_mismatch"
		return record
	}
	if !strings.EqualFold(strings.TrimSpace(envelope.EventKind), "FIELDS") {
		record.ErrorCode = "event_kind_mismatch"
		return record
	}
	if strings.TrimSpace(envelope.FieldMapping) == "" {
		record.ErrorCode = "missing_field_mapping"
		return record
	}
	if record.VIN == "" {
		record.ErrorCode = "missing_vin_" + strings.ToLower(record.Protocol)
		return record
	}
	if record.SourceEventID == "" {
		record.ErrorCode = "missing_source_event_id"
		return record
	}
	if len(envelope.Fields) == 0 {
		record.ErrorCode = "missing_fields"
		return record
	}
	prefix := map[string]string{"GB32960": "gb32960.", "JT808": "jt808.", "YUTONG_MQTT": "yutong_mqtt."}[strings.ToUpper(record.Protocol)]
	for field := range envelope.Fields {
		if prefix != "" && (field != strings.TrimSpace(field) || !strings.HasPrefix(field, prefix) || len(field) <= len(prefix)) {
			record.ErrorCode = "invalid_field_name"
			return record
		}
	}
	if envelope.ReceivedAtMS <= 0 {
		record.ErrorCode = "missing_received_time"
		return record
	}
	receivedAt := time.UnixMilli(envelope.ReceivedAtMS)
	eventMS := envelope.EventTimeMS
	if eventMS <= 0 || eventMS > envelope.ReceivedAtMS+alertStreamMaxFutureSkew.Milliseconds() {
		eventMS = envelope.ReceivedAtMS
	}
	record.EventAt = time.UnixMilli(eventMS)
	record.ReceivedAt = receivedAt
	if lateness > 0 && receivedAt.Sub(record.EventAt) > lateness {
		record.Late = true
	}
	record.Valid = true
	return record
}

func alertStreamTopicProtocol(topic string) (string, bool) {
	switch strings.TrimSpace(topic) {
	case "vehicle.fields.go.gb32960.v1":
		return "GB32960", true
	case "vehicle.fields.go.jt808.v1":
		return "JT808", true
	case "vehicle.fields.go.yutong-mqtt.v1":
		return "YUTONG_MQTT", true
	default:
		return "", false
	}
}

type alertStreamPartition struct {
	topic     string
	partition int
}

type alertStreamCheckpointUpdate struct {
	key                                     alertStreamPartition
	nextOffset, highWatermark               int64
	processed, valid, invalid, late, replay int64
	lastEventAt, lastReceivedAt             any
	lastSourceEventID                       string
	lastInvalidCode                         string
}

func (s *ProductionStore) RecordAlertStreamBatch(ctx context.Context, consumerGroup string, records []AlertStreamRecord) (AlertStreamBatchResult, error) {
	result := AlertStreamBatchResult{Fetched: len(records)}
	consumerGroup = strings.TrimSpace(consumerGroup)
	if consumerGroup == "" || len(consumerGroup) > 128 {
		return result, fmt.Errorf("alert stream consumer group is empty or too long")
	}
	if len(records) == 0 {
		return result, nil
	}
	grouped := map[alertStreamPartition][]AlertStreamRecord{}
	for _, record := range records {
		if record.Topic == "" || len(record.Topic) > 255 || record.Partition < 0 || record.Offset < 0 {
			return result, fmt.Errorf("invalid alert stream message position topic=%q partition=%d offset=%d", record.Topic, record.Partition, record.Offset)
		}
		key := alertStreamPartition{topic: record.Topic, partition: record.Partition}
		grouped[key] = append(grouped[key], record)
	}
	keys := make([]alertStreamPartition, 0, len(grouped))
	for key := range grouped {
		keys = append(keys, key)
	}
	sort.Slice(keys, func(i, j int) bool {
		if keys[i].topic == keys[j].topic {
			return keys[i].partition < keys[j].partition
		}
		return keys[i].topic < keys[j].topic
	})
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
	if err != nil {
		return result, err
	}
	defer tx.Rollback()
	updates := make([]alertStreamCheckpointUpdate, 0, len(keys))
	accepted := make([]AlertStreamRecord, 0, len(records))
	for _, key := range keys {
		partitionRecords := grouped[key]
		sort.Slice(partitionRecords, func(i, j int) bool { return partitionRecords[i].Offset < partitionRecords[j].Offset })
		var nextOffset int64
		hasCheckpoint := true
		if err = tx.QueryRowContext(ctx, `SELECT next_offset FROM vehicle_alert_stream_checkpoint WHERE consumer_group=? AND topic=? AND partition_id=? FOR UPDATE`, consumerGroup, key.topic, key.partition).Scan(&nextOffset); err == sql.ErrNoRows {
			hasCheckpoint = false
			nextOffset = -1
		} else if err != nil {
			return result, err
		}
		update := alertStreamCheckpointUpdate{key: key}
		maxOffset := nextOffset - 1
		var lastEventTime time.Time
		var lastPosition int64
		hasPosition := false
		for _, record := range partitionRecords {
			if record.HighWatermark > update.highWatermark {
				update.highWatermark = record.HighWatermark
			}
			if hasCheckpoint && record.Offset < nextOffset {
				update.replay++
				continue
			}
			if hasPosition && record.Offset == lastPosition {
				update.replay++
				continue
			}
			lastPosition = record.Offset
			hasPosition = true
			update.processed++
			if record.Valid {
				update.valid++
				if record.Late {
					update.late++
				}
				if lastEventTime.IsZero() || record.EventAt.After(lastEventTime) {
					lastEventTime = record.EventAt
					update.lastEventAt = record.EventAt
					update.lastReceivedAt = record.ReceivedAt
					update.lastSourceEventID = record.SourceEventID
				}
				accepted = append(accepted, record)
			} else {
				update.invalid++
				update.lastInvalidCode = record.ErrorCode
			}
			if record.Offset > maxOffset {
				maxOffset = record.Offset
			}
		}
		update.nextOffset = nextOffset
		if maxOffset >= 0 && maxOffset+1 > update.nextOffset {
			update.nextOffset = maxOffset + 1
		}
		if update.nextOffset < 0 {
			update.nextOffset = 0
		}
		updates = append(updates, update)
		result.Processed += int(update.processed)
		result.Valid += int(update.valid)
		result.Invalid += int(update.invalid)
		result.Late += int(update.late)
		result.ReplaySkipped += int(update.replay)
	}
	if strings.EqualFold(s.alertStreamMode, "active") && len(accepted) > 0 {
		evaluation, evaluationErr := evaluateAlertStreamRecordsTx(ctx, tx, accepted, time.Now())
		if evaluationErr != nil {
			return result, evaluationErr
		}
		result.RulesEvaluated = evaluation.RulesEvaluated
		result.CandidatesAdvanced = evaluation.CandidatesAdvanced
		result.DuplicateObservations = evaluation.DuplicateObservations
		result.LateObservations = evaluation.LateObservations
		result.Opened = evaluation.Opened
		result.Recovered = evaluation.Recovered
	}
	for _, update := range updates {
		_, err = tx.ExecContext(ctx, `INSERT INTO vehicle_alert_stream_checkpoint(consumer_group,topic,partition_id,next_offset,high_watermark,processed_count,valid_count,invalid_count,late_count,replay_skipped_count,last_event_at,last_received_at,last_source_event_id,last_invalid_code,last_invalid_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,IF(?='',NULL,CURRENT_TIMESTAMP(3))) ON DUPLICATE KEY UPDATE next_offset=GREATEST(next_offset,VALUES(next_offset)),high_watermark=GREATEST(high_watermark,VALUES(high_watermark)),processed_count=processed_count+VALUES(processed_count),valid_count=valid_count+VALUES(valid_count),invalid_count=invalid_count+VALUES(invalid_count),late_count=late_count+VALUES(late_count),replay_skipped_count=replay_skipped_count+VALUES(replay_skipped_count),last_event_at=COALESCE(VALUES(last_event_at),last_event_at),last_received_at=COALESCE(VALUES(last_received_at),last_received_at),last_source_event_id=IF(VALUES(last_source_event_id)='',last_source_event_id,VALUES(last_source_event_id)),last_invalid_code=IF(VALUES(last_invalid_code)='',last_invalid_code,VALUES(last_invalid_code)),last_invalid_at=COALESCE(VALUES(last_invalid_at),last_invalid_at)`, consumerGroup, update.key.topic, update.key.partition, update.nextOffset, update.highWatermark, update.processed, update.valid, update.invalid, update.late, update.replay, update.lastEventAt, update.lastReceivedAt, update.lastSourceEventID, update.lastInvalidCode, update.lastInvalidCode)
		if err != nil {
			return result, err
		}
	}
	result.Partitions = len(keys)
	if err = tx.Commit(); err != nil {
		return result, err
	}
	return result, nil
}