Help us shape the future of Go! Take the 2026 Go Developer Survey

postgres

package
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Oct 28, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Index

Constants

View Source 
const JobsChannel = "qf_jobs"

JobsChannel is the Postgres LISTEN/NOTIFY channel that EnqueueJob signals on commit.

Variables

This section is empty.

Functions

func GenerateSecureToken 

func GenerateSecureToken() (string, []byte, string, error)

func HashSecret 

func HashSecret(secretStr string) []byte

func RunMigrations 

func RunMigrations(databaseURL string) error

RunMigrations applies versioned migrations using goose

Types

type APIKeyCreationResult 

type APIKeyCreationResult struct {
	APIKey    *domain.APIKey
	RawSecret string
}

type AuditEntry 

type AuditEntry struct {
	ID         int64           `json:"id"`
	OccurredAt time.Time       `json:"occurred_at"`
	TenantID   *uuid.UUID      `json:"tenant_id,omitempty"`
	ActorType  string          `json:"actor_type"` // "api_key" | "operator"
	ActorID    string          `json:"actor_id"`
	Action     string          `json:"action"`
	TargetID   string          `json:"target_id,omitempty"`
	StatusCode int             `json:"status_code,omitempty"`
	RequestID  string          `json:"request_id,omitempty"`
	RemoteAddr string          `json:"remote_addr,omitempty"`
	Details    json.RawMessage `json:"details,omitempty"`
}

AuditEntry is one control-plane action.

type AuditRepo 

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

func NewAuditRepo 

func NewAuditRepo(db *DB) *AuditRepo

func (*AuditRepo) List 

func (r *AuditRepo) List(ctx context.Context, tenantID uuid.UUID, limit int) ([]AuditEntry, error)

List returns a tenant's most recent entries, newest first. Always tenant-scoped.

func (*AuditRepo) Record 

func (r *AuditRepo) Record(ctx context.Context, e AuditEntry) error

Record appends an entry. The timestamp is assigned by the database clock.

type ClaimResult 

type ClaimResult struct {
	Job        *domain.Job
	LeaseToken uuid.UUID
	SlotNo     int
	// QueueWaitSeconds is how long the job was runnable before this claim, measured entirely on
	// the database clock (the worker's own clock is never consulted).
	QueueWaitSeconds float64
}

type CreateScheduleParams 

type CreateScheduleParams struct {
	TenantID       uuid.UUID
	Name           string
	QueueName      string
	TaskType       string
	Payload        json.RawMessage
	CronExpression string
	Timezone       string
	Priority       int16
	MaxAttempts    int
	TimeoutSeconds int
	MisfirePolicy  domain.MisfirePolicy
}

type CreateTenantParams 

type CreateTenantParams struct {
	ID                   uuid.UUID
	Name                 string
	EnqueueRatePerSecond float64
	EnqueueBurstCapacity float64
	MaxPendingJobs       int
	MaxConcurrentJobs    int
	MaxSchedules         int
}

type DB 

type DB struct {
	Pool *pgxpool.Pool
}

func NewDB 

func NewDB(ctx context.Context, cfg *config.Config) (*DB, error)

func (*DB) Close 

func (db *DB) Close()

func (*DB) Ping 

func (db *DB) Ping(ctx context.Context) error

type EnqueueParams 

type EnqueueParams struct {
	TenantID        uuid.UUID
	TaskType        string
	QueueName       string
	Payload         json.RawMessage
	Priority        int16
	RunAt           time.Time
	IsExplicitRunAt bool
	MaxAttempts     int
	TimeoutSeconds  int
	IdempotencyKey  string
	IdempotencyTTL  time.Duration
	ScheduleID      *uuid.UUID
	ScheduledFor    *time.Time
	TraceID         string // correlation id; generated if empty
}

type EnqueueResult 

type EnqueueResult struct {
	Job              *domain.Job
	IdempotentReplay bool
	RetryAfterHint   time.Duration
}

type HeartbeatParams 

type HeartbeatParams struct {
	WorkerID    uuid.UUID
	ProcessName string
	Hostname    string
	Version     string
	Queues      []string
	TaskTypes   []string
	Draining    bool
	Metadata    json.RawMessage
}

