Skip to content

Commit 4b0575d

Browse files
xernobylxernobyl
authored
Merge pull request #90 from formbricks/feat/ENG-1255_translate-feedback-records
feat: enrich feedback records with translated text
2 parents 2e9d171 + 9227920 commit 4b0575d

49 files changed

Lines changed: 4474 additions & 118 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.env.example‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,24 @@ WEBHOOK_MAX_COUNT=500
108108
# GOOGLE_APPLICATION_CREDENTIALS= (optional; for google-gemini when outside Google Cloud: path to service account key JSON)
109109
# EMBEDDING_MODEL=<model name> (required to enable embeddings; no default)
110110

111+
# 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.
112+
# Open-text feedback (value_text) is translated into each tenant's configured target_language (Hub tenant settings). Same providers/auth model as embeddings.
113+
# Configure identically in the API and worker processes (the API enqueues, the worker translates); otherwise jobs pile up unprocessed.
114+
# TRANSLATION_PROVIDER=openai
115+
# TRANSLATION_PROVIDER=google-gemini
116+
# TRANSLATION_PROVIDER_API_KEY=sk-... (required for openai and google; not used for google-gemini)
117+
# TRANSLATION_BASE_URL=https://llm.example.com/v1 (optional; only supported with TRANSLATION_PROVIDER=openai)
118+
# TRANSLATION_GOOGLE_CLOUD_PROJECT= (required for google-gemini; or use GOOGLE_CLOUD_PROJECT)
119+
# TRANSLATION_GOOGLE_CLOUD_LOCATION= (required for google-gemini, e.g. europe-west3; or use GOOGLE_CLOUD_LOCATION)
120+
# TRANSLATION_MODEL=<model name> (required to enable translation; no default; e.g. gemini-2.5-flash)
121+
# TRANSLATION_MAX_CONCURRENT=5 (worker concurrency; default 5)
122+
# TRANSLATION_MAX_ATTEMPTS=3 (River job retries before failing; default 3)
123+
124+
# Tenant settings cache (optional): an in-process LRU+TTL over tenant target_language reads on the translation enqueue path.
125+
# Set TENANT_SETTINGS_CACHE_SIZE=0 to disable the cache. Defaults: size 2048, TTL 60s.
126+
# TENANT_SETTINGS_CACHE_SIZE=2048
127+
# TENANT_SETTINGS_CACHE_TTL_SECONDS=60
128+
111129
# Tenant data purge (optional). How long DELETE /v1/tenants/{tenant_id}/data waits for in-flight
112130
# tenant-owned writes to drain before returning a retryable 409 (code tenant_write_conflict).
113131
# Must be a positive integer (seconds); non-positive values fall back to the default. Default: 5

‎cmd/api/app.go‎

Lines changed: 62 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -174,14 +174,16 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
174174
}
175175

176176
var (
177-
eventMetrics observability.EventMetrics
178-
webhookMetrics observability.WebhookMetrics
179-
embeddingMetrics observability.EmbeddingMetrics
177+
eventMetrics observability.EventMetrics
178+
webhookMetrics observability.WebhookMetrics
179+
embeddingMetrics observability.EmbeddingMetrics
180+
translationMetrics observability.TranslationMetrics
180181
)
181182
if metrics != nil {
182183
eventMetrics = metrics.Events
183184
webhookMetrics = metrics.Webhooks
184185
embeddingMetrics = metrics.Embeddings
186+
translationMetrics = metrics.Translation
185187
}
186188

187189
var tracerProvider *sdktrace.TracerProvider
@@ -277,6 +279,40 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
277279
searchHandler = handlers.NewSearchHandler(nil) // 503 when embeddings disabled
278280
}
279281

