lingniu-vehicle-ingest/go/vehicle-gateway/internal/history/writer.go

package history

import (
	"context"
	"crypto/sha1"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"sync"
	"time"
	"unicode/utf8"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime"
)

type Execer interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
}

type Writer struct {
	exec     Execer
	database string
	cache    tableCache
}

const (
	rawFramePayloadInlineLimit = 12_000
	rawFramePayloadChunkSize   = 16_000
)

type payloadChunk struct {
	Kind  string
	Index int
	Count int
	Text  string
}

type tableCache struct {
	mu       sync.Mutex
	seen     map[string]struct{}
	inflight map[string]*tableEnsure
}

type tableEnsure struct {
	done chan struct{}
	err  error
}

func (c *tableCache) doOnce(key string, fn func() error) error {
	c.mu.Lock()
	if _, ok := c.seen[key]; ok {
		c.mu.Unlock()
		return nil
	}
	if c.inflight == nil {
		c.inflight = map[string]*tableEnsure{}
	}
	if entry, ok := c.inflight[key]; ok {
		c.mu.Unlock()
		<-entry.done
		return entry.err
	}
	entry := &tableEnsure{done: make(chan struct{})}
	c.inflight[key] = entry
	c.mu.Unlock()

	err := fn()

	c.mu.Lock()
	if err == nil {
		c.seen[key] = struct{}{}
	}
	entry.err = err
	delete(c.inflight, key)
	close(entry.done)
	c.mu.Unlock()
	return err
}

func NewWriter(exec Execer) *Writer {
	if exec == nil {
		panic("history execer must not be nil")
	}
	return &Writer{exec: exec, cache: tableCache{seen: map[string]struct{}{}}}
}

func NewWriterWithDatabase(exec Execer, database string) *Writer {
	writer := NewWriter(exec)
	writer.database = normalizeIdentifier(database)
	return writer
}

func (w *Writer) EnsureSchema(ctx context.Context, database string) error {
	for _, statement := range SchemaStatements(database) {
		if _, err := w.exec.ExecContext(ctx, statement); err != nil {
			return err
		}
	}
	return nil
}

func (w *Writer) qualify(table string) string {
	table = normalizeIdentifier(table)
	if w.database == "" {
		return table
	}
	return w.database + "." + table
}

func (w *Writer) AppendAll(ctx context.Context, env envelope.FrameEnvelope) error {
	if err := w.AppendRawFrame(ctx, env); err != nil {
		return err
	}
	return w.AppendLocation(ctx, env)
}

func (w *Writer) AppendAllBatch(ctx context.Context, envelopes []envelope.FrameEnvelope) error {
	if len(envelopes) == 0 {
		return nil
	}
	if err := w.AppendRawFrameBatch(ctx, envelopes); err != nil {
		return err
	}
	return w.AppendLocationBatch(ctx, envelopes)
}

func (w *Writer) AppendRawFrame(ctx context.Context, env envelope.FrameEnvelope) error {
	table := tableName("raw", env)
	if err := w.ensureRawChild(ctx, table, "raw_frames", env); err != nil {
		return err
	}
	rawHex, rawHexChunks := chunkPayload(env, "raw_hex", env.RawHex)
	rawText, rawTextChunks := chunkPayload(env, "raw_text", env.RawText)
	parsedFields, parsedChunks := chunkPayload(env, "parsed_fields", parsedFieldsJSONString(env))
	_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
  (ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes,
   raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint)
VALUES (%s)`, w.qualify(table), joinLiterals(rawValues(env, rawHex, rawText, parsedFields))))
	if err != nil {
		return err
	}
	chunks := append(rawHexChunks, rawTextChunks...)
	chunks = append(chunks, parsedChunks...)
	if len(chunks) == 0 {
		return nil
	}
	chunkTable := tableName("chunk", env)
	if err := w.ensureRawChild(ctx, chunkTable, "raw_frame_payload_chunks", env); err != nil {
		return err
	}
	for _, chunk := range chunks {
		if _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
  (ts, event_id, frame_id, received_at, payload_kind, chunk_index, chunk_count, chunk_text)
VALUES (%s)`, w.qualify(chunkTable), joinLiterals(chunkValues(env, chunk)))); err != nil {
			return err
		}
	}
	return nil
}

