service

type BatchEmbeddingClient ¶

type BatchEmbeddingClient interface {
	CreateEmbeddings(ctx context.Context, inputs []string) ([][]float32, error)
}

BatchEmbeddingClient is an optional provider capability for embedding multiple documents in one request. Workers use it when batching is enabled; providers that do not implement it continue to receive one request per document. Implementations must return exactly one vector per input in the same order as inputs.

type BatchingEmbeddingClient ¶

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

BatchingEmbeddingClient coalesces concurrent document calls while leaving query calls synchronous. It is only constructed when the provider implements BatchEmbeddingClient.

func NewBatchingEmbeddingClient(
	client EmbeddingClient,
	cfg EmbeddingBatchConfig,
	metrics EmbeddingBatchMetrics,
) (*BatchingEmbeddingClient, bool)

func (b *BatchingEmbeddingClient) CreateEmbedding(ctx context.Context, input string) ([]float32, error)

func (b *BatchingEmbeddingClient) CreateEmbeddingForQuery(ctx context.Context, input string) ([]float32, error)

func (b *BatchingEmbeddingClient) Shutdown(ctx context.Context) error

type CachedTenantSettings ¶

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

CachedTenantSettings wraps a TenantSettingsReader with a per-process, size-bounded, TTL-expiring LRU cache. Feedback-record creation is high volume and the translation enqueue gate must resolve a tenant's target language per event, so caching avoids a tenant_settings read on every feedback event. Staleness is bounded by the TTL; because the worker persists the target it actually used, a changed target self-corrects on the next write. Safe for concurrent use (expirable.LRU is internally locked).

func NewCachedTenantSettings(
	delegate TenantSettingsReader, size int, ttl time.Duration, metrics observability.CacheMetrics,
) *CachedTenantSettings

func (c *CachedTenantSettings) GetSettings(
	ctx context.Context, tenantID string,
) (*models.TenantSettings, error)

func (c *CachedTenantSettings) Invalidate(tenantID string)

func (c *CachedTenantSettings) OnSettingsChanged(_ context.Context, tenantID string, _ []string)

type EmbeddingBatchConfig ¶

type EmbeddingBatchConfig struct {
	BatchSize   int
	MaxWait     time.Duration
	MaxInFlight int
}

EmbeddingBatchConfig controls document micro-batching. BatchSize <= 1 disables batching.

type EmbeddingBatchMetrics ¶

type EmbeddingBatchMetrics interface {
	RecordEmbeddingBatch(ctx context.Context, inputs int64, duration time.Duration, status string)
	AddEmbeddingBatchInFlight(ctx context.Context, delta int64)
}

EmbeddingBatchMetrics is the bounded observability seam used by the batching client.

type EmbeddingClient ¶

type EmbeddingClient interface {
	CreateEmbedding(ctx context.Context, input string) ([]float32, error)
	CreateEmbeddingForQuery(ctx context.Context, input string) ([]float32, error)
}

EmbeddingClient generates embedding vectors for text. CreateEmbedding is for embedding documents (e.g. feedback records) for storage. CreateEmbeddingForQuery is for embedding search queries; some providers (e.g. Google) use a different task type for asymmetric retrieval.

func NewEmbeddingClient(ctx context.Context, cfg EmbeddingClientConfig) (EmbeddingClient, error)

type EmbeddingClientConfig ¶

type EmbeddingClientConfig struct {
	Provider              string
	ProviderAPIKey        string // API key for openai/google providers; not logged or serialized
	Model                 string
	BaseURL               string
	HTTPDisableKeepAlives bool
	Normalize             bool
	GoogleCloudProject    string
	GoogleCloudLocation   string
	// UsageRecorder receives each provider call's token counts and duration. nil disables it.
	UsageRecorder llm.UsageRecorder
}

EmbeddingClientConfig holds configuration for creating an embedding client.

type EmbeddingProvider ¶

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

EmbeddingProvider implements eventPublisher by enqueueing one River job per feedback record event when the event is FeedbackRecordCreated (with non-empty value_text) or FeedbackRecordUpdated (with value_text in ChangedFields, including when value_text is now empty so the worker can clear).

func NewEmbeddingProvider(
	inserter RiverJobInserter,
	model string,
	queueName string,
	maxAttempts int,
	docPrefix string,
	metrics observability.EmbeddingMetrics,
) *EmbeddingProvider

func NewEmbeddingProviderForInputKind(
	inserter RiverJobInserter,
	model string,
	queueName string,
	maxAttempts int,
	docPrefix string,
	metrics observability.EmbeddingMetrics,
	inputKind models.EmbeddingInputKind,
) *EmbeddingProvider

func (p *EmbeddingProvider) PublishEvent(ctx context.Context, event Event)

type EmbeddingReconcileArgs ¶

type EmbeddingReconcileArgs struct{}

EmbeddingReconcileArgs is an argument-free, level-triggered taxonomy embedding sweep. Database state and deployment configuration are read when the job runs.

func (EmbeddingReconcileArgs) Kind() string

type EmbeddingReconcileRepository ¶

type EmbeddingReconcileRepository interface {
	ListPendingTaxonomyEmbeddingIDs(
		ctx context.Context, model string, retryBefore time.Time, limit int,
	) ([]uuid.UUID, error)
	CountRunnableEmbeddingJobs(ctx context.Context, queue string) (int, error)
}

EmbeddingReconcileRepository is the data boundary needed by the taxonomy embedding sweep.

type EmbeddingReconcileResult ¶

type EmbeddingReconcileResult struct {
	Found    int
	Enqueued int
	Depth    int
	AtTarget bool
}

EmbeddingReconcileResult reports one sweep's bounded queue action.

type EmbeddingReconcileService ¶

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

EmbeddingReconcileService keeps the low-priority repair lane topped up without materializing the complete deployment backlog or crowding live embedding work.

func NewEmbeddingReconcileService(
	repo EmbeddingReconcileRepository,
	model string,
	targetDepth int,
	maxAttempts int,
	retryAfter time.Duration,
) *EmbeddingReconcileService

func (s *EmbeddingReconcileService) SetInserter(inserter RiverBatchInserter)

func (s *EmbeddingReconcileService) Sweep(ctx context.Context) (EmbeddingReconcileResult, error)

type EmbeddingsRepository ¶

type EmbeddingsRepository interface {
	Upsert(
		ctx context.Context, feedbackRecordID uuid.UUID, model string, embedding []float32,
		stillCurrent func(fieldLabel, valueText, valueTextTranslated *string) bool,
	) error
	DeleteByFeedbackRecordAndModel(
		ctx context.Context, feedbackRecordID uuid.UUID, model string,
		stillCurrent func(fieldLabel, valueText, valueTextTranslated *string) bool,
	) error
	ListFeedbackRecordIDsForBackfill(
		ctx context.Context, model string, afterID uuid.UUID, limit int,
	) ([]uuid.UUID, error)
	ListFeedbackRecordIDsForBackfillByInputKind(
		ctx context.Context, model string, inputKind models.EmbeddingInputKind, afterID uuid.UUID, limit int,
	) ([]uuid.UUID, error)
	ListTenantFeedbackRecordIDsForBackfillByInputKind(
		ctx context.Context, tenantID, model string, inputKind models.EmbeddingInputKind, afterID uuid.UUID, limit int,
	) ([]uuid.UUID, error)
}