type JobRepo 

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

func NewJobRepo 

func NewJobRepo(db *DB) *JobRepo

func (*JobRepo) CancelJob 

func (r *JobRepo) CancelJob(ctx context.Context, tenantID, jobID uuid.UUID) (*domain.Job, error)

func (*JobRepo) ClaimJob 

func (r *JobRepo) ClaimJob(ctx context.Context, workerID uuid.UUID, allowedQueues, supportedTaskTypes []string, leaseDuration time.Duration) (*ClaimResult, error)

ClaimJob leases a single job. See ClaimJobs.

func (*JobRepo) ClaimJobs 

func (r *JobRepo) ClaimJobs(ctx context.Context, workerID uuid.UUID, allowedQueues, supportedTaskTypes []string, leaseDuration time.Duration, max int) ([]*ClaimResult, error)

ClaimJobs leases up to max jobs in one transaction (usually one statement). Jobs are chosen by priority then FIFO across all tenants, subject to each tenant's free execution slots. It returns fewer than max jobs when nothing more is claimable.

func (*JobRepo) CompleteJobCancelled 

func (r *JobRepo) CompleteJobCancelled(ctx context.Context, tenantID, jobID, leaseToken uuid.UUID, durationMs int64) error

func (*JobRepo) CompleteJobFailure 

func (r *JobRepo) CompleteJobFailure(ctx context.Context, tenantID, jobID, leaseToken uuid.UUID, errVal error, backoffPolicy *domain.BackoffPolicy, durationMs int64) error

func (*JobRepo) CompleteJobSuccess 

func (r *JobRepo) CompleteJobSuccess(ctx context.Context, tenantID, jobID, leaseToken uuid.UUID, result json.RawMessage, durationMs int64) error

CompleteJobSuccess records a successful attempt, releases the tenant slot and decrements the outstanding-jobs counter in ONE statement (data-modifying CTEs). One statement is atomic on its own, so there is no explicit transaction: 1 round trip and 1 commit instead of 6 round trips.

Fencing: every row change hangs off the job update, which matches only if the caller still holds the current lease_token. A worker that lost its lease updates nothing and gets ErrLeaseLost.

func (*JobRepo) CompleteJobsSuccessBatch 

func (r *JobRepo) CompleteJobsSuccessBatch(ctx context.Context, items []SuccessCompletion) ([]bool, error)

CompleteJobsSuccessBatch records many successful jobs in ONE statement (one round trip, one commit) with exactly the guarantees of CompleteJobSuccess: each row is fenced by its own lease token, its attempt row is finished, its execution slot is released and the tenant's outstanding counter drops by the number of jobs actually completed. The returned slice is aligned with items; false means that job's lease was no longer valid (stale completion, rejected).

Callers must fall back to CompleteJobSuccess on error: a batch can fail as a whole (for example a deadlock between two batches touching the same tenant rows) even though each job is fine.

func (*JobRepo) EnqueueBatch 

func (r *JobRepo) EnqueueBatch(ctx context.Context, items []EnqueueParams) (results []*EnqueueResult, failed *EnqueueResult, err error)

EnqueueBatch enqueues several jobs atomically in ONE transaction: either all are accepted or none is (a rate-limit, quota or other rejection rolls everything back). One commit for N jobs is much cheaper than N separate commits. On error, failed carries the rejected item's result (for RetryAfterHint) when there is one.

func (*JobRepo) EnqueueJob 

func (r *JobRepo) EnqueueJob(ctx context.Context, p EnqueueParams) (*EnqueueResult, error)

EnqueueJob enqueues a job in its own transaction.

func (*JobRepo) EnqueueJobTx 

func (r *JobRepo) EnqueueJobTx(ctx context.Context, tx pgx.Tx, p EnqueueParams) (*EnqueueResult, error)

EnqueueJobTx enqueues a job inside a transaction the CALLER owns, so the job and the caller's own business rows commit or roll back together (the transactional-outbox pattern). It uses a savepoint, so a rate-limit, quota or idempotency rejection undoes only the enqueue and leaves the caller's transaction usable. The job becomes visible to workers (and the NOTIFY is delivered) only when the caller commits. Only available when your code shares QueueForge's PostgreSQL database.

