store

package
v2.15.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 28, 2026 License: GPL-3.0 Imports: 43 Imported by: 0

Documentation

Overview

Package store consolidates small persistence stores that each wrap a persis.Collection.

Index

Constants

This section is empty.

Variables

View Source 
var ErrCorrupt = persis.ErrCorrupt

ErrCorrupt reports that a record exists but its payload could not be decoded. It is distinct from a backend read failure so callers can choose to recover (for example, start from a fresh value) rather than propagate the error.

Functions

This section is empty.

Types

type APIKeyStore 

type APIKeyStore struct {
	// contains filtered or unexported fields
}

APIKeyStore implements auth.APIKeyStore. Name and credential-digest lookups use in-memory indexes rebuilt from the collection on startup; all writes keep them in sync under mu.

func NewAPIKeyStore 

func NewAPIKeyStore(col persis.Collection) (*APIKeyStore, error)

NewAPIKeyStore creates a APIKeyStore backed by col.

func (*APIKeyStore) Create 

func (s *APIKeyStore) Create(ctx context.Context, key *auth.APIKey) error

Create stores a new API key. Returns auth.ErrAPIKeyAlreadyExists if a key with the same name exists.

func (*APIKeyStore) Delete 

func (s *APIKeyStore) Delete(ctx context.Context, id string) error

Delete removes an API key by its ID. Returns auth.ErrAPIKeyNotFound if the key does not exist.

func (*APIKeyStore) GetByDigest 

func (s *APIKeyStore) GetByDigest(ctx context.Context, digest string) (*auth.APIKey, error)

GetByDigest retrieves an API key by its credential digest. Returns auth.ErrAPIKeyNotFound if the key does not exist.

func (*APIKeyStore) GetByID 

func (s *APIKeyStore) GetByID(ctx context.Context, id string) (*auth.APIKey, error)

GetByID retrieves an API key by its unique ID. Returns auth.ErrAPIKeyNotFound if the key does not exist.

func (*APIKeyStore) List 

func (s *APIKeyStore) List(ctx context.Context) ([]*auth.APIKey, error)

List returns all API keys in the store.

func (*APIKeyStore) PromoteDigest 

func (s *APIKeyStore) PromoteDigest(ctx context.Context, id, digest string) error

PromoteDigest atomically assigns a credential digest to an API key. Repeating the promotion with the same digest is idempotent.

func (*APIKeyStore) Update 

func (s *APIKeyStore) Update(ctx context.Context, key *auth.APIKey) error

Update modifies an existing API key. Credential fields and a newer LastUsedAt value in storage are preserved. Returns auth.ErrAPIKeyNotFound if the key does not exist.

func (*APIKeyStore) UpdateLastUsed 

func (s *APIKeyStore) UpdateLastUsed(ctx context.Context, id string) error

UpdateLastUsed records recent API key use without persisting more than once per minute.

type ActiveDistributedRunStore 

type ActiveDistributedRunStore struct {
	// contains filtered or unexported fields
}

ActiveDistributedRunStore implements dispatch.ActiveDistributedRunStore on top of a persis.Collection. Record IDs intentionally match the file-backed distributed store SHA-256 key.

func NewActiveDistributedRunStore 

func NewActiveDistributedRunStore(col persis.Collection, opts ...DistributedStoreOption) *ActiveDistributedRunStore

NewActiveDistributedRunStore creates an ActiveDistributedRunStore backed by col.

func (*ActiveDistributedRunStore) Delete 

func (s *ActiveDistributedRunStore) Delete(ctx context.Context, attemptKey string) error

func (*ActiveDistributedRunStore) Get 

func (s *ActiveDistributedRunStore) Get(ctx context.Context, attemptKey string) (*dispatch.ActiveDistributedRun, error)

func (*ActiveDistributedRunStore) ListAll 

func (s *ActiveDistributedRunStore) ListAll(ctx context.Context) ([]dispatch.ActiveDistributedRun, error)

func (*ActiveDistributedRunStore) Upsert 

func (s *ActiveDistributedRunStore) Upsert(ctx context.Context, record dispatch.ActiveDistributedRun) error

Upsert writes the active-run record.

type DAGRunLeaseStore 

type DAGRunLeaseStore struct {
	// contains filtered or unexported fields
}

DAGRunLeaseStore implements dispatch.DAGRunLeaseStore on top of a persis.Collection. Record IDs use the file-backed distributed store SHA-256 key.

func NewDAGRunLeaseStore 

func NewDAGRunLeaseStore(col persis.Collection, opts ...DistributedStoreOption) *DAGRunLeaseStore

NewDAGRunLeaseStore creates a DAGRunLeaseStore backed by col.

func (*DAGRunLeaseStore) Delete 

func (s *DAGRunLeaseStore) Delete(ctx context.Context, attemptKey string) error

func (*DAGRunLeaseStore) Get 