EmbeddingsRepository defines the interface for embeddings table access.

type EmbeddingsRepositoryForSearch ¶

type EmbeddingsRepositoryForSearch interface {
	GetEmbeddingAndTenantByFeedbackRecordAndModel(
		ctx context.Context, feedbackRecordID uuid.UUID, model string,
	) ([]float32, string, error)
	NearestFeedbackRecordsByEmbedding(
		ctx context.Context, model string, queryEmbedding []float32, tenantID string, limit int, excludeID *uuid.UUID, minScore float64,
	) ([]models.FeedbackRecordWithScore, bool, error)
	NearestFeedbackRecordsByEmbeddingAfterCursor(
		ctx context.Context, model string, queryEmbedding []float32, tenantID string, limit int,
		lastDistance float64, lastFeedbackRecordID uuid.UUID, excludeID *uuid.UUID, minScore float64,
	) ([]models.FeedbackRecordWithScore, bool, error)
}

EmbeddingsRepositoryForSearch provides the embedding read operations needed for semantic search. HasMore is true when there may be additional results (full page returned or full fetch limit consumed).

type EmotionsClient ¶

type EmotionsClient interface {
	// Classify returns the emotions expressed in text. sourceLang is the record's BCP-47 language
	// and may be empty; it is passed to the model only as a hint (classification is multilingual).
	Classify(ctx context.Context, text, sourceLang string) (EmotionsResult, error)
}

EmotionsClient classifies open feedback text into zero or more emotion labels. Implementations call an LLM provider (OpenAI or Google) via structured output; the factory selects one from configuration. It mirrors the SentimentClient seam so the worker depends on the interface, not a provider.

func NewEmotionsClient(ctx context.Context, cfg EmotionsClientConfig) (EmotionsClient, error)

type EmotionsClientConfig ¶

type EmotionsClientConfig = EnrichmentClientConfig

EmotionsClientConfig aliases the shared classify client config (see EnrichmentClientConfig).

type EmotionsProvider ¶

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

EmotionsProvider enqueues one emotion job per eligible feedback record event, over the shared enrichmentProvider. Eligibility is a text field with non-empty open text; re-classification is triggered by a value_text change (emotions do not depend on source language, which is only a prompt hint). Like sentiment it resolves a per-tenant setting on the enqueue path — the per-directory emotions switch (ENG-1573) — and skips tenants that have turned emotions off.

func NewEmotionsProvider(
	inserter RiverJobInserter,
	resolver TenantSettingsReader,
	queueName string,
	maxAttempts int,
	metrics observability.EmotionsMetrics,
) *EmotionsProvider

func (p EmotionsProvider) PublishEvent(ctx context.Context, event Event)

type EmotionsResult ¶

type EmotionsResult struct {
	Labels []models.EmotionValue
}

EmotionsResult is the parsed, validated output of one emotion classification: the distinct, known emotion labels the text expresses (possibly empty when none apply).

type EnrichmentClearMetrics ¶

type EnrichmentClearMetrics interface {
	RecordOutputCleared(ctx context.Context, output string)
}

EnrichmentClearMetrics records enrichment outputs nulled by an edit's eager-clear, labeled by output. Optional: nil disables it; set via SetEnrichmentClearMetrics (the clear fires only on the API's UpdateFeedbackRecord path).

type EnrichmentClientConfig ¶

type EnrichmentClientConfig struct {
	Provider            string
	ProviderAPIKey      string // API key for openai/google providers; not logged or serialized
	Model               string
	BaseURL             string
	GoogleCloudProject  string
	GoogleCloudLocation string
	// UsageRecorder receives each provider call's token counts and duration. nil disables
	// recording; the backfill commands and tests leave it unset.
	UsageRecorder llm.UsageRecorder
}

EnrichmentClientConfig is the provider/model configuration shared by the classify clients — sentiment, emotions, and translation are structurally identical, so each exposes a readable alias of this one struct (embedding keeps its own: it adds Normalize).

type EnrichmentReconcileArgs ¶

type EnrichmentReconcileArgs struct{}

EnrichmentReconcileArgs is one sweep: find records still owed enrichments and top the reconcile queues up towards their target depth.

It carries no arguments. The sweep's inputs are the database's current state and the deployment config, both read at run time, so an argument would only be a chance for a queued job to act on a stale view.

Uniqueness is a bare marker across the in-flight states, so a tick that arrives while the previous sweep is still running collapses into it rather than running two scans concurrently. `completed` is deliberately NOT in that set: with no ByPeriod the window is unbounded, so including it would mean the first sweep is the only one that ever runs.

Deliberately an EMPTY struct: every insert encodes identically, so every insert site shares the one unique key and there is nothing a second site could get wrong. A discriminator field was tried and rejected — it created the drift channel it claimed to document, since an insert spelling the value differently would silently stop collapsing into the scheduled job.

func (EnrichmentReconcileArgs) Kind() string

type EnrichmentReconcileRepository ¶

type EnrichmentReconcileRepository interface {
	ListPendingEnrichment(
		ctx context.Context, enrichment, defaultLang string, limit int,
	) ([]repository.PendingEnrichmentTarget, error)
	CountInFlightByQueue(ctx context.Context, queues []string) (map[string]int64, error)
}

EnrichmentReconcileRepository is the pending-set half: which records still owe an enrichment, and how deep each backfill queue currently is.

type EnrichmentReconcileService ¶

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

EnrichmentReconcileService tops the backfill queues up towards a target depth with records that are still owed an enrichment.

It is the level-triggered half of enrichment coverage. The event path enqueues a job when a record changes and is what makes enrichment fast; this finds what that path missed — a record created while the provider was unconfigured, a job lost to a crash, a transient failure that used up its retries — and is what makes enrichment eventually COMPLETE.

func NewEnrichmentReconcileService(params NewEnrichmentReconcileServiceParams) *EnrichmentReconcileService

func (s *EnrichmentReconcileService) SetInserter(inserter RiverBatchInserter)

func (s *EnrichmentReconcileService) Sweep(ctx context.Context) (ReconcileResult, error)

type EnrichmentRetryRepository ¶

type EnrichmentRetryRepository interface {
	ClearTerminalMarkers(ctx context.Context, tenantID, enrichment string, window time.Duration) (claimed bool, cleared int64, err error)
	CooldownRemaining(ctx context.Context, tenantID, enrichment string, window time.Duration) (time.Duration, error)
}

EnrichmentRetryRepository clears terminal markers and tracks the cooldown. ClearTerminalMarkers makes the window decision itself, atomically with the delete; CooldownRemaining only reports the wait for a refused caller's response.

type EnrichmentRetryService ¶

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

EnrichmentRetryService gives permanently-failed records another chance.

The automatic sweep will not: a terminal failure is a property of the record's own text, so re-running it costs a provider call and fails identically. This is the deliberate override for when the premise has changed — the text was edited, or the provider changed its policy.

func NewEnrichmentRetryService(params NewEnrichmentRetryServiceParams) *EnrichmentRetryService

func (s *EnrichmentRetryService) Retry(
	ctx context.Context, tenantID string, enrichments []string,
) (*models.EnrichmentRetryResponse, error)

