Each dispatched poll is a durable job with two job log streams, so an unbounded poll history would grow the platform store by roughly two thousand jobs a day per server. A new poll now retires older terminal jobs of the same template together with their job log streams, leaving at most two rows per template while nothing is in flight.
810 lines
32 KiB
Go
810 lines
32 KiB
Go
package repo
|
|
|
|
import (
|
|
"errors"
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"browser.local/platform/domain"
|
|
)
|
|
|
|
var (
|
|
ErrDuplicate = errors.New("resource already exists")
|
|
ErrNotFound = errors.New("resource not found")
|
|
)
|
|
|
|
type UserRepository interface {
|
|
Create(domain.User) error
|
|
Get(id string) (domain.User, error)
|
|
List(domain.UserFilter) ([]domain.User, error)
|
|
Update(domain.User) error
|
|
}
|
|
|
|
type AuthSessionRepository interface {
|
|
Create(domain.AuthSessionRecord) error
|
|
Get(id string) (domain.AuthSessionRecord, error)
|
|
List(domain.AuthSessionFilter) ([]domain.AuthSessionRecord, error)
|
|
Update(domain.AuthSessionRecord) error
|
|
}
|
|
|
|
type RunControlSessionRepository interface {
|
|
Create(domain.RunControlSession) error
|
|
Get(id string) (domain.RunControlSession, error)
|
|
List(struct{}) ([]domain.RunControlSession, error)
|
|
Update(domain.RunControlSession) error
|
|
}
|
|
|
|
type AIProviderRepository interface {
|
|
Create(domain.AIProvider) error
|
|
Get(id string) (domain.AIProvider, error)
|
|
List(domain.AIProviderFilter) ([]domain.AIProvider, error)
|
|
Update(domain.AIProvider) error
|
|
}
|
|
|
|
type GamePluginRepository interface {
|
|
Create(domain.GamePlugin) error
|
|
Get(id string) (domain.GamePlugin, error)
|
|
List(domain.GamePluginFilter) ([]domain.GamePlugin, error)
|
|
Update(domain.GamePlugin) error
|
|
Delete(id string) error
|
|
}
|
|
|
|
type ServerInstanceRepository interface {
|
|
Create(domain.ServerInstance) error
|
|
Get(id string) (domain.ServerInstance, error)
|
|
List(domain.ServerInstanceFilter) ([]domain.ServerInstance, error)
|
|
Update(domain.ServerInstance) error
|
|
}
|
|
|
|
type RunEndpointRepository interface {
|
|
Create(domain.RunEndpoint) error
|
|
Get(id string) (domain.RunEndpoint, error)
|
|
List(domain.RunEndpointFilter) ([]domain.RunEndpoint, error)
|
|
Update(domain.RunEndpoint) error
|
|
Delete(id string) error
|
|
}
|
|
|
|
type JobRepository interface {
|
|
Create(domain.Job) error
|
|
Get(id string) (domain.Job, error)
|
|
GetByIdempotency(runEndpointID string, idempotencyKey string) (domain.Job, error)
|
|
List(domain.JobFilter) ([]domain.Job, error)
|
|
Update(domain.Job) error
|
|
Delete(id string) error
|
|
}
|
|
|
|
type ArtifactRepository interface {
|
|
Create(domain.Artifact) error
|
|
Get(id string) (domain.Artifact, error)
|
|
List(domain.ArtifactFilter) ([]domain.Artifact, error)
|
|
Update(domain.Artifact) error
|
|
}
|
|
|
|
type RuntimeBindingRepository interface {
|
|
Create(domain.RuntimeBinding) error
|
|
Get(id string) (domain.RuntimeBinding, error)
|
|
List(domain.RuntimeBindingFilter) ([]domain.RuntimeBinding, error)
|
|
Update(domain.RuntimeBinding) error
|
|
}
|
|
|
|
type EncryptedComponentKeyRepository interface {
|
|
Create(domain.EncryptedComponentKey) error
|
|
Get(id string) (domain.EncryptedComponentKey, error)
|
|
List(domain.EncryptedComponentKeyFilter) ([]domain.EncryptedComponentKey, error)
|
|
Update(domain.EncryptedComponentKey) error
|
|
}
|
|
|
|
type RunDistributionRepository interface {
|
|
Create(domain.RunDistribution) error
|
|
Get(id string) (domain.RunDistribution, error)
|
|
List(domain.RunDistributionFilter) ([]domain.RunDistribution, error)
|
|
Update(domain.RunDistribution) error
|
|
}
|
|
|
|
type DependencyStatusRepository interface {
|
|
Create(domain.DependencyStatus) error
|
|
Get(id string) (domain.DependencyStatus, error)
|
|
List(domain.DependencyStatusFilter) ([]domain.DependencyStatus, error)
|
|
Update(domain.DependencyStatus) error
|
|
}
|
|
|
|
type RunUpdateJobRepository interface {
|
|
Create(domain.RunUpdateJob) error
|
|
Get(id string) (domain.RunUpdateJob, error)
|
|
List(domain.RunUpdateJobFilter) ([]domain.RunUpdateJob, error)
|
|
Update(domain.RunUpdateJob) error
|
|
}
|
|
|
|
type LogStreamRepository interface {
|
|
Create(domain.LogStream) error
|
|
Get(id string) (domain.LogStream, error)
|
|
List(domain.LogStreamFilter) ([]domain.LogStream, error)
|
|
Update(domain.LogStream) error
|
|
Delete(id string) error
|
|
}
|
|
|
|
type MetricSampleRepository interface {
|
|
Create(domain.MetricSample) error
|
|
Get(id string) (domain.MetricSample, error)
|
|
List(domain.MetricSampleFilter) ([]domain.MetricSample, error)
|
|
Update(domain.MetricSample) error
|
|
Delete(id string) error
|
|
}
|
|
|
|
type BackupRepository interface {
|
|
Create(domain.BackupRecord) error
|
|
Get(id string) (domain.BackupRecord, error)
|
|
List(domain.BackupFilter) ([]domain.BackupRecord, error)
|
|
Update(domain.BackupRecord) error
|
|
Delete(id string) error
|
|
}
|
|
|
|
type PluginLifecycleRepository interface {
|
|
Create(domain.PluginLifecycleInstallation) error
|
|
Get(id string) (domain.PluginLifecycleInstallation, error)
|
|
List(domain.PluginLifecycleFilter) ([]domain.PluginLifecycleInstallation, error)
|
|
Update(domain.PluginLifecycleInstallation) error
|
|
}
|
|
|
|
type AIConfigDiffRepository interface {
|
|
Create(domain.AIConfigDiffPreview) error
|
|
Get(id string) (domain.AIConfigDiffPreview, error)
|
|
List(domain.AIConfigDiffFilter) ([]domain.AIConfigDiffPreview, error)
|
|
Update(domain.AIConfigDiffPreview) error
|
|
}
|
|
|
|
type GameClientBridgeCommandRepository interface {
|
|
Create(domain.GameClientBridgeCommand) error
|
|
Get(id string) (domain.GameClientBridgeCommand, error)
|
|
GetByIdempotency(serverInstanceID, requesterID, commandType, idempotencyKey string) (domain.GameClientBridgeCommand, error)
|
|
List(domain.GameClientBridgeCommandFilter) ([]domain.GameClientBridgeCommand, error)
|
|
Update(domain.GameClientBridgeCommand) error
|
|
Delete(id string) error
|
|
}
|
|
|
|
type GameClientBridgeSnapshotRepository interface {
|
|
Create(domain.GameClientBridgeSnapshot) error
|
|
Get(id string) (domain.GameClientBridgeSnapshot, error)
|
|
List(domain.GameClientBridgeSnapshotFilter) ([]domain.GameClientBridgeSnapshot, error)
|
|
Update(domain.GameClientBridgeSnapshot) error
|
|
Delete(id string) error
|
|
}
|
|
|
|
type GameClientBridgeSnapshotStreamRepository interface {
|
|
Create(domain.GameClientBridgeSnapshotStream) error
|
|
Get(id string) (domain.GameClientBridgeSnapshotStream, error)
|
|
List(domain.GameClientBridgeSnapshotStreamFilter) ([]domain.GameClientBridgeSnapshotStream, error)
|
|
Update(domain.GameClientBridgeSnapshotStream) error
|
|
Delete(id string) error
|
|
}
|
|
|
|
type PluginDataRecordRepository interface {
|
|
Create(domain.PluginDataRecord) error
|
|
Get(string) (domain.PluginDataRecord, error)
|
|
List(domain.PluginDataFilter) ([]domain.PluginDataRecord, error)
|
|
Update(domain.PluginDataRecord) error
|
|
Delete(string) error
|
|
Apply([]domain.PluginDataRecord, []string) error
|
|
}
|
|
|
|
type SCUMUserRepository interface {
|
|
Create(domain.SCUMUser) error
|
|
Get(string) (domain.SCUMUser, error)
|
|
List(domain.SCUMUserFilter) ([]domain.SCUMUser, error)
|
|
Update(domain.SCUMUser) error
|
|
Delete(string) error
|
|
}
|
|
|
|
type SCUMUserTrajectoryRepository interface {
|
|
Create(domain.SCUMUserTrajectory) error
|
|
Get(string) (domain.SCUMUserTrajectory, error)
|
|
List(domain.SCUMUserTrajectoryFilter) ([]domain.SCUMUserTrajectory, error)
|
|
Update(domain.SCUMUserTrajectory) error
|
|
Delete(string) error
|
|
}
|
|
|
|
type SCUMVehicleRepository interface {
|
|
Create(domain.SCUMVehicle) error
|
|
Get(string) (domain.SCUMVehicle, error)
|
|
List(domain.SCUMVehicleFilter) ([]domain.SCUMVehicle, error)
|
|
Update(domain.SCUMVehicle) error
|
|
Delete(string) error
|
|
}
|
|
|
|
type SCUMVehicleTrajectoryRepository interface {
|
|
Create(domain.SCUMVehicleTrajectory) error
|
|
Get(string) (domain.SCUMVehicleTrajectory, error)
|
|
List(domain.SCUMVehicleTrajectoryFilter) ([]domain.SCUMVehicleTrajectory, error)
|
|
Update(domain.SCUMVehicleTrajectory) error
|
|
Delete(string) error
|
|
}
|
|
|
|
type SCUMVehicleLockRepository interface {
|
|
Create(domain.SCUMVehicleLock) error
|
|
Get(string) (domain.SCUMVehicleLock, error)
|
|
List(domain.SCUMVehicleLockFilter) ([]domain.SCUMVehicleLock, error)
|
|
Update(domain.SCUMVehicleLock) error
|
|
Delete(string) error
|
|
}
|
|
|
|
type Store interface {
|
|
Users() UserRepository
|
|
AuthSessions() AuthSessionRepository
|
|
RunControlSessions() RunControlSessionRepository
|
|
AIProviders() AIProviderRepository
|
|
GamePlugins() GamePluginRepository
|
|
ServerInstances() ServerInstanceRepository
|
|
RunEndpoints() RunEndpointRepository
|
|
Jobs() JobRepository
|
|
Artifacts() ArtifactRepository
|
|
RuntimeBindings() RuntimeBindingRepository
|
|
EncryptedComponentKeys() EncryptedComponentKeyRepository
|
|
RunDistributions() RunDistributionRepository
|
|
DependencyStatuses() DependencyStatusRepository
|
|
RunUpdateJobs() RunUpdateJobRepository
|
|
LogStreams() LogStreamRepository
|
|
MetricSamples() MetricSampleRepository
|
|
Backups() BackupRepository
|
|
PluginLifecycles() PluginLifecycleRepository
|
|
AIConfigDiffs() AIConfigDiffRepository
|
|
GameClientBridgeCommands() GameClientBridgeCommandRepository
|
|
GameClientBridgeSnapshots() GameClientBridgeSnapshotRepository
|
|
GameClientBridgeSnapshotStreams() GameClientBridgeSnapshotStreamRepository
|
|
PluginDataRecords() PluginDataRecordRepository
|
|
SCUMUsers() SCUMUserRepository
|
|
SCUMUserTrajectories() SCUMUserTrajectoryRepository
|
|
SCUMVehicles() SCUMVehicleRepository
|
|
SCUMVehicleTrajectories() SCUMVehicleTrajectoryRepository
|
|
SCUMVehicleLocks() SCUMVehicleLockRepository
|
|
}
|
|
|
|
type MemoryStore struct {
|
|
users *memoryRepository[domain.User, domain.UserFilter]
|
|
authSessions *memoryRepository[domain.AuthSessionRecord, domain.AuthSessionFilter]
|
|
runSessions *memoryRepository[domain.RunControlSession, struct{}]
|
|
aiProviders *memoryRepository[domain.AIProvider, domain.AIProviderFilter]
|
|
gamePlugins *memoryRepository[domain.GamePlugin, domain.GamePluginFilter]
|
|
serverInstances *memoryRepository[domain.ServerInstance, domain.ServerInstanceFilter]
|
|
runEndpoints *memoryRepository[domain.RunEndpoint, domain.RunEndpointFilter]
|
|
jobs *memoryJobRepository
|
|
artifacts *memoryRepository[domain.Artifact, domain.ArtifactFilter]
|
|
runtimeBindings *memoryRepository[domain.RuntimeBinding, domain.RuntimeBindingFilter]
|
|
componentKeys *memoryRepository[domain.EncryptedComponentKey, domain.EncryptedComponentKeyFilter]
|
|
runDists *memoryRepository[domain.RunDistribution, domain.RunDistributionFilter]
|
|
dependencies *memoryRepository[domain.DependencyStatus, domain.DependencyStatusFilter]
|
|
updateJobs *memoryRepository[domain.RunUpdateJob, domain.RunUpdateJobFilter]
|
|
logStreams *memoryRepository[domain.LogStream, domain.LogStreamFilter]
|
|
metricSamples *memoryRepository[domain.MetricSample, domain.MetricSampleFilter]
|
|
backups *memoryRepository[domain.BackupRecord, domain.BackupFilter]
|
|
pluginLifecycle *memoryRepository[domain.PluginLifecycleInstallation, domain.PluginLifecycleFilter]
|
|
aiConfigDiffs *memoryRepository[domain.AIConfigDiffPreview, domain.AIConfigDiffFilter]
|
|
bridgeCommands *memoryGameClientBridgeCommandRepository
|
|
bridgeSnapshots *memoryGameClientBridgeSnapshotRepository
|
|
bridgeStreams *memoryRepository[domain.GameClientBridgeSnapshotStream, domain.GameClientBridgeSnapshotStreamFilter]
|
|
pluginDataRecords *memoryRepository[domain.PluginDataRecord, domain.PluginDataFilter]
|
|
scumUsers *memoryRepository[domain.SCUMUser, domain.SCUMUserFilter]
|
|
scumUserTracks *memoryRepository[domain.SCUMUserTrajectory, domain.SCUMUserTrajectoryFilter]
|
|
scumVehicles *memoryRepository[domain.SCUMVehicle, domain.SCUMVehicleFilter]
|
|
scumVehicleTracks *memoryRepository[domain.SCUMVehicleTrajectory, domain.SCUMVehicleTrajectoryFilter]
|
|
scumVehicleLocks *memoryRepository[domain.SCUMVehicleLock, domain.SCUMVehicleLockFilter]
|
|
}
|
|
|
|
func NewMemoryStore() *MemoryStore {
|
|
return &MemoryStore{
|
|
users: newMemoryRepository(
|
|
func(user domain.User) string { return user.ID },
|
|
domain.CopyUser,
|
|
matchUser,
|
|
),
|
|
authSessions: newMemoryRepository(
|
|
func(session domain.AuthSessionRecord) string { return session.ID },
|
|
domain.CopyAuthSessionRecord,
|
|
matchAuthSession,
|
|
),
|
|
runSessions: newMemoryRepository(
|
|
func(session domain.RunControlSession) string { return session.RunEndpointID },
|
|
domain.CopyRunControlSession,
|
|
func(domain.RunControlSession, struct{}) bool { return true },
|
|
),
|
|
aiProviders: newMemoryRepository(
|
|
func(provider domain.AIProvider) string { return provider.ID },
|
|
domain.CopyAIProvider,
|
|
matchAIProvider,
|
|
),
|
|
gamePlugins: newMemoryRepository(
|
|
func(plugin domain.GamePlugin) string { return plugin.ID },
|
|
domain.CopyGamePlugin,
|
|
matchGamePlugin,
|
|
),
|
|
serverInstances: newMemoryRepository(
|
|
func(instance domain.ServerInstance) string { return instance.ID },
|
|
domain.CopyServerInstance,
|
|
matchServerInstance,
|
|
),
|
|
runEndpoints: newMemoryRepository(
|
|
func(endpoint domain.RunEndpoint) string { return endpoint.ID },
|
|
domain.CopyRunEndpoint,
|
|
matchRunEndpoint,
|
|
),
|
|
jobs: newMemoryJobRepository(),
|
|
artifacts: newMemoryRepository(
|
|
func(artifact domain.Artifact) string { return artifact.ID },
|
|
domain.CopyArtifact,
|
|
matchArtifact,
|
|
),
|
|
runtimeBindings: newMemoryRepository(
|
|
func(binding domain.RuntimeBinding) string { return binding.ID },
|
|
domain.CopyRuntimeBinding,
|
|
matchRuntimeBinding,
|
|
),
|
|
componentKeys: newMemoryRepository(
|
|
func(key domain.EncryptedComponentKey) string { return key.ID },
|
|
domain.CopyEncryptedComponentKey,
|
|
matchEncryptedComponentKey,
|
|
),
|
|
runDists: newMemoryRepository(
|
|
func(distribution domain.RunDistribution) string { return distribution.ID },
|
|
domain.CopyRunDistribution,
|
|
matchRunDistribution,
|
|
),
|
|
dependencies: newMemoryRepository(
|
|
func(status domain.DependencyStatus) string { return status.ID },
|
|
domain.CopyDependencyStatus,
|
|
matchDependencyStatus,
|
|
),
|
|
updateJobs: newMemoryRepository(
|
|
func(job domain.RunUpdateJob) string { return job.ID },
|
|
domain.CopyRunUpdateJob,
|
|
matchRunUpdateJob,
|
|
),
|
|
logStreams: newMemoryRepository(
|
|
func(stream domain.LogStream) string { return stream.ID },
|
|
domain.CopyLogStream,
|
|
matchLogStream,
|
|
),
|
|
metricSamples: newMemoryRepository(
|
|
func(sample domain.MetricSample) string { return sample.ID },
|
|
domain.CopyMetricSample,
|
|
matchMetricSample,
|
|
),
|
|
backups: newMemoryRepository(
|
|
func(record domain.BackupRecord) string { return record.ID },
|
|
domain.CopyBackupRecord,
|
|
matchBackup,
|
|
),
|
|
pluginLifecycle: newMemoryRepository(
|
|
func(installation domain.PluginLifecycleInstallation) string { return installation.ID },
|
|
domain.CopyPluginLifecycleInstallation,
|
|
matchPluginLifecycle,
|
|
),
|
|
aiConfigDiffs: newMemoryRepository(
|
|
func(preview domain.AIConfigDiffPreview) string { return preview.ID },
|
|
domain.CopyAIConfigDiffPreview,
|
|
matchAIConfigDiff,
|
|
),
|
|
bridgeCommands: newMemoryGameClientBridgeCommandRepository(),
|
|
bridgeSnapshots: newMemoryGameClientBridgeSnapshotRepository(),
|
|
bridgeStreams: newMemoryRepository(
|
|
func(stream domain.GameClientBridgeSnapshotStream) string { return stream.ID },
|
|
domain.CopyGameClientBridgeSnapshotStream,
|
|
matchGameClientBridgeSnapshotStream,
|
|
),
|
|
pluginDataRecords: newMemoryRepository(func(value domain.PluginDataRecord) string { return value.ID }, domain.CopyPluginDataRecord, matchPluginDataRecord),
|
|
scumUsers: newMemoryRepository(func(value domain.SCUMUser) string { return value.ID }, domain.CopySCUMUser, matchSCUMUser),
|
|
scumUserTracks: newMemoryRepository(func(value domain.SCUMUserTrajectory) string { return value.ID }, domain.CopySCUMUserTrajectory, matchSCUMUserTrajectory),
|
|
scumVehicles: newMemoryRepository(func(value domain.SCUMVehicle) string { return value.ID }, domain.CopySCUMVehicle, matchSCUMVehicle),
|
|
scumVehicleTracks: newMemoryRepository(func(value domain.SCUMVehicleTrajectory) string { return value.ID }, domain.CopySCUMVehicleTrajectory, matchSCUMVehicleTrajectory),
|
|
scumVehicleLocks: newMemoryRepository(func(value domain.SCUMVehicleLock) string { return value.ID }, domain.CopySCUMVehicleLock, matchSCUMVehicleLock),
|
|
}
|
|
}
|
|
|
|
func (store *MemoryStore) Users() UserRepository { return store.users }
|
|
func (store *MemoryStore) AuthSessions() AuthSessionRepository { return store.authSessions }
|
|
func (store *MemoryStore) RunControlSessions() RunControlSessionRepository { return store.runSessions }
|
|
func (store *MemoryStore) AIProviders() AIProviderRepository { return store.aiProviders }
|
|
func (store *MemoryStore) GamePlugins() GamePluginRepository { return store.gamePlugins }
|
|
func (store *MemoryStore) ServerInstances() ServerInstanceRepository { return store.serverInstances }
|
|
func (store *MemoryStore) RunEndpoints() RunEndpointRepository { return store.runEndpoints }
|
|
func (store *MemoryStore) Jobs() JobRepository { return store.jobs }
|
|
func (store *MemoryStore) Artifacts() ArtifactRepository { return store.artifacts }
|
|
func (store *MemoryStore) RuntimeBindings() RuntimeBindingRepository { return store.runtimeBindings }
|
|
func (store *MemoryStore) EncryptedComponentKeys() EncryptedComponentKeyRepository {
|
|
return store.componentKeys
|
|
}
|
|
func (store *MemoryStore) RunDistributions() RunDistributionRepository { return store.runDists }
|
|
func (store *MemoryStore) DependencyStatuses() DependencyStatusRepository { return store.dependencies }
|
|
func (store *MemoryStore) RunUpdateJobs() RunUpdateJobRepository { return store.updateJobs }
|
|
func (store *MemoryStore) LogStreams() LogStreamRepository { return store.logStreams }
|
|
func (store *MemoryStore) MetricSamples() MetricSampleRepository { return store.metricSamples }
|
|
func (store *MemoryStore) Backups() BackupRepository { return store.backups }
|
|
func (store *MemoryStore) PluginLifecycles() PluginLifecycleRepository {
|
|
return store.pluginLifecycle
|
|
}
|
|
func (store *MemoryStore) AIConfigDiffs() AIConfigDiffRepository { return store.aiConfigDiffs }
|
|
func (store *MemoryStore) GameClientBridgeCommands() GameClientBridgeCommandRepository {
|
|
return store.bridgeCommands
|
|
}
|
|
func (store *MemoryStore) GameClientBridgeSnapshots() GameClientBridgeSnapshotRepository {
|
|
return store.bridgeSnapshots
|
|
}
|
|
func (store *MemoryStore) GameClientBridgeSnapshotStreams() GameClientBridgeSnapshotStreamRepository {
|
|
return store.bridgeStreams
|
|
}
|
|
func (store *MemoryStore) PluginDataRecords() PluginDataRecordRepository {
|
|
return store.pluginDataRecords
|
|
}
|
|
func (store *MemoryStore) SCUMUsers() SCUMUserRepository { return store.scumUsers }
|
|
func (store *MemoryStore) SCUMUserTrajectories() SCUMUserTrajectoryRepository {
|
|
return store.scumUserTracks
|
|
}
|
|
func (store *MemoryStore) SCUMVehicles() SCUMVehicleRepository { return store.scumVehicles }
|
|
func (store *MemoryStore) SCUMVehicleTrajectories() SCUMVehicleTrajectoryRepository {
|
|
return store.scumVehicleTracks
|
|
}
|
|
func (store *MemoryStore) SCUMVehicleLocks() SCUMVehicleLockRepository { return store.scumVehicleLocks }
|
|
|
|
type memoryRepository[T any, F any] struct {
|
|
mu sync.RWMutex
|
|
byID map[string]T
|
|
idOf func(T) string
|
|
copyOf func(T) T
|
|
match func(T, F) bool
|
|
}
|
|
|
|
func newMemoryRepository[T any, F any](idOf func(T) string, copyOf func(T) T, match func(T, F) bool) *memoryRepository[T, F] {
|
|
return &memoryRepository[T, F]{
|
|
byID: map[string]T{},
|
|
idOf: idOf,
|
|
copyOf: copyOf,
|
|
match: match,
|
|
}
|
|
}
|
|
|
|
func (repository *memoryRepository[T, F]) Create(value T) error {
|
|
repository.mu.Lock()
|
|
defer repository.mu.Unlock()
|
|
|
|
id := repository.idOf(value)
|
|
if _, exists := repository.byID[id]; exists {
|
|
return ErrDuplicate
|
|
}
|
|
repository.byID[id] = repository.copyOf(value)
|
|
return nil
|
|
}
|
|
|
|
func (repository *memoryRepository[T, F]) Get(id string) (T, error) {
|
|
repository.mu.RLock()
|
|
defer repository.mu.RUnlock()
|
|
|
|
value, exists := repository.byID[id]
|
|
if !exists {
|
|
var zero T
|
|
return zero, ErrNotFound
|
|
}
|
|
return repository.copyOf(value), nil
|
|
}
|
|
|
|
func (repository *memoryRepository[T, F]) List(filter F) ([]T, error) {
|
|
repository.mu.RLock()
|
|
defer repository.mu.RUnlock()
|
|
|
|
ids := make([]string, 0, len(repository.byID))
|
|
for id := range repository.byID {
|
|
ids = append(ids, id)
|
|
}
|
|
sort.Strings(ids)
|
|
|
|
values := make([]T, 0, len(ids))
|
|
for _, id := range ids {
|
|
value := repository.byID[id]
|
|
if repository.match(value, filter) {
|
|
values = append(values, repository.copyOf(value))
|
|
}
|
|
}
|
|
return values, nil
|
|
}
|
|
|
|
func (repository *memoryRepository[T, F]) Update(value T) error {
|
|
repository.mu.Lock()
|
|
defer repository.mu.Unlock()
|
|
|
|
id := repository.idOf(value)
|
|
if _, exists := repository.byID[id]; !exists {
|
|
return ErrNotFound
|
|
}
|
|
repository.byID[id] = repository.copyOf(value)
|
|
return nil
|
|
}
|
|
|
|
func (repository *memoryRepository[T, F]) Delete(id string) error {
|
|
repository.mu.Lock()
|
|
defer repository.mu.Unlock()
|
|
if _, exists := repository.byID[id]; !exists {
|
|
return ErrNotFound
|
|
}
|
|
delete(repository.byID, id)
|
|
return nil
|
|
}
|
|
|
|
// Apply makes a set of upserts and deletes visible as one repository change.
|
|
func (repository *memoryRepository[T, F]) Apply(upserts []T, deleteIDs []string) error {
|
|
repository.mu.Lock()
|
|
defer repository.mu.Unlock()
|
|
for _, value := range upserts {
|
|
repository.byID[repository.idOf(value)] = repository.copyOf(value)
|
|
}
|
|
for _, id := range deleteIDs {
|
|
delete(repository.byID, id)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type memoryJobRepository struct {
|
|
*memoryRepository[domain.Job, domain.JobFilter]
|
|
}
|
|
|
|
func newMemoryJobRepository() *memoryJobRepository {
|
|
return &memoryJobRepository{
|
|
memoryRepository: newMemoryRepository(
|
|
func(job domain.Job) string { return job.ID },
|
|
domain.CopyJob,
|
|
matchJob,
|
|
),
|
|
}
|
|
}
|
|
|
|
func (repository *memoryJobRepository) List(filter domain.JobFilter) ([]domain.Job, error) {
|
|
if filter.Limit <= 0 {
|
|
return repository.memoryRepository.List(filter)
|
|
}
|
|
|
|
repository.mu.RLock()
|
|
defer repository.mu.RUnlock()
|
|
|
|
values := make([]domain.Job, 0, min(filter.Limit, len(repository.byID)))
|
|
for _, job := range repository.byID {
|
|
if repository.match(job, filter) {
|
|
values = append(values, domain.CopyJob(job))
|
|
}
|
|
}
|
|
sort.SliceStable(values, func(i, j int) bool {
|
|
leftUpdated := values[i].UpdatedAt
|
|
rightUpdated := values[j].UpdatedAt
|
|
if !leftUpdated.Equal(rightUpdated) {
|
|
return leftUpdated.After(rightUpdated)
|
|
}
|
|
leftCreated := values[i].CreatedAt
|
|
rightCreated := values[j].CreatedAt
|
|
if !leftCreated.Equal(rightCreated) {
|
|
return leftCreated.After(rightCreated)
|
|
}
|
|
return values[i].ID < values[j].ID
|
|
})
|
|
if len(values) > filter.Limit {
|
|
values = values[:filter.Limit]
|
|
}
|
|
return values, nil
|
|
}
|
|
|
|
func (repository *memoryJobRepository) GetByIdempotency(runEndpointID string, idempotencyKey string) (domain.Job, error) {
|
|
repository.mu.RLock()
|
|
defer repository.mu.RUnlock()
|
|
|
|
for _, job := range repository.byID {
|
|
if job.RunEndpointID == runEndpointID && job.IdempotencyKey == idempotencyKey {
|
|
return domain.CopyJob(job), nil
|
|
}
|
|
}
|
|
return domain.Job{}, ErrNotFound
|
|
}
|
|
|
|
func matchUser(user domain.User, filter domain.UserFilter) bool {
|
|
return filter.Status == "" || user.Status == filter.Status
|
|
}
|
|
|
|
func matchAuthSession(session domain.AuthSessionRecord, filter domain.AuthSessionFilter) bool {
|
|
return (filter.UserID == "" || session.UserID == filter.UserID) &&
|
|
(filter.TokenHash == "" || session.TokenHash == filter.TokenHash) &&
|
|
(filter.Status == "" || session.Status == filter.Status)
|
|
}
|
|
|
|
func matchAIProvider(provider domain.AIProvider, filter domain.AIProviderFilter) bool {
|
|
return (filter.Kind == "" || provider.Kind == filter.Kind) &&
|
|
(filter.Status == "" || provider.Status == filter.Status)
|
|
}
|
|
|
|
func matchGamePlugin(plugin domain.GamePlugin, filter domain.GamePluginFilter) bool {
|
|
return (filter.ServerType == "" || plugin.ServerType == filter.ServerType) &&
|
|
(filter.Status == "" || plugin.Status == filter.Status)
|
|
}
|
|
|
|
func matchServerInstance(instance domain.ServerInstance, filter domain.ServerInstanceFilter) bool {
|
|
if instance.State == domain.ServerInstanceStateDeleted && filter.State != domain.ServerInstanceStateDeleted {
|
|
return false
|
|
}
|
|
return (filter.PluginID == "" || instance.PluginID == filter.PluginID) &&
|
|
(filter.RunEndpointID == "" || instance.RunEndpointID == filter.RunEndpointID) &&
|
|
(filter.State == "" || instance.State == filter.State) &&
|
|
(filter.VisibleToUserID == "" || instance.OwnerUserID == filter.VisibleToUserID || containsString(instance.AdminUserIDs, filter.VisibleToUserID))
|
|
}
|
|
|
|
func containsString(values []string, target string) bool {
|
|
for _, value := range values {
|
|
if value == target {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func matchRunEndpoint(endpoint domain.RunEndpoint, filter domain.RunEndpointFilter) bool {
|
|
return filter.Status == "" || endpoint.Status == filter.Status
|
|
}
|
|
|
|
func matchJob(job domain.Job, filter domain.JobFilter) bool {
|
|
return (filter.ServerInstanceID == "" || job.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.RunEndpointID == "" || job.RunEndpointID == filter.RunEndpointID) &&
|
|
matchJobState(job.State, filter)
|
|
}
|
|
|
|
func matchJobState(state domain.JobState, filter domain.JobFilter) bool {
|
|
if filter.State != "" && state != filter.State {
|
|
return false
|
|
}
|
|
if len(filter.States) == 0 {
|
|
return true
|
|
}
|
|
for _, candidate := range filter.States {
|
|
if state == candidate {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func matchArtifact(artifact domain.Artifact, filter domain.ArtifactFilter) bool {
|
|
return (filter.OwnerKind == "" || artifact.OwnerKind == filter.OwnerKind) &&
|
|
(filter.OwnerID == "" || artifact.OwnerID == filter.OwnerID) &&
|
|
(filter.State == "" || artifact.State == filter.State)
|
|
}
|
|
|
|
func matchRuntimeBinding(binding domain.RuntimeBinding, filter domain.RuntimeBindingFilter) bool {
|
|
return (filter.ServerInstanceID == "" || binding.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.ProfileKey == "" || binding.ProfileKey == filter.ProfileKey) &&
|
|
(filter.Status == "" || binding.Status == filter.Status)
|
|
}
|
|
|
|
func matchEncryptedComponentKey(key domain.EncryptedComponentKey, filter domain.EncryptedComponentKeyFilter) bool {
|
|
return (filter.ServerInstanceID == "" || key.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.ComponentKind == "" || key.ComponentKind == filter.ComponentKind) &&
|
|
(filter.ComponentKey == "" || key.ComponentKey == filter.ComponentKey) &&
|
|
(filter.Status == "" || key.Status == filter.Status)
|
|
}
|
|
|
|
func matchRunDistribution(distribution domain.RunDistribution, filter domain.RunDistributionFilter) bool {
|
|
return (filter.ServerInstanceID == "" || distribution.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.TargetOS == "" || distribution.TargetOS == filter.TargetOS) &&
|
|
(filter.TargetArch == "" || distribution.TargetArch == filter.TargetArch) &&
|
|
(filter.Status == "" || distribution.Status == filter.Status)
|
|
}
|
|
|
|
func matchDependencyStatus(status domain.DependencyStatus, filter domain.DependencyStatusFilter) bool {
|
|
return (filter.ServerInstanceID == "" || status.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.ProbeKey == "" || status.ProbeKey == filter.ProbeKey) &&
|
|
(filter.State == "" || status.State == filter.State)
|
|
}
|
|
|
|
func matchRunUpdateJob(job domain.RunUpdateJob, filter domain.RunUpdateJobFilter) bool {
|
|
return (filter.ServerInstanceID == "" || job.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.Status == "" || job.Status == filter.Status)
|
|
}
|
|
|
|
func matchLogStream(stream domain.LogStream, filter domain.LogStreamFilter) bool {
|
|
return (filter.ServerInstanceID == "" || stream.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.StreamKey == "" || stream.StreamKey == filter.StreamKey)
|
|
}
|
|
|
|
func matchMetricSample(sample domain.MetricSample, filter domain.MetricSampleFilter) bool {
|
|
return (filter.ServerInstanceID == "" || sample.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.After.IsZero() || sample.CollectedAt.After(filter.After)) &&
|
|
(filter.Before.IsZero() || !sample.CollectedAt.After(filter.Before))
|
|
}
|
|
|
|
func matchBackup(record domain.BackupRecord, filter domain.BackupFilter) bool {
|
|
return (filter.ServerInstanceID == "" || record.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.State == "" || record.State == filter.State)
|
|
}
|
|
|
|
func matchPluginLifecycle(installation domain.PluginLifecycleInstallation, filter domain.PluginLifecycleFilter) bool {
|
|
return (filter.PluginID == "" || installation.PluginID == filter.PluginID) &&
|
|
(filter.ServerInstanceID == "" || installation.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.CurrentState == "" || installation.CurrentState == filter.CurrentState)
|
|
}
|
|
|
|
func matchAIConfigDiff(preview domain.AIConfigDiffPreview, filter domain.AIConfigDiffFilter) bool {
|
|
return (filter.ServerInstanceID == "" || preview.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.PluginID == "" || preview.PluginID == filter.PluginID) &&
|
|
(filter.State == "" || preview.State == filter.State)
|
|
}
|
|
|
|
func matchGameClientBridgeCommand(command domain.GameClientBridgeCommand, filter domain.GameClientBridgeCommandFilter) bool {
|
|
return (filter.ServerInstanceID == "" || command.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.PluginID == "" || command.PluginID == filter.PluginID) &&
|
|
(filter.ProfileKey == "" || command.ProfileKey == filter.ProfileKey) &&
|
|
(filter.State == "" || command.State == filter.State) &&
|
|
(filter.RequesterID == "" || command.RequesterID == filter.RequesterID) &&
|
|
(filter.CommandType == "" || command.CommandType == filter.CommandType) &&
|
|
(filter.IdempotencyKey == "" || command.IdempotencyKey == filter.IdempotencyKey) &&
|
|
(filter.ExpiresBefore.IsZero() || !command.ExpiresAt.After(filter.ExpiresBefore)) &&
|
|
(filter.CompletedBefore.IsZero() || (!command.CompletedAt.IsZero() && !command.CompletedAt.After(filter.CompletedBefore)))
|
|
}
|
|
|
|
func matchGameClientBridgeSnapshot(snapshot domain.GameClientBridgeSnapshot, filter domain.GameClientBridgeSnapshotFilter) bool {
|
|
return (filter.ServerInstanceID == "" || snapshot.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.PluginID == "" || snapshot.PluginID == filter.PluginID) &&
|
|
(filter.ProfileKey == "" || snapshot.ProfileKey == filter.ProfileKey) &&
|
|
(filter.Type == "" || snapshot.Type == filter.Type) &&
|
|
(filter.StreamKey == "" || snapshot.StreamKey == filter.StreamKey) &&
|
|
(filter.ObservedAfter.IsZero() || snapshot.ObservedAt.After(filter.ObservedAfter)) &&
|
|
(filter.ExpiresBefore.IsZero() || !snapshot.ExpiresAt.After(filter.ExpiresBefore))
|
|
}
|
|
|
|
func matchGameClientBridgeSnapshotStream(stream domain.GameClientBridgeSnapshotStream, filter domain.GameClientBridgeSnapshotStreamFilter) bool {
|
|
return (filter.ServerInstanceID == "" || stream.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.PluginID == "" || stream.PluginID == filter.PluginID) &&
|
|
(filter.ProfileKey == "" || stream.ProfileKey == filter.ProfileKey) &&
|
|
(filter.Type == "" || stream.Type == filter.Type) &&
|
|
(filter.StreamKey == "" || stream.StreamKey == filter.StreamKey)
|
|
}
|
|
|
|
func matchPluginDataRecord(value domain.PluginDataRecord, filter domain.PluginDataFilter) bool {
|
|
return (filter.PluginID == "" || value.PluginID == filter.PluginID) && (filter.ServerInstanceID == "" || value.ServerInstanceID == filter.ServerInstanceID) && (filter.Collection == "" || value.Collection == filter.Collection) && (filter.Key == "" || value.Key == filter.Key)
|
|
}
|
|
|
|
func matchSCUMUser(value domain.SCUMUser, filter domain.SCUMUserFilter) bool {
|
|
return (filter.ServerInstanceID == "" || value.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.SteamID == "" || value.SteamID == filter.SteamID) &&
|
|
(filter.Online == nil || value.Online == *filter.Online) &&
|
|
(filter.ChangedAfter.IsZero() || value.UpdatedAt.After(filter.ChangedAfter) || value.LastActivityAt.After(filter.ChangedAfter)) &&
|
|
(filter.StaleBefore.IsZero() || scumUserActivityAt(value).Before(filter.StaleBefore))
|
|
}
|
|
|
|
// scumUserActivityAt reports the timestamp used for offline convergence.
|
|
func scumUserActivityAt(value domain.SCUMUser) time.Time {
|
|
if value.LastActivityAt.IsZero() {
|
|
return value.UpdatedAt
|
|
}
|
|
return value.LastActivityAt
|
|
}
|
|
|
|
func matchSCUMUserTrajectory(value domain.SCUMUserTrajectory, filter domain.SCUMUserTrajectoryFilter) bool {
|
|
return (filter.ServerInstanceID == "" || value.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.SCUMUserID == "" || value.SCUMUserID == filter.SCUMUserID) &&
|
|
(filter.SteamID == "" || value.SteamID == filter.SteamID) &&
|
|
(filter.After.IsZero() || value.ObservedAt.After(filter.After))
|
|
}
|
|
|
|
func matchSCUMVehicle(value domain.SCUMVehicle, filter domain.SCUMVehicleFilter) bool {
|
|
return (filter.ServerInstanceID == "" || value.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.GameVehicleID == "" || value.GameVehicleID == filter.GameVehicleID) &&
|
|
(filter.Exists == nil || value.Exists == *filter.Exists) &&
|
|
(filter.ChangedAfter.IsZero() || value.UpdatedAt.After(filter.ChangedAfter) || value.LastObservedAt.After(filter.ChangedAfter))
|
|
}
|
|
|
|
func matchSCUMVehicleTrajectory(value domain.SCUMVehicleTrajectory, filter domain.SCUMVehicleTrajectoryFilter) bool {
|
|
return (filter.ServerInstanceID == "" || value.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.SCUMVehicleID == "" || value.SCUMVehicleID == filter.SCUMVehicleID) &&
|
|
(filter.GameVehicleID == "" || value.GameVehicleID == filter.GameVehicleID) &&
|
|
(filter.After.IsZero() || value.ObservedAt.After(filter.After))
|
|
}
|
|
|
|
func matchSCUMVehicleLock(value domain.SCUMVehicleLock, filter domain.SCUMVehicleLockFilter) bool {
|
|
return (filter.ServerInstanceID == "" || value.ServerInstanceID == filter.ServerInstanceID) &&
|
|
(filter.SCUMVehicleID == "" || value.SCUMVehicleID == filter.SCUMVehicleID) &&
|
|
(filter.GameVehicleID == "" || value.GameVehicleID == filter.GameVehicleID) &&
|
|
(filter.SCUMUserID == "" || value.SCUMUserID == filter.SCUMUserID) &&
|
|
(filter.SteamID == "" || value.SteamID == filter.SteamID) &&
|
|
(filter.After.IsZero() || value.LockedAt.After(filter.After))
|
|
}
|