func (s *DAGRunLeaseStore) Get(ctx context.Context, attemptKey string) (*dispatch.DAGRunLease, error)

func (*DAGRunLeaseStore) ListAll 

func (s *DAGRunLeaseStore) ListAll(ctx context.Context) ([]dispatch.DAGRunLease, error)

func (*DAGRunLeaseStore) ListByQueue 

func (s *DAGRunLeaseStore) ListByQueue(ctx context.Context, queueName string) ([]dispatch.DAGRunLease, error)

func (*DAGRunLeaseStore) Touch 

func (s *DAGRunLeaseStore) Touch(ctx context.Context, attemptKey string, observedAt time.Time) error

Touch sets LastHeartbeatAt = observedAt. Returns ErrDAGRunLeaseNotFound when the lease is gone (initially or via concurrent delete).

func (*DAGRunLeaseStore) Upsert 

func (s *DAGRunLeaseStore) Upsert(ctx context.Context, lease dispatch.DAGRunLease) error

Upsert creates or refreshes a lease without changing its claim identity.

type DAGSettingsStore 

type DAGSettingsStore struct {
	// contains filtered or unexported fields
}

func NewDAGSettingsStore 

func NewDAGSettingsStore(col persis.Collection) (*DAGSettingsStore, error)

func (*DAGSettingsStore) Delete 

func (s *DAGSettingsStore) Delete(ctx context.Context, dagName string) error

func (*DAGSettingsStore) Get 

func (s *DAGSettingsStore) Get(ctx context.Context, dagName string) (*dagsettings.Settings, error)

func (*DAGSettingsStore) Upsert 

func (s *DAGSettingsStore) Upsert(ctx context.Context, settings *dagsettings.Settings) error

type DAGStateStore 

type DAGStateStore struct {
	// contains filtered or unexported fields
}

DAGStateStore persists DAG state entries in a persis collection.

func NewDAGStateStore 

func NewDAGStateStore(col persis.Collection) *DAGStateStore

NewDAGStateStore returns a DAG state store backed by the provided collection.

func (*DAGStateStore) Delete 

func (s *DAGStateStore) Delete(ctx context.Context, ref dagrun.StateRef) (bool, error)

func (*DAGStateStore) Get 

func (s *DAGStateStore) Get(ctx context.Context, ref dagrun.StateRef) (*dagrun.StateEntry, error)

func (*DAGStateStore) List 

func (s *DAGStateStore) List(ctx context.Context, opts dagrun.StateListOptions) ([]*dagrun.StateEntry, error)

func (*DAGStateStore) Put 

func (s *DAGStateStore) Put(ctx context.Context, ref dagrun.StateRef, value json.RawMessage, opts dagrun.StatePutOptions) (*dagrun.StateEntry, error)

type DispatchTaskStore 

type DispatchTaskStore struct {
	// contains filtered or unexported fields
}

DispatchTaskStore implements dispatch.DispatchTaskStore on top of a persis.Collection. Record IDs use "pending/" and "claims/" prefixes so a file collection rooted at the distributed directory uses the existing on-disk layout directly.

func NewDispatchTaskStore 

func NewDispatchTaskStore(col persis.Collection, opts ...DispatchTaskStoreOption) *DispatchTaskStore

NewDispatchTaskStore creates a DispatchTaskStore backed by col.

func (*DispatchTaskStore) BindAdmission 

func (s *DispatchTaskStore) BindAdmission(ctx context.Context, req dispatch.DispatchAdmissionBindRequest) error

func (*DispatchTaskStore) ClaimNext 

func (s *DispatchTaskStore) ClaimNext(ctx context.Context, claim dispatch.DispatchTaskClaim) (*dispatch.ClaimedDispatchTask, error)

ClaimNext atomically transitions one matching pending record into a claim. CompareAndDelete(pending) is the per-task atomicity point; concurrent pollers racing on the same pending see one winner and the losers clean up their orphan claim and continue to the next pending.

func (*DispatchTaskStore) CleanupAdmissions 

func (s *DispatchTaskStore) CleanupAdmissions(ctx context.Context, staleThreshold time.Duration) error

func (*DispatchTaskStore) CountOutstandingByQueue 

func (s *DispatchTaskStore) CountOutstandingByQueue(ctx context.Context, queueName string, _ time.Duration) (int, error)

CountOutstandingByQueue returns the number of pending+claimed dispatch records matching queueName. External store changes can be invisible until the lazy ID reconciliation interval elapses. A task transitioning between pending and claim during the scan may be counted as both for a sub-millisecond window, which only under-reports available capacity.

func (*DispatchTaskStore) DeleteClaim 

func (s *DispatchTaskStore) DeleteClaim(ctx context.Context, claimToken string) error

func (*DispatchTaskStore) Enqueue 

func (s *DispatchTaskStore) Enqueue(ctx context.Context, task *dispatch.DispatchTask) error

func (*DispatchTaskStore) FinalizeAdmissionAttempt 