func (w *Writer) AppendRawFrameBatch(ctx context.Context, envelopes []envelope.FrameEnvelope) error {
	rowsByTable := map[string][]string{}
	chunkRowsByTable := map[string][]string{}
	chunkEnvByTable := map[string]envelope.FrameEnvelope{}
	for _, env := range envelopes {
		table := tableName("raw", env)
		if err := w.ensureRawChild(ctx, table, "raw_frames", env); err != nil {
			return err
		}
		rawHex, rawHexChunks := chunkPayload(env, "raw_hex", env.RawHex)
		rawText, rawTextChunks := chunkPayload(env, "raw_text", env.RawText)
		parsedFields, parsedChunks := chunkPayload(env, "parsed_fields", parsedFieldsJSONString(env))
		rowsByTable[table] = append(rowsByTable[table], "("+joinLiterals(rawValues(env, rawHex, rawText, parsedFields))+")")
		chunks := append(rawHexChunks, rawTextChunks...)
		chunks = append(chunks, parsedChunks...)
		if len(chunks) == 0 {
			continue
		}
		chunkTable := tableName("chunk", env)
		if err := w.ensureRawChild(ctx, chunkTable, "raw_frame_payload_chunks", env); err != nil {
			return err
		}
		chunkEnvByTable[chunkTable] = env
		for _, chunk := range chunks {
			chunkRowsByTable[chunkTable] = append(chunkRowsByTable[chunkTable], "("+joinLiterals(chunkValues(env, chunk))+")")
		}
	}
	for table, rows := range rowsByTable {
		if len(rows) == 0 {
			continue
		}
		if _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
  (ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes,
   raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint)
VALUES %s`, w.qualify(table), strings.Join(rows, ","))); err != nil {
			return err
		}
	}
	for table, rows := range chunkRowsByTable {
		if len(rows) == 0 {
			continue
		}
		_ = chunkEnvByTable[table]
		if _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
  (ts, event_id, frame_id, received_at, payload_kind, chunk_index, chunk_count, chunk_text)
VALUES %s`, w.qualify(table), strings.Join(rows, ","))); err != nil {
			return err
		}
	}
	return nil
}

func (w *Writer) AppendLocation(ctx context.Context, env envelope.FrameEnvelope) error {
	if strings.TrimSpace(env.VIN) == "" {
		return nil
	}
	longitude, okLon := floatField(env, envelope.FieldLongitude)
	latitude, okLat := floatField(env, envelope.FieldLatitude)
	if !okLon || !okLat {
		return nil
	}
	table := locationTableName(env)
	if err := w.ensureLocationChild(ctx, table, env); err != nil {
		return err
	}
	_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
  (ts, event_id, received_at, longitude, latitude, altitude_m, speed_kmh,
   direction_deg, alarm_flag, status_flag, total_mileage_km)
VALUES (%s)`, w.qualify(table), joinLiterals(locationValues(env, longitude, latitude))))
	return err
}

func (w *Writer) AppendLocationBatch(ctx context.Context, envelopes []envelope.FrameEnvelope) error {
	rowsByTable := map[string][]string{}
	for _, env := range envelopes {
		if strings.TrimSpace(env.VIN) == "" {
			continue
		}
		longitude, okLon := floatField(env, envelope.FieldLongitude)
		latitude, okLat := floatField(env, envelope.FieldLatitude)
		if !okLon || !okLat {
			continue
		}
		table := locationTableName(env)
		if err := w.ensureLocationChild(ctx, table, env); err != nil {
			return err
		}
		rowsByTable[table] = append(rowsByTable[table], "("+joinLiterals(locationValues(env, longitude, latitude))+")")
	}
	for table, rows := range rowsByTable {
		if len(rows) == 0 {
			continue
		}
		if _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
  (ts, event_id, received_at, longitude, latitude, altitude_m, speed_kmh,
   direction_deg, alarm_flag, status_flag, total_mileage_km)
VALUES %s`, w.qualify(table), strings.Join(rows, ","))); err != nil {
			return err
		}
	}
	return nil
}

