feat: add tenant settings for language enrichment

.env.example

@@ -108,6 +108,25 @@ WEBHOOK_MAX_COUNT=500
 # GOOGLE_APPLICATION_CREDENTIALS=    (optional; for google-gemini when outside Google Cloud: path to service account key JSON)
 # EMBEDDING_MODEL=<model name>        (required to enable embeddings; no default)
 
+# Translation (language enrichment) is optional. To enable, set both TRANSLATION_PROVIDER and TRANSLATION_MODEL; if either is unset, translation is disabled and no translation jobs run.
+# Open-text feedback (value_text) is translated into each tenant's configured target_language (Hub tenant settings), falling back to TRANSLATION_DEFAULT_LANGUAGE when a tenant has none. Same providers/auth model as embeddings.
+# Configure identically in the API and worker processes (the API enqueues, the worker translates); otherwise jobs pile up unprocessed.
+# TRANSLATION_PROVIDER=openai
+# TRANSLATION_PROVIDER=google-gemini
+# TRANSLATION_PROVIDER_API_KEY=sk-...   (required for openai and google; not used for google-gemini)
+# TRANSLATION_BASE_URL=https://llm.example.com/v1  (optional; only supported with TRANSLATION_PROVIDER=openai)
+# TRANSLATION_GOOGLE_CLOUD_PROJECT=     (required for google-gemini; or use GOOGLE_CLOUD_PROJECT)
+# TRANSLATION_GOOGLE_CLOUD_LOCATION=    (required for google-gemini, e.g. europe-west3; or use GOOGLE_CLOUD_LOCATION)
+# TRANSLATION_MODEL=<model name>        (required to enable translation; no default; e.g. gemini-2.5-flash)
+# TRANSLATION_DEFAULT_LANGUAGE=         (optional BCP-47 fallback target, e.g. en-US; used when a tenant has no target_language of its own. Empty = per-tenant opt-in only)
+# TRANSLATION_MAX_CONCURRENT=5          (worker concurrency; default 5)
+# TRANSLATION_MAX_ATTEMPTS=3            (River job retries before failing; default 3)
+
+# Tenant settings cache (optional): an in-process LRU+TTL over tenant target_language reads on the translation enqueue path.
+# Set TENANT_SETTINGS_CACHE_SIZE=0 to disable the cache. Defaults: size 2048, TTL 60s.
+# TENANT_SETTINGS_CACHE_SIZE=2048
+# TENANT_SETTINGS_CACHE_TTL_SECONDS=60
+
 # Tenant data purge (optional). How long DELETE /v1/tenants/{tenant_id}/data waits for in-flight
 # tenant-owned writes to drain before returning a retryable 409 (code tenant_write_conflict).
 # Must be a positive integer (seconds); non-positive values fall back to the default. Default: 5

cmd/api/app.go

@@ -174,14 +174,16 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
 	}
 
 	var (
-		eventMetrics     observability.EventMetrics
-		webhookMetrics   observability.WebhookMetrics
-		embeddingMetrics observability.EmbeddingMetrics
+		eventMetrics       observability.EventMetrics
+		webhookMetrics     observability.WebhookMetrics
+		embeddingMetrics   observability.EmbeddingMetrics
+		translationMetrics observability.TranslationMetrics
 	)
 	if metrics != nil {
 		eventMetrics = metrics.Events
 		webhookMetrics = metrics.Webhooks
 		embeddingMetrics = metrics.Embeddings
+		translationMetrics = metrics.Translation
 	}
 
 	var tracerProvider *sdktrace.TracerProvider
@@ -241,6 +243,7 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
 		nil, // riverClient set below after creation
 		service.EmbeddingsQueueName,
 		cfg.Embedding.MaxAttempts,
+		cfg.Translation.DefaultLanguage,
 	)
 
 	// Shared worker/queue registration first (webhook + optional embedding added below).
@@ -277,6 +280,40 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
 		searchHandler = handlers.NewSearchHandler(nil) // 503 when embeddings disabled
 	}
 