func (s *DispatchTaskStore) FinalizeAdmissionAttempt(ctx context.Context, attemptKey string) error

func (*DispatchTaskStore) GetClaim 

func (s *DispatchTaskStore) GetClaim(ctx context.Context, claimToken string) (*dispatch.ClaimedDispatchTask, error)

func (*DispatchTaskStore) HasOutstandingAttempt 

func (s *DispatchTaskStore) HasOutstandingAttempt(ctx context.Context, attemptKey string, _ time.Duration) (bool, error)

HasOutstandingAttempt reports whether any pending or claimed record matches attemptKey. It uses the same bounded-staleness contract as [CountOutstandingByQueue].

func (*DispatchTaskStore) ListBundleDigests added in v2.15.4 

func (s *DispatchTaskStore) ListBundleDigests(ctx context.Context) ([]string, error)

ListBundleDigests returns bundle digests referenced by outstanding tasks.

func (*DispatchTaskStore) ReleaseAdmissionToken 

func (s *DispatchTaskStore) ReleaseAdmissionToken(ctx context.Context, reservationToken string) error

func (*DispatchTaskStore) ReleaseClaim 

func (s *DispatchTaskStore) ReleaseClaim(ctx context.Context, claimToken string) error

func (*DispatchTaskStore) ReserveAdmission 

func (s *DispatchTaskStore) ReserveAdmission(
	ctx context.Context,
	req dispatch.DispatchAdmissionRequest,
) (*dispatch.DispatchAdmissionDecision, error)

type DispatchTaskStoreOption 

type DispatchTaskStoreOption func(*DispatchTaskStore)

DispatchTaskStoreOption configures a DispatchTaskStore.

func WithDispatchAdmissionLiveness 

func WithDispatchAdmissionLiveness(
	leaseStore dispatch.DAGRunLeaseStore,
	activeRunStore dispatch.ActiveDistributedRunStore,
) DispatchTaskStoreOption

WithDispatchAdmissionLiveness enables admission cleanup against shared distributed liveness stores.

func WithDispatchReservationTTL 

func WithDispatchReservationTTL(ttl time.Duration) DispatchTaskStoreOption

WithDispatchReservationTTL sets how long pending and claimed dispatch records can remain outstanding before cleanup recycles or removes them.

func WithDispatchTransitionLock added in v2.15.4 

func WithDispatchTransitionLock(
	lock func(context.Context, func(context.Context) error) error,
) DispatchTaskStoreOption

WithDispatchTransitionLock serializes task transitions with external users of the same shared lock.

type DistributedStoreOption 

type DistributedStoreOption func(*distributedStoreOptions)

DistributedStoreOption configures file-corruption recovery for distributed control-plane stores.

func WithCorruptRecordGracePeriod 

func WithCorruptRecordGracePeriod(period time.Duration) DistributedStoreOption

WithCorruptRecordGracePeriod sets how long a corrupt distributed record remains fail-closed before it can be removed as stale.

type IncidentStore added in v2.14.0 

type IncidentStore struct {
	// contains filtered or unexported fields
}

IncidentStore persists incident configuration and state in a collection.

func NewIncidentStore added in v2.14.0 

func NewIncidentStore(col persis.Collection, enc *crypto.Encryptor) (*IncidentStore, error)

NewIncidentStore creates an incident store backed by col.

func (*IncidentStore) DeletePolicySet added in v2.14.0 

func (s *IncidentStore) DeletePolicySet(
	ctx context.Context,
	scope incident.PolicyScope,
	workspaceName, dagName string,
) error

func (*IncidentStore) DeleteProvider added in v2.14.0 

func (s *IncidentStore) DeleteProvider(ctx context.Context, providerID string) error

func (*IncidentStore) DeleteState added in v2.14.0 

func (s *IncidentStore) DeleteState(ctx context.Context, providerID, dedupKey string) error

func (*IncidentStore) GetPolicySet added in v2.14.0 

func (s *IncidentStore) GetPolicySet(
	ctx context.Context,
	scope incident.PolicyScope,
	workspaceName, dagName string,
) (*incident.PolicySet, error)

func (*IncidentStore) GetProvider added in v2.14.0 

func (s *IncidentStore) GetProvider(ctx context.Context, providerID string) (*incident.Provider, error)

func (*IncidentStore) GetState added in v2.14.0 

func (s *IncidentStore) GetState(ctx context.Context, providerID, dedupKey string) (*incident.IncidentState, error)

func (*IncidentStore) ListOpenStates added in v2.14.0 

func (s *IncidentStore) ListOpenStates(ctx context.Context) ([]*incident.IncidentState, error)

func (*IncidentStore) ListPolicySets added in v2.14.0 

func (s *IncidentStore) ListPolicySets(ctx context.Context) ([]*incident.PolicySet, error)

func (*IncidentStore) ListProviders added in v2.14.0 

