Documentation
¶
Overview ¶
Package workers provides River job workers (e.g. webhook delivery, feedback embedding).
Index ¶
- Constants
- func NewRiverWorkersAndQueues(cfg *config.Config, deps RiverDeps) (*river.Workers, map[string]river.QueueConfig)
- type FeedbackEmbeddingWorker
- type FeedbackEmotionsWorker
- type FeedbackSentimentWorker
- type FeedbackTranslationWorker
- type RiverDeps
- type TenantTranslationBackfillWorker
- type WebhookDispatchWorker
Constants ¶
const WebhookDeliveryTimeoutBuffer = 5 * time.Second
WebhookDeliveryTimeoutBuffer is added to HTTP timeout for the River job timeout.
Variables ¶
This section is empty.
Functions ¶
func NewRiverWorkersAndQueues ¶
func NewRiverWorkersAndQueues( cfg *config.Config, deps RiverDeps, ) (*river.Workers, map[string]river.QueueConfig)
NewRiverWorkersAndQueues builds River workers and queue config from cfg and deps. Each optional worker is registered only when its client is set, and its queue is declared alongside it, so a disabled enrichment leaves no queue for jobs to pile up on unworked.
Only hub-worker calls this: the API inserts jobs through an insert-only River client and registers nothing (see cmd/api/app.go), so the queue MaxWorkers below are always the real configured concurrency.
Types ¶
type FeedbackEmbeddingWorker ¶
type FeedbackEmbeddingWorker struct {
river.WorkerDefaults[service.FeedbackEmbeddingArgs]
// contains filtered or unexported fields
}
FeedbackEmbeddingWorker generates and stores embeddings for feedback records.
func NewFeedbackEmbeddingWorker ¶
func NewFeedbackEmbeddingWorker( embeddingService feedbackEmbeddingService, embeddingClient service.EmbeddingClient, docPrefix string, metrics observability.EmbeddingMetrics, ) *FeedbackEmbeddingWorker
NewFeedbackEmbeddingWorker creates a worker that fetches the record, calls the embedding client, and stores the result. docPrefix is the prefix for document text. Can be empty for some providers. metrics may be nil when metrics are disabled.
func (*FeedbackEmbeddingWorker) Timeout ¶
func (w *FeedbackEmbeddingWorker) Timeout(*river.Job[service.FeedbackEmbeddingArgs]) time.Duration
Timeout limits how long a single embedding job can run.
func (*FeedbackEmbeddingWorker) Work ¶
func (w *FeedbackEmbeddingWorker) Work(ctx context.Context, job *river.Job[service.FeedbackEmbeddingArgs]) error
Work loads the record, generates or clears the embedding, and persists it.
type FeedbackEmotionsWorker ¶
type FeedbackEmotionsWorker = enrichmentWorker[service.FeedbackEmotionsArgs, service.EmotionsResult]
FeedbackEmotionsWorker classifies a feedback record's value_text into a set of emotion labels and stores it — a configured enrichmentWorker. It borrows the shared rate-limit snooze (it, too, calls a rate-limited LLM provider); it has no per-tenant target, but its persist is guarded against content supersession (a job that read older text skips instead of landing its labels last — stale non-NULL labels would escape the NULL-rows-only backfill forever).
func NewFeedbackEmotionsWorker ¶
func NewFeedbackEmotionsWorker( svc emotionsWorkerService, resolver tenantSettingsReader, client service.EmotionsClient, metrics observability.EmotionsMetrics, ) *FeedbackEmotionsWorker
NewFeedbackEmotionsWorker creates a worker that fetches the record, classifies its value_text, and stores the emotion labels. metrics may be nil when metrics are disabled.
type FeedbackSentimentWorker ¶
type FeedbackSentimentWorker = enrichmentWorker[service.FeedbackSentimentArgs, service.SentimentResult]
FeedbackSentimentWorker classifies a feedback record's value_text into a sentiment label and score and stores it — a configured enrichmentWorker. It borrows the shared rate-limit snooze (it, too, calls a rate-limited LLM provider); it has no per-tenant target, but its persist is guarded against content supersession (a job that read older text skips instead of landing its label last — a stale non-NULL label would escape the NULL-rows-only backfill forever).
func NewFeedbackSentimentWorker ¶
func NewFeedbackSentimentWorker( svc sentimentWorkerService, resolver tenantSettingsReader, client service.SentimentClient, metrics observability.SentimentMetrics, ) *FeedbackSentimentWorker
NewFeedbackSentimentWorker creates a worker that fetches the record, classifies its value_text, and stores the result. metrics may be nil when metrics are disabled.
type FeedbackTranslationWorker ¶
type FeedbackTranslationWorker = enrichmentWorker[service.FeedbackTranslationArgs, string]
FeedbackTranslationWorker translates a feedback record's value_text into the tenant's target language and stores it — a configured enrichmentWorker. It borrows the shared rate-limit snooze (it calls a rate-limited LLM provider) and uses the supersession skip: a stale-target OR stale-content write is a no-op once a newer job owns the row.
func NewFeedbackTranslationWorker ¶
func NewFeedbackTranslationWorker( svc translationWorkerService, client service.TranslationClient, metrics observability.TranslationMetrics, ) *FeedbackTranslationWorker
NewFeedbackTranslationWorker creates a worker that fetches the record, translates its value_text into the target language (or copies it when the source already matches), and stores the result. metrics may be nil when metrics are disabled.
type RiverDeps ¶
type RiverDeps struct {
// Webhook worker
WebhooksRepo webhookDispatchRepo
WebhookSender service.WebhookSender
WebhookHTTPTimeout time.Duration
WebhookMetrics observability.WebhookMetrics
// Embedding worker (optional; if EmbeddingClient is nil, embedding worker is not registered)
EmbeddingService feedbackEmbeddingService
EmbeddingClient service.EmbeddingClient
EmbeddingDocPrefix string
EmbeddingMetrics observability.EmbeddingMetrics
// Translation worker (optional; if TranslationClient is nil, translation worker is not registered)
TranslationService translationWorkerService
TranslationClient service.TranslationClient
TranslationMetrics observability.TranslationMetrics
// Per-tenant translation backfill worker (registered alongside the translation worker).
TranslationBackfillService tenantTranslationBackfillService
TranslationMaxAttempts int
// Sentiment worker (optional; if SentimentClient is nil, sentiment worker is not registered)
SentimentService sentimentWorkerService
SentimentResolver tenantSettingsReader
SentimentClient service.SentimentClient
SentimentMetrics observability.SentimentMetrics
// Emotions worker (optional; if EmotionsClient is nil, emotions worker is not registered)
EmotionsService emotionsWorkerService
EmotionsResolver tenantSettingsReader
EmotionsClient service.EmotionsClient
EmotionsMetrics observability.EmotionsMetrics
}
RiverDeps holds dependencies required to build River workers and queue config. Each optional group is gated on its client: leave a client nil and neither its worker nor its queue is registered.
type TenantTranslationBackfillWorker ¶
type TenantTranslationBackfillWorker struct {
river.WorkerDefaults[service.TenantTranslationBackfillArgs]
// contains filtered or unexported fields
}
TenantTranslationBackfillWorker fans out a per-tenant re-translation: it lists the tenant's stale text records and enqueues a FeedbackTranslationArgs job for each. It is enqueued when a tenant's translation settings change (see service.TenantTranslationBackfillArgs) and runs off the request path.
func NewTenantTranslationBackfillWorker ¶
func NewTenantTranslationBackfillWorker( svc tenantTranslationBackfillService, maxAttempts int, ) *TenantTranslationBackfillWorker
NewTenantTranslationBackfillWorker creates the worker. maxAttempts is applied to the per-record translation jobs it enqueues.
func (*TenantTranslationBackfillWorker) Timeout ¶
func (w *TenantTranslationBackfillWorker) Timeout(*river.Job[service.TenantTranslationBackfillArgs]) time.Duration
Timeout limits how long a single tenant backfill fan-out can run.
func (*TenantTranslationBackfillWorker) Work ¶
func (w *TenantTranslationBackfillWorker) Work( ctx context.Context, job *river.Job[service.TenantTranslationBackfillArgs], ) error
Work lists the tenant's stale records and enqueues per-record translation jobs onto the translations queue. The River client is obtained from the context (the only place River sets it) and handed to the service as the inserter, preserving the injected-inserter seam.
type WebhookDispatchWorker ¶
type WebhookDispatchWorker struct {
river.WorkerDefaults[service.WebhookDispatchArgs]
// contains filtered or unexported fields
}
WebhookDispatchWorker delivers one event to one webhook endpoint.
func NewWebhookDispatchWorker ¶
func NewWebhookDispatchWorker( repo webhookDispatchRepo, sender service.WebhookSender, httpTimeout time.Duration, metrics observability.WebhookMetrics, ) *WebhookDispatchWorker
NewWebhookDispatchWorker creates a worker that uses the given repo and sender. httpTimeout is the webhook HTTP client timeout; job timeout is httpTimeout + WebhookDeliveryTimeoutBuffer. metrics may be nil when metrics are disabled.
func (*WebhookDispatchWorker) Timeout ¶
func (w *WebhookDispatchWorker) Timeout(*river.Job[service.WebhookDispatchArgs]) time.Duration
Timeout limits how long a single delivery can run (HTTP timeout + buffer).
func (*WebhookDispatchWorker) Work ¶
func (w *WebhookDispatchWorker) Work(ctx context.Context, job *river.Job[service.WebhookDispatchArgs]) error
Work loads the webhook, builds the payload, and sends once.
Source Files