+	// Register the translation worker and declare its queue so the River client can
+	// enqueue translation jobs (River requires the job kind registered and the queue
+	// declared at insert time); the jobs are processed by hub-worker, not in this
+	// process. Gated on TRANSLATION_PROVIDER+MODEL like embeddings; the enqueue
+	// provider is registered below, after the River client and tenant settings exist.
+	if cfg.Translation.Provider != "" && cfg.Translation.Model != "" {
+		translationCfg := service.TranslationClientConfig{
+			Provider:            cfg.Translation.Provider,
+			ProviderAPIKey:      cfg.Translation.ProviderAPIKey,
+			Model:               cfg.Translation.Model,
+			BaseURL:             cfg.Translation.BaseURL,
+			GoogleCloudProject:  cfg.Translation.GoogleCloudProject,
+			GoogleCloudLocation: cfg.Translation.GoogleCloudLocation,
+		}
+
+		translationClient, translationErr := service.NewTranslationClient(context.Background(), translationCfg)
+		if translationErr != nil {
+			cleanupNewAppStartupFailure(context.Background(), messageManager, nil, tracerProvider, meterProvider)
+
+			return nil, fmt.Errorf("translation config: %w", translationErr)
+		}
+
+		river.AddWorker(riverWorkers, workers.NewFeedbackTranslationWorker(feedbackRecordsService, translationClient, translationMetrics))
+
+		queues[service.TranslationsQueueName] = river.QueueConfig{MaxWorkers: 1}
+
+		// Per-tenant re-translation backfill, enqueued by the settings-change listener
+		// below. Registered here only so the River client can validate the kind and queue
+		// at insert time; the fan-out is processed by hub-worker.
+		river.AddWorker(riverWorkers, workers.NewTenantTranslationBackfillWorker(feedbackRecordsService, cfg.Translation.MaxAttempts))
+
+		queues[service.TranslationBackfillsQueueName] = river.QueueConfig{MaxWorkers: 1}
+	}
+
 	riverClient, err := river.NewClient(riverpgxv5.New(db), &river.Config{
 		Queues:  queues,
 		Workers: riverWorkers,
@@ -321,6 +358,39 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
 	tenantDataService := service.NewTenantDataService(tenantDataRepo)
 	tenantDataHandler := handlers.NewTenantDataHandler(tenantDataService)
 
+	tenantSettingsRepo := repository.NewTenantSettingsRepository(db)
+	tenantSettingsService := service.NewTenantSettingsService(tenantSettingsRepo)
+	tenantSettingsHandler := handlers.NewTenantSettingsHandler(tenantSettingsService)
+
+	// Translation enqueue provider: on a feedback-record create/update it resolves the
+	// tenant's target language (through a short-TTL cache over tenant settings) and
+	// enqueues a translation job. Gated on TRANSLATION_PROVIDER+MODEL.
+	if cfg.Translation.Provider != "" && cfg.Translation.Model != "" {
+		var translationCacheMetrics observability.CacheMetrics
+		if metrics != nil {
+			translationCacheMetrics = metrics.Cache
+		}
+
+		translationCache := service.NewCachedTenantSettings(
+			tenantSettingsService,
+			cfg.TenantSettingsCache.Size, cfg.TenantSettingsCache.TTL.Duration(),
+			translationCacheMetrics,
+		)
+		messageManager.RegisterProvider(service.NewTranslationProvider(
+			riverClient, translationCache, service.TranslationsQueueName, cfg.Translation.MaxAttempts,
+			cfg.Translation.DefaultLanguage, translationMetrics))
+
+		// On a settings write: evict the tenant's cached settings (so a changed/enabled
+		// target is visible to the enqueue gate immediately, not after TTL expiry) and
+		// enqueue a per-tenant re-translation backfill (so existing records pick up a new
+		// target, not only newly ingested ones).
+		tenantSettingsService.SetSettingsChangeListener(service.NewCompositeSettingsChangeListener(
+			translationCache,
+			service.NewTranslationSettingsListener(
+				riverClient, service.TranslationBackfillsQueueName, cfg.Translation.MaxAttempts),
+		))
+	}
+
 	taxonomyRepo := repository.NewTaxonomyRepository(db)
 
 	var taxonomyStarter service.TaxonomyRunStarter
@@ -358,7 +428,8 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
 	}
 
 	server := newHTTPServer(
-		cfg, healthHandler, openapiHandler, feedbackRecordsHandler, webhooksHandler, tenantDataHandler, searchHandler,
+		cfg, healthHandler, openapiHandler, feedbackRecordsHandler, webhooksHandler, tenantDataHandler,
+		tenantSettingsHandler, searchHandler,
 		taxonomyHandler, taxonomyInternalHandler,
 		meterProvider, tracerProvider,
 	)
@@ -385,6 +456,7 @@ func newHTTPServer(
 	feedback *handlers.FeedbackRecordsHandler,
 	webhooks *handlers.WebhooksHandler,
 	tenantData *handlers.TenantDataHandler,
+	tenantSettings *handlers.TenantSettingsHandler,
 	search *handlers.SearchHandler,
 	taxonomy *handlers.TaxonomyHandler,
 	taxonomyInternal *handlers.TaxonomyInternalHandler,
@@ -410,6 +482,9 @@ func newHTTPServer(
 	protected.HandleFunc("PATCH /v1/webhooks/{id}", webhooks.Update)
 	protected.HandleFunc("DELETE /v1/webhooks/{id}", webhooks.Delete)
 	protected.HandleFunc("DELETE /v1/tenants/{tenant_id}/data", tenantData.Delete)
+	protected.HandleFunc("GET /v1/tenants/{tenant_id}/settings", tenantSettings.Get)
+	protected.HandleFunc("PUT /v1/tenants/{tenant_id}/settings", tenantSettings.Update)
+	protected.HandleFunc("PATCH /v1/tenants/{tenant_id}/settings", tenantSettings.Patch)
 
 	// Search endpoints are always registered; when embeddings are disabled, the handler returns 503.
 	protected.HandleFunc("POST /v1/feedback-records/search/semantic", search.SemanticSearch)

cmd/api/app_test.go

@@ -374,6 +374,7 @@ func newTestHTTPServerWithConfig(t *testing.T, publicBaseURL string, taxonomy co
 		handlers.NewFeedbackRecordsHandler(nil),
 		handlers.NewWebhooksHandler(nil),
 		handlers.NewTenantDataHandler(nil),
+		handlers.NewTenantSettingsHandler(nil),
 		handlers.NewSearchHandler(nil),
 		handlers.NewTaxonomyHandler(nil),
 		handlers.NewTaxonomyInternalHandler(),

cmd/backfill-embeddings/main.go

@@ -110,6 +110,7 @@ func run() int {
 		nil, // inserter set below after River client is created
 		service.EmbeddingsQueueName,
 		maxAttempts,
+		"", // translation default unused: embeddings backfill only
 	)
 
 	embeddingClient, err := service.NewEmbeddingClient(ctx, embeddingCfg)

cmd/backfill-translations/main.go

@@ -0,0 +1,131 @@
+// backfill-translations enqueues River translation jobs for feedback records whose
+// tenant has a target language configured and whose value_text is not yet translated
+// to it (missing or stale). Run this one-off after enabling translation or changing a
+// tenant's target language; hub-worker (or the API process) runs the jobs.
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/riverdriver/riverpgxv5"
+
+	"github.com/formbricks/hub/internal/config"
+	"github.com/formbricks/hub/internal/repository"
+	"github.com/formbricks/hub/internal/service"
+	"github.com/formbricks/hub/internal/workers"
+	"github.com/formbricks/hub/pkg/database"
+)
+
+var (
+	errTranslationProviderRequired = errors.New("TRANSLATION_PROVIDER is required")
+	errTranslationModelRequired    = errors.New("TRANSLATION_MODEL is required")
+)
+
+const (
+	defaultTranslationMaxAttempts = 3
+	exitSuccess                   = 0
+	exitFailure                   = 1
+)
+
+func main() {
+	os.Exit(run())
+}
+
+func run() int {
+	cfg, err := config.Load()
+	if err != nil {
+		slog.Error("Failed to load configuration", "error", err)
+
+		return exitFailure
+	}
+
+	if cfg.Database.URL == "" || cfg.Database.URL == config.DefaultDatabaseURL {
+		slog.Error("DATABASE_URL must be set explicitly for this binary (do not use the default test URL)")
+
+		return exitFailure
+	}
+
+	if cfg.Translation.Provider == "" {
+		slog.Error(errTranslationProviderRequired.Error())
+
+		return exitFailure
+	}
+
+	if cfg.Translation.Model == "" {
+		slog.Error(errTranslationModelRequired.Error())
+
+		return exitFailure
+	}
+
+	maxAttempts := cfg.Translation.MaxAttempts
+	if maxAttempts <= 0 {
+		maxAttempts = defaultTranslationMaxAttempts
+	}
+
+	ctx := context.Background()
+
+	db, err := database.NewPostgresPool(ctx, cfg.Database.URL,
+		database.WithPoolConfig(cfg.Database.PoolConfig()),
+	)
+	if err != nil {
+		slog.Error("Failed to connect to database", "error", err)
+
+		return exitFailure
+	}
+	defer db.Close()
+
+	translationCfg := service.TranslationClientConfig{
+		Provider:            cfg.Translation.Provider,
+		ProviderAPIKey:      cfg.Translation.ProviderAPIKey,
+		Model:               cfg.Translation.Model,
+		BaseURL:             cfg.Translation.BaseURL,
+		GoogleCloudProject:  cfg.Translation.GoogleCloudProject,
+		GoogleCloudLocation: cfg.Translation.GoogleCloudLocation,
+	}
+
+	translationClient, err := service.NewTranslationClient(ctx, translationCfg)
+	if err != nil {
+		slog.Error("Failed to create translation client", "error", err)
+
+		return exitFailure
+	}
+
+	repo := repository.NewFeedbackRecordsRepository(db)
+	feedbackRecordsService := service.NewFeedbackRecordsService(repo, nil, "", nil, nil, "", 0, cfg.Translation.DefaultLanguage)
+
+	// Producer-only: we only enqueue jobs; workers run in hub-worker (or the API process).
+	// River requires the job kind registered (worker added) and MaxWorkers > 0 for a declared queue.
+	riverWorkers := river.NewWorkers()
+	river.AddWorker(riverWorkers, workers.NewFeedbackTranslationWorker(feedbackRecordsService, translationClient, nil))
+
+	riverClient, err := river.NewClient(riverpgxv5.New(db), &river.Config{
+		Queues: map[string]river.QueueConfig{
+			service.TranslationsQueueName: {MaxWorkers: 1},
+		},
+		Workers: riverWorkers,
+	})
+	if err != nil {
+		slog.Error("Failed to create River client", "error", err)
+
+		return exitFailure
+	}
+
+	enqueued, err := feedbackRecordsService.BackfillTranslations(
+		ctx, riverClient, service.TranslationsQueueName, maxAttempts)
+	if err != nil {
+		slog.Error("Backfill failed", "error", err)
+
+		return exitFailure
+	}
+
+	slog.Info("Backfill complete", "enqueued", enqueued)
+
+	fmt.Printf("Enqueued %d translation job(s).\n", enqueued)
+
+	return exitSuccess
+}

cmd/worker/app.go

@@ -68,13 +68,15 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) {
 	webhooksRepo := repository.NewWebhooksRepository(db)
 
 	var (
-		webhookMetrics   observability.WebhookMetrics
-		embeddingMetrics observability.EmbeddingMetrics
+		webhookMetrics     observability.WebhookMetrics
+		embeddingMetrics   observability.EmbeddingMetrics
+		translationMetrics observability.TranslationMetrics
 	)
 
 	if metrics != nil {
 		webhookMetrics = metrics.Webhooks
 		embeddingMetrics = metrics.Embeddings
+		translationMetrics = metrics.Translation
 	}
 
 	webhookSender := service.NewWebhookSenderImpl(
@@ -121,6 +123,7 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) {
 			nil,
 			service.EmbeddingsQueueName,
 			cfg.Embedding.MaxAttempts,
+			"", // translation default unused: this service handles embeddings only
 		)
 		docPrefix := service.EmbeddingPrefixForProvider(providerName)
 
@@ -130,6 +133,36 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) {
 		deps.EmbeddingMetrics = embeddingMetrics
 	}
 
+	if cfg.Translation.Provider != "" && cfg.Translation.Model != "" {
+		translationCfg := service.TranslationClientConfig{
+			Provider:            cfg.Translation.Provider,
+			ProviderAPIKey:      cfg.Translation.ProviderAPIKey,
+			Model:               cfg.Translation.Model,
+			BaseURL:             cfg.Translation.BaseURL,
+			GoogleCloudProject:  cfg.Translation.GoogleCloudProject,
+			GoogleCloudLocation: cfg.Translation.GoogleCloudLocation,
+		}
+
+		translationClient, err := service.NewTranslationClient(context.Background(), translationCfg)
+		if err != nil {
+			shutdownObservability(context.Background(), meterProvider, tracerProvider)
+
+			return nil, fmt.Errorf("translation config: %w", err)
+		}
+
+		// The translation worker only reads the record and writes the translation, so
+		// the embedding-specific service params are unused here.
+		translationRecordsRepo := repository.NewFeedbackRecordsRepository(db)
+		translationRecordsService := service.NewFeedbackRecordsService(
+			translationRecordsRepo, nil, "", nil, nil, "", 0, cfg.Translation.DefaultLanguage)
+
+		deps.TranslationService = translationRecordsService
+		deps.TranslationClient = translationClient
+		deps.TranslationMetrics = translationMetrics
+		deps.TranslationBackfillService = translationRecordsService
+		deps.TranslationMaxAttempts = cfg.Translation.MaxAttempts
+	}
+
 	riverWorkers, queues := workers.NewRiverWorkersAndQueues(cfg, deps, 0)
 
 	riverCfg := &river.Config{