type EnrichmentSettingsListener ¶

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

EnrichmentSettingsListener implements SettingsChangeListener by dispatching each changed setting key to the enrichment backfill it triggers. The dispatch table is built from what the deployment has enabled (e.g. translation), so a disabled enrichment registers no handler and an unknown/irrelevant key is ignored.

It deliberately does not use the webhook MessagePublisher: this is an internal, best-effort side-effect, not a customer-facing event. Enqueue errors are logged and swallowed — the global backfill command remains the guaranteed recovery path.

Adding a future setting is one map entry (e.g. a "sentiment_enabled" → sentiment backfill handler); TenantSettingsService does not change.

func NewTranslationSettingsListener(
	inserter RiverJobInserter, queueName string, maxAttempts int,
) *EnrichmentSettingsListener

func (l *EnrichmentSettingsListener) OnSettingsChanged(ctx context.Context, tenantID string, changedKeys []string)

type EnrichmentStatusRepository ¶

type EnrichmentStatusRepository interface {
	CountEnrichmentStatus(ctx context.Context, tenantID, defaultLang string) (repository.EnrichmentStatusCounts, error)
}

EnrichmentStatusRepository is the count surface the status service needs.

type EnrichmentStatusService ¶

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

EnrichmentStatusService reports a tenant's enrichment progress. It resolves the per-tenant enable state from settings (via TenantSettingsReader) and overlays the deployment-level (provider/model) gate; the repo supplies the counts.

func NewEnrichmentStatusService(params NewEnrichmentStatusServiceParams) *EnrichmentStatusService

func (s *EnrichmentStatusService) GetEnrichmentStatus(
	ctx context.Context, tenantID string,
) (*models.EnrichmentStatusResponse, error)

type EnrichmentSweepSpec ¶

type EnrichmentSweepSpec struct {
	Name string
	// MaxAttempts mirrors the live path, so a reconciled job is retried exactly as many times as
	// an event-driven one before it writes its failure marker.
	MaxAttempts int
}

EnrichmentSweepSpec is one enrichment's sweep configuration. One struct rather than parallel collections, because the parallel version had a silent failure mode: an enrichment present in the enabled list but missing from the attempts map handed River MaxAttempts 0, which it treats as its default of 25 — quietly betraying the "retried like an event-driven job" promise by a factor of eight in provider calls.

type Event ¶

type Event struct {
	ID            uuid.UUID           // Unique event id (UUID v7, time-ordered)
	Type          datatypes.EventType // Event type enum (e.g., FeedbackRecordCreated, WebhookCreated)
	Timestamp     time.Time           // Event creation time
	Data          any                 // Event data (FeedbackRecord, Webhook, etc.)
	ChangedFields []string            // Only for updates
}

Event represents an event that can be published to message providers (webhooks, email, etc.)

type FeedbackEmbeddingArgs ¶

type FeedbackEmbeddingArgs struct {
	FeedbackRecordID uuid.UUID `json:"feedback_record_id" river:"unique"`
	// EventID correlates this job to the domain event that enqueued it (uuid.Nil for backfill
	// jobs, which have no originating event). Not part of the dedupe key — observability only.
	EventID uuid.UUID `json:"event_id"`
	// Model is the embedding model name; stored in embeddings.model.
	Model string `json:"model" river:"unique"`
	// InputKind selects which feedback text is embedded. Empty is treated as raw for legacy queued jobs.
	InputKind models.EmbeddingInputKind `json:"input_kind,omitempty"`
	// ValueTextHash is a hash of the input (trimmed value_text, or "empty"/"backfill") for dedupe semantics.
	ValueTextHash string `json:"value_text_hash" river:"unique"`
}

FeedbackEmbeddingArgs is the job payload for generating and storing an embedding for one feedback record. Used by EmbeddingProvider and the backfill flow to enqueue, and by FeedbackEmbeddingWorker to run. Uniqueness is by (FeedbackRecordID, Model, ValueTextHash) so that edits within the uniqueness window get a new job when value_text changes; same content within 24h is deduped; one job per record+model.

func (FeedbackEmbeddingArgs) Kind() string

type FeedbackEmotionsArgs ¶

type FeedbackEmotionsArgs struct {
	FeedbackRecordID uuid.UUID `json:"feedback_record_id" river:"unique"`
	// EventID correlates this job to the domain event that enqueued it (uuid.Nil for backfill
	// jobs, which have no originating event). Not part of the dedupe key — observability only.
	EventID uuid.UUID `json:"event_id"`
	// ValueTextHash is a hash of the normalized value_text, or "empty" when value_text is blank.
	ValueTextHash string `json:"value_text_hash" river:"unique"`
}

FeedbackEmotionsArgs is the job payload for classifying one feedback record's value_text into emotion labels. The river:"unique" tags define the dedupe key (FeedbackRecordID, ValueTextHash) but only take effect where the insert passes UniqueOpts — the backfill path, whose per-run ValueTextHash discriminator stops a rescued fan-out from double-enqueuing. The event-driven path deliberately inserts WITHOUT UniqueOpts (River's completed unique state would swallow legitimate re-enrichment after an edit). Like sentiment there is no target language — emotions are classified directly from the text, independent of language.

func (FeedbackEmotionsArgs) Kind() string

type FeedbackRecordsPurgeArgs ¶

type FeedbackRecordsPurgeArgs struct {
	TenantID string `json:"tenant_id" river:"unique"`
}

FeedbackRecordsPurgeArgs deletes every feedback record for one tenant, everything derived from those records, and the taxonomy built on them — leaving the tenant's configuration (its webhooks and its settings) intact.

The purge runs as a job rather than on the request path because it is unbounded: feedback_records has no per-tenant cap, and the API server's write timeout would cut the response long before a large tenant finished deleting.

Uniqueness is by TenantID across the in-flight states only, spelled out explicitly at the enqueue site (see feedbackRecordsPurgeUniqueStates). A second request while a purge is running collapses into it; a purge requested after the previous one finished starts a new run.

Do not fall back to River's default state set here. It includes `completed`, and with no ByPeriod the uniqueness window is unbounded, so the first purge of a tenant would be the only one that ever runs — every later request skipped as a duplicate while still returning 202.

The worker re-reads the tenant from these args and scopes every statement by it; nothing about the purge is resolved at enqueue time.

func (FeedbackRecordsPurgeArgs) Kind() string

type FeedbackRecordsPurgeRepository ¶

type FeedbackRecordsPurgeRepository interface {
	PurgeFeedbackRecordsByTenant(ctx context.Context, tenantID string) (*models.FeedbackRecordsPurgeCounts, error)
}

FeedbackRecordsPurgeRepository is the repository surface the purge needs.

type FeedbackRecordsPurgeService ¶

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