282+
// Register the translation worker and declare its queue so the River client can
283+
// enqueue translation jobs (River requires the job kind registered and the queue
284+
// declared at insert time); the jobs are processed by hub-worker, not in this
285+
// process. Gated on TRANSLATION_PROVIDER+MODEL like embeddings; the enqueue
286+
// provider is registered below, after the River client and tenant settings exist.
287+
if cfg.Translation.Provider != "" && cfg.Translation.Model != "" {
288+
translationCfg := service.TranslationClientConfig{
289+
Provider: cfg.Translation.Provider,
290+
ProviderAPIKey: cfg.Translation.ProviderAPIKey,
291+
Model: cfg.Translation.Model,
292+
BaseURL: cfg.Translation.BaseURL,
293+
GoogleCloudProject: cfg.Translation.GoogleCloudProject,
294+
GoogleCloudLocation: cfg.Translation.GoogleCloudLocation,
295+
}
296+
297+
translationClient, translationErr := service.NewTranslationClient(context.Background(), translationCfg)
298+
if translationErr != nil {
299+
cleanupNewAppStartupFailure(context.Background(), messageManager, nil, tracerProvider, meterProvider)
300+
301+
return nil, fmt.Errorf("translation config: %w", translationErr)
302+
}
303+
304+
river.AddWorker(riverWorkers, workers.NewFeedbackTranslationWorker(feedbackRecordsService, translationClient, translationMetrics))
305+
306+
queues[service.TranslationsQueueName] = river.QueueConfig{MaxWorkers: 1}
307+
308+
// Per-tenant re-translation backfill, enqueued by the settings-change listener
309+
// below. Registered here only so the River client can validate the kind and queue
310+
// at insert time; the fan-out is processed by hub-worker.
311+
river.AddWorker(riverWorkers, workers.NewTenantTranslationBackfillWorker(feedbackRecordsService, cfg.Translation.MaxAttempts))
312+
313+
queues[service.TranslationBackfillsQueueName] = river.QueueConfig{MaxWorkers: 1}
314+
}
315+
280316
riverClient, err := river.NewClient(riverpgxv5.New(db), &river.Config{
281317
Queues: queues,
282318
Workers: riverWorkers,
@@ -325,6 +361,29 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
325361
tenantSettingsService := service.NewTenantSettingsService(tenantSettingsRepo)
326362
tenantSettingsHandler := handlers.NewTenantSettingsHandler(tenantSettingsService)
327363

364+
// Translation enqueue provider: on a feedback-record create/update it resolves the
365+
// tenant's target language (through a short-TTL cache over tenant settings) and
366+
// enqueues a translation job. Gated on TRANSLATION_PROVIDER+MODEL.
367+
if cfg.Translation.Provider != "" && cfg.Translation.Model != "" {
368+
var translationCacheMetrics observability.CacheMetrics
369+
if metrics != nil {
370+
translationCacheMetrics = metrics.Cache
371+
}
372+
373+
translationCache := service.NewCachedTenantSettings(
374+
tenantSettingsService,
375+
cfg.TenantSettingsCache.Size, cfg.TenantSettingsCache.TTL.Duration(),
376+
translationCacheMetrics,
377+
)
378+
messageManager.RegisterProvider(service.NewTranslationProvider(
379+
riverClient, translationCache, service.TranslationsQueueName, cfg.Translation.MaxAttempts, translationMetrics))
380+
381+
// On a target_language change, enqueue a per-tenant re-translation backfill so
382+
// existing records pick up the new target (not only newly ingested ones).
383+
tenantSettingsService.SetSettingsChangeListener(service.NewTranslationSettingsListener(
384+
riverClient, service.TranslationBackfillsQueueName, cfg.Translation.MaxAttempts))
385+
}
386+
328387
taxonomyRepo := repository.NewTaxonomyRepository(db)
329388

330389
var taxonomyStarter service.TaxonomyRunStarter

‎cmd/backfill-translations/main.go‎

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,131 @@
1+
// backfill-translations enqueues River translation jobs for feedback records whose
2+
// tenant has a target language configured and whose value_text is not yet translated
3+
// to it (missing or stale). Run this one-off after enabling translation or changing a
4+
// tenant's target language; hub-worker (or the API process) runs the jobs.
5+
package main
6+
7+
import (
8+
"context"
9+
"errors"
10+
"fmt"
11+
"log/slog"
12+
"os"
13+
14+
"github.com/riverqueue/river"
15+
"github.com/riverqueue/river/riverdriver/riverpgxv5"
16+
17+
"github.com/formbricks/hub/internal/config"
18+
"github.com/formbricks/hub/internal/repository"
19+
"github.com/formbricks/hub/internal/service"
20+
"github.com/formbricks/hub/internal/workers"
21+
"github.com/formbricks/hub/pkg/database"
22+
)
23+
24+
var (
25+
errTranslationProviderRequired = errors.New("TRANSLATION_PROVIDER is required")
26+
errTranslationModelRequired = errors.New("TRANSLATION_MODEL is required")
27+
)
28+
29+
const (
30+
defaultTranslationMaxAttempts = 3
31+
exitSuccess = 0
32+
exitFailure = 1
33+
)
34+
35+
func main() {
36+
os.Exit(run())
37+
}
38+
39+
func run() int {
40+
cfg, err := config.Load()
41+
if err != nil {
42+
slog.Error("Failed to load configuration", "error", err)
43+
44+
return exitFailure
45+
}
46+
47+
if cfg.Database.URL == "" || cfg.Database.URL == config.DefaultDatabaseURL {
48+
slog.Error("DATABASE_URL must be set explicitly for this binary (do not use the default test URL)")
49+
50+
return exitFailure
51+
}
52+
53+
if cfg.Translation.Provider == "" {
54+
slog.Error(errTranslationProviderRequired.Error())
55+
56+
return exitFailure
57+
}
58+
59+
if cfg.Translation.Model == "" {
60+
slog.Error(errTranslationModelRequired.Error())
61+
62+
return exitFailure
63+
}
64+
65+
maxAttempts := cfg.Translation.MaxAttempts
66+
if maxAttempts <= 0 {
67+
maxAttempts = defaultTranslationMaxAttempts
68+
}
69+
70+
ctx := context.Background()
71+
72+
db, err := database.NewPostgresPool(ctx, cfg.Database.URL,
73+
database.WithPoolConfig(cfg.Database.PoolConfig()),
74+
)
75+
if err != nil {
76+
slog.Error("Failed to connect to database", "error", err)
77+
78+
return exitFailure
79+
}
80+
defer db.Close()
81+
82+
translationCfg := service.TranslationClientConfig{
83+
Provider: cfg.Translation.Provider,
84+
ProviderAPIKey: cfg.Translation.ProviderAPIKey,
85+
Model: cfg.Translation.Model,
86+
BaseURL: cfg.Translation.BaseURL,
87+
GoogleCloudProject: cfg.Translation.GoogleCloudProject,
88+
GoogleCloudLocation: cfg.Translation.GoogleCloudLocation,
89+
}
90+
91+
translationClient, err := service.NewTranslationClient(ctx, translationCfg)
92+
if err != nil {
93+
slog.Error("Failed to create translation client", "error", err)
94+
95+
return exitFailure
96+
}
97+
98+
repo := repository.NewFeedbackRecordsRepository(db)
99+
feedbackRecordsService := service.NewFeedbackRecordsService(repo, nil, "", nil, nil, "", 0)
100+
101+
// Producer-only: we only enqueue jobs; workers run in hub-worker (or the API process).
102+
// River requires the job kind registered (worker added) and MaxWorkers > 0 for a declared queue.
103+
riverWorkers := river.NewWorkers()
104+
river.AddWorker(riverWorkers, workers.NewFeedbackTranslationWorker(feedbackRecordsService, translationClient, nil))
105+
106+
riverClient, err := river.NewClient(riverpgxv5.New(db), &river.Config{
107+
Queues: map[string]river.QueueConfig{
108+
service.TranslationsQueueName: {MaxWorkers: 1},
109+
},
110+
Workers: riverWorkers,
111+
})
112+
if err != nil {
113+
slog.Error("Failed to create River client", "error", err)
114+
115+
return exitFailure
116+
}
117+
118+
enqueued, err := feedbackRecordsService.BackfillTranslations(
119+
ctx, riverClient, service.TranslationsQueueName, maxAttempts)
120+
if err != nil {
121+
slog.Error("Backfill failed", "error", err)
122+
123+
return exitFailure
124+
}
125+
126+
slog.Info("Backfill complete", "enqueued", enqueued)
127+
128+
fmt.Printf("Enqueued %d translation job(s).\n", enqueued)
129+
130+
return exitSuccess
131+
}

‎cmd/worker/app.go‎

Lines changed: 34 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,13 +68,15 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) {
6868
webhooksRepo := repository.NewWebhooksRepository(db)
6969

7070
var (
71-
webhookMetrics observability.WebhookMetrics
72-
embeddingMetrics observability.EmbeddingMetrics
71+
webhookMetrics observability.WebhookMetrics
72+
embeddingMetrics observability.EmbeddingMetrics
73+
translationMetrics observability.TranslationMetrics
7374
)
7475

7576
if metrics != nil {
7677
webhookMetrics = metrics.Webhooks
7778
embeddingMetrics = metrics.Embeddings
79+
translationMetrics = metrics.Translation
7880
}
7981

8082
webhookSender := service.NewWebhookSenderImpl(
@@ -130,6 +132,36 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) {
130132
deps.EmbeddingMetrics = embeddingMetrics
131133
}
132134

135+
if cfg.Translation.Provider != "" && cfg.Translation.Model != "" {
136+
translationCfg := service.TranslationClientConfig{
137+
Provider: cfg.Translation.Provider,
138+
ProviderAPIKey: cfg.Translation.ProviderAPIKey,
139+
Model: cfg.Translation.Model,
140+
BaseURL: cfg.Translation.BaseURL,
141+
GoogleCloudProject: cfg.Translation.GoogleCloudProject,
142+
GoogleCloudLocation: cfg.Translation.GoogleCloudLocation,
143+
}
144+
145+
translationClient, err := service.NewTranslationClient(context.Background(), translationCfg)
146+
if err != nil {
147+
shutdownObservability(context.Background(), meterProvider, tracerProvider)
148+
149+
return nil, fmt.Errorf("translation config: %w", err)
150+
}
151+
152+
// The translation worker only reads the record and writes the translation, so
153+
// the embedding-specific service params are unused here.
154+
translationRecordsRepo := repository.NewFeedbackRecordsRepository(db)
155+
translationRecordsService := service.NewFeedbackRecordsService(
156+
translationRecordsRepo, nil, "", nil, nil, "", 0)
157+
158+
deps.TranslationService = translationRecordsService
159+
deps.TranslationClient = translationClient
160+
deps.TranslationMetrics = translationMetrics
161+
deps.TranslationBackfillService = translationRecordsService
162+
deps.TranslationMaxAttempts = cfg.Translation.MaxAttempts
163+
}
164+
133165
riverWorkers, queues := workers.NewRiverWorkersAndQueues(cfg, deps, 0)
134166

135167
riverCfg := &river.Config{

0 commit comments

Comments
 (0)