func (w *Writer) ensureRawChild(ctx context.Context, table string, stable string, env envelope.FrameEnvelope) error {
	key := stable + "." + table
	return w.cache.doOnce(key, func() error {
		statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING %s TAGS ('%s', '%s', '%s', '%s', '%s')",
			w.qualify(table), w.qualify(stable),
			quote(string(env.Protocol)),
			quote(env.VehicleKey()),
			quote(env.VIN),
			quote(normalizePhoneTag(env.Phone)),
			quote(env.DeviceID))
		if _, err := w.exec.ExecContext(ctx, statement); err != nil {
			return err
		}
		if phone := normalizePhoneTag(env.Phone); phone != "" {
			if _, err := w.exec.ExecContext(ctx, fmt.Sprintf("ALTER TABLE %s SET TAG phone = '%s'", w.qualify(table), quote(phone))); err != nil {
				return err
			}
		}
		return nil
	})
}

func (w *Writer) ensureLocationChild(ctx context.Context, table string, env envelope.FrameEnvelope) error {
	key := "vehicle_locations." + table
	return w.cache.doOnce(key, func() error {
		statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING %s TAGS ('%s', '%s')",
			w.qualify(table),
			w.qualify("vehicle_locations"),
			quote(string(env.Protocol)),
			quote(strings.TrimSpace(env.VIN)))
		if _, err := w.exec.ExecContext(ctx, statement); err != nil {
			return err
		}
		return nil
	})
}

func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedFields string) []any {
	received := millis(env.ReceivedAtMS)
	eventTime := millis(env.EventTimeMS)
	return []any{
		received,
		frameID(env),
		env.StableEventID(),
		messageIDInt(env.MessageID),
		eventTime,
		received,
		rawSizeBytes(env),
		rawHex,
		rawText,
		parsedFields,
		string(env.ParseStatus),
		env.ParseError,
		env.SourceEndpoint,
	}
}

func chunkValues(env envelope.FrameEnvelope, chunk payloadChunk) []any {
	received := millis(env.ReceivedAtMS)
	return []any{
		received,
		env.StableEventID(),
		frameID(env),
		received,
		chunk.Kind,
		chunk.Index,
		chunk.Count,
		chunk.Text,
	}
}

func chunkPayload(env envelope.FrameEnvelope, kind string, value string) (string, []payloadChunk) {
	if len(value) <= rawFramePayloadInlineLimit {
		return value, nil
	}
	chunks := make([]payloadChunk, 0, (len(value)+rawFramePayloadChunkSize-1)/rawFramePayloadChunkSize)
	for index, start := 0, 0; start < len(value); index++ {
		end := safeChunkEnd(value, start, rawFramePayloadChunkSize)
		chunks = append(chunks, payloadChunk{
			Kind:  kind,
			Index: index,
			Text:  value[start:end],
		})
		start = end
	}
	count := len(chunks)
	for index := range chunks {
		chunks[index].Count = count
	}
	manifest := map[string]any{
		"chunked":      true,
		"payload_kind": kind,
		"event_id":     env.StableEventID(),
		"chunk_count":  count,
	}
	return jsonString(manifest), chunks
}

func safeChunkEnd(value string, start int, maxBytes int) int {
	end := start + maxBytes
	if end >= len(value) {
		return len(value)
	}
	for end > start && !utf8.RuneStart(value[end]) {
		end--
	}
	if end == start {
		return start + maxBytes
	}
	return end
}

func locationValues(env envelope.FrameEnvelope, longitude float64, latitude float64) []any {
	received := millis(env.ReceivedAtMS)
	return []any{
		eventTimeOrReceived(env),
		env.StableEventID(),
		received,
		longitude,
		latitude,
		floatFieldOrNil(env, "altitude_m"),
		floatFieldOrNil(env, envelope.FieldSpeedKMH),
		intFieldOrNil(env, "direction_deg"),
		intFieldOrNil(env, "alarm_flag"),
		intFieldOrNil(env, "status_flag"),
		floatFieldOrNil(env, envelope.FieldTotalMileageKM),
	}
}

func frameID(env envelope.FrameEnvelope) string {
	return "go_" + env.StableEventID()
}

func tableName(prefix string, env envelope.FrameEnvelope) string {
	return prefix + "_" + strings.ToLower(string(env.Protocol)) + "_" + hash16(env.VehicleKey())
}

func locationTableName(env envelope.FrameEnvelope) string {
	return "loc_" + strings.ToLower(string(env.Protocol)) + "_" + hash16(strings.TrimSpace(env.VIN))
}

func normalizeIdentifier(value string) string {
	value = strings.TrimSpace(value)
	value = strings.Trim(value, "`")
	return value
}