func (s *IncidentStore) ListProviders(ctx context.Context) ([]*incident.Provider, error)

func (*IncidentStore) SavePolicySet added in v2.14.0 

func (s *IncidentStore) SavePolicySet(ctx context.Context, policySet *incident.PolicySet) error

func (*IncidentStore) SaveProvider added in v2.14.0 

func (s *IncidentStore) SaveProvider(ctx context.Context, provider *incident.Provider) error

func (*IncidentStore) SaveState added in v2.14.0 

func (s *IncidentStore) SaveState(ctx context.Context, state *incident.IncidentState) error

type LicenseStore 

type LicenseStore struct {
	// contains filtered or unexported fields
}

LicenseStore implements license.ActivationStore over a single persis.Collection record.

func NewLicenseStore 

func NewLicenseStore(col persis.Collection) *LicenseStore

NewLicenseStore creates a LicenseStore backed by col.

func (*LicenseStore) Load 

func (s *LicenseStore) Load() (*license.ActivationData, error)

Load returns the activation data, or (nil, nil) when no record exists.

func (*LicenseStore) Remove 

func (s *LicenseStore) Remove() error

Remove deletes the activation record. Returns nil when no record exists.

func (*LicenseStore) Save 

func (s *LicenseStore) Save(ad *license.ActivationData) error

Save replaces the stored activation data.

type NotificationStore added in v2.14.0 

type NotificationStore struct {
	// contains filtered or unexported fields
}

NotificationStore persists notification configuration in a collection.

func NewNotificationStore added in v2.14.0 

func NewNotificationStore(col persis.Collection, enc *crypto.Encryptor) (*NotificationStore, error)

NewNotificationStore creates a notification store backed by col.

func (*NotificationStore) DeleteByDAGName added in v2.14.0 

func (s *NotificationStore) DeleteByDAGName(ctx context.Context, dagName string) error

func (*NotificationStore) DeleteChannel added in v2.14.0 

func (s *NotificationStore) DeleteChannel(ctx context.Context, channelID string) error

func (*NotificationStore) DeleteRouteSet added in v2.14.0 

func (s *NotificationStore) DeleteRouteSet(ctx context.Context, scope notification.RouteScope, workspace string) error

func (*NotificationStore) GetByDAGName added in v2.14.0 

func (s *NotificationStore) GetByDAGName(ctx context.Context, dagName string) (*notification.Settings, error)

func (*NotificationStore) GetChannel added in v2.14.0 

func (s *NotificationStore) GetChannel(ctx context.Context, channelID string) (*notification.Channel, error)

func (*NotificationStore) GetRouteSet added in v2.14.0 

func (s *NotificationStore) GetRouteSet(
	ctx context.Context,
	scope notification.RouteScope,
	workspace string,
) (*notification.RouteSet, error)

func (*NotificationStore) GetWorkspaceSettings added in v2.14.0 

func (s *NotificationStore) GetWorkspaceSettings(ctx context.Context) (*notification.WorkspaceSettings, error)

func (*NotificationStore) List added in v2.14.0 

func (s *NotificationStore) List(ctx context.Context) ([]*notification.Settings, error)

func (*NotificationStore) ListChannels added in v2.14.0 

func (s *NotificationStore) ListChannels(ctx context.Context) ([]*notification.Channel, error)

func (*NotificationStore) ListRouteSets added in v2.14.0 

func (s *NotificationStore) ListRouteSets(ctx context.Context) ([]*notification.RouteSet, error)

func (*NotificationStore) Save added in v2.14.0 

func (s *NotificationStore) Save(ctx context.Context, settings *notification.Settings) error

func (*NotificationStore) SaveChannel added in v2.14.0 

func (s *NotificationStore) SaveChannel(ctx context.Context, channel *notification.Channel) error

func (*NotificationStore) SaveRouteSet added in v2.14.0 

func (s *NotificationStore) SaveRouteSet(ctx context.Context, routeSet *notification.RouteSet) error

func (*NotificationStore) SaveWorkspaceSettings added in v2.14.0 

func (s *NotificationStore) SaveWorkspaceSettings(ctx context.Context, settings *notification.WorkspaceSettings) error

type ProfileStore 

type ProfileStore struct {
	// contains filtered or unexported fields
}

func NewProfileStore 

func NewProfileStore(col persis.Collection) (*ProfileStore, error)

func (*ProfileStore) Create 

func (s *ProfileStore) Create(ctx context.Context, p *profile.Profile) error

func (*ProfileStore) Delete 

func (s *ProfileStore) Delete(ctx context.Context, name string) error

func (*ProfileStore) GetByName 

func (s *ProfileStore) GetByName(ctx context.Context, name string) (*profile.Profile, error)

func (*ProfileStore) GetInherited 

func (s *ProfileStore) GetInherited(ctx context.Context, ref profile.InheritedRef) (*profile.Profile, error)

func (*ProfileStore) List 

