Rebuild SCUM plugin data ownership
This commit is contained in:
npc0-hue
@@ -171,6 +171,9 @@ func (store *MySQLStore) GameClientBridgeSnapshots() GameClientBridgeSnapshotRep
|
||||
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) GamePlayers() GamePlayerRepository {
|
||||
return &persistentRepository[domain.GamePlayer, domain.GamePlayerFilter]{repository: store.MemoryStore.gamePlayers, persist: store.persist}
|
||||
}
|
||||
@@ -205,10 +208,7 @@ func (store *MySQLStore) GameGiftGrants() GameGiftGrantRepository {
|
||||
return &persistentRepository[domain.GameGiftGrant, domain.GameGiftGrantFilter]{repository: store.MemoryStore.gameGiftGrants, persist: store.persist}
|
||||
}
|
||||
func (store *MySQLStore) SCUMDataObservations() SCUMDataObservationRepository {
|
||||
return &mysqlSCUMObservationRepository{repository: store.MemoryStore.scumDataObservations, store: store}
|
||||
}
|
||||
func (store *MySQLStore) SCUMDataRows() SCUMDataRowRepository {
|
||||
return &mysqlSCUMDataRowRepository{repository: store.MemoryStore.scumDataRows, store: store}
|
||||
return &persistentRepository[domain.SCUMDataObservation, domain.SCUMProjectionFilter]{repository: store.MemoryStore.scumDataObservations, persist: store.persist}
|
||||
}
|
||||
func (store *MySQLStore) SCUMPlayerLiveStates() SCUMPlayerLiveStateRepository {
|
||||
return &persistentRepository[domain.SCUMPlayerLiveState, domain.SCUMProjectionFilter]{repository: store.MemoryStore.scumPlayerLiveStates, persist: store.persist}
|
||||
@@ -253,158 +253,9 @@ CREATE TABLE IF NOT EXISTS platform_metadata_snapshots (
|
||||
if err != nil {
|
||||
return fmt.Errorf("create mysql metadata snapshot table: %w", err)
|
||||
}
|
||||
for _, table := range []string{"scum_sync_runs", "scum_users", "scum_squads", "scum_squad_members", "scum_vehicles", "scum_flags", "scum_activity_events", "scum_gift_events", "scum_map_points"} {
|
||||
statement := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (
|
||||
id VARCHAR(191) PRIMARY KEY,
|
||||
server_instance_id VARCHAR(191) NOT NULL,
|
||||
upsert_key VARCHAR(512) NOT NULL,
|
||||
fields_json JSON NOT NULL,
|
||||
payload_json JSON NOT NULL,
|
||||
plugin_id VARCHAR(191) NOT NULL,
|
||||
query_key VARCHAR(191) NOT NULL,
|
||||
freshness_json JSON NOT NULL,
|
||||
created_at DATETIME(6) NOT NULL,
|
||||
updated_at DATETIME(6) NOT NULL,
|
||||
INDEX %s_server_updated (server_instance_id, updated_at)
|
||||
)`, table, table)
|
||||
if _, err := store.db.ExecContext(ctx, statement); err != nil {
|
||||
return fmt.Errorf("create mysql %s table: %w", table, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type mysqlSCUMDataRowRepository struct {
|
||||
repository mutableRepository[domain.SCUMDataRow, domain.SCUMProjectionFilter]
|
||||
store *MySQLStore
|
||||
}
|
||||
|
||||
func (repository *mysqlSCUMDataRowRepository) Create(value domain.SCUMDataRow) error {
|
||||
if err := repository.repository.Create(value); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := repository.store.persistSCUMDataRow(value); err != nil {
|
||||
return err
|
||||
}
|
||||
return repository.store.persist()
|
||||
}
|
||||
func (repository *mysqlSCUMDataRowRepository) Get(id string) (domain.SCUMDataRow, error) {
|
||||
return repository.repository.Get(id)
|
||||
}
|
||||
func (repository *mysqlSCUMDataRowRepository) List(filter domain.SCUMProjectionFilter) ([]domain.SCUMDataRow, error) {
|
||||
return repository.repository.List(filter)
|
||||
}
|
||||
func (repository *mysqlSCUMDataRowRepository) Update(value domain.SCUMDataRow) error {
|
||||
if err := repository.repository.Update(value); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := repository.store.persistSCUMDataRow(value); err != nil {
|
||||
return err
|
||||
}
|
||||
return repository.store.persist()
|
||||
}
|
||||
|
||||
type mysqlSCUMObservationRepository struct {
|
||||
repository mutableRepository[domain.SCUMDataObservation, domain.SCUMProjectionFilter]
|
||||
store *MySQLStore
|
||||
}
|
||||
|
||||
func (repository *mysqlSCUMObservationRepository) Create(value domain.SCUMDataObservation) error {
|
||||
if err := repository.repository.Create(value); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := repository.store.persistSCUMSyncRun(value); err != nil {
|
||||
return err
|
||||
}
|
||||
return repository.store.persist()
|
||||
}
|
||||
func (repository *mysqlSCUMObservationRepository) Get(id string) (domain.SCUMDataObservation, error) {
|
||||
return repository.repository.Get(id)
|
||||
}
|
||||
func (repository *mysqlSCUMObservationRepository) List(filter domain.SCUMProjectionFilter) ([]domain.SCUMDataObservation, error) {
|
||||
return repository.repository.List(filter)
|
||||
}
|
||||
func (repository *mysqlSCUMObservationRepository) Update(value domain.SCUMDataObservation) error {
|
||||
if err := repository.repository.Update(value); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := repository.store.persistSCUMSyncRun(value); err != nil {
|
||||
return err
|
||||
}
|
||||
return repository.store.persist()
|
||||
}
|
||||
|
||||
func (store *MySQLStore) persistSCUMDataRow(value domain.SCUMDataRow) error {
|
||||
table, ok := mysqlSCUMTable(value.TargetTable)
|
||||
if !ok {
|
||||
return fmt.Errorf("unsupported SCUM data target %q", value.TargetTable)
|
||||
}
|
||||
fields, err := json.Marshal(value.Fields)
|
||||
if err != nil {
|
||||
return fmt.Errorf("encode SCUM fields: %w", err)
|
||||
}
|
||||
payload, err := json.Marshal(value.Payload)
|
||||
if err != nil {
|
||||
return fmt.Errorf("encode SCUM payload: %w", err)
|
||||
}
|
||||
freshness, err := json.Marshal(value.Freshness)
|
||||
if err != nil {
|
||||
return fmt.Errorf("encode SCUM freshness: %w", err)
|
||||
}
|
||||
return store.upsertSCUMPhysicalRow(table, value.ID, value.ServerInstanceID, value.UpsertKey, fields, payload, value.PluginID, value.QueryKey, freshness, value.CreatedAt, value.UpdatedAt)
|
||||
}
|
||||
|
||||
func (store *MySQLStore) persistSCUMSyncRun(value domain.SCUMDataObservation) error {
|
||||
freshness, err := json.Marshal(domain.SCUMProjectionFreshnessState{Status: domain.SCUMProjectionFreshness(value.Status), Source: value.Source, QueryKey: value.QueryKey, Sequence: value.Sequence, Checksum: value.Checksum, ObservedAt: value.ObservedAt, ReceivedAt: value.ReceivedAt})
|
||||
if err != nil {
|
||||
return fmt.Errorf("encode SCUM sync freshness: %w", err)
|
||||
}
|
||||
fields, _ := json.Marshal(map[string]any{"status": value.Status, "errorCode": value.ErrorCode, "observedAt": value.ObservedAt, "receivedAt": value.ReceivedAt})
|
||||
payload, _ := json.Marshal(value.SafeSummary)
|
||||
return store.upsertSCUMPhysicalRow("scum_sync_runs", value.ID, value.ServerInstanceID, value.QueryKey, fields, payload, value.PluginID, value.QueryKey, freshness, value.ReceivedAt, value.ReceivedAt)
|
||||
}
|
||||
|
||||
func (store *MySQLStore) upsertSCUMPhysicalRow(table, id, serverInstanceID, upsertKey string, fields, payload []byte, pluginID, queryKey string, freshness []byte, createdAt, updatedAt time.Time) error {
|
||||
if createdAt.IsZero() {
|
||||
createdAt = time.Now().UTC()
|
||||
}
|
||||
if updatedAt.IsZero() {
|
||||
updatedAt = createdAt
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
statement := fmt.Sprintf(`INSERT INTO %s (id, server_instance_id, upsert_key, fields_json, payload_json, plugin_id, query_key, freshness_json, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON DUPLICATE KEY UPDATE server_instance_id=VALUES(server_instance_id), upsert_key=VALUES(upsert_key), fields_json=VALUES(fields_json), payload_json=VALUES(payload_json), plugin_id=VALUES(plugin_id), query_key=VALUES(query_key), freshness_json=VALUES(freshness_json), updated_at=VALUES(updated_at)`, table)
|
||||
if _, err := store.db.ExecContext(ctx, statement, id, serverInstanceID, upsertKey, string(fields), string(payload), pluginID, queryKey, string(freshness), createdAt, updatedAt); err != nil {
|
||||
return fmt.Errorf("write mysql %s row: %w", table, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func mysqlSCUMTable(target domain.SCUMDataSet) (string, bool) {
|
||||
switch target {
|
||||
case domain.SCUMDataSetUsers:
|
||||
return "scum_users", true
|
||||
case domain.SCUMDataSetSquads:
|
||||
return "scum_squads", true
|
||||
case domain.SCUMDataSetMembers:
|
||||
return "scum_squad_members", true
|
||||
case domain.SCUMDataSetVehicles:
|
||||
return "scum_vehicles", true
|
||||
case domain.SCUMDataSetFlags:
|
||||
return "scum_flags", true
|
||||
case domain.SCUMDataSetActivity:
|
||||
return "scum_activity_events", true
|
||||
case domain.SCUMDataSetGiftEvents:
|
||||
return "scum_gift_events", true
|
||||
case domain.SCUMDataSetMapPoints:
|
||||
return "scum_map_points", true
|
||||
default:
|
||||
return "", false
|
||||
}
|
||||
}
|
||||
|
||||
func (store *MySQLStore) load() error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -476,7 +327,8 @@ func (store *MySQLStore) snapshot() StoreSnapshot {
|
||||
GameClientBridgeCommands: snapshotRepository(store.MemoryStore.bridgeCommands.memoryRepository),
|
||||
GameClientBridgeSnapshots: snapshotRepository(store.MemoryStore.bridgeSnapshots.memoryRepository),
|
||||
GameClientBridgeStreams: snapshotRepository(store.MemoryStore.bridgeStreams),
|
||||
GamePlayers: snapshotRepository(store.MemoryStore.gamePlayers), GamePlayerAliases: snapshotRepository(store.MemoryStore.gamePlayerAliases), GamePlayerSessions: snapshotRepository(store.MemoryStore.gamePlayerSessions), GameAccessAttempts: snapshotRepository(store.MemoryStore.gameAccessAttempts), GameSecuritySignals: snapshotRepository(store.MemoryStore.gameSecuritySignals), GamePlayerStatePatches: snapshotRepository(store.MemoryStore.gamePlayerStatePatches), GameMapTrackPoints: snapshotRepository(store.MemoryStore.gameMapTrackPoints), GamePlayerVehicleSegments: snapshotRepository(store.MemoryStore.gamePlayerVehicleSegments), GameGiftCatalogs: snapshotRepository(store.MemoryStore.gameGiftCatalogs), GameGiftRevisions: snapshotRepository(store.MemoryStore.gameGiftRevisions), GameGiftGrants: snapshotRepository(store.MemoryStore.gameGiftGrants), SCUMDataObservations: snapshotRepository(store.MemoryStore.scumDataObservations), SCUMDataRows: snapshotRepository(store.MemoryStore.scumDataRows), SCUMPlayerLiveStates: snapshotRepository(store.MemoryStore.scumPlayerLiveStates), SCUMSquads: snapshotRepository(store.MemoryStore.scumSquads), SCUMSquadMembers: snapshotRepository(store.MemoryStore.scumSquadMembers), SCUMVehicles: snapshotRepository(store.MemoryStore.scumVehicles), SCUMFlags: snapshotRepository(store.MemoryStore.scumFlags), SCUMCurrentPositions: snapshotRepository(store.MemoryStore.scumCurrentPositions), SCUMOperationRequests: snapshotRepository(store.MemoryStore.scumOperationRequests), SCUMWorkflowInstances: snapshotRepository(store.MemoryStore.scumWorkflowInstances), SCUMWorkflowSteps: snapshotRepository(store.MemoryStore.scumWorkflowSteps),
|
||||
PluginDataRecords: snapshotRepository(store.MemoryStore.pluginDataRecords),
|
||||
GamePlayers: snapshotRepository(store.MemoryStore.gamePlayers), GamePlayerAliases: snapshotRepository(store.MemoryStore.gamePlayerAliases), GamePlayerSessions: snapshotRepository(store.MemoryStore.gamePlayerSessions), GameAccessAttempts: snapshotRepository(store.MemoryStore.gameAccessAttempts), GameSecuritySignals: snapshotRepository(store.MemoryStore.gameSecuritySignals), GamePlayerStatePatches: snapshotRepository(store.MemoryStore.gamePlayerStatePatches), GameMapTrackPoints: snapshotRepository(store.MemoryStore.gameMapTrackPoints), GamePlayerVehicleSegments: snapshotRepository(store.MemoryStore.gamePlayerVehicleSegments), GameGiftCatalogs: snapshotRepository(store.MemoryStore.gameGiftCatalogs), GameGiftRevisions: snapshotRepository(store.MemoryStore.gameGiftRevisions), GameGiftGrants: snapshotRepository(store.MemoryStore.gameGiftGrants), SCUMDataObservations: snapshotRepository(store.MemoryStore.scumDataObservations), SCUMPlayerLiveStates: snapshotRepository(store.MemoryStore.scumPlayerLiveStates), SCUMSquads: snapshotRepository(store.MemoryStore.scumSquads), SCUMSquadMembers: snapshotRepository(store.MemoryStore.scumSquadMembers), SCUMVehicles: snapshotRepository(store.MemoryStore.scumVehicles), SCUMFlags: snapshotRepository(store.MemoryStore.scumFlags), SCUMCurrentPositions: snapshotRepository(store.MemoryStore.scumCurrentPositions), SCUMOperationRequests: snapshotRepository(store.MemoryStore.scumOperationRequests), SCUMWorkflowInstances: snapshotRepository(store.MemoryStore.scumWorkflowInstances), SCUMWorkflowSteps: snapshotRepository(store.MemoryStore.scumWorkflowSteps),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -510,6 +362,7 @@ func (store *MySQLStore) loadSnapshot(snapshot StoreSnapshot) {
|
||||
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.gamePlayers, snapshot.GamePlayers)
|
||||
loadRepository(store.MemoryStore.gamePlayerAliases, snapshot.GamePlayerAliases)
|
||||
loadRepository(store.MemoryStore.gamePlayerSessions, snapshot.GamePlayerSessions)
|
||||
@@ -522,7 +375,6 @@ func (store *MySQLStore) loadSnapshot(snapshot StoreSnapshot) {
|
||||
loadRepository(store.MemoryStore.gameGiftRevisions, snapshot.GameGiftRevisions)
|
||||
loadRepository(store.MemoryStore.gameGiftGrants, snapshot.GameGiftGrants)
|
||||
loadRepository(store.MemoryStore.scumDataObservations, snapshot.SCUMDataObservations)
|
||||
loadRepository(store.MemoryStore.scumDataRows, snapshot.SCUMDataRows)
|
||||
loadRepository(store.MemoryStore.scumPlayerLiveStates, snapshot.SCUMPlayerLiveStates)
|
||||
loadRepository(store.MemoryStore.scumSquads, snapshot.SCUMSquads)
|
||||
loadRepository(store.MemoryStore.scumSquadMembers, snapshot.SCUMSquadMembers)
|
||||
|
||||
Reference in New Issue
Block a user