343 lines
16 KiB
Go
343 lines
16 KiB
Go
package repo
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"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
|
|
}
|
|
|
|
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", 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) SCUMUsers() SCUMUserRepository {
|
|
return &persistentRepository[domain.SCUMUser, domain.SCUMUserFilter]{repository: store.MemoryStore.scumUsers, persist: store.persist}
|
|
}
|
|
func (store *MySQLStore) SCUMUserTrajectories() SCUMUserTrajectoryRepository {
|
|
return &persistentRepository[domain.SCUMUserTrajectory, domain.SCUMUserTrajectoryFilter]{repository: store.MemoryStore.scumUserTracks, persist: store.persist}
|
|
}
|
|
func (store *MySQLStore) SCUMVehicles() SCUMVehicleRepository {
|
|
return &persistentRepository[domain.SCUMVehicle, domain.SCUMVehicleFilter]{repository: store.MemoryStore.scumVehicles, persist: store.persist}
|
|
}
|
|
func (store *MySQLStore) SCUMVehicleTrajectories() SCUMVehicleTrajectoryRepository {
|
|
return &persistentRepository[domain.SCUMVehicleTrajectory, domain.SCUMVehicleTrajectoryFilter]{repository: store.MemoryStore.scumVehicleTracks, persist: store.persist}
|
|
}
|
|
func (store *MySQLStore) SCUMVehicleLocks() SCUMVehicleLockRepository {
|
|
return &persistentRepository[domain.SCUMVehicleLock, domain.SCUMVehicleLockFilter]{repository: store.MemoryStore.scumVehicleLocks, 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)
|
|
}
|
|
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)
|
|
return store.loadRuntime()
|
|
}
|
|
|
|
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 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),
|
|
SCUMUsers: snapshotRepository(store.MemoryStore.scumUsers),
|
|
SCUMUserTrajectories: snapshotRepository(store.MemoryStore.scumUserTracks),
|
|
SCUMVehicles: snapshotRepository(store.MemoryStore.scumVehicles),
|
|
SCUMVehicleTrajectories: snapshotRepository(store.MemoryStore.scumVehicleTracks),
|
|
SCUMVehicleLocks: snapshotRepository(store.MemoryStore.scumVehicleLocks),
|
|
}
|
|
}
|
|
|
|
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) {
|
|
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)
|
|
}
|