374 lines
12 KiB
Go
374 lines
12 KiB
Go
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 (store *SCUMSQLStore) Close() error {
|
|
if store == nil || store.db == nil {
|
|
return nil
|
|
}
|
|
return store.db.Close()
|
|
}
|
|
|
|
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)`
|