func (s *ProfileStore) List(ctx context.Context) ([]*profile.Profile, error)

func (*ProfileStore) Update 

func (s *ProfileStore) Update(ctx context.Context, p *profile.Profile) error

type QueueStore 

type QueueStore struct {
	// contains filtered or unexported fields
}

QueueStore implements queue.QueueStore on top of a persis.Collection. Records are keyed as "{queueName}/{itemID}", while item IDs exposed through queue.QueuedItemData intentionally stay as "{itemID}" for caller compatibility.

func NewQueueStore 

func NewQueueStore(col persis.Collection) *QueueStore

NewQueueStore creates a QueueStore backed by col.

func (*QueueStore) All 

func (s *QueueStore) All(ctx context.Context) ([]queue.QueuedItemData, error)

All returns all queued items across all queues.

func (*QueueStore) DeleteByItemIDs 

func (s *QueueStore) DeleteByItemIDs(ctx context.Context, name string, itemIDs []string) (int, error)

DeleteByItemIDs removes exact queue item IDs from the named queue.

func (*QueueStore) DequeueByDAGRunID 

func (s *QueueStore) DequeueByDAGRunID(ctx context.Context, name string, dagRun ir.DAGRunRef) ([]queue.QueuedItemData, error)

DequeueByDAGRunID removes all queued items matching dagRun from the named queue.

func (*QueueStore) Enqueue 

func (s *QueueStore) Enqueue(ctx context.Context, name string, priority queue.QueuePriority, dagRun ir.DAGRunRef) error

Enqueue adds a DAG-run reference to the named queue.

func (*QueueStore) GetByItemID added in v2.15.3 

func (s *QueueStore) GetByItemID(ctx context.Context, name, itemID string) (queue.QueuedItemData, error)

GetByItemID returns an exact queued item from the named queue.

func (*QueueStore) Len 

func (s *QueueStore) Len(ctx context.Context, name string) (int, error)

Len returns the number of queued items in the named queue.

func (*QueueStore) List 

func (s *QueueStore) List(ctx context.Context, name string) ([]queue.QueuedItemData, error)

List returns all queued items in the named queue.

func (*QueueStore) ListByDAGName 

func (s *QueueStore) ListByDAGName(ctx context.Context, name, dagName string) ([]queue.QueuedItemData, error)

ListByDAGName returns all items in a queue for a DAG name.

func (*QueueStore) ListCursor 

func (s *QueueStore) ListCursor(ctx context.Context, name, cursor string, limit int) (pagination.CursorResult[queue.QueuedItemData], error)

ListCursor returns one forward-only page of queued items.

func (*QueueStore) QueueList 

func (s *QueueStore) QueueList(ctx context.Context) ([]string, error)

QueueList lists queue names that currently have at least one item record.

func (*QueueStore) QueueWatcher 

func (s *QueueStore) QueueWatcher(ctx context.Context) queue.QueueWatcher

QueueWatcher returns a backend-neutral polling watcher.

func (*QueueStore) Revision added in v2.15.3 

func (s *QueueStore) Revision(ctx context.Context, name string) (int64, error)

Revision returns the current ordered membership revision of the named queue.

type RemoteNodeStore 

type RemoteNodeStore struct {
	// contains filtered or unexported fields
}

RemoteNodeStore implements remotenode.Store over persis.Collection. Credential fields are encrypted before Put and decrypted after Get.

func NewRemoteNodeStore 

func NewRemoteNodeStore(col persis.Collection, enc *crypto.Encryptor) (*RemoteNodeStore, error)

NewRemoteNodeStore creates a RemoteNodeStore backed by col.

func (*RemoteNodeStore) Create 

func (s *RemoteNodeStore) Create(ctx context.Context, node *remotenode.RemoteNode) error

Create stores a new remote node, encrypting credentials.

func (*RemoteNodeStore) Delete 

func (s *RemoteNodeStore) Delete(ctx context.Context, id string) error

Delete removes a remote node by ID.

func (*RemoteNodeStore) GetByID 

func (s *RemoteNodeStore) GetByID(ctx context.Context, id string) (*remotenode.RemoteNode, error)

GetByID retrieves a remote node by its ID and decrypts credentials.

func (*RemoteNodeStore) GetByName 

func (s *RemoteNodeStore) GetByName(ctx context.Context, name string) (*remotenode.RemoteNode, error)

GetByName retrieves a remote node by its name.

func (*RemoteNodeStore) List 

func (s *RemoteNodeStore) List(ctx context.Context) ([]*remotenode.RemoteNode, error)

List returns all remote nodes (with decrypted credentials).

func (*RemoteNodeStore) Update 

func (s *RemoteNodeStore) Update(ctx context.Context, node *remotenode.RemoteNode) error

Update modifies an existing remote node.

type SchedulerStateStore added in v2.14.0 

type SchedulerStateStore struct {
	// contains filtered or unexported fields
}

SchedulerStateStore persists scheduler state as collection records.