FeedbackRecordsPurgeService owns the two halves of a tenant feedback-records purge: accepting the request (enqueue) and performing it (the worker's call).

They are split because the purge is unbounded and must not run on the request path. The API process only ever reaches Enqueue — its River client is insert-only — while hub-worker reaches Purge. Keeping both here means the tenant normalization rule is written once and applies to whichever entry point is used. The queue and retry budget are not constructor parameters: unlike the enrichment jobs they have no config knob, and both are properties of the job kind itself (see FeedbackRecordsPurgeArgs).

func NewFeedbackRecordsPurgeService(
	repo FeedbackRecordsPurgeRepository, inserter RiverJobInserter,
) *FeedbackRecordsPurgeService

func (s *FeedbackRecordsPurgeService) Enqueue(
	ctx context.Context, tenantID string,
) (*models.FeedbackRecordsPurgeAcceptedResponse, error)

func (s *FeedbackRecordsPurgeService) Purge(
	ctx context.Context, tenantID string,
) (*models.FeedbackRecordsPurgeCounts, error)

type FeedbackRecordsRepository ¶

type FeedbackRecordsRepository interface {
	Create(ctx context.Context, req *models.CreateFeedbackRecordRequest) (*models.FeedbackRecord, error)
	GetByID(ctx context.Context, id uuid.UUID) (*models.FeedbackRecord, error)
	List(ctx context.Context, filters *models.ListFeedbackRecordsFilters) ([]models.FeedbackRecord, bool, error)
	ListAfterCursor(
		ctx context.Context, filters *models.ListFeedbackRecordsFilters,
		cursorCollectedAt time.Time, cursorID uuid.UUID,
	) ([]models.FeedbackRecord, bool, error)
	Update(ctx context.Context, id uuid.UUID, req *models.UpdateFeedbackRecordRequest,
	) (updated, previous *models.FeedbackRecord, err error)
	SetTranslation(ctx context.Context, feedbackRecordID uuid.UUID, translated *string, langKey, defaultLang string,
		stillCurrent func(valueText *string) bool) error
	SetSentiment(ctx context.Context, feedbackRecordID uuid.UUID, sentiment *models.SentimentValue, score *float64,
		stillCurrent func(valueText *string) bool) error
	SetEmotions(ctx context.Context, feedbackRecordID uuid.UUID, emotions []models.EmotionValue,
		stillCurrent func(valueText *string) bool) error
	ClearEmotions(ctx context.Context, feedbackRecordID uuid.UUID,
		stillCurrent func(valueText *string) bool) error
	ListTranslationBackfillTargets(
		ctx context.Context, afterID uuid.UUID, limit int, defaultLang string,
	) ([]models.TranslationBackfillTarget, error)
	ListTranslationBackfillTargetsForTenant(
		ctx context.Context, tenantID string, afterID uuid.UUID, limit int, defaultLang string,
	) ([]models.TranslationBackfillTarget, error)
	ListSentimentBackfillTargets(ctx context.Context, afterID uuid.UUID, limit int) ([]uuid.UUID, error)
	ListEmotionsBackfillTargets(ctx context.Context, afterID uuid.UUID, limit int) ([]uuid.UUID, error)
	Count(ctx context.Context, filters *models.ListFeedbackRecordsFilters) (int, error)
	Delete(ctx context.Context, id uuid.UUID) error
	DeleteByUser(ctx context.Context, filters *models.DeleteFeedbackRecordsByUserFilters) ([]models.DeletedFeedbackRecordsByTenant, error)
}

FeedbackRecordsRepository defines the interface for feedback records data access.

type FeedbackRecordsService ¶

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

FeedbackRecordsService handles business logic for feedback records.

func NewFeedbackRecordsService(
	repo FeedbackRecordsRepository,
	embeddingsRepo EmbeddingsRepository,
	embeddingModel string,
	publisher MessagePublisher,
	embeddingInserter RiverJobInserter,
	embeddingQueueName string,
	embeddingMaxAttempts int,
	translationDefaultLang string,
) *FeedbackRecordsService

func (s *FeedbackRecordsService) BackfillEmbeddings(ctx context.Context, model string) (int, error)

func (s *FeedbackRecordsService) BackfillEmbeddingsWithInputKind(
	ctx context.Context,
	model string,
	inputKind models.EmbeddingInputKind,
) (int, error)

func (s *FeedbackRecordsService) BackfillEmbeddingsWithInputKindLimit(
	ctx context.Context,
	model string,
	inputKind models.EmbeddingInputKind,
	maxRecords int,
) (int, error)

func (s *FeedbackRecordsService) BackfillEmotions(
	ctx context.Context, inserter RiverJobInserter, queueName string, maxAttempts int, runID string,
) (int, error)

func (s *FeedbackRecordsService) BackfillSentiment(
	ctx context.Context, inserter RiverJobInserter, queueName string, maxAttempts int, runID string,
) (int, error)

func (s *FeedbackRecordsService) BackfillTenantEmbeddingsWithInputKind(
	ctx context.Context,
	tenantID string,
	model string,
	inputKind models.EmbeddingInputKind,
	maxRecords int,
) (int, error)

func (s *FeedbackRecordsService) BackfillTranslations(
	ctx context.Context, inserter RiverJobInserter, queueName string, maxAttempts int, runID string,
) (int, error)

func (s *FeedbackRecordsService) BackfillTranslationsForTenant(
	ctx context.Context, inserter RiverJobInserter, queueName string, maxAttempts int, tenantID, runID string,
) (int, error)

func (s *FeedbackRecordsService) ClearEmotions(
	ctx context.Context, feedbackRecordID uuid.UUID, stillCurrent func(valueText *string) bool,
) error

func (s *FeedbackRecordsService) CountFeedbackRecords(
	ctx context.Context, filters *models.ListFeedbackRecordsFilters,
) (int, error)

func (s *FeedbackRecordsService) CreateFeedbackRecord(
	ctx context.Context, req *models.CreateFeedbackRecordRequest,
) (*models.FeedbackRecord, error)

func (s *FeedbackRecordsService) DeleteFeedbackRecord(ctx context.Context, id uuid.UUID) error

func (s *FeedbackRecordsService) DeleteFeedbackRecordsByUser(
	ctx context.Context, filters *models.DeleteFeedbackRecordsByUserFilters,
) (int, error)

func (s *FeedbackRecordsService) GetFeedbackRecord(ctx context.Context, id uuid.UUID) (*models.FeedbackRecord, error)

func (s *FeedbackRecordsService) ListFeedbackRecords(
	ctx context.Context, filters *models.ListFeedbackRecordsFilters,
) (*models.ListFeedbackRecordsResponse, error)

func (s *FeedbackRecordsService) SetEmbedding(
	ctx context.Context, feedbackRecordID uuid.UUID, model string, embedding []float32,
	stillCurrent func(fieldLabel, valueText, valueTextTranslated *string) bool,
) error

func (s *FeedbackRecordsService) SetEmbeddingInserter(inserter RiverJobInserter)

func (s *FeedbackRecordsService) SetEmotions(
	ctx context.Context, feedbackRecordID uuid.UUID, emotions []models.EmotionValue,
	stillCurrent func(valueText *string) bool,
) error

func (s *FeedbackRecordsService) SetEnrichmentClearMetrics(m EnrichmentClearMetrics)

func (s *FeedbackRecordsService) SetSentiment(
	ctx context.Context, feedbackRecordID uuid.UUID, sentiment *models.SentimentValue, score *float64,
	stillCurrent func(valueText *string) bool,
) error

func (s *FeedbackRecordsService) SetTaxonomyEmbeddingModel(model string)

func (s *FeedbackRecordsService) SetTranslation(
	ctx context.Context, feedbackRecordID uuid.UUID, translated *string, langKey string,
	stillCurrent func(valueText *string) bool,
) error

func (s *FeedbackRecordsService) UpdateFeedbackRecord(
	ctx context.Context, id uuid.UUID, req *models.UpdateFeedbackRecordRequest,
) (*models.FeedbackRecord, error)

type FeedbackSentimentArgs ¶

type FeedbackSentimentArgs struct {
	FeedbackRecordID uuid.UUID `json:"feedback_record_id" river:"unique"`
	// EventID correlates this job to the domain event that enqueued it (uuid.Nil for backfill
	// jobs, which have no originating event). Not part of the dedupe key — observability only.
	EventID uuid.UUID `json:"event_id"`
	// ValueTextHash is a hash of the normalized value_text, or "empty" when value_text is blank.
	ValueTextHash string `json:"value_text_hash" river:"unique"`
}

FeedbackSentimentArgs is the job payload for classifying one feedback record's value_text. The river:"unique" tags define the dedupe key (FeedbackRecordID, ValueTextHash) but only take effect where the insert passes UniqueOpts — the backfill path, whose per-run ValueTextHash discriminator stops a rescued fan-out from double-enqueuing. The event-driven path deliberately inserts WITHOUT UniqueOpts (River's completed unique state would swallow legitimate re-enrichment after an edit). Unlike translation there is no target language — sentiment is classified directly from the text, independent of language.

