feat: serialize tenant-owned writes against tenant data purges

.env.example

@@ -108,6 +108,11 @@ 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)
 
+# 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
+# TENANT_PURGE_LOCK_TIMEOUT_SECONDS=5
+
 # Local River UI basic auth (optional, used by docker compose). Change these for your local setup as needed.
 # compose.yml defaults to admin / changeme if these are unset.
 RIVER_BASIC_AUTH_USER=admin

AGENTS.md

@@ -36,6 +36,7 @@
 - Async jobs must carry the tenant boundary when the source data has one, and workers must re-check tenant scope before doing side effects. Do not rely only on enqueue-time filtering.
 - Global resources may intentionally have `tenant_id = NULL` only when the domain explicitly documents them as non-tenant-owned. Webhooks are tenant-owned and must require a non-empty `tenant_id`; a missing tenant on event data must not match any webhook.
 - When changing access rules for a tenant-owned model, search for all alternate access paths by resource name and by derived side effects: list, get, update, delete, bulk delete, webhook fan-out, River jobs, workers, embeddings, search, exports, cache invalidation, logs, and metrics.
+- Every tenant-owned mutation must run through the tenant write lock helper (`internal/repository/tenant_write_lock.go`): a transaction that first try-acquires the shared advisory lock on `tenant_write|<len>:<tenant_id>` (resolve the row's tenant inside the transaction for ID-based mutations, then mutate with a tenant-scoped WHERE). The tenant data purge holds the same key exclusively. Lock order is fixed: tenant shared lock → scope-specific advisory locks (e.g. taxonomy run scope) → row locks; never block on an advisory lock while holding row locks. A path that skips the helper silently reopens the purge/write race.
 - For any tenant-scoping change, include verification at the boundary where the leak could happen: database query behavior, service fan-out, worker execution, and API behavior when relevant. Include at least one alternate-path regression test proving that data allowed through the primary path cannot leak through async dispatch, bulk operations, derived indexes, exports, or background workers. Tests are the evidence; the invariant belongs in the architecture.
 
 ## Testing Guidelines

cmd/api/app.go

@@ -229,7 +229,7 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
 
 	feedbackRecordsRepo := repository.NewFeedbackRecordsRepository(db)
 	embeddingsRepo := repository.NewEmbeddingsRepository(db)
-	tenantDataRepo := repository.NewTenantDataRepository(db)
+	tenantDataRepo := repository.NewTenantDataRepository(db, cfg.TenantData.PurgeLockTimeout.Duration())
 	embeddingProviderName, embeddingModel := embeddingProviderAndModel(cfg)
 	embeddingModelForDB := embeddingModel
 

internal/api/response/errors.go

@@ -83,6 +83,15 @@ func problemFromError(err error) ProblemDetails {
 		return problem
 	}
 
+	var tenantWriteConflictErr *huberrors.TenantWriteConflictError
+	if errors.As(err, &tenantWriteConflictErr) {
+		problem := newProblem(http.StatusConflict, tenantWriteConflictErr.Error())
+		problem.Type = ProblemTypeTenantWriteConflict
+		problem.Code = CodeTenantWriteConflict
+
+		return problem
+	}
+
 	var conflictErr *huberrors.ConflictError
 	if errors.As(err, &conflictErr) {
 		return newProblem(http.StatusConflict, conflictErr.Error())

internal/api/response/problem.go

@@ -14,6 +14,7 @@ const (
 	ProblemTypeForbidden           = "https://hub.formbricks.com/problems/forbidden"
 	ProblemTypeNotFound            = "https://hub.formbricks.com/problems/not-found"
 	ProblemTypeConflict            = "https://hub.formbricks.com/problems/conflict"
+	ProblemTypeTenantWriteConflict = "https://hub.formbricks.com/problems/tenant-write-conflict"
 	ProblemTypeMethodNotAllowed    = "https://hub.formbricks.com/problems/method-not-allowed"
 	ProblemTypeServiceUnavailable  = "https://hub.formbricks.com/problems/service-unavailable"
 	ProblemTypeInternalServerError = "https://hub.formbricks.com/problems/internal-server-error"
@@ -31,6 +32,7 @@ const (
 	CodeForbidden           = "forbidden"
 	CodeNotFound            = "not_found"
 	CodeConflict            = "conflict"
+	CodeTenantWriteConflict = "tenant_write_conflict"
 	CodeMethodNotAllowed    = "method_not_allowed"
 	CodeServiceUnavailable  = "service_unavailable"
 	CodeInternalServerError = "internal_server_error"

internal/api/response/response_test.go

@@ -66,6 +66,15 @@ func TestRespondErrorMapping(t *testing.T) {
 			name: "conflict", err: huberrors.NewConflictError("already exists"),
 			wantStatus: http.StatusConflict, wantCode: CodeConflict, wantType: ProblemTypeConflict,
 		},
+		{
+			name: "tenant write conflict", err: huberrors.NewTenantWriteConflictError("tenant data purge in progress; retry later"),
+			wantStatus: http.StatusConflict, wantCode: CodeTenantWriteConflict, wantType: ProblemTypeTenantWriteConflict,
+		},
+		{
+			name:       "wrapped tenant write conflict keeps dedicated code",
+			err:        fmt.Errorf("delete feedback record: %w", huberrors.NewTenantWriteConflictError("")),
+			wantStatus: http.StatusConflict, wantCode: CodeTenantWriteConflict, wantType: ProblemTypeTenantWriteConflict,
+		},
 		{
 			name: "limit exceeded", err: huberrors.NewLimitExceededError("webhook limit reached"),
 			wantStatus: http.StatusForbidden, wantCode: CodeForbidden, wantType: ProblemTypeForbidden,

internal/config/config.go

@@ -46,6 +46,7 @@ type Config struct {
 	MessagePublisher MessagePublisherConfig
 	Embedding        EmbeddingConfig
 	Taxonomy         TaxonomyConfig
+	TenantData       TenantDataConfig
 	Observability    ObservabilityConfig
 }
 
@@ -134,6 +135,14 @@ type TaxonomyConfig struct {
 	MinimumEmbeddedRecords int    `env:"TAXONOMY_MIN_EMBEDDED_RECORDS" env-default:"20"`
 }
 
+// TenantDataConfig holds tenant data purge settings.
+type TenantDataConfig struct {
+	// PurgeLockTimeout bounds how long a tenant data purge waits for in-flight
+	// tenant-owned writes to drain before giving up with a retryable conflict.
+	// Must be positive: 0 would mean "wait forever" at the database level.
+	PurgeLockTimeout DurationSec `env:"TENANT_PURGE_LOCK_TIMEOUT_SECONDS" env-default:"5"`
+}
+
 // ObservabilityConfig holds OpenTelemetry settings.
 type ObservabilityConfig struct {
 	MetricsExporter string `env:"OTEL_METRICS_EXPORTER"`
@@ -283,6 +292,11 @@ func applyDefaults(cfg *Config) {
 	if cfg.Taxonomy.MinimumEmbeddedRecords <= 0 {
 		cfg.Taxonomy.MinimumEmbeddedRecords = 20
 	}
+
+	const defaultPurgeLockTimeoutSec = 5
+	if cfg.TenantData.PurgeLockTimeout.Duration() <= 0 {
+		cfg.TenantData.PurgeLockTimeout = DurationSec(time.Duration(defaultPurgeLockTimeoutSec) * time.Second)
+	}
 }
 
 func isNotExist(err error) bool {

internal/config/config_test.go

@@ -346,6 +346,37 @@ func TestLoad_TaxonomyConfig(t *testing.T) {
 	}
 }
 
+func TestLoad_TenantDataPurgeLockTimeout(t *testing.T) {
+	tests := []struct {
+		name  string
+		value string
+		set   bool
+		want  time.Duration
+	}{
+		{name: "defaults to 5 seconds", set: false, want: 5 * time.Second},
+		{name: "honors env override", value: "9", set: true, want: 9 * time.Second},
+		{name: "coerces zero to default", value: "0", set: true, want: 5 * time.Second},
+		{name: "coerces negative to default", value: "-3", set: true, want: 5 * time.Second},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if tt.set {
+				t.Setenv("TENANT_PURGE_LOCK_TIMEOUT_SECONDS", tt.value)
+			}
+
+			cfg, err := Load()
+			if err != nil {
+				t.Fatalf("Load() error = %v", err)
+			}
+
+			if got := cfg.TenantData.PurgeLockTimeout.Duration(); got != tt.want {
+				t.Errorf("TenantData.PurgeLockTimeout = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 func TestLoad_TaxonomyServiceURLValidation(t *testing.T) {
 	tests := []struct {
 		name  string

internal/huberrors/errors.go

@@ -135,3 +135,37 @@ func (e *ConflictError) Is(target error) bool {
 
 	return ok
 }
+
+// ErrTenantWriteConflict is the sentinel for tenant write coordination conflicts:
+// a tenant-owned write rejected because a tenant data purge is in progress, or a
+// purge that could not acquire its lock while tenant-owned writes were in flight.
+// Both directions are retryable.
+var ErrTenantWriteConflict = &TenantWriteConflictError{}
+
+// TenantWriteConflictError is a sentinel error for tenant write coordination conflicts.
+// It is deliberately distinct from ConflictError so retryable lock conflicts are never
+// confused with terminal resource conflicts (e.g. duplicate submissions).
+type TenantWriteConflictError struct {
+	Message string
+}
+
+// NewTenantWriteConflictError creates a TenantWriteConflictError with a custom message.
+func NewTenantWriteConflictError(message string) *TenantWriteConflictError {
+	return &TenantWriteConflictError{Message: message}
+}
+
+// Error implements the error interface.
+func (e *TenantWriteConflictError) Error() string {
+	if e.Message != "" {
+		return e.Message
+	}
+
+	return "tenant write conflict"
+}
+
+// Is implements the error interface for error comparison.
+func (e *TenantWriteConflictError) Is(target error) bool {
+	_, ok := target.(*TenantWriteConflictError)
+
+	return ok
+}

internal/repository/embeddings_repository.go

@@ -51,33 +51,48 @@ func (r *EmbeddingsRepository) Upsert(
 	vec := pgvector.NewHalfVector(embedding)
 	now := time.Now()
 
-	_, err := r.db.Exec(ctx, `
-		INSERT INTO embeddings (feedback_record_id, embedding, model, created_at, updated_at)
-		VALUES ($1, $2, $3, $4, $5)
-		ON CONFLICT (feedback_record_id, model)
-		DO UPDATE SET embedding = EXCLUDED.embedding, updated_at = $5`,
-		feedbackRecordID, vec, model, now, now,
-	)
-	if err != nil {
-		return fmt.Errorf("embeddings upsert: %w", err)
-	}
+	return withTenantWritePoolTx(ctx, r.db, nil, func(dbTx tenantWriteTx) error {
+		// Embeddings derive their tenant boundary from the parent feedback
+		// record; resolve and lock it so embedding writes cannot race a tenant
+		// data purge. A missing parent means the record was deleted or purged.
+		if _, err := lockFeedbackRecordTenantShared(ctx, dbTx, feedbackRecordID); err != nil {
+			return err
+		}
 
-	return nil
+		_, err := dbTx.Exec(ctx, `
+			INSERT INTO embeddings (feedback_record_id, embedding, model, created_at, updated_at)
+			VALUES ($1, $2, $3, $4, $5)
+			ON CONFLICT (feedback_record_id, model)
+			DO UPDATE SET embedding = EXCLUDED.embedding, updated_at = $5`,
+			feedbackRecordID, vec, model, now, now,
+		)
+		if err != nil {
+			return fmt.Errorf("embeddings upsert: %w", err)
+		}
+
+		return nil
+	})
 }
 
 // DeleteByFeedbackRecordAndModel removes the embedding row for the given feedback record and model.
 func (r *EmbeddingsRepository) DeleteByFeedbackRecordAndModel(
 	ctx context.Context, feedbackRecordID uuid.UUID, model string,
 ) error {
-	_, err := r.db.Exec(ctx,
-		`DELETE FROM embeddings WHERE feedback_record_id = $1 AND model = $2`,
-		feedbackRecordID, model,
-	)
-	if err != nil {
-		return fmt.Errorf("embeddings delete: %w", err)
-	}
+	return withTenantWritePoolTx(ctx, r.db, nil, func(dbTx tenantWriteTx) error {
+		if _, err := lockFeedbackRecordTenantShared(ctx, dbTx, feedbackRecordID); err != nil {
+			return err
+		}
+
+		_, err := dbTx.Exec(ctx,
+			`DELETE FROM embeddings WHERE feedback_record_id = $1 AND model = $2`,
+			feedbackRecordID, model,
+		)
+		if err != nil {
+			return fmt.Errorf("embeddings delete: %w", err)
+		}
 
-	return nil
+		return nil
+	})
 }
 
 // ListFeedbackRecordIDsForBackfill returns IDs of feedback records that have non-empty value_text