func NewSchedulerStateStore added in v2.14.0 

func NewSchedulerStateStore(col persis.Collection) *SchedulerStateStore

NewSchedulerStateStore creates a scheduler state store backed by col.

func (*SchedulerStateStore) Load added in v2.14.0 

func (s *SchedulerStateStore) Load(ctx context.Context) (*schedulerstate.State, error)

Load reads scheduler state, returning a fresh state when stored data is missing or unusable.

func (*SchedulerStateStore) Save added in v2.14.0 

func (s *SchedulerStateStore) Save(ctx context.Context, state *schedulerstate.State) error

Save writes scheduler state when its durable records have changed.

type SecretStore 

type SecretStore struct {
	// contains filtered or unexported fields
}

SecretStore implements secret.Store. byRef is an in-memory index rebuilt on startup; misses are resolved from persistence.

func NewSecretStore 

func NewSecretStore(col persis.Collection, enc *crypto.Encryptor) (*SecretStore, error)

NewSecretStore creates a SecretStore backed by col.

func (*SecretStore) Create 

func (s *SecretStore) Create(ctx context.Context, sec *secret.Secret, initialValue *secret.WriteValueInput) error

Create stores a new secret, optionally writing an initial encrypted value.

func (*SecretStore) Delete 

func (s *SecretStore) Delete(ctx context.Context, id string) error

Delete removes a secret by its ID.

func (*SecretStore) GetByID 

func (s *SecretStore) GetByID(ctx context.Context, id string) (*secret.Secret, error)

GetByID retrieves a secret by its unique ID.

func (*SecretStore) GetByRef 

func (s *SecretStore) GetByRef(ctx context.Context, workspace, ref string) (*secret.Secret, error)

GetByRef retrieves a secret by workspace and ref.

func (*SecretStore) GetCurrentVersion 

func (s *SecretStore) GetCurrentVersion(ctx context.Context, id string) (*secret.VersionMetadata, error)

GetCurrentVersion returns metadata about the current version without decrypting.

func (*SecretStore) List 

func (s *SecretStore) List(ctx context.Context, opts secret.ListOptions) ([]*secret.Secret, error)

List returns all secrets, optionally filtered by workspace.

func (*SecretStore) ResolveValue 

func (s *SecretStore) ResolveValue(ctx context.Context, id string) (string, *secret.VersionMetadata, error)

ResolveValue decrypts and returns the current plaintext value.

func (*SecretStore) Update 

func (s *SecretStore) Update(ctx context.Context, sec *secret.Secret) error

Update modifies an existing secret's metadata.

func (*SecretStore) WriteValue 

func (s *SecretStore) WriteValue(ctx context.Context, id string, input secret.WriteValueInput) (*secret.Secret, error)

WriteValue appends a new encrypted version to the secret.

type SingleRecord 

type SingleRecord[T any] struct {
	// contains filtered or unexported fields
}

SingleRecord persists one value at a fixed record ID within a collection.

It owns the mechanical encode/decode and Get/Put/Delete against the collection; callers layer their own policy — defaults, error handling, and concurrency — on top. It is the shared core of the single-record control-plane stores (license activation, upgrade-check cache, agent config, GitHub-dispatch tracking, scheduler watermark).

SingleRecord performs no locking of its own. The underlying collection is safe for concurrent use, but a caller needing read-modify-write atomicity must serialize access itself.

func NewSingleRecord 

func NewSingleRecord[T any](col persis.Collection, id string) *SingleRecord[T]

NewSingleRecord returns a SingleRecord addressing record id within col.

func (*SingleRecord[T]) Delete 

func (s *SingleRecord[T]) Delete(ctx context.Context) error

Delete removes the record. A missing record is not an error.

func (*SingleRecord[T]) Load 

func (s *SingleRecord[T]) Load(ctx context.Context, dst *T) (found bool, err error)

Load fetches the record and decodes it into dst.

When no record exists it returns found=false and leaves dst untouched, so a caller may pre-populate dst with defaults before calling Load and keep them when the record is absent. A backend read failure is returned unchanged. A payload that cannot be decoded is reported as an error satisfying errors.Is(err, ErrCorrupt).

func (*SingleRecord[T]) Save 

func (s *SingleRecord[T]) Save(ctx context.Context, v *T) error

Save encodes v and writes it as the record, stamping the current time.

type UpgradeCheckStore 

type UpgradeCheckStore struct {
	// contains filtered or unexported fields
}

UpgradeCheckStore implements upgrade.CacheStore over a single persis.Collection record.

func NewUpgradeCheckStore 

func NewUpgradeCheckStore(col persis.Collection) *UpgradeCheckStore

NewUpgradeCheckStore creates an UpgradeCheckStore backed by col.

func (*UpgradeCheckStore) Load 

func (s *UpgradeCheckStore) Load() (*upgrade.UpgradeCheckCache, error)

