mongodb

package
v0.7.0 Latest Latest
Warning

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

 Go to latest Published: Oct 5, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Overview

Package mongodb holds everything MongoDB-specific: the $jsonSchema validator translation, the repository implementation and the provisioner.

Index

Constants

View Source 
const (
	DeploymentDatabase   = "backd___deployment"
	DeploymentCollection = "realms"
)

DeploymentDatabase holds what provisioning applied: per realm, the fingerprint of the config it provisioned. Its name can't clash with a realm's databases (<realm>__<database>, where a database name can't start with "_").

View Source 
const (
	UsersCollection         = "users"
	IdentitiesCollection    = "identities"
	SessionsCollection      = "sessions"
	APIKeysCollection       = "api_keys"
	InvitationsCollection   = "invitations"
	LoginAttemptsCollection = "login_attempts"
	// AuditCollection is append-only: backd inserts and reads, and MongoDB
	// removes records when they expire.
	AuditCollection = "audit"
	// SecretsCollection holds encrypted values for functions (roadmap F5).
	SecretsCollection = "secrets"
	// InvocationsCollection is append-only, like AuditCollection: one
	// record per function call (roadmap F10).
	InvocationsCollection = "invocations"
	// JobsCollection holds async function calls (roadmap F11), from
	// enqueue through a worker's claim to their result, until it's
	// removed by expiry.
	JobsCollection = "jobs"
	// IdempotencyCollection ties an Idempotency-Key to its outcome
	// (roadmap F12), scoped to the function and the caller.
	IdempotencyCollection = "idempotency"
	// EmailTokensCollection holds the hashes of the tokens sent in emails.
	EmailTokensCollection = "email_tokens"
)

System collections of a realm's system database.

Variables

View Source 
var ErrConfigMismatch = errors.New("the config differs from the one last provisioned")

ErrConfigMismatch means this instance's config isn't the one last provisioned.

View Source 
var ErrStandalone = errors.New("mongodb: the server is a standalone instance; backd needs a replica set (a single node is fine): start mongod with --replSet and run rs.initiate(), then add ?replicaSet=<name> to MONGO_URI")

ErrStandalone means MongoDB isn't a replica set member (or a sharded cluster's router), so transactions aren't available.

Functions

func Connect 

func Connect(ctx context.Context, uri string, monitor ...*event.CommandMonitor) (*mongo.Client, error)

