Documentation
¶
Overview ¶
Package jobs wires ragit's Processor into a River job queue.
ragit does not own a river.Client — a host app builds and starts its own (often with a direct, non-pooled Postgres connection, since River's LISTEN/NOTIFY needs one; that's a host-app concern documented here, not something this package can enforce) and calls river.AddWorker with the workers this package constructs. This mirrors the "Registration" contribution pattern the reference implementation (valiro-go) uses to let a subsystem add workers/queues to a shared client without the platform layer importing the subsystem — simplified here since a standalone library has no such platform layer to invert around.
Index ¶
Constants ¶
const DefaultRetentionSweepTimeout = 5 * time.Minute
DefaultRetentionSweepTimeout bounds a single retention pass. A pass is capped at ragit.DeleteExpiredBatchSize documents, so this is generous rather than tight.
const RateLimitBackoff = 5 * time.Minute
RateLimitBackoff is the snooze duration used when the embedding provider rate-limits a batch (embed.ErrRateLimited). A longer, deliberate wait rather than River's normal exponential backoff.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DeleteDocumentArgs ¶
type DeleteDocumentArgs struct {
DocumentID uuid.UUID `json:"document_id"`
TenantID uuid.UUID `json:"tenant_id"`
}
DeleteDocumentArgs identifies which document to delete.
func (DeleteDocumentArgs) InsertOpts ¶
func (DeleteDocumentArgs) InsertOpts() river.InsertOpts
func (DeleteDocumentArgs) Kind ¶
func (DeleteDocumentArgs) Kind() string
type DeleteDocumentWorker ¶
type DeleteDocumentWorker struct {
river.WorkerDefaults[DeleteDocumentArgs]
// contains filtered or unexported fields
}
DeleteDocumentWorker runs Processor.DeleteDocument as a River job.
func NewDeleteDocumentWorker ¶
func NewDeleteDocumentWorker(processor *ragit.Processor) *DeleteDocumentWorker
NewDeleteDocumentWorker builds a worker around an existing Processor.
func (*DeleteDocumentWorker) Work ¶
func (w *DeleteDocumentWorker) Work(ctx context.Context, job *river.Job[DeleteDocumentArgs]) error
type DeleteExpiredArgs ¶
type DeleteExpiredArgs struct{}
DeleteExpiredArgs carries no state: the sweep finds its own work. It is meant to be scheduled periodically rather than enqueued per document.
Wire it up with River's periodic jobs, e.g.
river.NewPeriodicJob(
river.PeriodicInterval(15*time.Minute),
func() (river.JobArgs, *river.InsertOpts) { return jobs.DeleteExpiredArgs{}, nil },
&river.PeriodicJobOpts{RunOnStart: true},
)
func (DeleteExpiredArgs) InsertOpts ¶
func (DeleteExpiredArgs) InsertOpts() river.InsertOpts
func (DeleteExpiredArgs) Kind ¶
func (DeleteExpiredArgs) Kind() string
type DeleteExpiredWorker ¶
type DeleteExpiredWorker struct {
river.WorkerDefaults[DeleteExpiredArgs]
// contains filtered or unexported fields
}
DeleteExpiredWorker runs Processor.DeleteExpired as a River job, clearing documents and chunks whose retention clock has run out (design.md §8's ephemeral attachment scope).
func NewDeleteExpiredWorker ¶
func NewDeleteExpiredWorker(processor *ragit.Processor) *DeleteExpiredWorker
NewDeleteExpiredWorker builds the retention sweep worker.
func (*DeleteExpiredWorker) Timeout ¶
func (w *DeleteExpiredWorker) Timeout(*river.Job[DeleteExpiredArgs]) time.Duration
Timeout implements river.Worker.
func (*DeleteExpiredWorker) Work ¶
func (w *DeleteExpiredWorker) Work(ctx context.Context, job *river.Job[DeleteExpiredArgs]) error
Work runs one sweep pass.
A failure to purge object storage is reported but does not fail the job: the rows are already committed gone, so retrying the job would re-run the sweep against a corpus that no longer contains them and would never revisit the orphaned objects. Returning an error here would produce noisy retries that cannot fix anything. Orphans are a storage-lifecycle concern.
type ProcessDocumentArgs ¶
type ProcessDocumentArgs struct {
DocumentID uuid.UUID `json:"document_id"`
TenantID uuid.UUID `json:"tenant_id"`
}
ProcessDocumentArgs identifies which document to process. Kind and Queue are namespaced ("ragit_...") to avoid colliding with a host app's own job kinds/queues in the same River client.
func (ProcessDocumentArgs) InsertOpts ¶
func (ProcessDocumentArgs) InsertOpts() river.InsertOpts
func (ProcessDocumentArgs) Kind ¶
func (ProcessDocumentArgs) Kind() string
type ProcessDocumentWorker ¶
type ProcessDocumentWorker struct {
river.WorkerDefaults[ProcessDocumentArgs]
// contains filtered or unexported fields
}
ProcessDocumentWorker runs Processor.ProcessDocument as a River job.
func NewProcessDocumentWorker ¶
func NewProcessDocumentWorker(processor *ragit.Processor) *ProcessDocumentWorker
NewProcessDocumentWorker builds a worker around an existing Processor.
func (*ProcessDocumentWorker) Work ¶
func (w *ProcessDocumentWorker) Work(ctx context.Context, job *river.Job[ProcessDocumentArgs]) error
Work classifies ProcessDocument's error per docs/design.md §7: a transport/deployment failure (extract or embed unavailable) is retried with River's normal backoff; a rate limit gets a longer, deliberate snooze; anything else is a verdict on the document and is not retried.
Source Files