func normalizePhoneTag(phone string) string {
	trimmed := strings.TrimLeft(strings.TrimSpace(phone), "0")
	if trimmed == "" {
		return strings.TrimSpace(phone)
	}
	return trimmed
}

func hash16(value string) string {
	sum := sha1.Sum([]byte(value))
	return hex.EncodeToString(sum[:8])
}

func messageIDInt(value string) int64 {
	value = strings.TrimSpace(strings.ToLower(value))
	value = strings.TrimPrefix(value, "0x")
	parsed, err := strconv.ParseInt(value, 16, 32)
	if err == nil {
		return parsed
	}
	parsed, _ = strconv.ParseInt(value, 10, 32)
	return parsed
}

func rawSizeBytes(env envelope.FrameEnvelope) int {
	rawHex := strings.TrimSpace(env.RawHex)
	if len(rawHex)%2 != 0 {
		return len(rawHex) / 2
	}
	if rawHex != "" {
		return len(rawHex) / 2
	}
	return len([]byte(env.RawText))
}

func jsonString(value any) string {
	data, err := json.Marshal(value)
	if err != nil {
		return "{}"
	}
	return string(data)
}

func parsedFieldsJSONString(env envelope.FrameEnvelope) string {
	fields, _, ok := realtime.ParsedFieldsForEnvelope(env)
	if !ok || len(fields) == 0 {
		return ""
	}
	return jsonString(fields)
}

func floatField(env envelope.FrameEnvelope, key string) (float64, bool) {
	if env.Fields == nil {
		return 0, false
	}
	value, ok := env.Fields[key]
	if !ok || value == nil {
		return 0, false
	}
	switch typed := value.(type) {
	case float64:
		return typed, true
	case float32:
		return float64(typed), true
	case int:
		return float64(typed), true
	case int64:
		return float64(typed), true
	case uint16:
		return float64(typed), true
	case uint32:
		return float64(typed), true
	case string:
		parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64)
		return parsed, err == nil
	default:
		return 0, false
	}
}

func floatFieldOrNil(env envelope.FrameEnvelope, key string) any {
	value, ok := floatField(env, key)
	if !ok {
		return nil
	}
	return value
}

func intFieldOrNil(env envelope.FrameEnvelope, key string) any {
	if env.Fields == nil {
		return nil
	}
	value, ok := env.Fields[key]
	if !ok || value == nil {
		return nil
	}
	switch typed := value.(type) {
	case int:
		return typed
	case int64:
		return typed
	case uint16:
		return int64(typed)
	case uint32:
		return int64(typed)
	case float64:
		return int64(typed)
	case string:
		parsed, err := strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
		if err == nil {
			return parsed
		}
	}
	return nil
}

func eventTimeOrReceived(env envelope.FrameEnvelope) time.Time {
	if env.EventTimeMS > 0 {
		return millis(env.EventTimeMS)
	}
	return millis(env.ReceivedAtMS)
}

func millis(value int64) time.Time {
	if value <= 0 {
		return time.UnixMilli(time.Now().UnixMilli()).UTC()
	}
	return time.UnixMilli(value).UTC()
}

func quote(value string) string {
	return strings.ReplaceAll(value, "'", "''")
}

func joinLiterals(values []any) string {
	out := make([]string, 0, len(values))
	for _, value := range values {
		out = append(out, literal(value))
	}
	return strings.Join(out, ", ")
}

func literal(value any) string {
	if value == nil {
		return "NULL"
	}
	switch typed := value.(type) {
	case time.Time:
		return strconv.FormatInt(typed.UnixMilli(), 10)
	case string:
		return "'" + quote(typed) + "'"
	case envelope.ParseStatus:
		return "'" + quote(string(typed)) + "'"
	case int:
		return strconv.FormatInt(int64(typed), 10)
	case int64:
		return strconv.FormatInt(typed, 10)
	case uint16:
		return strconv.FormatUint(uint64(typed), 10)
	case uint32:
		return strconv.FormatUint(uint64(typed), 10)
	case float64:
		return strconv.FormatFloat(typed, 'f', -1, 64)
	case float32:
		return strconv.FormatFloat(float64(typed), 'f', -1, 64)
	default:
		return "'" + quote(fmt.Sprint(typed)) + "'"
	}
}