Move SCUM trajectories to plugin SQL storage

2026-08-28 17:50:45 +08:00
parent bb187f6007
commit aafbec4c96
21 changed files with 611 additions and 194 deletions
@@ -0,0 +1,366 @@
+package companion
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"math"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	_ "github.com/go-sql-driver/mysql"
+)
+
+const PlatformMySQLDSNEnvironment = "PLATFORM_MYSQL_DSN"
+
+var openSQL = sql.Open
+
+type SCUMSQLStore struct{ db *sql.DB }
+
+type TrajectorySample struct {
+	ServerInstanceID string
+	SubjectType      string
+	SubjectID        string
+	SteamID          string
+	UserProfileID    string
+	GamePlayerID     string
+	VehicleID        string
+	EntityID         string
+	BaseID           string
+	DisplayName      string
+	Label            string
+	ClassName        string
+	WorldX           float64
+	WorldY           float64
+	WorldZ           *float64
+	ObservedAt       time.Time
+	SampledAt        time.Time
+	Source           string
+}
+
+func NewSCUMSQLStore(db *sql.DB) (*SCUMSQLStore, error) {
+	if db == nil {
+		return nil, fmt.Errorf("platform SQL handle is required")
+	}
+	return &SCUMSQLStore{db: db}, nil
+}
+
+func OpenSCUMSQLStoreFromEnv(envName string) (*SCUMSQLStore, error) {
+	name := strings.TrimSpace(envName)
+	if name == "" {
+		name = PlatformMySQLDSNEnvironment
+	}
+	dsn := strings.TrimSpace(os.Getenv(name))
+	if dsn == "" {
+		return nil, fmt.Errorf("%s is required for SCUM plugin SQL storage", name)
+	}
+	db, err := openSQL("mysql", dsn)
+	if err != nil {
+		return nil, fmt.Errorf("open SCUM plugin SQL storage: %w", err)
+	}
+	return NewSCUMSQLStore(db)
+}
+
+func (store *SCUMSQLStore) EnsureSchema(ctx context.Context) error {
+	if store == nil || store.db == nil {
+		return fmt.Errorf("SCUM plugin SQL store is not configured")
+	}
+	for _, statement := range scumSQLStoreMigrations {
+		if _, err := store.db.ExecContext(ctx, statement); err != nil {
+			return fmt.Errorf("apply SCUM plugin SQL migration: %w", err)
+		}
+	}
+	return nil
+}
+
+func (store *SCUMSQLStore) StorePositionRows(ctx context.Context, serverInstanceID string, rows []map[string]any, sampledAt time.Time) (int, error) {
+	samples, err := TrajectorySamplesFromPositionRows(serverInstanceID, rows, sampledAt)
+	if err != nil {
+		return 0, err
+	}
+	return store.StoreTrajectorySamples(ctx, samples)
+}
+
+func (store *SCUMSQLStore) StoreVehicleRows(ctx context.Context, serverInstanceID string, rows []map[string]any, sampledAt time.Time) (int, error) {
+	samples, err := TrajectorySamplesFromVehicleRows(serverInstanceID, rows, sampledAt)
+	if err != nil {
+		return 0, err
+	}
+	return store.StoreTrajectorySamples(ctx, samples)
+}
+
+func (store *SCUMSQLStore) StoreTrajectorySamples(ctx context.Context, samples []TrajectorySample) (int, error) {
+	if store == nil || store.db == nil {
+		return 0, fmt.Errorf("SCUM plugin SQL store is not configured")
+	}
+	normalized := make([]TrajectorySample, 0, len(samples))
+	for _, sample := range samples {
+		value, err := normalizeTrajectorySample(sample)
+		if err != nil {
+			return 0, err
+		}
+		normalized = append(normalized, value)
+	}
+	if len(normalized) == 0 {
+		return 0, nil
+	}
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return 0, fmt.Errorf("begin SCUM trajectory write: %w", err)
+	}
+	defer tx.Rollback()
+	stamp := time.Now().UTC()
+	for _, sample := range normalized {
+		args := scumTrajectoryInsertArgs(sample, stamp)
+		if _, err := tx.ExecContext(ctx, scumTrajectoryInsertSQL, args...); err != nil {
+			return 0, fmt.Errorf("write SCUM trajectory sample: %w", err)
+		}
+	}
+	if err := tx.Commit(); err != nil {
+		return 0, fmt.Errorf("commit SCUM trajectory samples: %w", err)
+	}
+	return len(normalized), nil
+}
+
+func TrajectorySamplesFromPositionRows(serverInstanceID string, rows []map[string]any, sampledAt time.Time) ([]TrajectorySample, error) {
+	result := make([]TrajectorySample, 0, len(rows))
+	for _, row := range rows {
+		subjectType := textFromRow(row["subjectType"])
+		if subjectType != "player" && subjectType != "vehicle" {
+			continue
+		}
+		x, xOK := numberFromRow(row["x"])
+		y, yOK := numberFromRow(row["y"])
+		if !xOK || !yOK {
+			continue
+		}
+		z := optionalNumberFromRow(row["z"])
+		subjectID := textFromRow(row["subjectId"])
+		sample := TrajectorySample{
+			ServerInstanceID: serverInstanceID,
+			SubjectType:      subjectType,
+			SubjectID:        subjectID,
+			UserProfileID:    textFromRow(row["userProfileId"]),
+			GamePlayerID:     textFromRow(row["gamePlayerId"]),
+			VehicleID:        textFromRow(row["vehicleId"]),
+			EntityID:         textFromRow(row["entityId"]),
+			BaseID:           textFromRow(row["baseId"]),
+			WorldX:           x,
+			WorldY:           y,
+			WorldZ:           z,
+			ObservedAt:       timestampFromRow(row["observedAt"], sampledAt),
+			SampledAt:        sampledAt,
+			Source:           "plugin.sql.scum.positions",
+		}
+		if sample.SubjectType == "player" {
+			sample.SteamID = sample.SubjectID
+		}
+		result = append(result, sample)
+	}
+	return result, nil
+}
+
+func TrajectorySamplesFromVehicleRows(serverInstanceID string, rows []map[string]any, sampledAt time.Time) ([]TrajectorySample, error) {
+	result := make([]TrajectorySample, 0, len(rows))
+	for _, row := range rows {
+		x, xOK := numberFromRow(row["x"])
+		y, yOK := numberFromRow(row["y"])
+		if !xOK || !yOK {
+			continue
+		}
+		vehicleID := textFromRow(row["vehicleId"])
+		result = append(result, TrajectorySample{
+			ServerInstanceID: serverInstanceID,
+			SubjectType:      "vehicle",
+			SubjectID:        vehicleID,
+			VehicleID:        vehicleID,
+			EntityID:         textFromRow(row["entityId"]),
+			Label:            textFromRow(row["label"]),
+			ClassName:        textFromRow(row["className"]),
+			WorldX:           x,
+			WorldY:           y,
+			WorldZ:           optionalNumberFromRow(row["z"]),
+			ObservedAt:       timestampFromRow(row["lastAccessTime"], sampledAt),
+			SampledAt:        sampledAt,
+			Source:           "plugin.sql.scum.vehicles",
+		})
+	}
+	return result, nil
+}
+
+func normalizeTrajectorySample(sample TrajectorySample) (TrajectorySample, error) {
+	sample.ServerInstanceID = strings.TrimSpace(sample.ServerInstanceID)
+	sample.SubjectType = strings.TrimSpace(sample.SubjectType)
+	sample.SubjectID = strings.TrimSpace(sample.SubjectID)
+	sample.SteamID = strings.TrimSpace(sample.SteamID)
+	sample.UserProfileID = strings.TrimSpace(sample.UserProfileID)
+	sample.GamePlayerID = strings.TrimSpace(sample.GamePlayerID)
+	sample.VehicleID = strings.TrimSpace(sample.VehicleID)
+	sample.EntityID = strings.TrimSpace(sample.EntityID)
+	sample.BaseID = strings.TrimSpace(sample.BaseID)
+	sample.DisplayName = strings.TrimSpace(sample.DisplayName)
+	sample.Label = strings.TrimSpace(sample.Label)
+	sample.ClassName = strings.TrimSpace(sample.ClassName)
+	sample.Source = strings.TrimSpace(sample.Source)
+	if sample.ServerInstanceID == "" || sample.SubjectID == "" || sample.SampledAt.IsZero() || !validTrajectorySubjectType(sample.SubjectType) || !finite(sample.WorldX) || !finite(sample.WorldY) || sample.WorldZ != nil && !finite(*sample.WorldZ) {
+		return TrajectorySample{}, fmt.Errorf("SCUM trajectory sample is invalid")
+	}
+	if sample.ObservedAt.IsZero() {
+		sample.ObservedAt = sample.SampledAt
+	}
+	if sample.Source == "" {
+		sample.Source = "plugin.sql.scum"
+	}
+	if !boundedTrajectoryTexts(sample) {
+		return TrajectorySample{}, fmt.Errorf("SCUM trajectory sample text is too long")
+	}
+	sample.ObservedAt = sample.ObservedAt.UTC()
+	sample.SampledAt = sample.SampledAt.UTC()
+	return sample, nil
+}
+
+func scumTrajectoryInsertArgs(sample TrajectorySample, stamp time.Time) []any {
+	return []any{
+		scumTrajectoryRecordKey(sample), sample.ServerInstanceID, sample.SubjectType, sample.SubjectID,
+		nullText(sample.SteamID), nullText(sample.UserProfileID), nullText(sample.GamePlayerID), nullText(sample.VehicleID), nullText(sample.EntityID), nullText(sample.BaseID),
+		nullText(sample.DisplayName), nullText(sample.Label), nullText(sample.ClassName), sample.WorldX, sample.WorldY, nullFloat(sample.WorldZ), sample.ObservedAt, sample.SampledAt, sample.Source, stamp, stamp,
+	}
+}
+
+func scumTrajectoryRecordKey(sample TrajectorySample) string {
+	digest := sha256.Sum256([]byte(strings.Join([]string{sample.ServerInstanceID, sample.SubjectType, sample.SubjectID, sample.SampledAt.UTC().Format(time.RFC3339Nano)}, "\x00")))
+	return hex.EncodeToString(digest[:])
+}
+
+func validTrajectorySubjectType(value string) bool { return value == "player" || value == "vehicle" }
+func finite(value float64) bool                    { return !math.IsNaN(value) && !math.IsInf(value, 0) }
+func nullText(value string) any {
+	if value == "" {
+		return nil
+	}
+	return value
+}
+func nullFloat(value *float64) any {
+	if value == nil {
+		return nil
+	}
+	return *value
+}
+
+func boundedTrajectoryTexts(sample TrajectorySample) bool {
+	return len(sample.ServerInstanceID) <= 96 && len(sample.SubjectType) <= 16 && len(sample.SubjectID) <= 128 && len(sample.SteamID) <= 32 && len(sample.UserProfileID) <= 96 && len(sample.GamePlayerID) <= 96 && len(sample.VehicleID) <= 96 && len(sample.EntityID) <= 96 && len(sample.BaseID) <= 96 && len(sample.DisplayName) <= 120 && len(sample.Label) <= 120 && len(sample.ClassName) <= 160 && len(sample.Source) <= 80
+}
+
+func textFromRow(value any) string {
+	switch typed := value.(type) {
+	case nil:
+		return ""
+	case string:
+		return strings.TrimSpace(typed)
+	case json.Number:
+		return typed.String()
+	default:
+		return strings.TrimSpace(fmt.Sprint(typed))
+	}
+}
+
+func numberFromRow(value any) (float64, bool) {
+	switch typed := value.(type) {
+	case nil:
+		return 0, false
+	case float64:
+		return typed, finite(typed)
+	case float32:
+		value := float64(typed)
+		return value, finite(value)
+	case int:
+		return float64(typed), true
+	case int64:
+		return float64(typed), true
+	case int32:
+		return float64(typed), true
+	case json.Number:
+		value, err := typed.Float64()
+		return value, err == nil && finite(value)
+	case string:
+		value, err := strconv.ParseFloat(strings.TrimSpace(typed), 64)
+		return value, err == nil && finite(value)
+	default:
+		return 0, false
+	}
+}
+
+func optionalNumberFromRow(value any) *float64 {
+	number, ok := numberFromRow(value)
+	if !ok {
+		return nil
+	}
+	return &number
+}
+
+func timestampFromRow(value any, fallback time.Time) time.Time {
+	text := textFromRow(value)
+	if text == "" {
+		return fallback
+	}
+	parsed, err := time.Parse(time.RFC3339Nano, text)
+	if err != nil {
+		return fallback
+	}
+	return parsed
+}
+
+var scumSQLStoreMigrations = []string{`
+CREATE TABLE IF NOT EXISTS scum_trajectories (
+  record_key CHAR(64) PRIMARY KEY,
+  server_instance_id VARCHAR(96) NOT NULL,
+  subject_type VARCHAR(16) NOT NULL,
+  subject_id VARCHAR(128) NOT NULL,
+  steam_id VARCHAR(32) NULL,
+  user_profile_id VARCHAR(96) NULL,
+  game_player_id VARCHAR(96) NULL,
+  vehicle_id VARCHAR(96) NULL,
+  entity_id VARCHAR(96) NULL,
+  base_id VARCHAR(96) NULL,
+  display_name VARCHAR(120) NULL,
+  label VARCHAR(120) NULL,
+  class_name VARCHAR(160) NULL,
+  world_x DOUBLE NOT NULL,
+  world_y DOUBLE NOT NULL,
+  world_z DOUBLE NULL,
+  observed_at DATETIME(6) NOT NULL,
+  sampled_at DATETIME(6) NOT NULL,
+  source VARCHAR(80) NOT NULL,
+  created_at DATETIME(6) NOT NULL,
+  updated_at DATETIME(6) NOT NULL,
+  UNIQUE KEY scum_trajectories_sample_uq (server_instance_id, subject_type, subject_id, sampled_at),
+  KEY scum_trajectories_subject_idx (server_instance_id, subject_type, subject_id, observed_at),
+  KEY scum_trajectories_sampled_idx (server_instance_id, sampled_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`}
+
+const scumTrajectoryInsertSQL = `
+INSERT INTO scum_trajectories (
+  record_key, server_instance_id, subject_type, subject_id, steam_id, user_profile_id, game_player_id, vehicle_id, entity_id, base_id,
+  display_name, label, class_name, world_x, world_y, world_z, observed_at, sampled_at, source, created_at, updated_at
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ON DUPLICATE KEY UPDATE
+  steam_id = VALUES(steam_id),
+  user_profile_id = VALUES(user_profile_id),
+  game_player_id = VALUES(game_player_id),
+  vehicle_id = VALUES(vehicle_id),
+  entity_id = VALUES(entity_id),
+  base_id = VALUES(base_id),
+  display_name = VALUES(display_name),
+  label = VALUES(label),
+  class_name = VALUES(class_name),
+  world_x = VALUES(world_x),
+  world_y = VALUES(world_y),
+  world_z = VALUES(world_z),
+  observed_at = VALUES(observed_at),
+  source = VALUES(source),
+  updated_at = VALUES(updated_at)`