func (FeedbackSentimentArgs) Kind() string

type FeedbackTranslationArgs ¶

type FeedbackTranslationArgs struct {
	FeedbackRecordID uuid.UUID `json:"feedback_record_id" river:"unique"`
	// EventID correlates this job to the domain event that enqueued it (uuid.Nil for backfill
	// jobs, which have no originating event). Not part of the dedupe key — observability only.
	EventID uuid.UUID `json:"event_id"`
	// TargetLang is the tenant's configured target language (BCP-47) at enqueue time.
	TargetLang string `json:"target_lang" river:"unique"`
	// ValueTextHash is a hash of the inputs that determine the translation — the
	// normalized value_text and the source language — or "empty" when value_text is blank.
	ValueTextHash string `json:"value_text_hash" river:"unique"`
}

FeedbackTranslationArgs is the job payload for translating one feedback record's value_text into the tenant's target language. Uniqueness is by (FeedbackRecordID, TargetLang, ValueTextHash): a value_text edit or a target-language change yields a new job, while identical content for the same target within the window is deduped.

func (FeedbackTranslationArgs) Kind() string

type JobKindSpec ¶

type JobKindSpec struct {
	// Args is a zero value of the job's argument type; Args.Kind() is the River job kind.
	Args river.JobArgs
	// Queue is the River queue the kind is inserted on by the event path.
	Queue string
	// ReconcileQueue is the second, lower-concurrency queue the same kind is inserted on when the
	// work is repaired historical data rather than event-driven, or "" for kinds with no such lane.
	//
	// A kind, not a worker, is what River registers, so the same worker serves both lanes and the
	// queue is purely a concurrency budget. That is the entire mechanism keeping a large repair
	// backlog from starving a record submitted right now: the two lanes draw from different
	// MaxWorkers.
	ReconcileQueue string
}

JobKindSpec pairs a River job kind with the queue its inserts land on.

func JobKindSpecs() []JobKindSpec

func (s JobKindSpec) Kind() string

type ListPaginationMeta ¶

type ListPaginationMeta struct {
	Limit      int
	NextCursor string
}

ListPaginationMeta holds pagination metadata for list endpoints (feedback records, webhooks).

func BuildListPaginationMeta(
	limit int, hasMore bool, encodeLast func() (string, error),
) (ListPaginationMeta, error)

type MessagePublisher ¶

type MessagePublisher interface {
	// PublishEvent publishes a single event with data (no changed fields)
	PublishEvent(ctx context.Context, eventType datatypes.EventType, data any)
	// PublishEventWithChangedFields publishes a single event with data and optional changed fields (for updates)
	PublishEventWithChangedFields(ctx context.Context, eventType datatypes.EventType, data any, changedFields []string)
}

MessagePublisher defines the interface for publishing events.

type MessagePublisherManager ¶

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

MessagePublisherManager coordinates multiple message providers.

func NewMessagePublisherManager(
	bufferSize int, perEventTimeout time.Duration, metrics observability.EventMetrics,
) *MessagePublisherManager

func (m *MessagePublisherManager) PublishEvent(ctx context.Context, eventType datatypes.EventType, data any)

func (m *MessagePublisherManager) PublishEventWithChangedFields(
	ctx context.Context, eventType datatypes.EventType, data any, changedFields []string,
)

func (m *MessagePublisherManager) RegisterProvider(provider eventPublisher)

func (m *MessagePublisherManager) Shutdown()

type NewEnrichmentReconcileServiceParams ¶

type NewEnrichmentReconcileServiceParams struct {
	Repo        EnrichmentReconcileRepository
	Inserter    RiverBatchInserter
	DefaultLang string
	TargetDepth int
	Specs       []EnrichmentSweepSpec
}

NewEnrichmentReconcileServiceParams configures the sweep.

type NewEnrichmentRetryServiceParams ¶

type NewEnrichmentRetryServiceParams struct {
	Repo                  EnrichmentRetryRepository
	Settings              TenantSettingsReader
	DefaultLang           string
	TranslationConfigured bool
	SentimentConfigured   bool
	EmotionsConfigured    bool
	// Cooldown overrides the default window. Zero uses EnrichmentRetryCooldown; tests set it short.
	Cooldown time.Duration
	// ReconcileEnabled is cfg.EnrichmentReconcile.Enabled — whether anything will ever act on a
	// clear. False makes Retry refuse outright rather than accept a no-op.
	ReconcileEnabled bool
	// Metrics counts each enrichment's retry outcome. nil disables them; retries still work.
	Metrics observability.EnrichmentReconcileMetrics
}

NewEnrichmentRetryServiceParams configures the retry service.

type NewEnrichmentStatusServiceParams ¶

type NewEnrichmentStatusServiceParams struct {
	Repo                  EnrichmentStatusRepository
	Settings              TenantSettingsReader
	DefaultLang           string
	TranslationConfigured bool
	SentimentConfigured   bool
	EmotionsConfigured    bool
}

NewEnrichmentStatusServiceParams configures an EnrichmentStatusService. The *Configured flags are the deployment-level gates (provider+model set), mirroring how the enrichment providers are constructed; defaultLang is TRANSLATION_DEFAULT_LANGUAGE.

type NewTaxonomyServiceParams ¶

type NewTaxonomyServiceParams struct {
	Repo                  TaxonomyRepository
	Starter               TaxonomyRunStarter
	EmbeddingModel        string
	MinimumEmbeddingCount int
	Metrics               observability.TaxonomyMetrics
}

NewTaxonomyServiceParams configures a TaxonomyService.

type ReconcileResult ¶

type ReconcileResult struct {
	// Enqueued counts jobs actually inserted — after River dropped the ones already queued, so
	// this is new work rather than attempts.
	Enqueued map[string]int
	// Skipped names enrichments whose queue was already at or above the target depth. A sweep that
	// skips everything is the steady state once a backlog is draining, not a problem.
	Skipped []string
}

