diff --git a/cmd/api/app.go b/cmd/api/app.go index f837f235..00218ded 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "net/http" + "strings" "time" lru "github.com/hashicorp/golang-lru/v2" @@ -52,6 +53,18 @@ var ( const ( riverQueueDepthInterval = 15 * time.Second startupCleanupTimeout = 5 * time.Second + // enrichmentBacklogInterval is deliberately much slower than the River depth poll: the backlog + // query is a full-table aggregate over feedback_records (a high-write table), so it runs + // infrequently to minimize shared-DB load and keep the MVCC snapshot it holds short-lived + // relative to VACUUM. Backlog is a slow-moving trend signal, so 5-minute resolution is ample. + // The poller only runs at all when metrics are enabled (see App.Run). + enrichmentBacklogInterval = 5 * time.Minute + // enrichmentBacklogQueryTimeout bounds each aggregate scan so a slow query cannot pin a pool + // connection, stall the ticker, or hold a long snapshot that delays VACUUM on feedback_records. + enrichmentBacklogQueryTimeout = 30 * time.Second + // enrichmentBacklogFailuresBeforeError is how many consecutive failed refreshes escalate the + // log from warn to error (a transient blip is expected; a sustained run means a stale gauge). + enrichmentBacklogFailuresBeforeError = 3 ) // embeddingProviderAndModel returns (provider, model) when embeddings are enabled: both EMBEDDING_PROVIDER @@ -545,6 +558,17 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) { taxonomyHandler := handlers.NewTaxonomyHandler(taxonomyService) feedbackRecordsHandler := handlers.NewFeedbackRecordsHandler(feedbackRecordsService) taxonomyInternalHandler := handlers.NewTaxonomyInternalHandler(taxonomyService) + + enrichmentStatusService := service.NewEnrichmentStatusService(service.NewEnrichmentStatusServiceParams{ + Repo: repository.NewEnrichmentStatusRepository(db), + Settings: tenantSettingsService, + DefaultLang: cfg.Translation.DefaultLanguage, + TranslationConfigured: cfg.Translation.Provider != "" && cfg.Translation.Model != "", + SentimentConfigured: cfg.Sentiment.Enabled(), + EmotionsConfigured: cfg.Emotions.Enabled(), + }) + enrichmentStatusHandler := handlers.NewEnrichmentStatusHandler(enrichmentStatusService) + healthHandler := handlers.NewHealthHandler() openapiHandler, err := handlers.NewOpenAPIHandler(handlers.ResolveOpenAPISpecPath(), cfg.Server.PublicBaseURL) @@ -557,7 +581,7 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) { server := newHTTPServer( cfg, healthHandler, openapiHandler, feedbackRecordsHandler, webhooksHandler, tenantDataHandler, tenantSettingsHandler, searchHandler, - taxonomyHandler, taxonomyInternalHandler, + taxonomyHandler, taxonomyInternalHandler, enrichmentStatusHandler, meterProvider, tracerProvider, ) @@ -588,6 +612,7 @@ func newHTTPServer( search *handlers.SearchHandler, taxonomy *handlers.TaxonomyHandler, taxonomyInternal *handlers.TaxonomyInternalHandler, + enrichmentStatus *handlers.EnrichmentStatusHandler, meterProvider *sdkmetric.MeterProvider, tracerProvider *sdktrace.TracerProvider, ) *http.Server { @@ -615,6 +640,8 @@ func newHTTPServer( protected.HandleFunc("PUT /v1/tenants/{tenant_id}/settings", tenantSettings.Update) protected.HandleFunc("PATCH /v1/tenants/{tenant_id}/settings", tenantSettings.Patch) + protected.HandleFunc("GET /v1/enrichment-status", enrichmentStatus.GetStatus) + // Search endpoints are always registered; when embeddings are disabled, the handler returns 503. protected.HandleFunc("POST /v1/feedback-records/search/semantic", search.SemanticSearch) protected.HandleFunc("GET /v1/feedback-records/{id}/similar", search.SimilarFeedback) @@ -692,6 +719,17 @@ func (a *App) Run(ctx context.Context) error { go runRiverQueueDepthPoller(ctx, a.db, a.metrics.Events) } + if a.metrics != nil && a.metrics.EnrichmentBacklog != nil { + go runEnrichmentBacklogPoller(ctx, a.db, a.metrics.EnrichmentBacklog, enrichmentBacklogPollConfig{ + // Trim to stay consistent with NewEnrichmentStatusService (config already canonicalizes + // this, so it's defensive symmetry) — the endpoint and the gauge resolve the same target. + defaultLang: strings.TrimSpace(a.cfg.Translation.DefaultLanguage), + translationConfigured: a.cfg.Translation.Provider != "" && a.cfg.Translation.Model != "", + sentimentConfigured: a.cfg.Sentiment.Enabled(), + emotionsConfigured: a.cfg.Emotions.Enabled(), + }) + } + // Reap taxonomy runs orphaned in a non-terminal state, but only when the taxonomy service is wired // (no runs exist otherwise, so the sweep would be pointless). if a.taxonomyRepo != nil && (a.cfg.Taxonomy.ServiceURL != "" || a.cfg.Taxonomy.ServiceToken != "") { @@ -730,6 +768,104 @@ var riverDepthQueues = []string{ service.EmotionsQueueName, } +// enrichmentBacklogPollConfig configures runEnrichmentBacklogPoller: the deployment default target +// language and which enrichments are deployment-configured (only those emit a gauge). +type enrichmentBacklogPollConfig struct { + defaultLang string + translationConfigured bool + sentimentConfigured bool + emotionsConfigured bool +} + +// runEnrichmentBacklogPoller periodically refreshes the aggregate enrichment-backlog gauge +// (eligible-but-unenriched records per enrichment, summed across all tenants) — a durable +// completeness signal complementing the transient River queue-depth gauge. Only +// deployment-configured enrichments are reported, and each scan is bounded by its own timeout. +func runEnrichmentBacklogPoller( + ctx context.Context, + db *pgxpool.Pool, + backlog observability.EnrichmentBacklogMetrics, + cfg enrichmentBacklogPollConfig, +) { + leader := repository.NewEnrichmentBacklogLeader(db) + defer leader.Close(ctx) + + ticker := time.NewTicker(enrichmentBacklogInterval) + defer ticker.Stop() + + consecutiveFailures := 0 + + update := func() { + queryCtx, cancel := context.WithTimeout(ctx, enrichmentBacklogQueryTimeout) + defer cancel() + + // Exactly one replica holds leadership and scans; the rest skip until it goes away. + counts, isLeader, err := leader.CountIfLeader(queryCtx, cfg.defaultLang) + if err != nil { + // Shutdown cancels the scan mid-flight. That is not a poll failure, and counting it + // would fire the very alert this counter exists for on every rolling deploy. + if ctx.Err() != nil { + return + } + + consecutiveFailures++ + + // A failed scan also costs this process its leadership, so withdraw the series rather + // than leave it frozen at the last good reading while the new leader publishes its own. + backlog.ClearEnrichmentPending() + + // Always count the failure so a stale gauge is alertable, then escalate the log from + // warn to error once failures persist: a single blip is noise, a run of them means the + // gauge is frozen at its last value and silently lying about the backlog. + backlog.RecordPollError(ctx) + + if consecutiveFailures >= enrichmentBacklogFailuresBeforeError { + slog.ErrorContext(ctx, "enrichment backlog poll failing repeatedly; gauge is stale", + "error", err, "consecutive_failures", consecutiveFailures) + } else { + slog.WarnContext(ctx, "enrichment backlog poll failed", + "error", err, "consecutive_failures", consecutiveFailures) + } + + return + } + + consecutiveFailures = 0 + + if !isLeader { + // Another replica owns this gauge and exports the single global series for it. Drop + // anything this process exported while it was previously the leader, so a handover + // leaves exactly one series rather than a live one plus a frozen one. + backlog.ClearEnrichmentPending() + + return + } + + if cfg.translationConfigured { + backlog.SetEnrichmentPending(observability.EnrichmentTypeTranslation, counts.TranslationEligible-counts.TranslationDone) + } + + if cfg.sentimentConfigured { + backlog.SetEnrichmentPending(observability.EnrichmentTypeSentiment, counts.SentimentEligible-counts.SentimentDone) + } + + if cfg.emotionsConfigured { + backlog.SetEnrichmentPending(observability.EnrichmentTypeEmotions, counts.EmotionsEligible-counts.EmotionsDone) + } + } + + update() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + update() + } + } +} + // runRiverQueueDepthPoller periodically updates the per-queue River backlog gauge. Covering // every declared queue (not just default) means a provider outage or a backfill piling tens of // thousands of jobs into an enrichment queue is visible in metrics before users notice the lag. diff --git a/cmd/api/app_test.go b/cmd/api/app_test.go index e2337745..075f4c83 100644 --- a/cmd/api/app_test.go +++ b/cmd/api/app_test.go @@ -380,6 +380,7 @@ func newTestHTTPServerWithConfig(t *testing.T, publicBaseURL string, taxonomy co handlers.NewSearchHandler(nil), handlers.NewTaxonomyHandler(nil), handlers.NewTaxonomyInternalHandler(), + handlers.NewEnrichmentStatusHandler(nil), nil, nil, ) diff --git a/go.mod b/go.mod index 2e267b88..85a4966d 100644 --- a/go.mod +++ b/go.mod @@ -27,8 +27,8 @@ require ( go.opentelemetry.io/otel/sdk v1.43.0 go.opentelemetry.io/otel/sdk/metric v1.43.0 go.opentelemetry.io/otel/trace v1.43.0 - golang.org/x/sync v0.20.0 - golang.org/x/text v0.37.0 + golang.org/x/sync v0.21.0 + golang.org/x/text v0.39.0 google.golang.org/genai v1.54.0 ) diff --git a/go.sum b/go.sum index f447f872..4a587470 100644 --- a/go.sum +++ b/go.sum @@ -187,14 +187,14 @@ golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= +golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= -golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= -golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c= -golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI= +golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus= +golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM= +golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q= +golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/api v0.276.0 h1:nVArUtfLEihtW+b0DdcqRGK1xoEm2+ltAihyztq7MKY= diff --git a/internal/api/handlers/enrichment_status_handler.go b/internal/api/handlers/enrichment_status_handler.go new file mode 100644 index 00000000..47885fe9 --- /dev/null +++ b/internal/api/handlers/enrichment_status_handler.go @@ -0,0 +1,47 @@ +package handlers + +import ( + "context" + "net/http" + + "github.com/formbricks/hub/internal/api/response" + "github.com/formbricks/hub/internal/models" +) + +// EnrichmentStatusService is the application service used by the enrichment status handler. +type EnrichmentStatusService interface { + GetEnrichmentStatus(ctx context.Context, tenantID string) (*models.EnrichmentStatusResponse, error) +} + +// EnrichmentStatusHandler hosts the public enrichment-status endpoint. +type EnrichmentStatusHandler struct { + service EnrichmentStatusService +} + +// NewEnrichmentStatusHandler creates an enrichment status handler. +func NewEnrichmentStatusHandler(service EnrichmentStatusService) *EnrichmentStatusHandler { + return &EnrichmentStatusHandler{service: service} +} + +// GetStatus handles GET /v1/enrichment-status?tenant_id=. It returns per-enrichment +// eligible/done counts for the tenant. tenant_id is required; a missing/blank value yields 400. +func (h *EnrichmentStatusHandler) GetStatus(w http.ResponseWriter, r *http.Request) { + if h.service == nil { + response.RespondServiceUnavailable(w, r, "Enrichment status is not available.") + + return + } + + tenantID := r.URL.Query().Get("tenant_id") + + result, err := h.service.GetEnrichmentStatus(r.Context(), tenantID) + if err != nil { + // RespondError maps a missing/blank tenant_id (huberrors.ValidationError) to 400 and + // any repository/DB failure to a generic 500 — no internals leak to the client. + response.RespondError(w, r, err) + + return + } + + response.RespondJSON(w, http.StatusOK, result) +} diff --git a/internal/api/handlers/enrichment_status_handler_test.go b/internal/api/handlers/enrichment_status_handler_test.go new file mode 100644 index 00000000..f8cc0885 --- /dev/null +++ b/internal/api/handlers/enrichment_status_handler_test.go @@ -0,0 +1,66 @@ +package handlers + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/formbricks/hub/internal/huberrors" + "github.com/formbricks/hub/internal/models" +) + +type fakeEnrichmentStatusService struct { + resp *models.EnrichmentStatusResponse + err error +} + +func (f *fakeEnrichmentStatusService) GetEnrichmentStatus( + _ context.Context, _ string, +) (*models.EnrichmentStatusResponse, error) { + return f.resp, f.err +} + +func TestEnrichmentStatusHandler_GetStatus(t *testing.T) { + t.Run("service unavailable when not configured", func(t *testing.T) { + h := NewEnrichmentStatusHandler(nil) + rec := httptest.NewRecorder() + h.GetStatus(rec, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/v1/enrichment-status?tenant_id=t1", nil)) + + assert.Equal(t, http.StatusServiceUnavailable, rec.Code) + }) + + t.Run("missing tenant_id maps to 400", func(t *testing.T) { + // The service returns a huberrors.ValidationError for a blank tenant_id; the handler must + // surface it as a 400 via response.RespondError, not a 500. + h := NewEnrichmentStatusHandler(&fakeEnrichmentStatusService{ + err: huberrors.NewValidationError("tenant_id", "tenant_id is required and cannot be empty"), + }) + rec := httptest.NewRecorder() + h.GetStatus(rec, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/v1/enrichment-status", nil)) + + assert.Equal(t, http.StatusBadRequest, rec.Code) + }) + + t.Run("returns 200 with the status body", func(t *testing.T) { + want := &models.EnrichmentStatusResponse{ + TenantID: "t1", + Translation: models.EnrichmentTypeStatus{Enabled: true, Eligible: 10, Done: 4}, + Sentiment: models.EnrichmentTypeStatus{Enabled: true, Eligible: 8, Done: 8}, + Emotions: models.EnrichmentTypeStatus{Enabled: false}, + } + h := NewEnrichmentStatusHandler(&fakeEnrichmentStatusService{resp: want}) + rec := httptest.NewRecorder() + h.GetStatus(rec, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/v1/enrichment-status?tenant_id=t1", nil)) + + require.Equal(t, http.StatusOK, rec.Code) + + var got models.EnrichmentStatusResponse + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &got)) + assert.Equal(t, *want, got) + }) +} diff --git a/internal/models/enrichment_status.go b/internal/models/enrichment_status.go new file mode 100644 index 00000000..6ca3bc53 --- /dev/null +++ b/internal/models/enrichment_status.go @@ -0,0 +1,23 @@ +package models + +// EnrichmentTypeStatus is one tenant's progress for a single record-level enrichment +// (translation, sentiment, or emotions). Enabled reports whether the enrichment is both +// deployment-configured and switched on for the tenant (for translation: has a resolvable +// target language). Eligible is the number of feedback records that qualify for the +// enrichment; Done is how many have been enriched. The UI derives "in progress" as +// Eligible - Done. When Enabled is false, Eligible and Done are zero (no work will run). +type EnrichmentTypeStatus struct { + Enabled bool `json:"enabled"` + Eligible int64 `json:"eligible"` + Done int64 `json:"done"` +} + +// EnrichmentStatusResponse reports a tenant's enrichment progress across the record-level +// enrichments. Counts are directory-level totals; the response never includes record +// identifiers or feedback content. +type EnrichmentStatusResponse struct { + TenantID string `json:"tenant_id"` + Translation EnrichmentTypeStatus `json:"translation"` + Sentiment EnrichmentTypeStatus `json:"sentiment"` + Emotions EnrichmentTypeStatus `json:"emotions"` +} diff --git a/internal/observability/aggregate.go b/internal/observability/aggregate.go index 1ffc4459..0194bd0b 100644 --- a/internal/observability/aggregate.go +++ b/internal/observability/aggregate.go @@ -20,6 +20,8 @@ type Metrics struct { Cache CacheMetrics // EnrichmentClear counts enrichment outputs nulled by an edit's eager-clear. EnrichmentClear EnrichmentClearMetrics + // EnrichmentBacklog gauges the aggregate eligible-but-unenriched record count per enrichment. + EnrichmentBacklog EnrichmentBacklogMetrics } // NewMetrics creates EventMetrics, WebhookMetrics, EmbeddingMetrics, TranslationMetrics, and CacheMetrics from the given meter. @@ -70,14 +72,20 @@ func NewMetrics(meter metric.Meter) (*Metrics, error) { return nil, fmt.Errorf("enrichment clear metrics: %w", err) } + enrichmentBacklog, err := NewEnrichmentBacklogMetrics(meter) + if err != nil { + return nil, fmt.Errorf("enrichment backlog metrics: %w", err) + } + return &Metrics{ - Events: events, - Webhooks: webhooks, - Embeddings: embeddings, - Translation: translation, - Sentiment: sentiment, - Emotions: emotions, - Cache: cache, - EnrichmentClear: enrichmentClear, + Events: events, + Webhooks: webhooks, + Embeddings: embeddings, + Translation: translation, + Sentiment: sentiment, + Emotions: emotions, + Cache: cache, + EnrichmentClear: enrichmentClear, + EnrichmentBacklog: enrichmentBacklog, }, nil } diff --git a/internal/observability/enrichment_backlog.go b/internal/observability/enrichment_backlog.go new file mode 100644 index 00000000..3509b1dd --- /dev/null +++ b/internal/observability/enrichment_backlog.go @@ -0,0 +1,123 @@ +package observability + +import ( + "context" + "fmt" + "sync" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// Enrichment type label values for MetricNameEnrichmentPendingRecords — a fixed, bounded set that +// keeps the gauge's cardinality low. Callers pass these to SetEnrichmentPending. +const ( + EnrichmentTypeTranslation = "translation" + EnrichmentTypeSentiment = "sentiment" + EnrichmentTypeEmotions = "emotions" +) + +// EnrichmentBacklogMetrics reports the aggregate (cross-tenant) count of eligible-but-unenriched +// feedback records per enrichment type — a data-derived "how far behind is enrichment" gauge that +// complements the transient River queue-depth gauge. A background poller refreshes the values; an +// async gauge observes them. Labeled by enrichment type only (a fixed, bounded set); tenant_id is +// deliberately NOT a label, so per-tenant detail stays in the API and metric cardinality is bounded. +type EnrichmentBacklogMetrics interface { + SetEnrichmentPending(enrichment string, count int64) + // ClearEnrichmentPending withdraws every series this process exports for the gauge. The async + // callback re-observes the stored values on EVERY collection, so a process that stops being the + // leader would otherwise keep exporting its final reading forever: the new leader's live series + // and the old leader's frozen one would coexist, a sum would double-count, and the frozen copy + // would look like a permanently stuck backlog. Callers must clear as soon as they are no longer + // the leader. Exporting nothing is the honest state — absence is visible, a stale value is not. + ClearEnrichmentPending() + // RecordPollError counts a failed refresh. The gauge holds its last value when a poll fails, + // which is indistinguishable from a healthy steady backlog on a dashboard — this counter is + // the signal to alert on (rate > 0 means the gauge is going stale). + RecordPollError(ctx context.Context) +} + +// enrichmentBacklogMetrics implements EnrichmentBacklogMetrics. The latest per-enrichment value is +// stored under mu and read by the gauge callback. +type enrichmentBacklogMetrics struct { + mu sync.Mutex + pending map[string]int64 + gauge metric.Int64ObservableGauge + pollErrors metric.Int64Counter +} + +// NewEnrichmentBacklogMetrics registers the pending-records gauge. Returns (nil, nil) when meter is +// nil (metrics disabled); the caller translates that into a nil interface. +func NewEnrichmentBacklogMetrics(meter metric.Meter) (EnrichmentBacklogMetrics, error) { + if meter == nil { + //nolint:nilnil // intentional: callers use "if metrics != nil" when metrics disabled + return nil, nil + } + + m := &enrichmentBacklogMetrics{pending: make(map[string]int64)} + + gauge, err := meter.Int64ObservableGauge( + MetricNameEnrichmentPendingRecords, + metric.WithDescription( + "Eligible-but-unenriched feedback records per enrichment type (translation, sentiment, "+ + "emotions), aggregated across all tenants. A data-derived backlog/completeness signal; "+ + "unlike the River queue depth it persists across queue drains.", + ), + metric.WithUnit("1"), + metric.WithInt64Callback(func(_ context.Context, observer metric.Int64Observer) error { + m.mu.Lock() + defer m.mu.Unlock() + + for enrichment, count := range m.pending { + observer.Observe(count, metric.WithAttributes(attribute.String(AttrEnrichment, enrichment))) + } + + return nil + }), + ) + if err != nil { + return nil, fmt.Errorf("create %s: %w", MetricNameEnrichmentPendingRecords, err) + } + + m.gauge = gauge + + pollErrors, err := meter.Int64Counter( + MetricNameEnrichmentBacklogPollErrs, + metric.WithDescription( + "Failed refreshes of the enrichment backlog gauge. A non-zero rate means the gauge is "+ + "stale (holding its last value), which on a dashboard is indistinguishable from a "+ + "steady backlog — alert on this rather than trusting a flat gauge.", + ), + metric.WithUnit("1"), + ) + if err != nil { + return nil, fmt.Errorf("create %s: %w", MetricNameEnrichmentBacklogPollErrs, err) + } + + m.pollErrors = pollErrors + + return m, nil +} + +// ClearEnrichmentPending drops every stored value so the next collection observes nothing and this +// process stops exporting the gauge entirely. +func (m *enrichmentBacklogMetrics) ClearEnrichmentPending() { + m.mu.Lock() + defer m.mu.Unlock() + + clear(m.pending) +} + +// RecordPollError counts one failed backlog refresh. +func (m *enrichmentBacklogMetrics) RecordPollError(ctx context.Context) { + m.pollErrors.Add(ctx, 1) +} + +// SetEnrichmentPending stores the latest backlog count for an enrichment type; the registered gauge +// callback reports it on the next collection. +func (m *enrichmentBacklogMetrics) SetEnrichmentPending(enrichment string, count int64) { + m.mu.Lock() + defer m.mu.Unlock() + + m.pending[enrichment] = count +} diff --git a/internal/observability/enrichment_backlog_test.go b/internal/observability/enrichment_backlog_test.go new file mode 100644 index 00000000..368b815a --- /dev/null +++ b/internal/observability/enrichment_backlog_test.go @@ -0,0 +1,136 @@ +package observability + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +func TestNewEnrichmentBacklogMetricsNilMeterDisabled(t *testing.T) { + metrics, err := NewEnrichmentBacklogMetrics(nil) + require.NoError(t, err) + assert.Nil(t, metrics, "a nil meter disables metrics") +} + +// TestEnrichmentBacklogMetricsGauge verifies the async gauge reports the latest per-enrichment +// value under the enrichment label, and that a later Set overwrites (a gauge, not a counter). +func TestEnrichmentBacklogMetricsGauge(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + + metrics, err := NewEnrichmentBacklogMetrics(provider.Meter("test")) + require.NoError(t, err) + require.NotNil(t, metrics) + + metrics.SetEnrichmentPending("translation", 12) + metrics.SetEnrichmentPending("sentiment", 5) + metrics.SetEnrichmentPending("emotions", 0) + + assert.Equal(t, int64(12), backlogGaugeValue(t, reader, "translation")) + assert.Equal(t, int64(5), backlogGaugeValue(t, reader, "sentiment")) + assert.Equal(t, int64(0), backlogGaugeValue(t, reader, "emotions")) + + // A gauge reports the latest value, not a running sum. + metrics.SetEnrichmentPending("translation", 3) + assert.Equal(t, int64(3), backlogGaugeValue(t, reader, "translation")) +} + +// TestEnrichmentBacklogMetricsClearWithdrawsSeries verifies a demoted leader stops exporting +// entirely rather than freezing at its last reading. The async gauge re-observes stored values on +// every collection, so without this a former leader's stale series would coexist with the new +// leader's live one — a sum would double-count and the frozen copy would look like a stuck backlog. +func TestEnrichmentBacklogMetricsClearWithdrawsSeries(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + + metrics, err := NewEnrichmentBacklogMetrics(provider.Meter("test")) + require.NoError(t, err) + require.NotNil(t, metrics) + + metrics.SetEnrichmentPending(EnrichmentTypeTranslation, 42) + assert.Equal(t, int64(42), backlogGaugeValue(t, reader, EnrichmentTypeTranslation)) + + metrics.ClearEnrichmentPending() + + var collected metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &collected)) + + for _, scope := range collected.ScopeMetrics { + for _, m := range scope.Metrics { + assert.NotEqual(t, MetricNameEnrichmentPendingRecords, m.Name, + "a cleared gauge must export no data points at all, not a stale value") + } + } +} + +// TestEnrichmentBacklogMetricsPollErrors verifies failed refreshes are counted, so a gauge frozen +// at its last value (which looks like a healthy steady backlog) is still alertable. +func TestEnrichmentBacklogMetricsPollErrors(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + + metrics, err := NewEnrichmentBacklogMetrics(provider.Meter("test")) + require.NoError(t, err) + require.NotNil(t, metrics) + + ctx := context.Background() + metrics.RecordPollError(ctx) + metrics.RecordPollError(ctx) + + var collected metricdata.ResourceMetrics + require.NoError(t, reader.Collect(ctx, &collected)) + + var total int64 + + for _, scope := range collected.ScopeMetrics { + for _, m := range scope.Metrics { + if m.Name != MetricNameEnrichmentBacklogPollErrs { + continue + } + + sum, ok := m.Data.(metricdata.Sum[int64]) + require.True(t, ok, "expected Sum[int64] for %s", MetricNameEnrichmentBacklogPollErrs) + + for _, point := range sum.DataPoints { + total += point.Value + } + } + } + + assert.Equal(t, int64(2), total, "each failed poll increments the error counter") +} + +// backlogGaugeValue collects metrics and returns the pending-records gauge value for one enrichment. +func backlogGaugeValue(t *testing.T, reader sdkmetric.Reader, enrichment string) int64 { + t.Helper() + + var collected metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &collected)) + + for _, scope := range collected.ScopeMetrics { + for _, m := range scope.Metrics { + if m.Name != MetricNameEnrichmentPendingRecords { + continue + } + + gauge, ok := m.Data.(metricdata.Gauge[int64]) + require.True(t, ok, "expected Gauge[int64] for %s", MetricNameEnrichmentPendingRecords) + + for _, point := range gauge.DataPoints { + if value, present := point.Attributes.Value(attribute.Key(AttrEnrichment)); present && + value.AsString() == enrichment { + return point.Value + } + } + } + } + + t.Fatalf("gauge %q has no data point for enrichment %q", MetricNameEnrichmentPendingRecords, enrichment) + + return 0 +} diff --git a/internal/observability/names.go b/internal/observability/names.go index e21b6454..f78c6c96 100644 --- a/internal/observability/names.go +++ b/internal/observability/names.go @@ -15,6 +15,8 @@ const ( MetricNameProviderPanics = "hub_provider_panics_total" MetricNameHNSWIterativeScanDegraded = "hub_hnsw_iterative_scan_degraded" MetricNameEnrichmentOutputsCleared = "hub_enrichment_outputs_cleared_total" + MetricNameEnrichmentPendingRecords = "hub_enrichment_pending_records" + MetricNameEnrichmentBacklogPollErrs = "hub_enrichment_backlog_poll_errors_total" MetricNameWebhookJobsEnqueued = "hub_webhook_jobs_enqueued_total" MetricNameWebhookProviderErrors = "hub_webhook_provider_errors_total" MetricNameWebhookDeliveries = "hub_webhook_deliveries_total" @@ -65,6 +67,10 @@ const ( // AttrQueue labels the River queue-depth gauge; values come from the poller's fixed queue // set, so cardinality is bounded. AttrQueue = "queue" + // AttrEnrichment labels the enrichment-backlog gauge; values are the fixed enrichment types + // (translation, sentiment, emotions). tenant_id is deliberately NOT a label — the gauge is + // aggregated across all tenants to keep cardinality bounded. + AttrEnrichment = "enrichment" ) // AllowedEventTypes returns event type strings allowed for metric attributes (bounded cardinality). diff --git a/internal/repository/enrichment_status_repository.go b/internal/repository/enrichment_status_repository.go new file mode 100644 index 00000000..9c744328 --- /dev/null +++ b/internal/repository/enrichment_status_repository.go @@ -0,0 +1,299 @@ +package repository + +import ( + "context" + "fmt" + "log/slog" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// EnrichmentStatusRepository computes data-derived enrichment progress counts over +// feedback_records. It is a read-only sibling of FeedbackRecordsRepository, kept separate +// so the status/observability concern doesn't grow the primary records repository. +type EnrichmentStatusRepository struct { + db *pgxpool.Pool +} + +// NewEnrichmentStatusRepository creates an enrichment status repository. +func NewEnrichmentStatusRepository(db *pgxpool.Pool) *EnrichmentStatusRepository { + return &EnrichmentStatusRepository{db: db} +} + +// EnrichmentStatusCounts holds the raw per-enrichment eligible/done counts. The counts already +// reflect the per-tenant gates: sentiment/emotions include only tenants with the enrichment +// switched on, translation only records with a resolvable effective target language. The +// deployment-level (provider/model) gate is applied by the caller. +type EnrichmentStatusCounts struct { + TranslationEligible int64 + TranslationDone int64 + SentimentEligible int64 + SentimentDone int64 + EmotionsEligible int64 + EmotionsDone int64 +} + +// enrichmentEligibleText is the data-level eligibility predicate: an open-text field with content. +// +// It trims the full ASCII whitespace set (space, tab, VT, FF, CR, LF). This deliberately does NOT +// match the backfill queries (classifyBackfillEligibleSQL / translationBackfillSelectSQL), whose +// bare btrim() strips spaces only: a value of "\t\n" is enqueued by those but counted ineligible +// here. That asymmetry is intentional -- what matters for a progress count is agreeing with the +// WORKER, which gates on Go strings.TrimSpace (HasOpenText) and would clear such a record rather +// than enrich it. Counting it eligible would leave it pending forever. Do not "restore parity" by +// weakening this to bare btrim. +// +// It remains an approximation in one direction: strings.TrimSpace also strips exotic Unicode +// whitespace (NBSP U+00A0, ideographic space U+3000, ...), so a value composed ENTIRELY of those is +// still counted eligible while the worker treats it as empty. Rare enough to accept; expressing the +// full Unicode set here would mean embedding invisible characters in this source file. +// +// field_type = 'text' is load-bearing: matrix/multi-choice expansion writes value_text on +// categorical/number rows that are not enrichable. +const enrichmentEligibleText = `fr.field_type = 'text' AND fr.value_text IS NOT NULL AND btrim(fr.value_text, E' \t\n\v\f\r') <> ''` + +// enrichmentEffectiveTarget resolves a tenant's effective translation target: its own +// target_language, falling back to the deployment default ($1). An empty result means translation +// is not enabled for the tenant. Mirrors translationBackfillSelectSQL. +// enrichmentSentimentOn / enrichmentEmotionsOn read the tri-state per-directory switch, defaulting +// to enabled when the key is absent — matching EnrichmentSettings.SentimentEnrichmentEnabled / +// EmotionsEnrichmentEnabled (parity is covered by test). $1 is the deployment default target in +// both the per-tenant and aggregate queries. +const ( + enrichmentEffectiveTarget = `COALESCE(NULLIF(ts.settings->>'target_language', ''), $1)` + enrichmentSentimentOn = `COALESCE((ts.settings->>'sentiment_enabled')::boolean, true)` + enrichmentEmotionsOn = `COALESCE((ts.settings->>'emotions_enabled')::boolean, true)` +) + +// enrichmentEmotionsDone reports that emotion classification has completed for a record. It cannot +// be `emotions IS NOT NULL` alone: a successful classification that detects no emotion is stored as +// NULL (the 015 CHECK rejects the empty array), so such records would count as pending forever and +// the backlog would never drain. emotions_classified_at (migration 020) records completion +// independently of the labels found. The `emotions IS NOT NULL` arm covers rows classified BEFORE +// that column existed -- a non-NULL label set is itself proof of completion -- which is why the +// migration needs no bulk backfill; only historical classified-empty rows remain unresolved until +// the classify backfill re-processes them. +const enrichmentEmotionsDone = `(fr.emotions_classified_at IS NOT NULL OR fr.emotions IS NOT NULL)` + +// enrichmentCountSelect is the shared six-column SELECT list — {sentiment, emotions, translation} × +// {eligible, done} — used by both the per-tenant and aggregate queries so the predicates can't +// drift between them. Column order must match the EnrichmentStatusCounts scan order below. All +// fragments are static constants (never user input). +const enrichmentCountSelect = ` + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + `), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + ` AND fr.sentiment IS NOT NULL), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + `), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + ` AND ` + enrichmentEmotionsDone + `), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEffectiveTarget + ` <> ''), + COUNT(*) FILTER ( + WHERE ` + enrichmentEligibleText + ` + AND ` + enrichmentEffectiveTarget + ` <> '' + AND fr.translation_lang_key = ` + enrichmentEffectiveTarget + `)` + +const enrichmentCountFrom = ` + FROM feedback_records fr + LEFT JOIN tenant_settings ts ON ts.tenant_id = fr.tenant_id` + +// The `fr.field_type = 'text'` predicate in the outer WHERE of both queries below is redundant +// with enrichmentEligibleText inside every FILTER (so it can never change a count); it is hoisted +// out so the planner can use it as an access-path predicate rather than only as a per-row filter. + +// countEnrichmentStatusSQL counts eligible/done per enrichment for ONE tenant. $1 = deployment +// default target language, $2 = tenant_id. The (tenant_id, field_type) pair matches +// idx_feedback_records_tenant_field_type, so cost scales with the tenant's text-record count. +const countEnrichmentStatusSQL = `SELECT ` + enrichmentCountSelect + enrichmentCountFrom + ` + WHERE fr.tenant_id = $2 AND fr.field_type = 'text'` + +// countEnrichmentBacklogAggregateSQL is the same SELECT without the tenant filter: it sums +// eligible/done per enrichment across ALL tenants (for the observability gauge). $1 = deployment +// default target language. The per-tenant enable gates still apply, so a tenant that switched an +// enrichment off, or has no resolvable target, never inflates the backlog. +// +// NOTE: unlike the per-tenant query this one canNOT use idx_feedback_records_tenant_field_type -- +// tenant_id is that index's leading column and this query has no tenant predicate -- so Postgres +// plans a sequential scan of feedback_records (confirmed via EXPLAIN). Accepted rather than fixed +// with a new index: the aggregate has to read every text row regardless, so an index scan covering +// most of the table would not be cheaper. (Migration 016 does already maintain partial indexes over +// unenriched text rows, so a further index is not unthinkable -- it is simply unlikely to pay off +// for a whole-table aggregate.) The cost is instead bounded by running the scan infrequently +// (enrichmentBacklogInterval), under a statement timeout, and on exactly ONE replica via the +// leader election below. +const countEnrichmentBacklogAggregateSQL = `SELECT ` + enrichmentCountSelect + enrichmentCountFrom + ` + WHERE fr.field_type = 'text'` + +// enrichmentBacklogLockKey names the advisory lock that elects the single backlog-poller process +// across API replicas. Hashed with hashtextextended like the other advisory locks in this package. +const enrichmentBacklogLockKey = "hub:enrichment-backlog-poller" + +// enrichmentBacklogUnlockTimeout bounds the best-effort advisory unlock on shutdown. +const enrichmentBacklogUnlockTimeout = 5 * time.Second + +const ( + // trySessionLockSQL takes a SESSION-scoped advisory lock without blocking. Session scope is + // deliberate -- see EnrichmentBacklogLeader for why a transaction-scoped lock cannot work here. + trySessionLockSQL = `SELECT pg_try_advisory_lock(hashtextextended($1, 0))` + // sessionUnlockSQL releases it. A session lock outlives returning the connection to the pool, + // so it must be released explicitly. + sessionUnlockSQL = `SELECT pg_advisory_unlock(hashtextextended($1, 0))` + // setLeaderIdleTimeoutSQL caps how long a stalled leader session can hold the lock. Applies to + // this session only (SET, not ALTER ROLE); comfortably above enrichmentBacklogInterval so a + // healthy leader, which queries every interval, is never affected. Requires PG14+. + setLeaderIdleTimeoutSQL = `SET idle_session_timeout = '30min'` +) + +// CountEnrichmentStatus returns one tenant's eligible/done counts per enrichment. defaultLang is +// the deployment translation fallback ("" disables the fallback, so only tenants with their own +// target language have eligible translation records). Always scoped to the given tenant_id. +func (r *EnrichmentStatusRepository) CountEnrichmentStatus( + ctx context.Context, tenantID, defaultLang string, +) (EnrichmentStatusCounts, error) { + return scanEnrichmentCounts( + r.db.QueryRow(ctx, countEnrichmentStatusSQL, defaultLang, tenantID), "count enrichment status") +} + +// CountEnrichmentBacklogAggregate returns eligible/done counts per enrichment summed across all +// tenants. defaultLang is the deployment translation fallback. The result carries no tenant +// dimension. Prefer CountEnrichmentBacklogAggregateIfLeader from the poller so only one replica +// runs the scan. +func (r *EnrichmentStatusRepository) CountEnrichmentBacklogAggregate( + ctx context.Context, defaultLang string, +) (EnrichmentStatusCounts, error) { + return scanEnrichmentCounts( + r.db.QueryRow(ctx, countEnrichmentBacklogAggregateSQL, defaultLang), "count enrichment backlog aggregate") +} + +// EnrichmentBacklogLeader elects ONE process to refresh the cross-tenant backlog gauge. +// +// Production runs several API replicas per region. Without election each replica repeats the same +// full-table aggregate every tick and exports its own copy of a value that is global by definition: +// N times the DB work, and N identical series that a dashboard summing them silently over-counts. +// +// Leadership is a SESSION-scoped advisory lock held on a dedicated pooled connection for the +// process lifetime, NOT a lock taken around each scan. That distinction is the whole point: +// replicas tick on independent, unsynchronized schedules (each ticker starts at its own boot), and +// a scan-scoped lock is held for only a couple of seconds out of every interval, so replicas would +// virtually never collide -- suppressing nothing, while occasionally blanking one replica's series +// when they did collide. Sticky leadership instead means exactly one replica scans and exports, and +// the series stays put instead of flapping between replicas. +// +// A non-leader never holds a connection: it acquires one, loses the race, and hands it straight +// back. If the leader's connection dies its backend session ends and Postgres drops the lock +// automatically, so the next tick re-elects; the process also drops leadership itself whenever a +// scan fails, so it cannot keep believing it is the leader after losing the session. +type EnrichmentBacklogLeader struct { + pool *pgxpool.Pool + conn *pgxpool.Conn // non-nil only while this process holds leadership +} + +// NewEnrichmentBacklogLeader creates a leader-elected reader for the aggregate backlog counts. +func NewEnrichmentBacklogLeader(pool *pgxpool.Pool) *EnrichmentBacklogLeader { + return &EnrichmentBacklogLeader{pool: pool} +} + +// CountIfLeader returns the cross-tenant counts when this process holds (or wins) leadership, and +// reports whether it did. Not being the leader is the normal steady state for all but one replica, +// so it is signalled by a false second return rather than an error. +func (l *EnrichmentBacklogLeader) CountIfLeader( + ctx context.Context, defaultLang string, +) (EnrichmentStatusCounts, bool, error) { + if l.conn == nil { + acquired, err := l.tryAcquire(ctx) + if err != nil || !acquired { + return EnrichmentStatusCounts{}, false, err + } + } + + // Deliberately run on the leader connection: a broken session then surfaces here as a scan + // error rather than silently leaving this process convinced it still holds the lock. + counts, err := scanEnrichmentCounts( + l.conn.QueryRow(ctx, countEnrichmentBacklogAggregateSQL, defaultLang), "count enrichment backlog aggregate") + if err != nil { + l.release(ctx) + + return EnrichmentStatusCounts{}, false, err + } + + return counts, true, nil +} + +// Close relinquishes leadership so another replica can take over promptly instead of waiting for +// this process's session to time out. Safe to call when not the leader. +func (l *EnrichmentBacklogLeader) Close(ctx context.Context) { + l.release(ctx) +} + +// tryAcquire takes a connection and attempts the session lock, keeping the connection only on +// success -- a non-leader must not pin a pooled connection it will not use. +func (l *EnrichmentBacklogLeader) tryAcquire(ctx context.Context) (bool, error) { + conn, err := l.pool.Acquire(ctx) + if err != nil { + return false, fmt.Errorf("acquire enrichment backlog leader connection: %w", err) + } + + var acquired bool + if err := conn.QueryRow(ctx, trySessionLockSQL, enrichmentBacklogLockKey).Scan(&acquired); err != nil { + conn.Release() + + return false, fmt.Errorf("try enrichment backlog advisory lock: %w", err) + } + + if !acquired { + conn.Release() + + return false, nil + } + + // Bound how long a LOST leader can keep the lock. A session lock lives until its backend exits; + // a graceful shutdown releases it via Close, and a killed pod's socket closes, but a node + // failure or network partition leaves a zombie backend holding it until TCP keepalives reap the + // connection -- hours under Linux defaults, during which no replica can take over and the gauge + // just goes absent. An idle timeout on this session alone caps that at one timeout period. The + // leader queries every enrichmentBacklogInterval, so the timeout is set well above it and can + // only fire on a session that has genuinely stopped polling. Best effort: on an older server or + // a restricted role this simply does not apply and behaviour is as before. + if _, err := conn.Exec(ctx, setLeaderIdleTimeoutSQL); err != nil { + slog.WarnContext(ctx, "enrichment backlog: could not bound leader session idle timeout", + "error", err) + } + + l.conn = conn + + return true, nil +} + +// release unlocks and returns the leader connection. The unlock uses a context detached from the +// caller's, so shutdown (whose context is already cancelled) still releases the lock rather than +// leaving it held until the backend session is reaped. +func (l *EnrichmentBacklogLeader) release(ctx context.Context) { + if l.conn == nil { + return + } + + unlockCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), enrichmentBacklogUnlockTimeout) + defer cancel() + + // Best effort: if this fails the connection is broken, and ending that session releases the + // lock anyway. Returning the connection WITHOUT unlocking would be the real leak, since a + // session lock survives being handed back to the pool. + _, _ = l.conn.Exec(unlockCtx, sessionUnlockSQL, enrichmentBacklogLockKey) + + l.conn.Release() + l.conn = nil +} + +// scanEnrichmentCounts reads the six-column count row; the scan order matches enrichmentCountSelect. +func scanEnrichmentCounts(row pgx.Row, what string) (EnrichmentStatusCounts, error) { + var counts EnrichmentStatusCounts + + if err := row.Scan( + &counts.SentimentEligible, &counts.SentimentDone, + &counts.EmotionsEligible, &counts.EmotionsDone, + &counts.TranslationEligible, &counts.TranslationDone, + ); err != nil { + return EnrichmentStatusCounts{}, fmt.Errorf("%s: %w", what, err) + } + + return counts, nil +} diff --git a/internal/repository/feedback_records_repository.go b/internal/repository/feedback_records_repository.go index 0585c131..15b9d6af 100644 --- a/internal/repository/feedback_records_repository.go +++ b/internal/repository/feedback_records_repository.go @@ -426,36 +426,21 @@ func (r *FeedbackRecordsRepository) SetEmotions( emotionsArg = labels } - return withTenantWritePoolTx(ctx, r.db, nil, func(dbTx tenantWriteTx) error { - // Emotions ride the feedback record's tenant boundary; resolve and lock it so the write - // cannot race a tenant data purge. - if _, err := lockFeedbackRecordTenantShared(ctx, dbTx, feedbackRecordID); err != nil { - return err - } - - if err := guardValueTextCurrent(ctx, dbTx, feedbackRecordID, stillCurrent, - huberrors.ErrClassificationSuperseded); err != nil { - return err - } - - tag, err := dbTx.Exec(ctx, ` - UPDATE feedback_records - SET emotions = $2, updated_at = NOW() - WHERE id = $1`, - feedbackRecordID, emotionsArg, - ) - if err != nil { - return fmt.Errorf("set feedback record emotions: %w", err) - } - - // Locked above, so zero rows means the record was deleted between the lock and this write: - // surface NotFound so the worker treats it as a benign skip. - if tag.RowsAffected() == 0 { - return huberrors.NewNotFoundError("feedback record", "feedback record not found") - } + // A classifier result -- even an empty one -- means the record HAS been classified, so stamp + // the completion marker. Without it an empty result is stored as a bare NULL and is + // indistinguishable from "never classified", which pins the record in every pending count + // forever (ENG-1670). Use ClearEmotions for the not-classified transition. + return r.writeEmotions(ctx, feedbackRecordID, emotionsArg, true, stillCurrent) +} - return nil - }) +// ClearEmotions removes a record's emotion enrichment AND its completion marker, returning it to +// the "not classified" state. Used when the source content is gone (an empty-content job clears +// rather than classifies), so the record is correctly excluded from progress counts instead of +// masquerading as classified-with-no-emotions. +func (r *FeedbackRecordsRepository) ClearEmotions( + ctx context.Context, feedbackRecordID uuid.UUID, stillCurrent func(valueText *string) bool, +) error { + return r.writeEmotions(ctx, feedbackRecordID, nil, false, stillCurrent) } // translationBackfillSelectSQL selects feedback records that need (re)translation: text @@ -557,8 +542,14 @@ const sentimentBackfillSelectSQL = classifyBackfillEligibleSQL + ` ORDER BY id LIMIT $2` +// Both NULL checks are required. emotions IS NULL alone would re-send every record whose +// classification legitimately found no emotion (stored as NULL) to the provider on every run; +// emotions_classified_at IS NULL alone would re-classify rows enriched before that column existed +// (migration 020). Together they select exactly the records never classified. The 016 partial index +// covers the emotions IS NULL arm; the marker is an additional filter on top. const emotionsBackfillSelectSQL = classifyBackfillEligibleSQL + ` AND emotions IS NULL + AND emotions_classified_at IS NULL AND id > $1 ORDER BY id LIMIT $2` @@ -907,6 +898,9 @@ func buildUpdateQuery( clearColumnWhen("sentiment", valueTextChanged), clearColumnWhen("sentiment_score", valueTextChanged), clearColumnWhen("emotions", valueTextChanged), + // The completion marker must be cleared with the value it describes, otherwise an + // edited record keeps counting as classified while its emotions are gone (ENG-1670). + clearColumnWhen("emotions_classified_at", valueTextChanged), ) } @@ -1215,3 +1209,46 @@ func (r *FeedbackRecordsRepository) fetchFeedbackRecords( return records, nil } + +// writeEmotions is the shared write path for SetEmotions/ClearEmotions. classified controls the +// completion marker: NOW() for a classifier result, NULL when clearing. +func (r *FeedbackRecordsRepository) writeEmotions( + ctx context.Context, feedbackRecordID uuid.UUID, emotionsArg any, classified bool, + stillCurrent func(valueText *string) bool, +) error { + return withTenantWritePoolTx(ctx, r.db, nil, func(dbTx tenantWriteTx) error { + // Emotions ride the feedback record's tenant boundary; resolve and lock it so the write + // cannot race a tenant data purge. + if _, err := lockFeedbackRecordTenantShared(ctx, dbTx, feedbackRecordID); err != nil { + return err + } + + if err := guardValueTextCurrent(ctx, dbTx, feedbackRecordID, stillCurrent, + huberrors.ErrClassificationSuperseded); err != nil { + return err + } + + // The marker is stamped from the DB clock, like updated_at beside it, rather than the pod's + // clock -- every other timestamp on this table is server-generated, and the two should not + // be able to disagree. CASE keeps it NULL when clearing. + tag, err := dbTx.Exec(ctx, ` + UPDATE feedback_records + SET emotions = $2, + emotions_classified_at = CASE WHEN $3::boolean THEN NOW() END, + updated_at = NOW() + WHERE id = $1`, + feedbackRecordID, emotionsArg, classified, + ) + if err != nil { + return fmt.Errorf("set feedback record emotions: %w", err) + } + + // Locked above, so zero rows means the record was deleted between the lock and this write: + // surface NotFound so the worker treats it as a benign skip. + if tag.RowsAffected() == 0 { + return huberrors.NewNotFoundError("feedback record", "feedback record not found") + } + + return nil + }) +} diff --git a/internal/repository/feedback_records_repository_test.go b/internal/repository/feedback_records_repository_test.go index f132955c..9e50cc80 100644 --- a/internal/repository/feedback_records_repository_test.go +++ b/internal/repository/feedback_records_repository_test.go @@ -34,7 +34,10 @@ func TestBuildUpdateQuery_ClearsStaleEnrichmentOnContentChange(t *testing.T) { // Enrichment output columns, grouped by what invalidates them. translationCols := []string{"value_text_translated", "translation_lang_key"} - textOnlyCols := []string{"sentiment", "sentiment_score", "emotions"} + // emotions_classified_at must clear with emotions: it is the completion marker, so leaving it + // behind would keep an edited record counting as classified. It must equally NOT clear on a + // language-only edit, which the negative cases below cover. + textOnlyCols := []string{"sentiment", "sentiment_score", "emotions", "emotions_classified_at"} allCols := append(append([]string{}, translationCols...), textOnlyCols...) cases := []struct { diff --git a/internal/service/enrichment_status_service.go b/internal/service/enrichment_status_service.go new file mode 100644 index 00000000..2d9af72c --- /dev/null +++ b/internal/service/enrichment_status_service.go @@ -0,0 +1,104 @@ +package service + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/formbricks/hub/internal/models" + "github.com/formbricks/hub/internal/repository" +) + +// enrichmentStatusQueryTimeout bounds the settings lookup + count query for one request. The +// endpoint is polled and there is no HTTP handler timeout, so a slow count must not pin a DB +// pool connection indefinitely. +const enrichmentStatusQueryTimeout = 5 * time.Second + +// EnrichmentStatusRepository is the count surface the status service needs. +type EnrichmentStatusRepository interface { + CountEnrichmentStatus(ctx context.Context, tenantID, defaultLang string) (repository.EnrichmentStatusCounts, error) +} + +// 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. +type EnrichmentStatusService 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 NewEnrichmentStatusServiceParams struct { + Repo EnrichmentStatusRepository + Settings TenantSettingsReader + DefaultLang string + TranslationConfigured bool + SentimentConfigured bool + EmotionsConfigured bool +} + +// NewEnrichmentStatusService creates an enrichment status service. +func NewEnrichmentStatusService(params NewEnrichmentStatusServiceParams) *EnrichmentStatusService { + return &EnrichmentStatusService{ + repo: params.Repo, + settings: params.Settings, + defaultLang: strings.TrimSpace(params.DefaultLang), + translationConfigured: params.TranslationConfigured, + sentimentConfigured: params.SentimentConfigured, + emotionsConfigured: params.EmotionsConfigured, + } +} + +// GetEnrichmentStatus returns the tenant's per-enrichment progress. tenant_id is required and +// validated; the query is scoped to that tenant alone. +func (s *EnrichmentStatusService) GetEnrichmentStatus( + ctx context.Context, tenantID string, +) (*models.EnrichmentStatusResponse, error) { + normalizedTenantID, err := normalizeRequiredTenantIDValue(tenantID) + if err != nil { + return nil, err + } + + ctx, cancel := context.WithTimeout(ctx, enrichmentStatusQueryTimeout) + defer cancel() + + settings, err := s.settings.GetSettings(ctx, normalizedTenantID) + if err != nil { + return nil, fmt.Errorf("resolve tenant settings: %w", err) + } + + counts, err := s.repo.CountEnrichmentStatus(ctx, normalizedTenantID, s.defaultLang) + if err != nil { + return nil, fmt.Errorf("count enrichment status: %w", err) + } + + translationEnabled := s.translationConfigured && + resolveTargetLang(settings.Settings.TargetLanguage, s.defaultLang) != "" + + return &models.EnrichmentStatusResponse{ + TenantID: normalizedTenantID, + Translation: enrichmentTypeStatus(translationEnabled, + counts.TranslationEligible, counts.TranslationDone), + Sentiment: enrichmentTypeStatus(s.sentimentConfigured && settings.Settings.SentimentEnrichmentEnabled(), + counts.SentimentEligible, counts.SentimentDone), + Emotions: enrichmentTypeStatus(s.emotionsConfigured && settings.Settings.EmotionsEnrichmentEnabled(), + counts.EmotionsEligible, counts.EmotionsDone), + }, nil +} + +// enrichmentTypeStatus assembles one enrichment's status, zeroing the counts when it is not +// enabled for the tenant so the API never reports a backlog for work that will never run. +func enrichmentTypeStatus(enabled bool, eligible, done int64) models.EnrichmentTypeStatus { + if !enabled { + return models.EnrichmentTypeStatus{Enabled: false} + } + + return models.EnrichmentTypeStatus{Enabled: true, Eligible: eligible, Done: done} +} diff --git a/internal/service/enrichment_status_service_test.go b/internal/service/enrichment_status_service_test.go new file mode 100644 index 00000000..0193a9fb --- /dev/null +++ b/internal/service/enrichment_status_service_test.go @@ -0,0 +1,155 @@ +package service + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/formbricks/hub/internal/huberrors" + "github.com/formbricks/hub/internal/models" + "github.com/formbricks/hub/internal/repository" +) + +type fakeStatusRepo struct { + counts repository.EnrichmentStatusCounts + err error + gotTenantID string + gotDefaultLang string +} + +func (f *fakeStatusRepo) CountEnrichmentStatus( + _ context.Context, tenantID, defaultLang string, +) (repository.EnrichmentStatusCounts, error) { + f.gotTenantID = tenantID + f.gotDefaultLang = defaultLang + + return f.counts, f.err +} + +type fakeSettingsResolver struct { + settings *models.TenantSettings + err error +} + +func (f *fakeSettingsResolver) GetSettings(_ context.Context, tenantID string) (*models.TenantSettings, error) { + if f.err != nil { + return nil, f.err + } + + if f.settings != nil { + return f.settings, nil + } + + return &models.TenantSettings{TenantID: tenantID}, nil +} + +// fullCounts is a repo result with a distinct value in every bucket, so a test that expects a +// bucket to be zeroed proves the SERVICE zeroed it (not that the repo happened to return 0). +var fullCounts = repository.EnrichmentStatusCounts{ + TranslationEligible: 10, TranslationDone: 4, + SentimentEligible: 8, SentimentDone: 3, + EmotionsEligible: 6, EmotionsDone: 2, +} + +func TestEnrichmentStatusService_GetEnrichmentStatus(t *testing.T) { + emotionsOff := false + + cases := []struct { + name string + params NewEnrichmentStatusServiceParams + settings *models.TenantSettings + wantTrans models.EnrichmentTypeStatus + wantSent models.EnrichmentTypeStatus + wantEmotion models.EnrichmentTypeStatus + }{ + { + name: "all configured and enabled", + params: NewEnrichmentStatusServiceParams{ + DefaultLang: "en-US", TranslationConfigured: true, SentimentConfigured: true, EmotionsConfigured: true, + }, + settings: &models.TenantSettings{Settings: models.EnrichmentSettings{TargetLanguage: "de-DE"}}, + wantTrans: models.EnrichmentTypeStatus{Enabled: true, Eligible: 10, Done: 4}, + wantSent: models.EnrichmentTypeStatus{Enabled: true, Eligible: 8, Done: 3}, + wantEmotion: models.EnrichmentTypeStatus{Enabled: true, Eligible: 6, Done: 2}, + }, + { + name: "sentiment not deployment-configured is zeroed", + params: NewEnrichmentStatusServiceParams{ + DefaultLang: "en-US", TranslationConfigured: true, SentimentConfigured: false, EmotionsConfigured: true, + }, + settings: &models.TenantSettings{Settings: models.EnrichmentSettings{TargetLanguage: "de-DE"}}, + wantTrans: models.EnrichmentTypeStatus{Enabled: true, Eligible: 10, Done: 4}, + wantSent: models.EnrichmentTypeStatus{Enabled: false}, + wantEmotion: models.EnrichmentTypeStatus{Enabled: true, Eligible: 6, Done: 2}, + }, + { + name: "tenant emotions switch off is zeroed", + params: NewEnrichmentStatusServiceParams{ + DefaultLang: "en-US", TranslationConfigured: true, SentimentConfigured: true, EmotionsConfigured: true, + }, + settings: &models.TenantSettings{Settings: models.EnrichmentSettings{ + TargetLanguage: "de-DE", EmotionsEnabled: &emotionsOff, + }}, + wantTrans: models.EnrichmentTypeStatus{Enabled: true, Eligible: 10, Done: 4}, + wantSent: models.EnrichmentTypeStatus{Enabled: true, Eligible: 8, Done: 3}, + wantEmotion: models.EnrichmentTypeStatus{Enabled: false}, + }, + { + name: "translation with no target and no default is zeroed", + params: NewEnrichmentStatusServiceParams{ + DefaultLang: "", TranslationConfigured: true, SentimentConfigured: true, EmotionsConfigured: true, + }, + settings: &models.TenantSettings{Settings: models.EnrichmentSettings{}}, + wantTrans: models.EnrichmentTypeStatus{Enabled: false}, + wantSent: models.EnrichmentTypeStatus{Enabled: true, Eligible: 8, Done: 3}, + wantEmotion: models.EnrichmentTypeStatus{Enabled: true, Eligible: 6, Done: 2}, + }, + { + name: "translation enabled via default-language fallback", + params: NewEnrichmentStatusServiceParams{ + DefaultLang: "en-US", TranslationConfigured: true, SentimentConfigured: true, EmotionsConfigured: true, + }, + settings: &models.TenantSettings{Settings: models.EnrichmentSettings{}}, + wantTrans: models.EnrichmentTypeStatus{Enabled: true, Eligible: 10, Done: 4}, + wantSent: models.EnrichmentTypeStatus{Enabled: true, Eligible: 8, Done: 3}, + wantEmotion: models.EnrichmentTypeStatus{Enabled: true, Eligible: 6, Done: 2}, + }, + } + + for _, testCase := range cases { + t.Run(testCase.name, func(t *testing.T) { + repo := &fakeStatusRepo{counts: fullCounts} + params := testCase.params + params.Repo = repo + params.Settings = &fakeSettingsResolver{settings: testCase.settings} + + svc := NewEnrichmentStatusService(params) + + got, err := svc.GetEnrichmentStatus(context.Background(), " tenant-1 ") + require.NoError(t, err) + + assert.Equal(t, "tenant-1", got.TenantID, "tenant_id is normalized (trimmed)") + assert.Equal(t, "tenant-1", repo.gotTenantID, "repo receives the normalized tenant_id") + assert.Equal(t, testCase.wantTrans, got.Translation) + assert.Equal(t, testCase.wantSent, got.Sentiment) + assert.Equal(t, testCase.wantEmotion, got.Emotions) + }) + } +} + +func TestEnrichmentStatusService_RequiresTenantID(t *testing.T) { + svc := NewEnrichmentStatusService(NewEnrichmentStatusServiceParams{ + Repo: &fakeStatusRepo{}, + Settings: &fakeSettingsResolver{}, + }) + + _, err := svc.GetEnrichmentStatus(context.Background(), " ") + require.Error(t, err) + + // normalizeRequiredTenantIDValue returns a *huberrors.ValidationError, which the response + // layer maps to a client 400 rather than a generic 500. + var validationErr *huberrors.ValidationError + assert.ErrorAs(t, err, &validationErr, "missing tenant_id must be a validation error") +} diff --git a/internal/service/feedback_records_service.go b/internal/service/feedback_records_service.go index cfb0f4c8..b4e84975 100644 --- a/internal/service/feedback_records_service.go +++ b/internal/service/feedback_records_service.go @@ -58,6 +58,8 @@ type FeedbackRecordsRepository interface { //nolint:interfacebloat // one cohesi 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) @@ -260,16 +262,18 @@ func (s *FeedbackRecordsService) SetSentiment( // (no enrichment loop). stillCurrent (optional) is the repository's content-supersession guard: // it is given the record's current value_text atomically with the write, and a false return skips // the write with huberrors.ErrClassificationSuperseded (nil ⇒ unconditional). Emotions are -// multi-label; an empty (or nil) set clears the column, so "no emotion detected" and "not yet -// enriched" share the same NULL representation. +// multi-label; an empty (or nil) set still records a COMPLETED classification (the labels column +// stays NULL, but the completion marker is stamped), which is what distinguishes "no emotion +// detected" from "not yet enriched" -- use ClearEmotions for the latter. func (s *FeedbackRecordsService) SetEmotions( ctx context.Context, feedbackRecordID uuid.UUID, emotions []models.EmotionValue, stillCurrent func(valueText *string) bool, ) error { - // An empty set clears (stored as NULL, never an empty array). + // An empty set stores NULL labels (never an empty array) but is still a classification result, + // so it goes through SetEmotions -- not ClearEmotions -- to stamp the completion marker. if len(emotions) == 0 { if err := s.repo.SetEmotions(ctx, feedbackRecordID, nil, stillCurrent); err != nil { - return fmt.Errorf("clear feedback record emotions: %w", err) + return fmt.Errorf("set empty feedback record emotions: %w", err) } return nil @@ -288,6 +292,19 @@ func (s *FeedbackRecordsService) SetEmotions( return nil } +// ClearEmotions returns a record to the "not classified" state, dropping both the labels and the +// completion marker. Used when the source content is gone, so the record is excluded from progress +// counts rather than counting as classified-with-no-emotions. +func (s *FeedbackRecordsService) ClearEmotions( + ctx context.Context, feedbackRecordID uuid.UUID, stillCurrent func(valueText *string) bool, +) error { + if err := s.repo.ClearEmotions(ctx, feedbackRecordID, stillCurrent); err != nil { + return fmt.Errorf("clear feedback record emotions: %w", err) + } + + return nil +} + // ListFeedbackRecords retrieves a list of feedback records with optional filters. // Uses cursor-based pagination: omit cursor for first page, use next_cursor for subsequent pages. func (s *FeedbackRecordsService) ListFeedbackRecords( diff --git a/internal/service/feedback_records_service_test.go b/internal/service/feedback_records_service_test.go index 92299cdd..cacccd12 100644 --- a/internal/service/feedback_records_service_test.go +++ b/internal/service/feedback_records_service_test.go @@ -38,8 +38,9 @@ type mockFeedbackRecordsRepo struct { setSentimentLabel *models.SentimentValue setSentimentScore *float64 - setEmotionsCalled bool - setEmotionsLabels []models.EmotionValue + setEmotionsCalled bool + setEmotionsLabels []models.EmotionValue + clearEmotionsCalled bool } func (m *mockFeedbackRecordsRepo) Create( @@ -103,6 +104,14 @@ func (m *mockFeedbackRecordsRepo) SetSentiment( return nil } +func (m *mockFeedbackRecordsRepo) ClearEmotions( + _ context.Context, _ uuid.UUID, _ func(valueText *string) bool, +) error { + m.clearEmotionsCalled = true + + return nil +} + func (m *mockFeedbackRecordsRepo) SetEmotions( _ context.Context, _ uuid.UUID, emotions []models.EmotionValue, _ func(valueText *string) bool, ) error { diff --git a/internal/workers/feedback_emotions.go b/internal/workers/feedback_emotions.go index 653c6454..66ff1e59 100644 --- a/internal/workers/feedback_emotions.go +++ b/internal/workers/feedback_emotions.go @@ -24,6 +24,10 @@ type emotionsWorkerService interface { GetFeedbackRecord(ctx context.Context, id uuid.UUID) (*models.FeedbackRecord, error) SetEmotions(ctx context.Context, feedbackRecordID uuid.UUID, emotions []models.EmotionValue, stillCurrent func(valueText *string) bool) error + // ClearEmotions returns the record to "not classified" (used when the source content is gone), + // as opposed to SetEmotions with an empty result, which is a completed classification. + ClearEmotions(ctx context.Context, feedbackRecordID uuid.UUID, + stillCurrent func(valueText *string) bool) error } // tenantSettingsReader resolves a tenant's enrichment settings for the worker's authoritative @@ -58,11 +62,14 @@ func NewFeedbackEmotionsWorker( }, persist: func(ctx context.Context, record *models.FeedbackRecord, _ service.FeedbackEmotionsArgs, result *service.EmotionsResult) error { // Guard the write against content churn since the Work-time read: a stale job's labels - // (or clear) must not land last over a newer job's write. A nil result (empty content) - // or an empty label set both clear the column: absence is NULL, never an empty array. + // (or clear) must not land last over a newer job's write. Both paths leave the column + // NULL when there are no labels (absence is NULL, never an empty array), but they are + // NOT the same state: a nil result means the content is gone, whereas an empty label + // set is a completed classification. ClearEmotions/SetEmotions record that difference + // in emotions_classified_at so progress counts can tell them apart (ENG-1670). stillCurrent := valueTextStillCurrent(record.ValueText) if result == nil { - return svc.SetEmotions(ctx, record.ID, nil, stillCurrent) + return svc.ClearEmotions(ctx, record.ID, stillCurrent) } return svc.SetEmotions(ctx, record.ID, result.Labels, stillCurrent) diff --git a/internal/workers/feedback_emotions_test.go b/internal/workers/feedback_emotions_test.go index 09426f14..6477259d 100644 --- a/internal/workers/feedback_emotions_test.go +++ b/internal/workers/feedback_emotions_test.go @@ -58,10 +58,11 @@ func (m *countingEmotionsMetrics) RecordEmotionsDuration(_ context.Context, _ ti var _ observability.EmotionsMetrics = (*countingEmotionsMetrics)(nil) type mockEmotionsWorkerService struct { - record *models.FeedbackRecord - getErr error - setErr error - setCalls [][]models.EmotionValue + record *models.FeedbackRecord + getErr error + setErr error + setCalls [][]models.EmotionValue + clearCalls int } func (m *mockEmotionsWorkerService) GetFeedbackRecord(_ context.Context, _ uuid.UUID) (*models.FeedbackRecord, error) { @@ -76,6 +77,16 @@ func (m *mockEmotionsWorkerService) SetEmotions( return m.setErr } +// ClearEmotions is the "content is gone" path, distinct from SetEmotions with an empty result +// (a completed classification). Counted separately so tests can assert which one the worker took. +func (m *mockEmotionsWorkerService) ClearEmotions( + _ context.Context, _ uuid.UUID, _ func(valueText *string) bool, +) error { + m.clearCalls++ + + return m.setErr +} + type stubEmotionsClient struct { result service.EmotionsResult err error @@ -186,8 +197,15 @@ func TestFeedbackEmotionsWorker_EmptyValueTextClears(t *testing.T) { t.Fatalf("Classify calls = %d, want 0 (empty text is not classified)", client.calls) } - if len(svc.setCalls) != 1 || len(svc.setCalls[0]) != 0 { - t.Fatalf("setCalls = %+v, want one clear (empty set)", svc.setCalls) + // Empty content must take the CLEAR path, not a classification with an empty result: + // both leave the labels NULL, but only a classification stamps the completion marker, + // and an unclassifiable record must not be reported as "done" (ENG-1670). + if svc.clearCalls != 1 { + t.Fatalf("clearCalls = %d, want 1 (empty text clears rather than classifies)", svc.clearCalls) + } + + if len(svc.setCalls) != 0 { + t.Fatalf("setCalls = %+v, want none (empty text must not record a classification)", svc.setCalls) } }) } diff --git a/migrations/020_add_emotions_classified_at.sql b/migrations/020_add_emotions_classified_at.sql new file mode 100644 index 00000000..04571901 --- /dev/null +++ b/migrations/020_add_emotions_classified_at.sql @@ -0,0 +1,51 @@ +-- +goose NO TRANSACTION +-- +goose up +-- Emotion completion marker (ENG-1670). `emotions` cannot express "classified, found nothing": +-- the 015 CHECK rejects the empty array, so a successful classification that detects no emotion is +-- persisted as NULL — indistinguishable from "not classified yet". Anything deriving progress from +-- the data (the enrichment-status endpoint, the backlog gauge) therefore counted those records as +-- permanently pending, and the classify backfill re-sent them to the LLM on every run. +-- +-- emotions_classified_at records WHEN the classifier last produced a result for the record, +-- independently of whether that result had any labels. It is processing state, deliberately kept +-- out of the API surface (not in feedbackRecordColumns), so no response shape or SDK type changes. +-- The eager-clear on a value_text edit nulls it alongside `emotions`, so an edited record correctly +-- returns to "not classified". +-- +-- Deliberately NOT backfilled: a bulk UPDATE would rewrite every row of the primary high-write +-- table (WAL + bloat) for no benefit. Readers instead treat a non-NULL `emotions` as proof of +-- completion for pre-existing rows (see the enrichment-status query), which leaves only historical +-- classified-empty rows unresolved; the classify backfill re-processes those once and stamps them. +-- +-- Runs without a transaction like the sibling enrichment migrations. ADD COLUMN of a nullable +-- column with no default is metadata-only (instant, no table rewrite, no long lock), and +-- IF NOT EXISTS keeps the statement re-runnable after an interrupted deploy. +ALTER TABLE feedback_records + ADD COLUMN IF NOT EXISTS emotions_classified_at TIMESTAMPTZ; + +-- Realign the 016 emotions-backfill index with the new completion semantics. That index was built +-- on `emotions IS NULL` precisely so a row would "leave the index the moment it is enriched", +-- keeping it near-empty once a backfill drains. A classified-empty row now keeps emotions NULL +-- forever (the point of the column above), so under the old predicate it would stay indexed +-- permanently: the index would grow monotonically, and a drained backfill would have to scan the +-- whole retained set and discard every row instead of finishing near-instantly. Adding the marker +-- to the predicate restores the drain property, and also lets the backfill query use the index +-- directly instead of re-checking emotions_classified_at as a post-filter. +-- +-- Built CONCURRENTLY, and DROP-then-CREATE (not IF NOT EXISTS) so an interrupted build leaves an +-- INVALID index that a re-run replaces, matching 016. +DROP INDEX CONCURRENTLY IF EXISTS idx_feedback_records_emotions_backfill; +CREATE INDEX CONCURRENTLY idx_feedback_records_emotions_backfill + ON feedback_records (id) + WHERE field_type = 'text' AND value_text IS NOT NULL + AND emotions IS NULL AND emotions_classified_at IS NULL; + +-- +goose down +-- Restore the 016 predicate before dropping the column it references. +DROP INDEX CONCURRENTLY IF EXISTS idx_feedback_records_emotions_backfill; +CREATE INDEX CONCURRENTLY idx_feedback_records_emotions_backfill + ON feedback_records (id) + WHERE field_type = 'text' AND value_text IS NOT NULL AND emotions IS NULL; + +ALTER TABLE feedback_records + DROP COLUMN IF EXISTS emotions_classified_at; diff --git a/openapi.yaml b/openapi.yaml index ffc87700..b7f7ebd1 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -54,6 +54,8 @@ tags: description: Tenant-scoped enrichment settings - name: Taxonomy description: Automatic topic/subtopic taxonomy generation, run history, tree browsing, and node edits + - name: Enrichment Status + description: Tenant-scoped enrichment progress (translation, sentiment, emotions) security: - ApiKeyAuth: [] paths: @@ -1380,6 +1382,64 @@ paths: application/problem+json: schema: $ref: '#/components/schemas/ErrorModel' + /v1/enrichment-status: + get: + tags: + - Enrichment Status + summary: Get tenant enrichment status + description: | + Returns a tenant's enrichment progress across the record-level enrichments + (translation, sentiment, emotions). For each, `enabled` reports whether the enrichment + is active for the tenant (deployment-configured and switched on / with a resolvable + target language), and `eligible`/`done` are directory-level counts of feedback records + that qualify and that have been enriched — the UI derives "in progress" as + `eligible - done`. When an enrichment is not enabled its counts are zero. The response + contains counts only (no record identifiers or content). + operationId: get-enrichment-status + parameters: + - name: tenant_id + in: query + required: true + description: Tenant whose enrichment status should be returned. + schema: + type: string + minLength: 1 + maxLength: 255 + pattern: '^[^\x00]*$' + example: "org-123" + responses: + "200": + description: Tenant enrichment status + content: + application/json: + schema: + $ref: '#/components/schemas/EnrichmentStatusOutputBody' + "400": + description: Bad Request (e.g. missing or invalid tenant_id) + content: + application/problem+json: + schema: + $ref: '#/components/schemas/ErrorModel' + "401": + description: Unauthorized (missing or invalid API key) + content: + application/problem+json: + schema: + $ref: '#/components/schemas/ErrorModel' + "503": + description: | + Service Unavailable – the enrichment status service is not available + (code `service_unavailable`). + content: + application/problem+json: + schema: + $ref: '#/components/schemas/ErrorModel' + default: + description: Error + content: + application/problem+json: + schema: + $ref: '#/components/schemas/ErrorModel' /v1/taxonomy/fields: get: tags: @@ -3362,6 +3422,48 @@ components: $ref: '#/components/schemas/TaxonomyFieldOption' required: - data + EnrichmentTypeStatus: + type: object + additionalProperties: false + # Reused for translation/sentiment/emotions; declare a model so the SDK emits one shared + # type instead of three structurally identical inline copies (Model/Recommended). + # Model paths are snake_case (Name/NotSnakeCase), unlike the resource/method names. + x-stainless-model: enrichment_status.type_status + description: One enrichment's progress for a tenant. When `enabled` is false, `eligible` and `done` are zero. + properties: + enabled: + type: boolean + description: Whether the enrichment is active for the tenant (deployment-configured and switched on / with a resolvable target language). + eligible: + type: integer + format: int64 + description: Feedback records that qualify for this enrichment. + done: + type: integer + format: int64 + description: Eligible records that have been enriched. + required: + - enabled + - eligible + - done + EnrichmentStatusOutputBody: + type: object + additionalProperties: false + description: A tenant's enrichment progress across the record-level enrichments. Counts are directory-level totals. + properties: + tenant_id: + type: string + translation: + $ref: '#/components/schemas/EnrichmentTypeStatus' + sentiment: + $ref: '#/components/schemas/EnrichmentTypeStatus' + emotions: + $ref: '#/components/schemas/EnrichmentTypeStatus' + required: + - tenant_id + - translation + - sentiment + - emotions CreateTaxonomyRunInputBody: type: object additionalProperties: false diff --git a/tests/enrichment_status_test.go b/tests/enrichment_status_test.go new file mode 100644 index 00000000..89d5607c --- /dev/null +++ b/tests/enrichment_status_test.go @@ -0,0 +1,348 @@ +package tests + +import ( + "context" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/formbricks/hub/internal/config" + "github.com/formbricks/hub/internal/models" + "github.com/formbricks/hub/internal/repository" + "github.com/formbricks/hub/pkg/database" +) + +// TestCountEnrichmentStatus exercises the data-derived enrichment-status counts against Postgres, +// locking the gap-analysis edge cases: only text records with real content are eligible +// (field_type gate + whitespace trim), translation "done" means the stored lang key equals the +// effective target (stale != done), the sentiment/emotions per-tenant switch gates eligibility, +// the translation default-language fallback, and strict per-tenant isolation. +func TestCountEnrichmentStatus(t *testing.T) { + ctx := context.Background() + + cfg, err := config.Load() + require.NoError(t, err) + + db, err := database.NewPostgresPool(ctx, cfg.Database.URL, database.WithPoolConfig(cfg.Database.PoolConfig())) + require.NoError(t, err) + + defer db.Close() + + statusRepo := repository.NewEnrichmentStatusRepository(db) + frepo := repository.NewFeedbackRecordsRepository(db) + tsRepo := repository.NewTenantSettingsRepository(db) + + // mkRecord inserts one feedback record for a tenant with an explicit field type + value_text. + mkRecord := func(tenant string, fieldType models.FieldType, valueText string) *models.FeedbackRecord { + return seedEnrichmentRecord(t, frepo, tenant, fieldType, valueText) + } + + setSentiment := func(id uuid.UUID) { + label := models.SentimentPositive + score := 1.0 + require.NoError(t, frepo.SetSentiment(ctx, id, &label, &score, nil)) + } + setEmotions := func(id uuid.UUID) { + require.NoError(t, frepo.SetEmotions(ctx, id, []models.EmotionValue{models.EmotionJoy}, nil)) + } + // setTranslationLang writes the translation columns directly to a given lang key, so a test can + // stage both "done" (key == effective target) and "stale" (key != target) states precisely. + setTranslationLang := func(id uuid.UUID, langKey string) { + translated := "translated text" + _, execErr := db.Exec(ctx, + `UPDATE feedback_records SET value_text_translated = $1, translation_lang_key = $2 WHERE id = $3`, + translated, langKey, id) + require.NoError(t, execErr) + } + + t.Run("eligibility, done buckets, and translation staleness", func(t *testing.T) { + tenant := testTenantID("enrich-status-main") + _, err := tsRepo.Upsert(ctx, tenant, models.EnrichmentSettings{TargetLanguage: "de-DE"}) + require.NoError(t, err) + + // Three eligible text records with content. The first stays fully un-enriched (pending in + // every bucket); the other two are enriched below. + mkRecord(tenant, models.FieldTypeText, "great product, would recommend") + doneAll := mkRecord(tenant, models.FieldTypeText, "fully enriched record") + staleTrans := mkRecord(tenant, models.FieldTypeText, "sentiment set, translation stale") + + // doneAll: sentiment + emotions set, translated to the effective target. + setSentiment(doneAll.ID) + setEmotions(doneAll.ID) + setTranslationLang(doneAll.ID, "de-DE") + + // staleTrans: sentiment set (done), emotions NULL (pending), translated to a DIFFERENT + // language than the tenant's target — so translation is NOT done. + setSentiment(staleTrans.ID) + setTranslationLang(staleTrans.ID, "fr-FR") + + // Ineligible rows that must NOT be counted: + mkRecord(tenant, models.FieldTypeCategorical, "some choice") // non-text carries value_text + mkRecord(tenant, models.FieldTypeText, "\t\n ") // whitespace-only content + mkRecord(tenant, models.FieldTypeText, "") // empty content + + counts, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + + assert.Equal(t, int64(3), counts.SentimentEligible, "3 text records with content are eligible") + assert.Equal(t, int64(2), counts.SentimentDone, "doneAll + staleTrans have sentiment") + assert.Equal(t, int64(3), counts.EmotionsEligible) + assert.Equal(t, int64(1), counts.EmotionsDone, "only doneAll has emotions") + assert.Equal(t, int64(3), counts.TranslationEligible, "all 3 have the effective target de-DE") + assert.Equal(t, int64(1), counts.TranslationDone, "only doneAll's lang key matches; fr-FR is stale") + }) + + t.Run("sentiment switch off zeroes sentiment eligibility, emotions unaffected", func(t *testing.T) { + tenant := testTenantID("enrich-status-switch") + off := false + _, err := tsRepo.Upsert(ctx, tenant, models.EnrichmentSettings{SentimentEnabled: &off}) + require.NoError(t, err) + + mkRecord(tenant, models.FieldTypeText, "a") + mkRecord(tenant, models.FieldTypeText, "b") + + counts, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + + assert.Equal(t, int64(0), counts.SentimentEligible, "sentiment disabled → not eligible") + assert.Equal(t, int64(2), counts.EmotionsEligible, "emotions default-enabled → still eligible") + }) + + t.Run("emotions switch off zeroes emotions eligibility, sentiment unaffected", func(t *testing.T) { + tenant := testTenantID("enrich-status-emotions-off") + off := false + _, err := tsRepo.Upsert(ctx, tenant, models.EnrichmentSettings{EmotionsEnabled: &off}) + require.NoError(t, err) + + mkRecord(tenant, models.FieldTypeText, "a") + mkRecord(tenant, models.FieldTypeText, "b") + + counts, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + + assert.Equal(t, int64(0), counts.EmotionsEligible, "emotions disabled → not eligible") + assert.Equal(t, int64(2), counts.SentimentEligible, "sentiment default-enabled → still eligible") + }) + + t.Run("emotions completion is the marker, not the labels", func(t *testing.T) { + tenant := testTenantID("enrich-status-emotions-done") + + // A successful classification that detects NO emotion stores NULL labels (the 015 CHECK + // rejects an empty array). Without the completion marker it is indistinguishable from + // "never classified" and would count as pending forever, so the backlog would never drain. + classifiedEmpty := mkRecord(tenant, models.FieldTypeText, "the export ran at 3pm") + require.NoError(t, frepo.SetEmotions(ctx, classifiedEmpty.ID, nil, nil)) + + classifiedWithLabels := mkRecord(tenant, models.FieldTypeText, "I am thrilled") + setEmotions(classifiedWithLabels.ID) + + mkRecord(tenant, models.FieldTypeText, "not processed yet") + + // A pre-020 row: labels present but no marker (the column did not exist when it was + // enriched). Non-NULL labels are themselves proof of completion, which is why the migration + // needs no bulk backfill. + legacy := mkRecord(tenant, models.FieldTypeText, "enriched before the marker existed") + setEmotions(legacy.ID) + _, err := db.Exec(ctx, + `UPDATE feedback_records SET emotions_classified_at = NULL WHERE id = $1`, legacy.ID) + require.NoError(t, err) + + counts, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + + assert.Equal(t, int64(4), counts.EmotionsEligible, "all four text records are eligible") + assert.Equal(t, int64(3), counts.EmotionsDone, + "classified-empty, classified-with-labels and the legacy row are all done; only the unprocessed one is pending") + + // The eager-clear on a content edit must drop the marker with the labels, or an edited + // record would keep counting as classified while its emotions are gone. + newText := "completely different text now" + _, _, err = frepo.Update(ctx, classifiedWithLabels.ID, + &models.UpdateFeedbackRecordRequest{ValueText: &newText}) + require.NoError(t, err) + + afterEdit, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + assert.Equal(t, int64(2), afterEdit.EmotionsDone, "editing the text returns that record to pending") + }) + + t.Run("translation eligibility follows the effective target", func(t *testing.T) { + tenant := testTenantID("enrich-status-trans") + // No tenant settings row at all → no own target language. + mkRecord(tenant, models.FieldTypeText, "x") + mkRecord(tenant, models.FieldTypeText, "y") + + noDefault, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + assert.Equal(t, int64(0), noDefault.TranslationEligible, "no target + no default → not eligible") + + withDefault, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "en-US") + require.NoError(t, err) + assert.Equal(t, int64(2), withDefault.TranslationEligible, "default-language fallback makes records eligible") + }) + + t.Run("counts are strictly per-tenant", func(t *testing.T) { + tenantA := testTenantID("enrich-status-iso-a") + tenantB := testTenantID("enrich-status-iso-b") + + mkRecord(tenantA, models.FieldTypeText, "a1") + + for range 5 { + mkRecord(tenantB, models.FieldTypeText, "b") + } + + counts, err := statusRepo.CountEnrichmentStatus(ctx, tenantA, "") + require.NoError(t, err) + assert.Equal(t, int64(1), counts.SentimentEligible, "tenant A sees only its own record, not tenant B's") + }) +} + +// seedEnrichmentRecord inserts one feedback record and returns it. Shared by the per-tenant and +// aggregate enrichment-status tests. +func seedEnrichmentRecord( + t *testing.T, frepo *repository.FeedbackRecordsRepository, tenant string, fieldType models.FieldType, valueText string, +) *models.FeedbackRecord { + t.Helper() + + vt := valueText + rec, err := frepo.Create(context.Background(), &models.CreateFeedbackRecordRequest{ + SourceType: "formbricks", + FieldID: "q1", + FieldType: fieldType, + ValueText: &vt, + TenantID: tenant, + SubmissionID: testTenantID("sub"), + }) + require.NoError(t, err) + + return rec +} + +// TestCountEnrichmentBacklogAggregate covers the cross-tenant aggregate query that feeds the +// backlog gauge. The shared test DB holds records from other tests, so it asserts on the DELTA +// around a fresh seed rather than absolute totals — and verifies the per-tenant enable gate still +// applies (a sentiment-off tenant contributes to emotions but not sentiment). +func TestCountEnrichmentBacklogAggregate(t *testing.T) { + ctx := context.Background() + + cfg, err := config.Load() + require.NoError(t, err) + + db, err := database.NewPostgresPool(ctx, cfg.Database.URL, database.WithPoolConfig(cfg.Database.PoolConfig())) + require.NoError(t, err) + + defer db.Close() + + statusRepo := repository.NewEnrichmentStatusRepository(db) + frepo := repository.NewFeedbackRecordsRepository(db) + tsRepo := repository.NewTenantSettingsRepository(db) + + const defaultLang = "en-US" + + pending := func(c repository.EnrichmentStatusCounts) (sentiment, emotions, translation int64) { + return c.SentimentEligible - c.SentimentDone, + c.EmotionsEligible - c.EmotionsDone, + c.TranslationEligible - c.TranslationDone + } + + before, err := statusRepo.CountEnrichmentBacklogAggregate(ctx, defaultLang) + require.NoError(t, err) + + beforeSent, beforeEmo, beforeTrans := pending(before) + + // Tenant with everything enabled (no settings row → sentiment/emotions default-on; translation + // via the en-US default): 3 un-enriched text records. + enabled := testTenantID("agg-enabled") + for range 3 { + seedEnrichmentRecord(t, frepo, enabled, models.FieldTypeText, "pending record") + } + + // Tenant with sentiment switched OFF: 2 un-enriched text records. These must add to emotions and + // translation backlog but NOT sentiment. + sentimentOff := testTenantID("agg-sentiment-off") + off := false + _, err = tsRepo.Upsert(ctx, sentimentOff, models.EnrichmentSettings{SentimentEnabled: &off}) + require.NoError(t, err) + + for range 2 { + seedEnrichmentRecord(t, frepo, sentimentOff, models.FieldTypeText, "pending record") + } + + after, err := statusRepo.CountEnrichmentBacklogAggregate(ctx, defaultLang) + require.NoError(t, err) + + afterSent, afterEmo, afterTrans := pending(after) + + assert.Equal(t, int64(3), afterSent-beforeSent, "only the sentiment-enabled tenant's 3 records add to sentiment backlog") + assert.Equal(t, int64(5), afterEmo-beforeEmo, "emotions default-enabled for both tenants → 3+2") + assert.Equal(t, int64(5), afterTrans-beforeTrans, "en-US default makes all 5 records translation-pending") +} + +// TestCountEnrichmentBacklogAggregateIfLeader covers the single-flight advisory lock that stops +// every API replica from repeating the same cross-tenant scan: one caller wins and gets the counts, +// a concurrent caller is denied (not an error) and skips its tick, and the lock is released +// afterwards so the next tick can win again. +func TestCountEnrichmentBacklogAggregateIfLeader(t *testing.T) { + ctx := context.Background() + + cfg, err := config.Load() + require.NoError(t, err) + + // Two independent pools stand in for two API replicas: a transaction-scoped advisory lock is + // held per session, so the contention is only observable across separate connections. + dbLeader, err := database.NewPostgresPool(ctx, cfg.Database.URL, database.WithPoolConfig(cfg.Database.PoolConfig())) + require.NoError(t, err) + + defer dbLeader.Close() + + dbRival, err := database.NewPostgresPool(ctx, cfg.Database.URL, database.WithPoolConfig(cfg.Database.PoolConfig())) + require.NoError(t, err) + + defer dbRival.Close() + + leaderOne := repository.NewEnrichmentBacklogLeader(dbLeader) + leaderTwo := repository.NewEnrichmentBacklogLeader(dbRival) + + // Release leadership before the pools close. Close is idempotent, and without this an early + // require failure would unwind into pgxpool.Close with the leader's connection still checked + // out, wedging the whole suite until the test timeout instead of reporting the failure. + defer leaderOne.Close(ctx) + defer leaderTwo.Close(ctx) + + // First replica wins leadership and gets real counts. + counts, isLeader, err := leaderOne.CountIfLeader(ctx, "") + require.NoError(t, err) + require.True(t, isLeader, "the first replica to poll becomes the leader") + + // Prove the leader returns the real aggregate, not a zero value. + want, err := repository.NewEnrichmentStatusRepository(dbLeader).CountEnrichmentBacklogAggregate(ctx, "") + require.NoError(t, err) + assert.Equal(t, want.SentimentEligible, counts.SentimentEligible, "leader returns the true aggregate") + + // The second replica is denied — a normal skip, not an error — and must NOT publish counts. + zero, isLeader, err := leaderTwo.CountIfLeader(ctx, "") + require.NoError(t, err, "losing the leader election is not an error") + assert.False(t, isLeader, "a second replica must not scan or export the global gauge") + assert.Equal(t, repository.EnrichmentStatusCounts{}, zero, "non-leader returns no counts to publish") + + // Leadership is STICKY: unlike a scan-scoped lock, it persists across polls, so the same + // replica keeps exporting the series instead of it flapping between replicas. + _, stillLeader, err := leaderOne.CountIfLeader(ctx, "") + require.NoError(t, err) + assert.True(t, stillLeader, "leadership is held for the process lifetime, not per scan") + + _, stillDenied, err := leaderTwo.CountIfLeader(ctx, "") + require.NoError(t, err) + assert.False(t, stillDenied, "the follower stays a follower while the leader lives") + + // Releasing hands leadership over promptly rather than waiting for a session timeout. + leaderOne.Close(ctx) + + _, promoted, err := leaderTwo.CountIfLeader(ctx, "") + require.NoError(t, err) + assert.True(t, promoted, "a follower is promoted once the leader releases") + + leaderTwo.Close(ctx) +}