Documentation
¶
Overview ¶
Package postgres is the pgx implementation of metadata.Repository. It is the only code in the system that issues SQL.
Index ¶
- type Repo
- func (r *Repo) ClaimIdempotencyKey(ctx context.Context, key string) (bool, *metadata.IdempotencyRecord, error)
- func (r *Repo) ClaimJob(ctx context.Context) (*metadata.Job, error)
- func (r *Repo) CountPendingJobs(ctx context.Context) (int, error)
- func (r *Repo) Create(ctx context.Context, f *metadata.File) error
- func (r *Repo) CreateForKey(ctx context.Context, f *metadata.File, key string) error
- func (r *Repo) CreateJob(ctx context.Context, j *metadata.Job) (bool, string, error)
- func (r *Repo) Delete(ctx context.Context, id string) error
- func (r *Repo) FinishJob(ctx context.Context, id string, attempt int, res metadata.JobResult) error
- func (r *Repo) Get(ctx context.Context, id string) (*metadata.File, error)
- func (r *Repo) GetJob(ctx context.Context, id string) (*metadata.Job, error)
- func (r *Repo) HeartbeatJob(ctx context.Context, id string, attempt, done, total int) (bool, error)
- func (r *Repo) List(ctx context.Context, filter metadata.ListFilter) ([]*metadata.File, error)
- func (r *Repo) ListDeleted(ctx context.Context, limit int) ([]*metadata.File, error)
- func (r *Repo) ParkJob(ctx context.Context, id string, attempt, done, total int) error
- func (r *Repo) Purge(ctx context.Context, id string) error
- func (r *Repo) PurgeFinishedJobs(ctx context.Context, before time.Time) (int, error)
- func (r *Repo) PurgeIdempotencyKeys(ctx context.Context, before time.Time) (int, error)
- func (r *Repo) ReapJobs(ctx context.Context, ttl time.Duration) (int, error)
- func (r *Repo) ReleaseIdempotencyKey(ctx context.Context, key string) error
- func (r *Repo) RequestCancel(ctx context.Context, id string) (*metadata.Job, error)
- func (r *Repo) StorageKeys(ctx context.Context) (map[string]struct{}, error)
- func (r *Repo) UpdateMetadata(ctx context.Context, id string, tags map[string]any, merge bool) (*metadata.File, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Repo ¶
type Repo struct {
// contains filtered or unexported fields
}
Repo persists File records in Postgres.
func (*Repo) ClaimIdempotencyKey ¶
func (r *Repo) ClaimIdempotencyKey(ctx context.Context, key string) (bool, *metadata.IdempotencyRecord, error)
ClaimIdempotencyKey inserts the key as pending, claiming it. If the key already exists, the insert no-ops and we read back the existing row so the caller can replay (completed) or reject (pending).
func (*Repo) ClaimJob ¶ added in v1.3.0
ClaimJob takes the oldest pending job. SKIP LOCKED is what makes N replicas safe with no coordination: a row another claim holds locked is passed over rather than waited on, so two workers never claim one job.
func (*Repo) CountPendingJobs ¶ added in v1.3.0
CountPendingJobs is the queue depth, served by the pickup index.
func (*Repo) CreateForKey ¶
CreateForKey inserts the file row and marks the idempotency key completed in a single transaction. Atomicity matters: if the row committed but the key were left pending (two separate writes, crash in between), a retry after the key's TTL would create a duplicate file. One transaction closes that window.
func (*Repo) CreateJob ¶ added in v1.3.0
CreateJob inserts the job as pending. The partial unique index on (archive_id) where the status is active refuses a second live job for one archive; ON CONFLICT DO NOTHING turns that into "no row", and the active job's id is read back so the caller can point at it.
func (*Repo) Delete ¶
Delete soft-deletes the record by ID, stamping deleted_at. The object is reclaimed later by GC. Returns metadata.ErrNotFound if no live row matched.
func (*Repo) HeartbeatJob ¶ added in v1.3.0
HeartbeatJob records progress and liveness together and reads the cancel flag back, so the three cost one round trip.
func (*Repo) List ¶
List returns records matching the filter, newest first — live ones unless the filter asks for tombstones too — filtered by content type, JSONB tag containment and the extracted-entry scope, with keyset pagination. Ordering is by id descending: ids are UUIDv7, so id order is creation order and the primary-key index serves both the sort and the cursor bound.
func (*Repo) ListDeleted ¶
ListDeleted returns up to limit soft-deleted records, oldest deletion first, so GC purges the longest-pending objects first.
func (*Repo) ParkJob ¶ added in v1.3.0
ParkJob returns a running job to pending with its progress kept, for a worker that is shutting down. The next claim resumes it.
func (*Repo) Purge ¶
Purge hard-deletes a soft-deleted record. The deleted_at guard prevents ever removing a live row through this path.
func (*Repo) PurgeFinishedJobs ¶ added in v1.3.0
PurgeFinishedJobs deletes terminal rows past retention, returning the count.
func (*Repo) PurgeIdempotencyKeys ¶
PurgeIdempotencyKeys deletes keys older than the cutoff, returning the count.
func (*Repo) ReapJobs ¶ added in v1.3.0
ReapJobs returns stranded running rows — no heartbeat within ttl — to pending. Not to failed: extraction resumes, so the next pickup continues past the children already written. The cutoff is now() minus ttl on the database's clock, the clock that stamped the heartbeat, so no skew between this process and the database can make a live run look dead.
func (*Repo) ReleaseIdempotencyKey ¶
ReleaseIdempotencyKey removes a still-pending claim (after a failed upload).
func (*Repo) RequestCancel ¶ added in v1.3.0
RequestCancel flags a job. Every column reference on the right-hand side of a SET sees the row's old value, so one statement cancels a pending job outright, flags a running one, and leaves a terminal one untouched.
func (*Repo) StorageKeys ¶
StorageKeys returns every storage key in the table (live or soft-deleted) as a set, so GC can identify stored objects with no backing row.
Source Files