Connect opens a client and checks the server is reachable and part of a replica set (or a sharded cluster). A monitor, when given, sees every command the driver sends (backd's database latency metrics).

func SystemCollectionNames 

func SystemCollectionNames() []string

SystemCollectionNames returns the collections of every system database.

func SystemDatabaseName 

func SystemDatabaseName(realm string) string

SystemDatabaseName is the MongoDB database holding a realm's users, sessions and keys. Three underscores can't come from a valid database name, so it never collides with <realm>__<database>.

Types

type AuthStore 

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

AuthStore implements auth.Store on a realm's system database.

func NewAuthStore 

func NewAuthStore(client *mongo.Client, realm string) *AuthStore

NewAuthStore returns the auth store of realm.

func (*AuthStore) APIKeyByHash 

func (s *AuthStore) APIKeyByHash(ctx context.Context, hash string) (auth.APIKey, error)

func (*AuthStore) AcquireLoginLock added in v0.5.0 

func (s *AuthStore) AcquireLoginLock(ctx context.Context, key, owner string, now, until time.Time) (bool, error)

AcquireLoginLock stores the lock as a document of the attempts collection (so it also expires by TTL if a process dies holding it). The upsert only matches a lock whose lease ended, or inserts one: against a live lock it would insert a second document with the same _id, which fails as a duplicate key.

func (*AuthStore) AddEraseCount added in v0.4.0 

func (s *AuthStore) AddEraseCount(ctx context.Context, jobID, key string, n int64) error

AddEraseCount adds n to one of an erase job's counters. The counters are incremented, never read and written back, so two workers finishing the same batch can't lose each other's counts.

func (*AuthStore) AddRoles 

func (s *AuthStore) AddRoles(ctx context.Context, userID string, roles []string, now time.Time) error

func (*AuthStore) AppendAudit 

func (s *AuthStore) AppendAudit(ctx context.Context, r auth.AuditRecord) error

AppendAudit inserts a record. backd never updates or deletes audit records; MongoDB removes them when they expire.

func (*AuthStore) ClaimIdempotency 

func (s *AuthStore) ClaimIdempotency(ctx context.Context, rec auth.IdempotencyRecord) (auth.IdempotencyRecord, bool, error)

ClaimIdempotency inserts rec if its id isn't already used; a duplicate key means someone already claimed it, so the existing record is fetched and returned instead, claimed false.

func (*AuthStore) ClaimInvitation 

func (s *AuthStore) ClaimInvitation(ctx context.Context, hash string, now time.Time) (auth.Invitation, error)

func (*AuthStore) ClaimInvitationByID added in v0.4.0 

func (s *AuthStore) ClaimInvitationByID(ctx context.Context, id string, now time.Time) (auth.Invitation, error)

ClaimInvitationByID deletes and returns an unexpired invitation by id.

func (*AuthStore) ClaimJob 

func (s *AuthStore) ClaimJob(ctx context.Context, workerID string, at time.Time, margin time.Duration) (auth.Job, bool, error)

ClaimJob atomically claims the oldest claimable job: never claimed (status queued, no lease yet), or whose lease has expired (a dead worker's jobs are picked up again). The lease extends to at plus the job's own timeout_ms plus margin, computed in the same update so a claim never needs a second round trip to look the function's timeout up.

func (*AuthStore) ClearEraseEmail added in v0.4.0 

func (s *AuthStore) ClearEraseEmail(ctx context.Context, jobID string) error

ClearEraseEmail removes the email an erase job held.

func (*AuthStore) ClearLoginAttempts 

func (s *AuthStore) ClearLoginAttempts(ctx context.Context, key string) error

func (*AuthStore) CompleteIdempotency 

func (s *AuthStore) CompleteIdempotency(ctx context.Context, id string, result *auth.JobResult, jobID string, completedAt, expiresAt time.Time) error

CompleteIdempotency stores a claimed key's outcome: result for a sync call, jobID for async (mutually exclusive). Only the one that applies is set — the collection's validator types "result" as an object, so setting it to an explicit BSON null (rather than leaving it absent) would fail validation.

func (*AuthStore) CompleteJob 

func (s *AuthStore) CompleteJob(ctx context.Context, id string, result auth.JobResult, completedAt, expiresAt time.Time) error

CompleteJob records a job's result; expiresAt starts its retention countdown. The update is conditional on the job not already being done, so a worker whose lease expired and was reclaimed can't finish late and overwrite a result another worker already recorded.

func (*AuthStore) CreateAPIKey 

func (s *AuthStore) CreateAPIKey(ctx context.Context, k auth.APIKey) error

func (*AuthStore) CreateEmailToken added in v0.4.0 

func (s *AuthStore) CreateEmailToken(ctx context.Context, t auth.EmailToken) error

CreateEmailToken stores a token's hash.

func (*AuthStore) CreateInvitation 

func (s *AuthStore) CreateInvitation(ctx context.Context, inv auth.Invitation) error

func (*AuthStore) CreateSession 

func (s *AuthStore) CreateSession(ctx context.Context, x auth.Session) error

func (*AuthStore) CreateUser 

func (s *AuthStore) CreateUser(ctx context.Context, u auth.User) error

func (*AuthStore) DeleteAPIKey 

func (s *AuthStore) DeleteAPIKey(ctx context.Context, name string) error

func (*AuthStore) DeleteEmailJobsOfUser added in v0.4.0 

func (s *AuthStore) DeleteEmailJobsOfUser(ctx context.Context, userID string) error

DeleteEmailJobsOfUser removes the email jobs addressed to the user.

func (*AuthStore) DeleteEmailTokensOfUser added in v0.4.0 

func (s *AuthStore) DeleteEmailTokensOfUser(ctx context.Context, userID string) error

DeleteEmailTokensOfUser removes every token of the user.

func (*AuthStore) DeleteInvitation 

func (s *AuthStore) DeleteInvitation(ctx context.Context, id string) error

func (*AuthStore) DeleteOtherSessions 

func (s *AuthStore) DeleteOtherSessions(ctx context.Context, userID, keepID string) (int64, error)

func (*AuthStore) DeleteSecret 

func (s *AuthStore) DeleteSecret(ctx context.Context, database, name string) error

func (*AuthStore) DeleteSession 

func (s *AuthStore) DeleteSession(ctx context.Context, userID, id string) error

func (*AuthStore) DeleteSessions 

func (s *AuthStore) DeleteSessions(ctx context.Context, userID string) (int64, error)

func (*AuthStore) DeleteUser 

func (s *AuthStore) DeleteUser(ctx context.Context, id string) error

DeleteUser removes sessions and identities first, so an interrupted delete never leaves credentials without their user.

func (*AuthStore) EnqueueJob 

func (s *AuthStore) EnqueueJob(ctx context.Context, j auth.Job) error

EnqueueJob inserts a new job, queued for a worker to claim.

func (*AuthStore) EraseNeedsAttention added in v0.4.0 

func (s *AuthStore) EraseNeedsAttention(ctx context.Context) (int64, error)

EraseNeedsAttention counts the erase jobs that ended in failure: their attempts were used up, and an administrator has to repeat the erase.

func (*AuthStore) EraseUser added in v0.4.0 

func (s *AuthStore) EraseUser(ctx context.Context, id, placeholderEmail string, now time.Time) error

EraseUser turns the user into a tombstone. Sessions and sign-in methods go first, like DeleteUser, so an interrupted erase never leaves credentials for a user without their identity.

func (*AuthStore) GetEmailToken added in v0.4.0 

func (s *AuthStore) GetEmailToken(ctx context.Context, hash string) (auth.EmailToken, error)

GetEmailToken returns a token by hash, used or not.

func (*AuthStore) GetJob 

func (s *AuthStore) GetJob(ctx context.Context, id string) (auth.Job, bool, error)

GetJob returns one job by id.

func (*AuthStore) Identity 

func (s *AuthStore) Identity(ctx context.Context, provider, subject string) (auth.Identity, error)

func (*AuthStore) IncrementCounter 

func (s *AuthStore) IncrementCounter(ctx context.Context, key string, at, expires time.Time) (int, time.Time, error)

IncrementCounter reuses the login-throttling collection and pipeline (see RecordLoginFailure) but returns the resulting count and reset time, which a caller that must decide "over the limit or not" needs and login throttling itself doesn't.

func (*AuthStore) InvalidateEmailTokens added in v0.4.0 

func (s *AuthStore) InvalidateEmailTokens(ctx context.Context, userID, purpose string, now time.Time) error

InvalidateEmailTokens marks used every unused token of the user and purpose.

func (*AuthStore) JobStats added in v0.4.0 

func (s *AuthStore) JobStats(ctx context.Context, now time.Time) ([]metrics.JobStat, error)

JobStats summarizes the jobs that are not done, by kind and state: queued (claimable now), running (holding a lease) and waiting (for a retry's delay). It is the metrics refresher's query, never a scrape's.

func (*AuthStore) ListAPIKeys 

func (s *AuthStore) ListAPIKeys(ctx context.Context) ([]auth.APIKey, error)

func (*AuthStore) ListAudit 

func (s *AuthStore) ListAudit(ctx context.Context, f auth.AuditFilter) ([]auth.AuditRecord, bool, error)

func (*AuthStore) ListInvitations 

func (s *AuthStore) ListInvitations(ctx context.Context) ([]auth.Invitation, error)

func (*AuthStore) ListInvocations 

func (s *AuthStore) ListInvocations(ctx context.Context, f auth.InvocationFilter) ([]auth.InvocationRecord, bool, error)

func (*AuthStore) ListJobs 

func (s *AuthStore) ListJobs(ctx context.Context, f auth.JobFilter) ([]auth.Job, bool, error)

ListJobs returns the jobs matching f, newest first.

func (*AuthStore) ListSecrets 

func (s *AuthStore) ListSecrets(ctx context.Context) ([]auth.Secret, error)

func (*AuthStore) ListSessions 

func (s *AuthStore) ListSessions(ctx context.Context, userID string) ([]auth.Session, error)

func (*AuthStore) ListUnverifiedUsers added in v0.4.0 

func (s *AuthStore) ListUnverifiedUsers(ctx context.Context, before time.Time, limit int) ([]auth.User, error)

ListUnverifiedUsers returns the oldest users that never verified their address.

func (*AuthStore) ListUsers 

func (s *AuthStore) ListUsers(ctx context.Context) ([]auth.User, error)

func (*AuthStore) ListUsersPage added in v0.5.0 

func (s *AuthStore) ListUsersPage(ctx context.Context, contains, after string, skip, limit int) ([]auth.User, bool, error)

func (*AuthStore) LoginAttempts 

func (s *AuthStore) LoginAttempts(ctx context.Context, key string) (auth.Attempts, error)

func (*AuthStore) PutIdentity 

func (s *AuthStore) PutIdentity(ctx context.Context, id auth.Identity) error

func (*AuthStore) RecordInvocation 

func (s *AuthStore) RecordInvocation(ctx context.Context, r auth.InvocationRecord) error

RecordInvocation inserts a record. backd never updates or deletes invocation records; MongoDB removes them when they expire.

func (*AuthStore) RecordLoginFailure 

func (s *AuthStore) RecordLoginFailure(ctx context.Context, key string, at, expires time.Time) error

RecordLoginFailure increments the counter in one atomic update. A counter whose window ended (but that the TTL monitor, which runs about once a minute, hasn't deleted yet) restarts at 1.

func (*AuthStore) RedeemEmailToken added in v0.4.0 

func (s *AuthStore) RedeemEmailToken(ctx context.Context, hash, purpose string, now time.Time) (auth.EmailToken, error)

RedeemEmailToken marks the token used in one update that matches only an unused, unexpired token of this purpose, so two parallel redemptions can't both succeed.

func (*AuthStore) ReleaseIdempotency 

func (s *AuthStore) ReleaseIdempotency(ctx context.Context, id string) error

ReleaseIdempotency removes a claim that shouldn't be remembered.

func (*AuthStore) ReleaseLoginLock added in v0.5.0 

func (s *AuthStore) ReleaseLoginLock(ctx context.Context, key, owner string) error

func (*AuthStore) RemoveRole 

func (s *AuthStore) RemoveRole(ctx context.Context, userID, role string, now time.Time) error

func (*AuthStore) ReopenJob added in v0.4.0 

func (s *AuthStore) ReopenJob(ctx context.Context, id string, expiresAt time.Time) (bool, error)

ReopenJob queues a job that ended with an error again.

func (*AuthStore) RetryJob added in v0.4.0 

func (s *AuthStore) RetryJob(ctx context.Context, id string, notBefore time.Time) error

RetryJob counts a failed attempt and queues the job again: its lease runs until notBefore, so no worker claims it earlier. Conditional on the job not being done, like CompleteJob.

func (*AuthStore) SecretByScope 

func (s *AuthStore) SecretByScope(ctx context.Context, database, name string) (auth.Secret, error)

func (*AuthStore) SessionByTokenHash 

func (s *AuthStore) SessionByTokenHash(ctx context.Context, hash string) (auth.Session, auth.User, error)

SessionByTokenHash loads the session and its user in one round trip.

func (*AuthStore) TouchAPIKey 

func (s *AuthStore) TouchAPIKey(ctx context.Context, hash string, at time.Time) error

func (*AuthStore) TouchSession 

func (s *AuthStore) TouchSession(ctx context.Context, id string, lastUsed, expires time.Time) error

func (*AuthStore) UpdateUser 

func (s *AuthStore) UpdateUser(ctx context.Context, id string, upd auth.UserUpdate, now time.Time) error

func (*AuthStore) UpsertSecret 

func (s *AuthStore) UpsertSecret(ctx context.Context, sec auth.Secret) error

UpsertSecret creates or replaces the secret at (s.Database, s.Name), keeping its original created_at on an update.

func (*AuthStore) UserByEmail 

func (s *AuthStore) UserByEmail(ctx context.Context, email string) (auth.User, error)

func (*AuthStore) UserByID 

func (s *AuthStore) UserByID(ctx context.Context, id string) (auth.User, error)

type Omission 

type Omission struct {
	Path    string // JSON pointer of the schema containing the keyword
	Keyword string
}

Omission records a JSON Schema keyword that has no MongoDB $jsonSchema equivalent and was left out of the database validator.

func Validator 

func Validator(schema map[string]any) (map[string]any, []Omission)

Validator translates a collection's JSON Schema (as decoded by the registry) into a MongoDB $jsonSchema document, injecting the system fields _id and _meta at the top level. It never modifies schema.

type Provisioner 

type Provisioner struct {
	Client   *mongo.Client
	Registry *registry.Registry
	Log      *slog.Logger
	// Fingerprint of the config (registry.Fingerprint): Apply records it per
	// realm, Verify refuses a different one. Empty skips both.
	Fingerprint string
	Version     string // backd's version, recorded with the fingerprint
}

Provisioner makes MongoDB match the registry (Apply) or checks that it does (Verify). It never drops anything: objects in MongoDB that are not in the config are only reported as warnings.

func (*Provisioner) Apply 

func (p *Provisioner) Apply(ctx context.Context) error

Apply creates missing collections with their validators, updates validators that differ and creates missing declared indexes. It is idempotent.

func (*Provisioner) Verify 

func (p *Provisioner) Verify(ctx context.Context) error

Verify compares collections, validators and indexes with the registry without changing anything and returns an error listing every difference.

func (*Provisioner) VerifySystem 

func (p *Provisioner) VerifySystem(ctx context.Context, realm string) error

VerifySystem checks one realm's system database, for commands that write to it without provisioning first.

type Repository 

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

Repository implements storage.Repository for one MongoDB collection.

func (*Repository) AnonymizeOwned added in v0.4.0 

func (r *Repository) AnonymizeOwned(ctx context.Context, owner string, remove []string, replace map[string]any, limit int, now time.Time) (int64, error)

func (*Repository) ClearReference added in v0.4.0 

func (r *Repository) ClearReference(ctx context.Context, field, value string, limit int, now time.Time) (int64, error)

func (*Repository) CountOwned added in v0.4.0 

func (r *Repository) CountOwned(ctx context.Context, owner string) (int64, error)

func (*Repository) CountReferences added in v0.4.0 

func (r *Repository) CountReferences(ctx context.Context, field, value string, array bool) (int64, error)

func (*Repository) Create 

func (r *Repository) Create(ctx context.Context, doc storage.Document) error

func (*Repository) Delete 

func (r *Repository) Delete(ctx context.Context, id string, ifVersion *int64) error

func (*Repository) DeleteOwned added in v0.4.0 

func (r *Repository) DeleteOwned(ctx context.Context, owner string, limit int) (int64, error)

func (*Repository) Get 

func (r *Repository) Get(ctx context.Context, id string) (storage.Document, error)

func (*Repository) List 

func (r *Repository) List(ctx context.Context, q storage.Query) (storage.Page, error)

func (*Repository) PullReference added in v0.4.0 

func (r *Repository) PullReference(ctx context.Context, field, value string, limit int, now time.Time) (int64, error)

func (*Repository) Replace 

func (r *Repository) Replace(ctx context.Context, doc storage.Document, ifVersion int64) error

type Store 

type Store struct {
	Client *mongo.Client
}

Store hands out repositories backed by one MongoDB client.

func (*Store) Repository 

func (s *Store) Repository(c *registry.Collection) storage.Repository

Repository returns the repository of a configured collection.

func (*Store) Transact 

func (s *Store) Transact(ctx context.Context, fn func(ctx context.Context) error) error

Transact runs fn once inside a MongoDB transaction: every write made through a Repository, using the context fn receives, either all commit or none do (roadmap F7). The driver may retry fn on a transient error (e.g. a write conflict with another transaction), so fn must have no side effects beyond the repositories it's given.

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.