Load returns the cached upgrade-check data, or (nil, nil) when no usable record exists. Any read or decode failure is reported as a cache miss.

func (*UpgradeCheckStore) Save 

func (s *UpgradeCheckStore) Save(cache *upgrade.UpgradeCheckCache) error

Save replaces the cached upgrade-check data.

type UserStore 

type UserStore struct {
	// contains filtered or unexported fields
}

UserStore implements auth.UserStore. Secondary indices are rebuilt from the collection on startup and kept in sync under mu.

func NewUserStore 

func NewUserStore(col persis.Collection) (*UserStore, error)

NewUserStore creates a UserStore backed by col.

func (*UserStore) Count 

func (s *UserStore) Count(_ context.Context) (int64, error)

Count returns the total number of users.

func (*UserStore) Create 

func (s *UserStore) Create(ctx context.Context, user *auth.User) error

Create stores a new user. Returns auth.ErrUserAlreadyExists if a user with the same username exists.

func (*UserStore) Delete 

func (s *UserStore) Delete(ctx context.Context, id string) error

Delete removes a user by their ID. Returns auth.ErrUserNotFound if the user does not exist.

func (*UserStore) GetByID 

func (s *UserStore) GetByID(ctx context.Context, id string) (*auth.User, error)

GetByID retrieves a user by their unique ID. Returns auth.ErrUserNotFound if the user does not exist.

func (*UserStore) GetByOIDCIdentity 

func (s *UserStore) GetByOIDCIdentity(ctx context.Context, issuer, subject string) (*auth.User, error)

GetByOIDCIdentity retrieves a user by their OIDC identity. Returns auth.ErrOIDCIdentityNotFound if no user exists with the given identity.

func (*UserStore) GetByTrustedProxyIdentity 

func (s *UserStore) GetByTrustedProxyIdentity(ctx context.Context, source, identity string) (*auth.User, error)

GetByTrustedProxyIdentity retrieves a user by their proxy identity source and user. Returns auth.ErrTrustedProxyIdentityNotFound if no user exists with the given identity.

func (*UserStore) GetByUsername 

func (s *UserStore) GetByUsername(ctx context.Context, username string) (*auth.User, error)

GetByUsername retrieves a user by their username. Returns auth.ErrUserNotFound if the user does not exist.

func (*UserStore) List 

func (s *UserStore) List(ctx context.Context) ([]*auth.User, error)

List returns all users in the store.

func (*UserStore) Patch 

func (s *UserStore) Patch(ctx context.Context, id string, patch auth.UserPatch) (*auth.User, error)

Patch atomically applies selected account fields.

func (*UserStore) SyncAuthorization 

func (s *UserStore) SyncAuthorization(
	ctx context.Context,
	id string,
	role auth.Role,
	workspaceAccess *auth.WorkspaceAccess,
) (auth.AuthorizationSyncResult, error)

SyncAuthorization updates only externally managed authorization fields.

func (*UserStore) Update 

func (s *UserStore) Update(ctx context.Context, user *auth.User) error

Update modifies an existing user. Returns auth.ErrUserNotFound if the user does not exist.

type ViewStore 

type ViewStore struct {
	// contains filtered or unexported fields
}

ViewStore implements view.Store over a persis.Collection. Views are global and shared: each record is keyed by its view ID with no per-user scoping. Views have no secondary-key lookups, so no in-memory index is maintained; writes are serialized under mu.

func NewViewStore 

func NewViewStore(col persis.Collection) *ViewStore

NewViewStore creates a ViewStore backed by col.

func (*ViewStore) Create 

func (s *ViewStore) Create(ctx context.Context, v *view.View) error

Create stores a new view. Returns view.ErrViewExists if a view with the same ID already exists.

func (*ViewStore) Delete 

func (s *ViewStore) Delete(ctx context.Context, id string, expectedWorkspace string) error

Delete removes a view by ID. Returns view.ErrViewNotFound if absent.

func (*ViewStore) GetByID 

func (s *ViewStore) GetByID(ctx context.Context, id string) (*view.View, error)

GetByID retrieves a view by ID. Returns view.ErrViewNotFound if absent.

func (*ViewStore) List 

func (s *ViewStore) List(ctx context.Context) ([]*view.View, error)

List returns all views ordered by creation time ascending, tiebroken by ID.

func (*ViewStore) Update 

func (s *ViewStore) Update(ctx context.Context, v *view.View, expectedWorkspace string) error

Update replaces an existing view, preserving its original CreatedAt and stamping UpdatedAt with the current time. Returns view.ErrViewNotFound if the view does not exist.

type WebhookStore 

type WebhookStore struct {
	// contains filtered or unexported fields
}

WebhookStore implements auth.WebhookStore. DAG-name lookups use an in-memory index (byDAGName) rebuilt from the collection on startup; all writes keep it in sync under mu.

func NewWebhookStore 

func NewWebhookStore(col persis.Collection, enc *crypto.Encryptor) (*WebhookStore, error)