ReconcileResult reports what one sweep did, per enrichment.

type RiverBatchInserter ¶

type RiverBatchInserter interface {
	InsertMany(ctx context.Context, params []river.InsertManyParams) ([]*rivertype.JobInsertResult, error)
}

RiverBatchInserter inserts many River jobs in one round trip. Shared by the webhook fan-out and the enrichment reconcile sweep, both of which enqueue in batches; the concrete River client satisfies it, and keeping the seam this small is what lets their tests run without a database-backed queue.

InsertMany, never InsertManyFast. The fast variant skips the unique-options machinery and fails the entire batch on a conflict, and callers here depend on uniqueness to avoid enqueueing work that is already in flight.

type RiverJobInserter ¶

type RiverJobInserter interface {
	Insert(ctx context.Context, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error)
}

RiverJobInserter inserts a single River job. It is the shared seam the enrichment providers (embedding, translation) and their backfill flows use to enqueue work; satisfied by the River client.

type SSRFPolicy ¶

type SSRFPolicy struct {
	// Blacklist is a set of normalized hostnames/IPs that can never be used as webhook URLs.
	// An entry here is an explicit deny and wins over AllowedCIDRs.
	Blacklist map[string]struct{}

	// AllowedCIDRs re-permits specific private/reserved ranges that isPrivateOrReserved would
	// otherwise block, for operators whose webhook receivers legitimately live on internal
	// addresses (e.g. a tailnet in 100.64.0.0/10). It does not override Blacklist.
	AllowedCIDRs []netip.Prefix
}

SSRFPolicy holds the webhook URL restrictions applied at create/update time and again at dial time. The zero value still rejects private/reserved ranges; Blacklist and AllowedCIDRs are both optional.

func NewSSRFPolicy(blacklist map[string]struct{}, allowedCIDRs []netip.Prefix) SSRFPolicy

type SearchResult ¶

type SearchResult struct {
	Results    []models.FeedbackRecordWithScore
	NextCursor string // non-empty if there may be a next page (len(Results) == requested limit)
}

SearchResult holds the results and optional next-page cursor from semantic search or similar feedback.

type SearchService ¶

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

SearchService performs semantic search and similar-feedback lookups using embeddings.

func NewSearchService(p SearchServiceParams) *SearchService

func (s *SearchService) SemanticSearch(
	ctx context.Context, query, tenantID string, limit int, minScore float64, cursor string,
) (SearchResult, error)

func (s *SearchService) SimilarFeedback(
	ctx context.Context, feedbackRecordID uuid.UUID, limit int, minScore float64, cursor string,
) (SearchResult, error)

type SearchServiceParams ¶

type SearchServiceParams struct {
	EmbeddingClient EmbeddingClient
	EmbeddingsRepo  EmbeddingsRepositoryForSearch
	Model           string
	QueryCache      *lru.Cache[string, []float32]
	CacheMetrics    observability.CacheMetrics
	Logger          *slog.Logger
}

SearchServiceParams configures SearchService. QueryCache and CacheMetrics may be nil (no caching).

type SentimentClient ¶

type SentimentClient interface {
	// Classify returns the sentiment of text. sourceLang is the record's BCP-47 language and
	// may be empty; it is passed to the model only as a hint (classification is multilingual).
	Classify(ctx context.Context, text, sourceLang string) (SentimentResult, error)
}

SentimentClient classifies the polarity of open feedback text into a sentiment label and score. Implementations call an LLM provider (OpenAI or Google) via structured output; the factory selects one from configuration. It mirrors the TranslationClient seam so the worker depends on the interface, not a provider.

func NewSentimentClient(ctx context.Context, cfg SentimentClientConfig) (SentimentClient, error)

type SentimentClientConfig ¶

type SentimentClientConfig = EnrichmentClientConfig

SentimentClientConfig aliases the shared classify client config (see EnrichmentClientConfig).

type SentimentProvider ¶

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

SentimentProvider enqueues one sentiment job per eligible feedback record event, over the shared enrichmentProvider. Eligibility is a text field with non-empty open text; re-classification is triggered by a value_text change (sentiment does not depend on source language, unlike translation). Like translation it resolves a per-tenant setting on the enqueue path — the per-directory sentiment switch (ENG-1529) — and skips tenants that have turned sentiment off.

func NewSentimentProvider(
	inserter RiverJobInserter,
	resolver TenantSettingsReader,
	queueName string,
	maxAttempts int,
	metrics observability.SentimentMetrics,
) *SentimentProvider

func (p SentimentProvider) PublishEvent(ctx context.Context, event Event)

type SentimentResult ¶

type SentimentResult struct {
	Label models.SentimentValue
	Score float64
}

SentimentResult is the parsed, validated output of one sentiment classification: a known label and a score clamped to [models.SentimentScoreMin, SentimentScoreMax].

type SettingsChangeListener ¶

type SettingsChangeListener interface {
	OnSettingsChanged(ctx context.Context, tenantID string, changedKeys []string)
}

SettingsChangeListener is notified after a tenant's settings are successfully written, with the setting keys the write touched. It lets enrichment side-effects (e.g. re-translation) react to a settings change without TenantSettingsService depending on any enrichment concern — the service depends only on this translation-free port.

Implementations do the reaction themselves (a fast, durable enqueue) and own their error handling: the method returns nothing because the settings write has already committed and the side-effect must never fail it.

func NewCompositeSettingsChangeListener(listeners ...SettingsChangeListener) SettingsChangeListener

type TaxonomyClient ¶

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

TaxonomyClient calls the standalone taxonomy service.

func NewTaxonomyClient(cfg TaxonomyClientConfig, httpClient *http.Client) (*TaxonomyClient, error)

func (c *TaxonomyClient) StartRun(ctx context.Context, runID string) error

type TaxonomyClientConfig ¶

type TaxonomyClientConfig struct {
	ServiceURL   string
	ServiceToken string
}

TaxonomyClientConfig configures the outbound client Hub uses to call the taxonomy service.

type TaxonomyRepository ¶