func (*JobRepo) GetJob 

func (r *JobRepo) GetJob(ctx context.Context, tenantID, jobID uuid.UUID) (*domain.Job, error)

func (*JobRepo) GetJobAttempts 

func (r *JobRepo) GetJobAttempts(ctx context.Context, tenantID, jobID uuid.UUID) ([]domain.JobAttempt, error)

func (*JobRepo) GetQueueStats 

func (r *JobRepo) GetQueueStats(ctx context.Context, tenantID uuid.UUID, queueName string) (map[string]int64, error)

func (*JobRepo) ListJobs 

func (r *JobRepo) ListJobs(ctx context.Context, f ListJobsFilter) ([]domain.Job, error)

func (*JobRepo) ListenForJobs 

func (r *JobRepo) ListenForJobs(ctx context.Context, wake chan<- struct{})

ListenForJobs blocks until ctx is cancelled, sending a (non-blocking) signal on wake every time a job is enqueued. It holds one dedicated connection for LISTEN and transparently reconnects.

NOTIFY is only an optimisation: it removes the average poll-interval/2 pickup latency. Workers must still poll, because notifications are best-effort (a worker may be disconnected when one is sent) and delayed/retried jobs become runnable with the passage of time, not an event.

func (*JobRepo) RecoverExpiredLeases 

func (r *JobRepo) RecoverExpiredLeases(ctx context.Context, batchSize int, backoffPolicy *domain.BackoffPolicy) (int, error)

RecoverExpiredLeases finds running jobs whose lease has expired (the worker crashed, hung or was partitioned) and requeues them with backoff, or marks them dead once attempts are exhausted.

Each job is recovered in its OWN transaction. Doing the whole batch in one transaction (as an earlier version did) meant a single failing row aborted the Postgres transaction and silently rolled back every other recovery in the batch, on every tick, forever. Now a poison row is reported in the returned error while all healthy rows are still recovered.

func (*JobRepo) RenewLease 

func (r *JobRepo) RenewLease(ctx context.Context, jobID, leaseToken uuid.UUID, leaseDuration time.Duration) (*RenewLeaseResult, error)

func (*JobRepo) ReplayDeadJob 

func (r *JobRepo) ReplayDeadJob(ctx context.Context, tenantID, jobID uuid.UUID, actorKeyID *uuid.UUID, reason string) (*domain.Job, error)

type ListJobsFilter 

type ListJobsFilter struct {
	TenantID uuid.UUID
	Status   *domain.JobStatus
	Queue    *string
	TaskType *string
	Limit    int
	Cursor   *time.Time
}

type RenewLeaseResult 

type RenewLeaseResult struct {
	CancellationRequested bool
}

type RetentionCleanupResult 

type RetentionCleanupResult struct {
	ExpiredIdempotencyKeysDeleted int64
	TerminalJobsDeleted           int64
	StaleWorkersDeleted           int64
}

type RetentionRepo 

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

func NewRetentionRepo 

func NewRetentionRepo(db *DB) *RetentionRepo

func (*RetentionRepo) CleanupExpired 

func (r *RetentionRepo) CleanupExpired(ctx context.Context, terminalRetention time.Duration, batchSize int) (*RetentionCleanupResult, error)

CleanupExpired deletes expired idempotency keys, terminal jobs older than terminalRetention (with their attempts, audit events and idempotency keys) and stale worker registrations.

It keeps deleting batch after batch until the backlog is gone, the context ends, or maxRetentionRounds is reached. An earlier version deleted ONE batch of 100 per hourly tick, a ceiling of 2,400 jobs/day: any real workload grew the table (and the disk) without bound.

Each batch is one atomic statement. If a batch fails (say one poisoned row), the rows of that batch are retried one at a time so a single bad row cannot stall retention; every failure is returned (errors.Join), never swallowed.

type ScheduleRepo 

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

func NewScheduleRepo 

func NewScheduleRepo(db *DB) *ScheduleRepo

func (*ScheduleRepo) CreateSchedule 