NewWebhookStore creates a WebhookStore backed by col. enc may be nil when HMAC secrets are unused.

func (*WebhookStore) Create 

func (s *WebhookStore) Create(ctx context.Context, webhook *auth.Webhook) error

Create stores a new webhook. Returns auth.ErrWebhookAlreadyExists if a webhook for the DAG already exists.

func (*WebhookStore) Delete 

func (s *WebhookStore) Delete(ctx context.Context, id string) error

Delete removes a webhook by its ID. Returns auth.ErrWebhookNotFound if the webhook does not exist.

func (*WebhookStore) DeleteByDAGName 

func (s *WebhookStore) DeleteByDAGName(ctx context.Context, dagName string) error

DeleteByDAGName removes a webhook by its DAG name. Returns auth.ErrWebhookNotFound if no webhook exists for the DAG.

func (*WebhookStore) GetByDAGName 

func (s *WebhookStore) GetByDAGName(ctx context.Context, dagName string) (*auth.Webhook, error)

GetByDAGName retrieves the webhook for a specific DAG. Returns auth.ErrWebhookNotFound if no webhook exists for the DAG.

func (*WebhookStore) GetByID 

func (s *WebhookStore) GetByID(ctx context.Context, id string) (*auth.Webhook, error)

GetByID retrieves a webhook by its unique ID. Returns auth.ErrWebhookNotFound if the webhook does not exist.

func (*WebhookStore) List 

func (s *WebhookStore) List(ctx context.Context) ([]*auth.Webhook, error)

List returns all webhooks in the store.

func (*WebhookStore) Update 

func (s *WebhookStore) Update(ctx context.Context, webhook *auth.Webhook) error

Update modifies an existing webhook. Returns auth.ErrWebhookNotFound if the webhook does not exist.

func (*WebhookStore) UpdateLastUsed 

func (s *WebhookStore) UpdateLastUsed(ctx context.Context, id string) error

UpdateLastUsed updates the LastUsedAt timestamp for a webhook.

type WorkerHeartbeatStore 

type WorkerHeartbeatStore struct {
	// contains filtered or unexported fields
}

WorkerHeartbeatStore implements dispatch.WorkerHeartbeatStore. No secondary indices are needed; workerID is the primary key.

func NewWorkerHeartbeatStore 

func NewWorkerHeartbeatStore(col persis.Collection) *WorkerHeartbeatStore

NewWorkerHeartbeatStore creates a WorkerHeartbeatStore backed by col.

func (*WorkerHeartbeatStore) DeleteStale 

func (s *WorkerHeartbeatStore) DeleteStale(ctx context.Context, before time.Time) (int, error)

DeleteStale removes all records whose last heartbeat is before the given time. Returns the number of records deleted.

func (*WorkerHeartbeatStore) Get 

func (s *WorkerHeartbeatStore) Get(ctx context.Context, workerID string) (*dispatch.WorkerHeartbeatRecord, error)

Get retrieves the heartbeat record for a specific worker.

func (*WorkerHeartbeatStore) List 

func (s *WorkerHeartbeatStore) List(ctx context.Context) ([]dispatch.WorkerHeartbeatRecord, error)

List returns all heartbeat records.

func (*WorkerHeartbeatStore) Upsert 

func (s *WorkerHeartbeatStore) Upsert(ctx context.Context, record dispatch.WorkerHeartbeatRecord) error

Upsert inserts or overwrites the heartbeat record for a worker.

type WorkspaceStore 

type WorkspaceStore struct {
	// contains filtered or unexported fields
}

WorkspaceStore implements workspace.Store over persis.Collection, keeping an in-memory name→id index rebuilt on startup.

func NewWorkspaceStore 

func NewWorkspaceStore(col persis.Collection) (*WorkspaceStore, error)

NewWorkspaceStore creates a WorkspaceStore backed by col.

func (*WorkspaceStore) Create 

func (s *WorkspaceStore) Create(ctx context.Context, ws *workspace.Workspace) error

Create stores a new workspace.

func (*WorkspaceStore) Delete 

func (s *WorkspaceStore) Delete(ctx context.Context, id string) error

Delete removes the workspace with the given ID.

func (*WorkspaceStore) GetByID 

func (s *WorkspaceStore) GetByID(ctx context.Context, id string) (*workspace.Workspace, error)

GetByID returns the workspace with the given ID.

func (*WorkspaceStore) GetByName 

func (s *WorkspaceStore) GetByName(ctx context.Context, name string) (*workspace.Workspace, error)

GetByName returns the workspace with the given name.

func (*WorkspaceStore) List 

func (s *WorkspaceStore) List(ctx context.Context) ([]*workspace.Workspace, error)

List returns all workspaces.

func (*WorkspaceStore) Update 

func (s *WorkspaceStore) Update(ctx context.Context, ws *workspace.Workspace) error

Update modifies an existing workspace.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.