type TaxonomyRepository interface {
	ListFieldOptions(ctx context.Context, tenantID, embeddingModel string) ([]models.TaxonomyFieldOption, error)
	CountScopeInput(ctx context.Context, scope models.TaxonomyScope, embeddingModel string) (int, int, *string, error)
	CreateRunIfAvailable(ctx context.Context, params repository.CreateTaxonomyRunParams) (*models.TaxonomyRun, bool, error)
	MarkRunRunning(ctx context.Context, runID uuid.UUID, tenantID string) (*models.TaxonomyRun, error)
	MarkRunFailed(
		ctx context.Context,
		runID uuid.UUID,
		tenantID string,
		message string,
		errorCode models.TaxonomyRunFailureCode,
		metrics json.RawMessage,
	) (*models.TaxonomyRun, error)
	Heartbeat(ctx context.Context, runID uuid.UUID, tenantID string) error
	GetRunForInternalService(ctx context.Context, runID uuid.UUID) (*models.TaxonomyRun, error)
	GetRunForTenant(ctx context.Context, runID uuid.UUID, tenantID string) (*models.TaxonomyRun, error)
	GetActiveRun(ctx context.Context, scope models.TaxonomyScope) (*models.TaxonomyRun, error)
	ListRuns(ctx context.Context, filters models.ListTaxonomyRunsFilters) ([]models.TaxonomyRun, error)
	GetRunInput(
		ctx context.Context,
		runID uuid.UUID,
		tenantID string,
		embeddingModel string,
	) (*models.TaxonomyRunInputResponse, error)
	GetRunInputRecordIDs(
		ctx context.Context,
		runID uuid.UUID,
		tenantID string,
	) ([]uuid.UUID, error)
	StoreResultAndActivate(
		ctx context.Context,
		runID uuid.UUID,
		tenantID string,
		req models.TaxonomyRunResultRequest,
	) (*models.TaxonomyRun, error)
	GetTree(ctx context.Context, runID uuid.UUID, tenantID string) (*models.TaxonomyTreeResponse, error)
	RenameNode(ctx context.Context, nodeID uuid.UUID, tenantID, actorID, label string) (*models.TaxonomyNode, error)
	RemoveNode(ctx context.Context, nodeID uuid.UUID, tenantID, actorID string) (*models.TaxonomyNode, error)
	ListNodeRecords(ctx context.Context, nodeID uuid.UUID, tenantID string, limit int) ([]models.FeedbackRecord, int, error)
	CountNodeRecords(ctx context.Context, runID uuid.UUID, tenantID string) ([]models.TaxonomyNodeRecordCount, error)
}

TaxonomyRepository persists taxonomy run state and generated artifacts.

type TaxonomyRunStarter ¶

type TaxonomyRunStarter interface {
	StartRun(ctx context.Context, runID string) error
}

TaxonomyRunStarter starts asynchronous taxonomy compute work.

type TaxonomyService ¶

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

TaxonomyService coordinates taxonomy run lifecycle and edits.

func NewTaxonomyService(params NewTaxonomyServiceParams) *TaxonomyService

func (s *TaxonomyService) CompleteRun(
	ctx context.Context,
	runID uuid.UUID,
	req models.TaxonomyRunResultRequest,
) (*models.TaxonomyRun, error)

func (s *TaxonomyService) FailRun(
	ctx context.Context,
	runID uuid.UUID,
	req models.TaxonomyRunFailedRequest,
) (*models.TaxonomyRun, error)

func (s *TaxonomyService) GetActiveTree(
	ctx context.Context,
	scope models.TaxonomyScope,
) (*models.TaxonomyTreeResponse, error)

func (s *TaxonomyService) GetNodeRecordCounts(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
) (*models.TaxonomyRecordCountsResponse, error)

func (s *TaxonomyService) GetRun(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
) (*models.TaxonomyRun, error)

func (s *TaxonomyService) GetRunInput(
	ctx context.Context,
	runID uuid.UUID,
) (*models.TaxonomyRunInputResponse, error)

func (s *TaxonomyService) GetTree(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
) (*models.TaxonomyTreeResponse, error)

func (s *TaxonomyService) Heartbeat(
	ctx context.Context,
	runID uuid.UUID,
) error

func (s *TaxonomyService) ListFieldOptions(
	ctx context.Context,
	tenantID string,
) (*models.TaxonomyFieldsResponse, error)

func (s *TaxonomyService) ListNodeRecords(
	ctx context.Context,
	nodeID uuid.UUID,
	filters models.TaxonomyNodeRecordsFilters,
) (*models.TaxonomyNodeRecordsResponse, error)

func (s *TaxonomyService) ListRuns(
	ctx context.Context,
	filters models.ListTaxonomyRunsFilters,
) (*models.ListTaxonomyRunsResponse, error)

func (s *TaxonomyService) RemoveNode(
	ctx context.Context,
	nodeID uuid.UUID,
	filters models.RemoveTaxonomyNodeFilters,
) (*models.TaxonomyNode, error)

func (s *TaxonomyService) RenameNode(
	ctx context.Context,
	nodeID uuid.UUID,
	req models.RenameTaxonomyNodeRequest,
) (*models.TaxonomyNode, error)

func (s *TaxonomyService) StartManualRun(
	ctx context.Context,
	req models.CreateTaxonomyRunRequest,
) (*models.CreateTaxonomyRunResponse, error)

type TenantDataRepository ¶

type TenantDataRepository interface {
	DeleteByTenant(ctx context.Context, tenantID string) (*models.TenantDataDeleteCounts, error)
}

TenantDataRepository defines tenant data purge access.

type TenantDataService ¶

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

TenantDataService handles tenant data purge business logic.

func NewTenantDataService(repo TenantDataRepository) *TenantDataService

func (s *TenantDataService) DeleteTenantData(ctx context.Context, tenantID string) (*models.TenantDataDeleteResult, error)

type TenantSettingsReader ¶

type TenantSettingsReader interface {
	GetSettings(ctx context.Context, tenantID string) (*models.TenantSettings, error)
}

TenantSettingsReader is the read surface the cache wraps. *TenantSettingsService satisfies it, so the cache is a drop-in for any consumer that only needs reads (the translation enqueue gate and worker).

type TenantSettingsRepository ¶

type TenantSettingsRepository interface {
	Get(ctx context.Context, tenantID string) (*models.TenantSettings, bool, error)
	Upsert(ctx context.Context, tenantID string, settings models.EnrichmentSettings) (*models.TenantSettings, error)
	// Patch merges set into the tenant's settings and removes removeKeys (an RFC
	// 7396 merge patch: set + delete); the two are disjoint.
	Patch(
		ctx context.Context, tenantID string, set models.EnrichmentSettings, removeKeys []string,
	) (*models.TenantSettings, error)
}

TenantSettingsRepository is the persistence surface the settings service needs.

type TenantSettingsService ¶

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

TenantSettingsService reads and writes tenant-scoped enrichment settings. It is the accessor enrichment workflows will use to resolve a tenant's configuration.

func NewTenantSettingsService(repo TenantSettingsRepository) *TenantSettingsService

func (s *TenantSettingsService) GetSettings(ctx context.Context, tenantID string) (*models.TenantSettings, error)

func (s *TenantSettingsService) PatchSettings(
	ctx context.Context, tenantID string, req *models.PatchTenantSettingsRequest,
) (*models.TenantSettings, error)

func (s *TenantSettingsService) SetSettingsChangeListener(listener SettingsChangeListener)

func (s *TenantSettingsService) UpdateSettings(
	ctx context.Context, tenantID string, req *models.UpdateTenantSettingsRequest,
) (*models.TenantSettings, error)

type TenantTranslationBackfillArgs ¶

type TenantTranslationBackfillArgs struct {
	TenantID string `json:"tenant_id" river:"unique"`
}

TenantTranslationBackfillArgs fans out a re-translation backfill for one tenant: the worker lists the tenant's stale text records and enqueues a FeedbackTranslationArgs job for each. It is enqueued when a tenant's translation-relevant settings change (today: target_language).

Uniqueness is by TenantID across the default (in-flight) states — the enqueue site sets ByArgs without a ByPeriod — so rapid repeated settings changes collapse to a single in-flight backfill, while a change after the previous backfill has completed re-triggers. The worker resolves the tenant's current target at run time, so a coalesced backfill always targets the latest configured language.

func (TenantTranslationBackfillArgs) Kind() string

