feat: enrich feedback records with translated text

.env.example

@@ -108,6 +108,24 @@ 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). 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_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
@@ -277,6 +279,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,
@@ -325,6 +361,29 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
 	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, translationMetrics))
+
+		// On a target_language change, enqueue a per-tenant re-translation backfill so
+		// existing records pick up the new target (not only newly ingested ones).
+		tenantSettingsService.SetSettingsChangeListener(service.NewTranslationSettingsListener(
+			riverClient, service.TranslationBackfillsQueueName, cfg.Translation.MaxAttempts))
+	}
+
 	taxonomyRepo := repository.NewTaxonomyRepository(db)
 
 	var taxonomyStarter service.TaxonomyRunStarter

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)
+
+	// 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(
@@ -130,6 +132,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)
+
+		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{