func (r *ScheduleRepo) CreateSchedule(ctx context.Context, p CreateScheduleParams) (*domain.JobSchedule, error)

func (*ScheduleRepo) DisableSchedule 

func (r *ScheduleRepo) DisableSchedule(ctx context.Context, tenantID, scheduleID uuid.UUID) error

func (*ScheduleRepo) GetSchedule 

func (r *ScheduleRepo) GetSchedule(ctx context.Context, tenantID, scheduleID uuid.UUID) (*domain.JobSchedule, error)

func (*ScheduleRepo) ListSchedules 

func (r *ScheduleRepo) ListSchedules(ctx context.Context, tenantID uuid.UUID) ([]domain.JobSchedule, error)

func (*ScheduleRepo) ParseNextOccurrence 

func (r *ScheduleRepo) ParseNextOccurrence(cronExpr, timezone string, fromTime time.Time) (time.Time, error)

func (*ScheduleRepo) ProcessDueSchedules 

func (r *ScheduleRepo) ProcessDueSchedules(ctx context.Context, batchSize int, misfireGrace time.Duration) (int, error)

ProcessDueSchedules processes due recurring schedules with concurrency protection and misfire grace

func (*ScheduleRepo) SetSchedulePaused 

func (r *ScheduleRepo) SetSchedulePaused(ctx context.Context, tenantID, scheduleID uuid.UUID, paused bool) (*domain.JobSchedule, error)

type SuccessCompletion 

type SuccessCompletion struct {
	TenantID, JobID, LeaseToken uuid.UUID
	Result                      json.RawMessage
	DurationMs                  int64
}

SuccessCompletion is one finished job to record in CompleteJobsSuccessBatch.

type TenantRepo 

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

func NewTenantRepo 

func NewTenantRepo(db *DB) *TenantRepo

func (*TenantRepo) CreateAPIKey 

func (r *TenantRepo) CreateAPIKey(ctx context.Context, tenantID uuid.UUID, name string, permissions []string, expiresAt *time.Time) (*APIKeyCreationResult, error)

func (*TenantRepo) CreateAPIKeyExpiringIn 

func (r *TenantRepo) CreateAPIKeyExpiringIn(ctx context.Context, tenantID uuid.UUID, name string, permissions []string, d time.Duration) (*APIKeyCreationResult, error)

CreateAPIKeyExpiringIn creates a key that expires d after now, where "now" is the DATABASE clock (see ADR-014), so expiry is unaffected by the operator's own clock.

func (*TenantRepo) CreateTenant 

func (r *TenantRepo) CreateTenant(ctx context.Context, p CreateTenantParams) (*domain.Tenant, error)

func (*TenantRepo) GetLimits 

func (r *TenantRepo) GetLimits(ctx context.Context, tenantID uuid.UUID) (*domain.TenantLimits, error)

func (*TenantRepo) GetTenant 

func (r *TenantRepo) GetTenant(ctx context.Context, id uuid.UUID) (*domain.Tenant, error)

func (*TenantRepo) ListTenants 

func (r *TenantRepo) ListTenants(ctx context.Context) ([]domain.Tenant, error)

func (*TenantRepo) LookupAPIKeyBySecret 

func (r *TenantRepo) LookupAPIKeyBySecret(ctx context.Context, rawSecret string) (*domain.APIKey, *domain.Tenant, error)

func (*TenantRepo) ReconcileUsage 

func (r *TenantRepo) ReconcileUsage(ctx context.Context, tenantID uuid.UUID) (int64, error)

func (*TenantRepo) RevokeAPIKey 

func (r *TenantRepo) RevokeAPIKey(ctx context.Context, keyID uuid.UUID) error

type WorkerRepo 

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

func NewWorkerRepo 

func NewWorkerRepo(db *DB) *WorkerRepo

func (*WorkerRepo) Heartbeat 

func (r *WorkerRepo) Heartbeat(ctx context.Context, p HeartbeatParams) error

func (*WorkerRepo) ListActiveWorkers 

func (r *WorkerRepo) ListActiveWorkers(ctx context.Context, maxAge time.Duration) ([]domain.WorkerInstance, error)

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.