type TranslateRequest ¶

type TranslateRequest struct {
	Text       string
	SourceLang string
	TargetLang string
}

TranslateRequest is the input to a single translation. SourceLang and TargetLang are BCP-47 tags: TargetLang comes from the tenant's settings; SourceLang from the feedback record's language and may be empty when the source language is unknown.

type TranslationClient ¶

type TranslationClient interface {
	Translate(ctx context.Context, req TranslateRequest) (string, error)
}

TranslationClient translates TranslateRequest.Text from SourceLang into TargetLang and returns the translated text. Implementations call an LLM provider (OpenAI or Google); the factory selects one from configuration. It mirrors the EmbeddingClient seam so the worker depends on the interface, not a provider.

func NewTranslationClient(ctx context.Context, cfg TranslationClientConfig) (TranslationClient, error)

type TranslationClientConfig ¶

type TranslationClientConfig = EnrichmentClientConfig

TranslationClientConfig aliases the shared classify client config (see EnrichmentClientConfig).

type TranslationProvider ¶

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

TranslationProvider enqueues one translation job per eligible feedback record event, over the shared enrichmentProvider. Eligibility is a text field with non-empty open text; re-translation is triggered by a value_text OR source-language change (translation output depends on both, unlike sentiment/embedding). The per-tenant target language (settings.target_language, falling back to defaultLang) is both the gate — a record with no resolvable target is skipped — and part of the job args.

func NewTranslationProvider(
	inserter RiverJobInserter,
	resolver TenantSettingsReader,
	queueName string,
	maxAttempts int,
	defaultLang string,
	metrics observability.TranslationMetrics,
) *TranslationProvider

func (p TranslationProvider) PublishEvent(ctx context.Context, event Event)

type WebhookDispatchArgs ¶

type WebhookDispatchArgs struct {
	EventID       uuid.UUID `json:"event_id"                 river:"unique"`
	EventType     string    `json:"event_type"`
	Timestamp     time.Time `json:"timestamp"`
	Data          any       `json:"data"`
	ChangedFields []string  `json:"changed_fields,omitempty"`
	TenantID      *string   `json:"tenant_id,omitempty"`
	WebhookID     uuid.UUID `json:"webhook_id"               river:"unique"`
}

WebhookDispatchArgs is the job payload for one (event, webhook) delivery. Used by WebhookProvider to enqueue and by WebhookDispatchWorker to run. Only event_id and webhook_id are used for River uniqueness (river:"unique") so the hash is fast and does not include the potentially large data payload.

func (WebhookDispatchArgs) Kind() string

type WebhookDispatchInserter ¶

type WebhookDispatchInserter = RiverBatchInserter

WebhookDispatchInserter inserts webhook_dispatch jobs in batch (e.g. River client).

type WebhookPayload ¶

type WebhookPayload struct {
	ID            uuid.UUID `json:"id"`                       // Unique event id (UUID v7)
	Type          string    `json:"type"`                     // Event type as string (e.g., "feedback_record.created", "webhook.created")
	Timestamp     time.Time `json:"timestamp"`                // Event creation timestamp
	TenantID      *string   `json:"tenant_id,omitempty"`      // Tenant boundary for the event
	Data          any       `json:"data"`                     // Event data (FeedbackRecord, Webhook, etc.)
	ChangedFields []string  `json:"changed_fields,omitempty"` // Only for update events (optional)
}

WebhookPayload represents a generic webhook payload structure for all event types. The Data field can contain FeedbackRecord, Webhook, or other event data types.

func NewWebhookPayload(args WebhookDispatchArgs) *WebhookPayload

type WebhookProvider ¶

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

WebhookProvider implements eventPublisher by enqueueing one River job per (event, webhook).

func NewWebhookProvider(
	inserter RiverBatchInserter, repo WebhookProviderRepository,
	maxAttempts, maxFanOut int,
	enqueueMaxRetries int, enqueueInitialBackoff, enqueueMaxBackoff time.Duration,
	metrics observability.WebhookMetrics,
) *WebhookProvider

func (p *WebhookProvider) PublishEvent(ctx context.Context, event Event)

type WebhookProviderRepository ¶

type WebhookProviderRepository interface {
	ListEnabledForEventTypeAndTenant(ctx context.Context, eventType string, tenantID *string) ([]models.Webhook, error)
}

WebhookProviderRepository lists tenant-scoped webhooks eligible for event fan-out.

type WebhookSender ¶

type WebhookSender interface {
	Send(ctx context.Context, webhook *models.Webhook, payload *WebhookPayload) error
}

WebhookSender sends a single webhook payload to an endpoint (Standard Webhooks: signing, headers, 410 handling).

type WebhookSenderImpl ¶

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

WebhookSenderImpl implements WebhookSender with Standard Webhooks conformance.

func NewWebhookSenderImpl(
	repo WebhookSenderRepository, metrics observability.WebhookMetrics, ssrfPolicy SSRFPolicy,
	httpTimeout time.Duration, httpClient *http.Client,
) *WebhookSenderImpl

func (s *WebhookSenderImpl) Send(ctx context.Context, webhook *models.Webhook, payload *WebhookPayload) error

type WebhookSenderRepository ¶

type WebhookSenderRepository interface {
	Update(ctx context.Context, id uuid.UUID, req *models.UpdateWebhookRequest) (*models.Webhook, error)
}

WebhookSenderRepository persists webhook state changes caused by delivery.

type WebhooksRepository ¶

type WebhooksRepository interface {
	Create(ctx context.Context, req *models.CreateWebhookRequest) (*models.Webhook, error)
	GetByID(ctx context.Context, id uuid.UUID) (*models.Webhook, error)
	List(ctx context.Context, filters *models.ListWebhooksFilters) ([]models.Webhook, bool, error)
	ListAfterCursor(
		ctx context.Context, filters *models.ListWebhooksFilters,
		cursorCreatedAt time.Time, cursorID uuid.UUID,
	) ([]models.Webhook, bool, error)
	Count(ctx context.Context, filters *models.ListWebhooksFilters) (int64, error)
	Update(ctx context.Context, id uuid.UUID, req *models.UpdateWebhookRequest) (*models.Webhook, error)
	Delete(ctx context.Context, id uuid.UUID) (*models.DeletedWebhook, error)
}

WebhooksRepository defines the interface for webhooks data access.

type WebhooksService ¶

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

WebhooksService handles business logic for webhooks.

func NewWebhooksService(
	repo WebhooksRepository, publisher MessagePublisher, maxWebhooks int, ssrfPolicy SSRFPolicy,
) *WebhooksService

func (s *WebhooksService) CreateWebhook(ctx context.Context, req *models.CreateWebhookRequest) (*models.Webhook, error)

func (s *WebhooksService) DeleteWebhook(ctx context.Context, id uuid.UUID) error

func (s *WebhooksService) GetWebhook(ctx context.Context, id uuid.UUID) (*models.Webhook, error)

func (s *WebhooksService) ListWebhooks(ctx context.Context, filters *models.ListWebhooksFilters) (*models.ListWebhooksResponse, error)

func (s *WebhooksService) UpdateWebhook(ctx context.Context, id uuid.UUID, req *models.UpdateWebhookRequest) (*models.Webhook, error)