142 lines
3.8 KiB
Go
142 lines
3.8 KiB
Go
package realtime
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"strings"
|
|
"time"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
)
|
|
|
|
type SnapshotExecer interface {
|
|
ExecContext(context.Context, string, ...any) (sql.Result, error)
|
|
}
|
|
|
|
type SnapshotWriter struct {
|
|
exec SnapshotExecer
|
|
}
|
|
|
|
func NewSnapshotWriter(exec SnapshotExecer) *SnapshotWriter {
|
|
if exec == nil {
|
|
panic("snapshot execer must not be nil")
|
|
}
|
|
return &SnapshotWriter{exec: exec}
|
|
}
|
|
|
|
func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error {
|
|
_, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL)
|
|
return err
|
|
}
|
|
|
|
func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) error {
|
|
vehicleKey := strings.TrimSpace(env.VehicleKey())
|
|
if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") {
|
|
return nil
|
|
}
|
|
if !hasRealtimePayload(env) {
|
|
return nil
|
|
}
|
|
fieldsJSON, err := marshalObject(env.Fields)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
parsedJSON, err := marshalObject(env.Parsed)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
eventTime := nullableTime(env.EventTimeMS)
|
|
receivedAt := nullableTime(env.ReceivedAtMS)
|
|
_, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL,
|
|
string(env.Protocol),
|
|
vehicleKey,
|
|
strings.TrimSpace(env.VIN),
|
|
strings.TrimSpace(env.Phone),
|
|
strings.TrimSpace(env.DeviceID),
|
|
strings.TrimSpace(env.Plate),
|
|
strings.TrimSpace(env.MessageID),
|
|
env.Sequence,
|
|
strings.TrimSpace(env.SourceEndpoint),
|
|
eventTime,
|
|
parsedJSON,
|
|
fieldsJSON,
|
|
receivedAt,
|
|
env.StableEventID(),
|
|
)
|
|
return err
|
|
}
|
|
|
|
func marshalObject(value map[string]any) (string, error) {
|
|
if value == nil {
|
|
value = map[string]any{}
|
|
}
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return string(payload), nil
|
|
}
|
|
|
|
func nullableTime(ms int64) any {
|
|
if ms <= 0 {
|
|
return nil
|
|
}
|
|
return time.UnixMilli(ms)
|
|
}
|
|
|
|
func hasRealtimePayload(env envelope.FrameEnvelope) bool {
|
|
if len(env.Fields) > 0 {
|
|
return true
|
|
}
|
|
if _, ok := env.Parsed["data_units"]; ok {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot (
|
|
id BIGINT PRIMARY KEY AUTO_INCREMENT,
|
|
protocol VARCHAR(32) NOT NULL,
|
|
vehicle_key VARCHAR(96) NOT NULL,
|
|
vin VARCHAR(32) NOT NULL DEFAULT '',
|
|
phone VARCHAR(32) NOT NULL DEFAULT '',
|
|
device_id VARCHAR(96) NOT NULL DEFAULT '',
|
|
plate VARCHAR(32) NOT NULL DEFAULT '',
|
|
message_id VARCHAR(32) NOT NULL DEFAULT '',
|
|
sequence_id INT NOT NULL DEFAULT 0,
|
|
source_endpoint VARCHAR(128) NOT NULL DEFAULT '',
|
|
event_time DATETIME(3) NULL,
|
|
parsed_json JSON NOT NULL,
|
|
fields_json JSON NOT NULL,
|
|
received_at DATETIME(3) NULL,
|
|
event_id VARCHAR(64) NOT NULL DEFAULT '',
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
UNIQUE KEY uk_realtime_snapshot_vehicle (protocol, vehicle_key),
|
|
KEY idx_vehicle_key (vehicle_key),
|
|
KEY idx_vin (vin),
|
|
KEY idx_protocol_updated (protocol, updated_at)
|
|
)`
|
|
|
|
const upsertRealtimeSnapshotSQL = `
|
|
INSERT INTO vehicle_realtime_snapshot
|
|
(protocol, vehicle_key, vin, phone, device_id, plate, message_id, sequence_id,
|
|
source_endpoint, event_time, parsed_json, fields_json, received_at, event_id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), CAST(? AS JSON), ?, ?)
|
|
ON DUPLICATE KEY UPDATE
|
|
vin = IF(VALUES(vin) <> '', VALUES(vin), vin),
|
|
phone = IF(VALUES(phone) <> '', VALUES(phone), phone),
|
|
device_id = IF(VALUES(device_id) <> '', VALUES(device_id), device_id),
|
|
plate = IF(VALUES(plate) <> '', VALUES(plate), plate),
|
|
message_id = VALUES(message_id),
|
|
sequence_id = VALUES(sequence_id),
|
|
source_endpoint = VALUES(source_endpoint),
|
|
event_time = VALUES(event_time),
|
|
parsed_json = VALUES(parsed_json),
|
|
fields_json = VALUES(fields_json),
|
|
received_at = VALUES(received_at),
|
|
event_id = VALUES(event_id),
|
|
updated_at = CURRENT_TIMESTAMP
|
|
`
|