feat: expose per-tenant enrichment status and backlog metric (ENG-1670)

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.

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,
 	)

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
 )
 

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=

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=<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)
+}

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)
+	})
+}

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"`
+}