npc0-hue
a822e9250a
Run pushes typed SCUM facts through POST /api/v1/run/scum/facts, but the production router rejected that path before the signature middleware because runServiceRequest listed individual Run channel prefixes, and MySQL kept SCUM rows inside the whole metadata snapshot instead of platform tables. - /api/v1/run/ is now the signed machine channel space while /api/v1/run/endpoints keeps normal bearer/admin authorization. - MySQL gets real scum_user, scum_user_trajectory, scum_vehicle, scum_vehicle_trajectory and scum_vehicle_lock tables with parameterized per-row repositories instead of full snapshot rewrites. Snapshot-shaped tables from the unreleased interim build are replaced, and SCUM rows still inside a metadata snapshot are migrated once. - Facts ingest verifies the target server plugin type and converges stale online users to offline after SCUMUserOfflineAfter. - The plugin page and browser read one bounded /scum/surface response instead of five list calls per refresh. Call-count budget for one server: per 5s facts batch, one SELECT plus one INSERT/UPDATE per reported user and vehicle, one INSERT per moved trajectory sample or new lock row, one bounded stale-user SELECT, and a trajectory retention DELETE at most once per hour. One browser refresh issues one surface request every 15s instead of five list requests.
390 lines
17 KiB
Go
390 lines
17 KiB
Go
package repo
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"browser.local/platform/domain"
|
|
|
|
_ "github.com/go-sql-driver/mysql"
|
|
)
|
|
|
|
const (
|
|
mysqlSnapshotID = "current"
|
|
mysqlRuntimeSnapshotID = "runtime"
|
|
)
|
|
|
|
type MySQLStore struct {
|
|
*MemoryStore
|
|
db *sql.DB
|
|
persistMu sync.Mutex
|
|
runtimePersistMu sync.Mutex
|
|
scumUserTrackPruner scumTrajectoryPruner
|
|
scumVehicleTrackPruner scumTrajectoryPruner
|
|
}
|
|
|
|
func NewMySQLStore(dsn string) (*MySQLStore, error) {
|
|
dsn = strings.TrimSpace(dsn)
|
|
if dsn == "" {
|
|
return nil, fmt.Errorf("PLATFORM_MYSQL_DSN is required when PLATFORM_STORAGE_BACKEND=mysql")
|
|
}
|
|
db, err := sql.Open("mysql", mysqlDSNWithParseTime(dsn))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("open mysql metadata store: %w", err)
|
|
}
|
|
store := &MySQLStore{
|
|
MemoryStore: NewMemoryStore(),
|
|
db: db,
|
|
}
|
|
if err := store.initialize(); err != nil {
|
|
_ = db.Close()
|
|
return nil, err
|
|
}
|
|
if err := store.load(); err != nil {
|
|
_ = db.Close()
|
|
return nil, err
|
|
}
|
|
return store, nil
|
|
}
|
|
|
|
func (store *MySQLStore) Close() error {
|
|
return store.db.Close()
|
|
}
|
|
|
|
func (store *MySQLStore) Users() UserRepository {
|
|
return &persistentRepository[domain.User, domain.UserFilter]{repository: store.MemoryStore.users, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) AuthSessions() AuthSessionRepository {
|
|
return &persistentRepository[domain.AuthSessionRecord, domain.AuthSessionFilter]{repository: store.MemoryStore.authSessions, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) RunControlSessions() RunControlSessionRepository {
|
|
return &persistentRepository[domain.RunControlSession, struct{}]{repository: store.MemoryStore.runSessions, persist: store.persistRuntime}
|
|
}
|
|
|
|
func (store *MySQLStore) AIProviders() AIProviderRepository {
|
|
return &persistentRepository[domain.AIProvider, domain.AIProviderFilter]{repository: store.MemoryStore.aiProviders, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) GamePlugins() GamePluginRepository {
|
|
return &persistentRepository[domain.GamePlugin, domain.GamePluginFilter]{repository: store.MemoryStore.gamePlugins, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) ServerInstances() ServerInstanceRepository {
|
|
return &persistentRepository[domain.ServerInstance, domain.ServerInstanceFilter]{repository: store.MemoryStore.serverInstances, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) RunEndpoints() RunEndpointRepository {
|
|
return &persistentRepository[domain.RunEndpoint, domain.RunEndpointFilter]{repository: store.MemoryStore.runEndpoints, persist: store.persistRuntime}
|
|
}
|
|
|
|
func (store *MySQLStore) Jobs() JobRepository {
|
|
return &persistentJobRepository{
|
|
persistentRepository: &persistentRepository[domain.Job, domain.JobFilter]{repository: store.MemoryStore.jobs, persist: store.persist},
|
|
repository: store.MemoryStore.jobs,
|
|
}
|
|
}
|
|
|
|
func (store *MySQLStore) Artifacts() ArtifactRepository {
|
|
return &persistentRepository[domain.Artifact, domain.ArtifactFilter]{repository: store.MemoryStore.artifacts, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) RuntimeBindings() RuntimeBindingRepository {
|
|
return &persistentRepository[domain.RuntimeBinding, domain.RuntimeBindingFilter]{repository: store.MemoryStore.runtimeBindings, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) EncryptedComponentKeys() EncryptedComponentKeyRepository {
|
|
return &persistentRepository[domain.EncryptedComponentKey, domain.EncryptedComponentKeyFilter]{repository: store.MemoryStore.componentKeys, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) RunDistributions() RunDistributionRepository {
|
|
return &persistentRepository[domain.RunDistribution, domain.RunDistributionFilter]{repository: store.MemoryStore.runDists, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) DependencyStatuses() DependencyStatusRepository {
|
|
return &persistentRepository[domain.DependencyStatus, domain.DependencyStatusFilter]{repository: store.MemoryStore.dependencies, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) RunUpdateJobs() RunUpdateJobRepository {
|
|
return &persistentRepository[domain.RunUpdateJob, domain.RunUpdateJobFilter]{repository: store.MemoryStore.updateJobs, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) LogStreams() LogStreamRepository {
|
|
return &persistentRepository[domain.LogStream, domain.LogStreamFilter]{repository: store.MemoryStore.logStreams, persist: store.persistRuntime}
|
|
}
|
|
|
|
func (store *MySQLStore) MetricSamples() MetricSampleRepository {
|
|
return &persistentRepository[domain.MetricSample, domain.MetricSampleFilter]{repository: store.MemoryStore.metricSamples, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) Backups() BackupRepository {
|
|
return &persistentRepository[domain.BackupRecord, domain.BackupFilter]{repository: store.MemoryStore.backups, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) PluginLifecycles() PluginLifecycleRepository {
|
|
return &persistentRepository[domain.PluginLifecycleInstallation, domain.PluginLifecycleFilter]{repository: store.MemoryStore.pluginLifecycle, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) AIConfigDiffs() AIConfigDiffRepository {
|
|
return &persistentRepository[domain.AIConfigDiffPreview, domain.AIConfigDiffFilter]{repository: store.MemoryStore.aiConfigDiffs, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) GameClientBridgeCommands() GameClientBridgeCommandRepository {
|
|
return &persistentGameClientBridgeCommandRepository{
|
|
persistentRepository: &persistentRepository[domain.GameClientBridgeCommand, domain.GameClientBridgeCommandFilter]{repository: store.MemoryStore.bridgeCommands, persist: store.persist},
|
|
repository: store.MemoryStore.bridgeCommands,
|
|
}
|
|
}
|
|
|
|
func (store *MySQLStore) GameClientBridgeSnapshots() GameClientBridgeSnapshotRepository {
|
|
return &persistentRepository[domain.GameClientBridgeSnapshot, domain.GameClientBridgeSnapshotFilter]{repository: store.MemoryStore.bridgeSnapshots, persist: store.persist}
|
|
}
|
|
|
|
func (store *MySQLStore) GameClientBridgeSnapshotStreams() GameClientBridgeSnapshotStreamRepository {
|
|
return &persistentRepository[domain.GameClientBridgeSnapshotStream, domain.GameClientBridgeSnapshotStreamFilter]{repository: store.MemoryStore.bridgeStreams, persist: store.persist}
|
|
}
|
|
func (store *MySQLStore) PluginDataRecords() PluginDataRecordRepository {
|
|
return &persistentRepository[domain.PluginDataRecord, domain.PluginDataFilter]{repository: store.MemoryStore.pluginDataRecords, persist: store.persist}
|
|
}
|
|
func (store *MySQLStore) initialize() error {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
if err := store.db.PingContext(ctx); err != nil {
|
|
return fmt.Errorf("connect mysql metadata store: %w", err)
|
|
}
|
|
_, err := store.db.ExecContext(ctx, `
|
|
CREATE TABLE IF NOT EXISTS platform_metadata_snapshots (
|
|
id VARCHAR(64) PRIMARY KEY,
|
|
snapshot_json JSON NOT NULL,
|
|
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
|
|
)`)
|
|
if err != nil {
|
|
return fmt.Errorf("create mysql metadata snapshot table: %w", err)
|
|
}
|
|
if err := store.initializeSCUMTables(ctx); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (store *MySQLStore) load() error {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
var payload []byte
|
|
err := store.db.QueryRowContext(ctx, "SELECT snapshot_json FROM platform_metadata_snapshots WHERE id = ?", mysqlSnapshotID).Scan(&payload)
|
|
if err != nil {
|
|
if err == sql.ErrNoRows {
|
|
return store.loadRuntime()
|
|
}
|
|
return fmt.Errorf("read mysql metadata snapshot: %w", err)
|
|
}
|
|
var snapshot StoreSnapshot
|
|
if err := json.Unmarshal(payload, &snapshot); err != nil {
|
|
return fmt.Errorf("decode mysql metadata snapshot: %w", err)
|
|
}
|
|
store.loadSnapshot(snapshot)
|
|
if err := store.migrateSnapshotSCUM(ctx, snapshot); err != nil {
|
|
return err
|
|
}
|
|
return store.loadRuntime()
|
|
}
|
|
|
|
// migrateSnapshotSCUM copies SCUM rows that an older build kept inside the
|
|
// metadata snapshot into the typed SCUM tables. SCUM is no longer stored in
|
|
// the snapshot, so this runs at most once for a snapshot written before the
|
|
// typed tables existed.
|
|
func (store *MySQLStore) migrateSnapshotSCUM(ctx context.Context, snapshot StoreSnapshot) error {
|
|
if len(snapshot.SCUMUsers)+len(snapshot.SCUMVehicles)+len(snapshot.SCUMUserTrajectories)+len(snapshot.SCUMVehicleTrajectories)+len(snapshot.SCUMVehicleLocks) == 0 {
|
|
return nil
|
|
}
|
|
users := store.SCUMUsers()
|
|
for _, value := range snapshot.SCUMUsers {
|
|
if _, err := users.Get(value.ID); err == nil {
|
|
continue
|
|
} else if !errors.Is(err, ErrNotFound) {
|
|
return err
|
|
}
|
|
if err := users.Create(value); err != nil && !errors.Is(err, ErrDuplicate) {
|
|
return fmt.Errorf("migrate scum_user %s: %w", value.ID, err)
|
|
}
|
|
}
|
|
vehicles := store.SCUMVehicles()
|
|
for _, value := range snapshot.SCUMVehicles {
|
|
if _, err := vehicles.Get(value.ID); err == nil {
|
|
continue
|
|
} else if !errors.Is(err, ErrNotFound) {
|
|
return err
|
|
}
|
|
if err := vehicles.Create(value); err != nil && !errors.Is(err, ErrDuplicate) {
|
|
return fmt.Errorf("migrate scum_vehicle %s: %w", value.ID, err)
|
|
}
|
|
}
|
|
userTracks := store.SCUMUserTrajectories()
|
|
for _, value := range snapshot.SCUMUserTrajectories {
|
|
if err := userTracks.Create(value); err != nil && !errors.Is(err, ErrDuplicate) {
|
|
return fmt.Errorf("migrate scum_user_trajectory %s: %w", value.ID, err)
|
|
}
|
|
}
|
|
vehicleTracks := store.SCUMVehicleTrajectories()
|
|
for _, value := range snapshot.SCUMVehicleTrajectories {
|
|
if err := vehicleTracks.Create(value); err != nil && !errors.Is(err, ErrDuplicate) {
|
|
return fmt.Errorf("migrate scum_vehicle_trajectory %s: %w", value.ID, err)
|
|
}
|
|
}
|
|
locks := store.SCUMVehicleLocks()
|
|
for _, value := range snapshot.SCUMVehicleLocks {
|
|
if err := locks.Create(value); err != nil && !errors.Is(err, ErrDuplicate) {
|
|
return fmt.Errorf("migrate scum_vehicle_lock %s: %w", value.ID, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (store *MySQLStore) loadRuntime() error {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
var payload []byte
|
|
err := store.db.QueryRowContext(ctx, "SELECT snapshot_json FROM platform_metadata_snapshots WHERE id = ?", mysqlRuntimeSnapshotID).Scan(&payload)
|
|
if err != nil {
|
|
if err == sql.ErrNoRows {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("read mysql runtime metadata snapshot: %w", err)
|
|
}
|
|
var snapshot runtimeSnapshot
|
|
if err := json.Unmarshal(payload, &snapshot); err != nil {
|
|
return fmt.Errorf("decode mysql runtime metadata snapshot: %w", err)
|
|
}
|
|
store.loadRuntimeSnapshot(snapshot)
|
|
return nil
|
|
}
|
|
|
|
func (store *MySQLStore) persist() error {
|
|
store.persistMu.Lock()
|
|
defer store.persistMu.Unlock()
|
|
|
|
snapshot := store.snapshot()
|
|
payload, err := json.Marshal(snapshot)
|
|
if err != nil {
|
|
return fmt.Errorf("encode mysql metadata snapshot: %w", err)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
_, err = store.db.ExecContext(ctx, `
|
|
INSERT INTO platform_metadata_snapshots (id, snapshot_json)
|
|
VALUES (?, ?)
|
|
ON DUPLICATE KEY UPDATE snapshot_json = ?, updated_at = CURRENT_TIMESTAMP`, mysqlSnapshotID, string(payload), string(payload))
|
|
if err != nil {
|
|
return fmt.Errorf("write mysql metadata snapshot: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (store *MySQLStore) persistRuntime() error {
|
|
store.runtimePersistMu.Lock()
|
|
defer store.runtimePersistMu.Unlock()
|
|
|
|
snapshot := store.runtimeSnapshot()
|
|
payload, err := json.Marshal(snapshot)
|
|
if err != nil {
|
|
return fmt.Errorf("encode mysql runtime metadata snapshot: %w", err)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
_, err = store.db.ExecContext(ctx, `
|
|
INSERT INTO platform_metadata_snapshots (id, snapshot_json)
|
|
VALUES (?, ?)
|
|
ON DUPLICATE KEY UPDATE snapshot_json = ?, updated_at = CURRENT_TIMESTAMP`, mysqlRuntimeSnapshotID, string(payload), string(payload))
|
|
if err != nil {
|
|
return fmt.Errorf("write mysql runtime metadata snapshot: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (store *MySQLStore) snapshot() StoreSnapshot {
|
|
return normalizeStoreSnapshot(StoreSnapshot{
|
|
Users: snapshotRepository(store.MemoryStore.users),
|
|
AuthSessions: snapshotRepository(store.MemoryStore.authSessions),
|
|
RunControlSessions: snapshotRepository(store.MemoryStore.runSessions),
|
|
AIProviders: snapshotRepository(store.MemoryStore.aiProviders),
|
|
GamePlugins: snapshotRepository(store.MemoryStore.gamePlugins),
|
|
ServerInstances: snapshotRepository(store.MemoryStore.serverInstances),
|
|
RunEndpoints: snapshotRepository(store.MemoryStore.runEndpoints),
|
|
Jobs: snapshotRepository(store.MemoryStore.jobs.memoryRepository),
|
|
Artifacts: snapshotRepository(store.MemoryStore.artifacts),
|
|
RuntimeBindings: snapshotRepository(store.MemoryStore.runtimeBindings),
|
|
EncryptedComponentKeys: snapshotRepository(store.MemoryStore.componentKeys),
|
|
RunDistributions: snapshotRepository(store.MemoryStore.runDists),
|
|
DependencyStatuses: snapshotRepository(store.MemoryStore.dependencies),
|
|
RunUpdateJobs: snapshotRepository(store.MemoryStore.updateJobs),
|
|
LogStreams: snapshotRepository(store.MemoryStore.logStreams),
|
|
MetricSamples: snapshotRepository(store.MemoryStore.metricSamples),
|
|
Backups: snapshotRepository(store.MemoryStore.backups),
|
|
PluginLifecycles: snapshotRepository(store.MemoryStore.pluginLifecycle),
|
|
AIConfigDiffs: snapshotRepository(store.MemoryStore.aiConfigDiffs),
|
|
GameClientBridgeCommands: snapshotRepository(store.MemoryStore.bridgeCommands.memoryRepository),
|
|
GameClientBridgeSnapshots: snapshotRepository(store.MemoryStore.bridgeSnapshots.memoryRepository),
|
|
GameClientBridgeStreams: snapshotRepository(store.MemoryStore.bridgeStreams),
|
|
PluginDataRecords: snapshotRepository(store.MemoryStore.pluginDataRecords),
|
|
// SCUM rows live in the typed scum_* tables, never in the metadata snapshot.
|
|
SCUMUsers: nil,
|
|
SCUMUserTrajectories: nil,
|
|
SCUMVehicles: nil,
|
|
SCUMVehicleTrajectories: nil,
|
|
SCUMVehicleLocks: nil,
|
|
})
|
|
}
|
|
|
|
func (store *MySQLStore) runtimeSnapshot() runtimeSnapshot {
|
|
return runtimeSnapshot{
|
|
RunControlSessions: snapshotRepository(store.MemoryStore.runSessions),
|
|
RunEndpoints: snapshotRepository(store.MemoryStore.runEndpoints),
|
|
LogStreams: snapshotRepository(store.MemoryStore.logStreams),
|
|
}
|
|
}
|
|
|
|
func (store *MySQLStore) loadSnapshot(snapshot StoreSnapshot) {
|
|
snapshot = normalizeStoreSnapshot(snapshot)
|
|
loadRepository(store.MemoryStore.users, snapshot.Users)
|
|
loadRepository(store.MemoryStore.authSessions, snapshot.AuthSessions)
|
|
loadRepository(store.MemoryStore.runSessions, snapshot.RunControlSessions)
|
|
loadRepository(store.MemoryStore.aiProviders, snapshot.AIProviders)
|
|
loadRepository(store.MemoryStore.gamePlugins, snapshot.GamePlugins)
|
|
loadRepository(store.MemoryStore.serverInstances, snapshot.ServerInstances)
|
|
loadRepository(store.MemoryStore.runEndpoints, snapshot.RunEndpoints)
|
|
loadRepository(store.MemoryStore.jobs.memoryRepository, snapshot.Jobs)
|
|
loadRepository(store.MemoryStore.artifacts, snapshot.Artifacts)
|
|
loadRepository(store.MemoryStore.runtimeBindings, snapshot.RuntimeBindings)
|
|
loadRepository(store.MemoryStore.componentKeys, snapshot.EncryptedComponentKeys)
|
|
loadRepository(store.MemoryStore.runDists, snapshot.RunDistributions)
|
|
loadRepository(store.MemoryStore.dependencies, snapshot.DependencyStatuses)
|
|
loadRepository(store.MemoryStore.updateJobs, snapshot.RunUpdateJobs)
|
|
loadRepository(store.MemoryStore.logStreams, snapshot.LogStreams)
|
|
loadRepository(store.MemoryStore.metricSamples, snapshot.MetricSamples)
|
|
loadRepository(store.MemoryStore.backups, snapshot.Backups)
|
|
loadRepository(store.MemoryStore.pluginLifecycle, snapshot.PluginLifecycles)
|
|
loadRepository(store.MemoryStore.aiConfigDiffs, snapshot.AIConfigDiffs)
|
|
loadRepository(store.MemoryStore.bridgeCommands.memoryRepository, snapshot.GameClientBridgeCommands)
|
|
loadRepository(store.MemoryStore.bridgeSnapshots.memoryRepository, snapshot.GameClientBridgeSnapshots)
|
|
loadRepository(store.MemoryStore.bridgeStreams, snapshot.GameClientBridgeStreams)
|
|
loadRepository(store.MemoryStore.pluginDataRecords, snapshot.PluginDataRecords)
|
|
loadRepository(store.MemoryStore.scumUsers, snapshot.SCUMUsers)
|
|
loadRepository(store.MemoryStore.scumUserTracks, snapshot.SCUMUserTrajectories)
|
|
loadRepository(store.MemoryStore.scumVehicles, snapshot.SCUMVehicles)
|
|
loadRepository(store.MemoryStore.scumVehicleTracks, snapshot.SCUMVehicleTrajectories)
|
|
loadRepository(store.MemoryStore.scumVehicleLocks, snapshot.SCUMVehicleLocks)
|
|
}
|
|
|
|
func (store *MySQLStore) loadRuntimeSnapshot(snapshot runtimeSnapshot) {
|
|
loadRepository(store.MemoryStore.runSessions, snapshot.RunControlSessions)
|
|
loadRepository(store.MemoryStore.runEndpoints, snapshot.RunEndpoints)
|
|
loadRepository(store.MemoryStore.logStreams, snapshot.LogStreams)
|
|
}
|