feat(dashboard-api): supabase auth users sync background runner (#2247)

* feat: supabase auth users sync background runner in dashboard api

* chore: auto-commit generated changes

* feat(db): enhance auth user sync triggers to retain direct operations while enqueuing for processing

* feat(sync): implement user sync queue with enhanced error handling and recovery mechanisms

- Updated the sync runner to use `RunWithRestart` for improved error recovery.
- Introduced a new `UserSyncQueue` model to manage user synchronization tasks.
- Added SQL migration for creating the `user_sync_queue` table with necessary triggers.
- Implemented tests for the processor and supervisor to ensure robust handling of retries and panics.
- Refactored existing queries to target the new `public.user_sync_queue` table.

* feat(sync): add supabase auth user sync configuration and secrets management

- Introduced `supabase_auth_user_sync_enabled` variable to control user synchronization.
- Updated Nomad job configuration to include the new sync setting.
- Added Google Secret Manager resources for managing the sync configuration securely.
- Enhanced the dashboard API to utilize the new sync configuration in processing logic.
- Refactored related components to improve error handling and logging for the sync process.

* chore: remove smoke test

* add: e2e runner test

* chore: change dashboard-api env variable management

* refactor(sync): update user sync logic to utilize new database structure

- Replaced the previous `Store` implementation with a new structure that integrates both authentication and main database queries.
- Updated the `Runner` and `NewRunner` functions to accommodate the new database client structure.
- Removed obsolete SQL queries and migration files related to the `user_sync_queue` table.
- Enhanced the test suite to reflect changes in the runner's initialization and database interactions.

* fix: lint

* test: apply database migrations in end-to-end test setup

- Updated the `TestSupabaseAuthUserSyncRunner_EndToEnd` to apply necessary database migrations before running tests.
- Refactored the `SetupDatabase` function to include a new method `ApplyMigrations` for better migration management.

* feat(sync): enhance user sync processing and acknowledgment

- Introduced a new process outcome `ready_to_ack` to streamline acknowledgment handling.
- Refactored the `process` method to prepare for batch acknowledgment of processed items.
- Added a new `AckBatch` method in the store to handle multiple acknowledgments efficiently.
- Updated the `Runner` to process items in batches and finalize acknowledgments accordingly.
- Removed obsolete SQL query for single item acknowledgment as part of the refactor.
- Enhanced tests to cover new deletion logic and acknowledgment scenarios.

* refactor(gcp): remove auth_db_connection_string resources and update references

- Deleted the `auth_db_connection_string` secret and its version from the GCP configuration.
- Updated references in `main.tf` and `nomad/main.tf` to use the `postgres_connection_string` instead.
- Removed the corresponding variable declaration from `variables.tf` to clean up unused configurations.

* chore(dashboard-api): refactor environment variable management

- Updated the dashboard API module to separate base and extra environment variables.
- Introduced a precondition to prevent conflicts with reserved keys in extra environment variables.
- Modified the HCL job configuration to iterate over environment variables dynamically.
- Adjusted variable declarations to reflect the new structure for extra environment variables.

* refactor: use river for queue worker

* chore: auto-commit generated changes

* chore: auto-commit generated changes

* chore: update pgx dependency to v5.9.1 and golang.org/x/mod to v0.34.0 across multiple packages

* chore: fix lint

* chore: auto-commit generated changes

* improve: retrying for river jobs

* refactor: enhance auth user sync worker and streamline environment variable handling

- Updated the AuthUserSyncWorker to utilize OpenTelemetry metrics for better monitoring.
- Refactored the Makefile to improve environment variable exportation and streamline build/run commands.
- Removed outdated SQL query files related to user sync queue operations to clean up the codebase.
- Adjusted the database migration script to drop the auth_custom schema if it exists, ensuring a cleaner migration process.

* chore: rename config and fix test

* improve: config

* chore: auto-commit generated changes

* fix: correct env var name in template to match Go config

Use AUTH_USER_SYNC_BACKGROUND_WORKER_ENABLED instead of
SUPABASE_AUTH_USER_SYNC_ENABLED to match the actual config
field in packages/dashboard-api/internal/cfg/model.go

Co-authored-by: Ben Fornefeld <ben-fornefeld@users.noreply.github.com>

* fix: update permissions for trigger_user on river_job and sequences in auth_custom schema

- Changed REVOKE statement to specify INSERT permission on river_job.
- Added REVOKE for USAGE and SELECT on all sequences in the auth_custom schema for trigger_user.

* refactor: update environment variable handling and improve auth user sync worker

- Renamed environment variable from AUTH_USER_SYNC_BACKGROUND_WORKER_ENABLED to ENABLE_AUTH_USER_SYNC_BACKGROUND_WORKER for clarity and consistency.
- Streamlined the AuthUserSyncWorker to enhance monitoring and error handling.
- Adjusted the handling of environment variables in the Nomad job configuration to allow for better integration with module defaults.
- Updated test cases to reflect changes in the database client naming convention from AuthDb to AuthDB for consistency across the codebase.

* chore: auto-commit generated changes

* fix: lint

* chore: auto-commit generated changes

* chore: auto-commit generated changes

* refactor(dashboard-api): align auth user sync worker with review feedback

* chore: auto-commit generated changes

* refactor(db): split supabase auth sync source client

* chore(db): address follow-up nits

* fix(dashboard-api): stop river worker gracefully on shutdown

* refactor(dashboard-api): simplify service lifecycle orchestration

* chore: auto-commit generated changes

* refactor(db): drop supabase replica client plumbing

* fix: use write db for supabase db get user read

* chore: fix lint

* fix(dashboard-api): lazy-init supabase worker client

* chore(dashboard-api): reduce supabase worker pool size

* test(dashboard-api): tighten auth user sync coverage

Replace the broad backlog case with smaller tests that cover stale upsert cleanup and same-email trigger behavior. This keeps the worker coverage focused on the branch's critical correctness signals without extra soak-style runtime.

* chore: auto-commit generated changes

* feat(dashboard-api): add supabase db config override

Allow dashboard-api to use a dedicated SUPABASE_DB_CONNECTION_STRING while keeping the existing fallback to POSTGRES_CONNECTION_STRING. Thread the new setting through Terraform so deployments can configure the worker without reusing the auth DB connection string.

* refactor(dashboard-api): narrow infra sync runner changes

Keep dashboard-api specific env wiring explicit, restore NODE_ID and PORT in the Nomad job spec, and remove generic dashboard-api env passthrough. Also trim unrelated lint churn from the branch and scope the remaining test/migration helpers to the auth user sync worker flow.

* chore: auto-commit generated changes

* chore: auto-commit generated changes

* refactor(supabase): drop unused trigger role setup

Remove the copied trigger_user scaffolding from the Supabase River migrations because these SECURITY DEFINER functions never use that role. Keep the test bootstrap aligned with the simplified migration flow.

* refactor(dashboard-api): use shared not-found helper

Replace direct pgx no-rows checks in dashboard-api handlers with dberrors.IsNotFoundError so the package uses the shared database error handling consistently.

* chore: remove env vars artifact

* refactor(dashboard-api): tie river worker to signal context

Start the auth user sync River client with the signal-aware run context so it begins graceful shutdown as soon as SIGINT or SIGTERM arrives, while preserving the explicit coordinated stop path.

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: Ben Fornefeld <ben-fornefeld@users.noreply.github.com>

Author: 3 people

.env.gcp.template

@@ -77,6 +77,10 @@ CLICKHOUSE_CLUSTER_SIZE=1
 
 # Dashboard API instance count (default: 0)
 DASHBOARD_API_COUNT=
+# Dashboard API Supabase DB connection string (default: POSTGRES_CONNECTION_STRING)
+SUPABASE_DB_CONNECTION_STRING=
+# Enable dashboard-api auth user sync background worker (default: false)
+ENABLE_AUTH_USER_SYNC_BACKGROUND_WORKER=
 
 # Filestore cache for builds shared across cluster (default:false)
 FILESTORE_CACHE_ENABLED=

iac/modules/job-dashboard-api/jobs/dashboard-api.hcl

@@ -71,20 +71,12 @@ job "dashboard-api" {
       }
 
       env {
-        GIN_MODE                               = "release"
-        ENVIRONMENT                            = "${environment}"
-        NODE_ID                                = "$${node.unique.id}"
-        PORT                                   = "$${NOMAD_PORT_api}"
-        POSTGRES_CONNECTION_STRING             = "${postgres_connection_string}"
-        AUTH_DB_CONNECTION_STRING              = "${auth_db_connection_string}"
-        AUTH_DB_READ_REPLICA_CONNECTION_STRING = "${auth_db_read_replica_connection_string}"
-        CLICKHOUSE_CONNECTION_STRING           = "${clickhouse_connection_string}"
-        SUPABASE_JWT_SECRETS                   = "${supabase_jwt_secrets}"
-        REDIS_URL                              = "${redis_url}"
-        REDIS_CLUSTER_URL                      = "${redis_cluster_url}"
-        REDIS_TLS_CA_BASE64                    = "${redis_tls_ca_base64}"
-        OTEL_COLLECTOR_GRPC_ENDPOINT           = "${otel_collector_grpc_endpoint}"
-        LOGS_COLLECTOR_ADDRESS                 = "${logs_collector_address}"
+        NODE_ID = "$${node.unique.id}"
+        PORT    = "$${NOMAD_PORT_api}"
+
+        %{ for key, value in env ~}
+        ${key} = "${value}"
+        %{ endfor ~}
       }
 
       config {

iac/modules/job-dashboard-api/main.tf

@@ -1,27 +1,35 @@
+locals {
+  base_env = {
+    GIN_MODE                                = "release"
+    ENVIRONMENT                             = var.environment
+    POSTGRES_CONNECTION_STRING              = var.postgres_connection_string
+    AUTH_DB_CONNECTION_STRING               = var.auth_db_connection_string
+    AUTH_DB_READ_REPLICA_CONNECTION_STRING  = var.auth_db_read_replica_connection_string
+    SUPABASE_DB_CONNECTION_STRING           = var.supabase_db_connection_string
+    CLICKHOUSE_CONNECTION_STRING            = var.clickhouse_connection_string
+    SUPABASE_JWT_SECRETS                    = var.supabase_jwt_secrets
+    REDIS_URL                               = var.redis_url
+    REDIS_CLUSTER_URL                       = var.redis_cluster_url
+    REDIS_TLS_CA_BASE64                     = var.redis_tls_ca_base64
+    ENABLE_AUTH_USER_SYNC_BACKGROUND_WORKER = tostring(var.enable_auth_user_sync_background_worker)
+    OTEL_COLLECTOR_GRPC_ENDPOINT            = "localhost:${var.otel_collector_grpc_port}"
+    LOGS_COLLECTOR_ADDRESS                  = "http://localhost:${var.logs_proxy_port.port}"
+  }
+}
+
 resource "nomad_job" "dashboard_api" {
   jobspec = templatefile("${path.module}/jobs/dashboard-api.hcl", {
     update_stanza = var.update_stanza
     node_pool     = var.node_pool
     image_name    = var.image
-    environment   = var.environment
 
     count = var.count_instances
 
     memory_mb = 512
     cpu_count = 1
 
-    postgres_connection_string             = var.postgres_connection_string
-    auth_db_connection_string              = var.auth_db_connection_string
-    auth_db_read_replica_connection_string = var.auth_db_read_replica_connection_string
-    clickhouse_connection_string           = var.clickhouse_connection_string
-    supabase_jwt_secrets                   = var.supabase_jwt_secrets
-    redis_url                              = var.redis_url
-    redis_cluster_url                      = var.redis_cluster_url
-    redis_tls_ca_base64                    = var.redis_tls_ca_base64
+    env = local.base_env
 
     subdomain = "dashboard-api"
-
-    otel_collector_grpc_endpoint = "localhost:${var.otel_collector_grpc_port}"
-    logs_collector_address       = "http://localhost:${var.logs_proxy_port.port}"
   })
 }

iac/modules/job-dashboard-api/variables.tf

@@ -34,6 +34,12 @@ variable "auth_db_read_replica_connection_string" {
   default   = ""
 }
 
+variable "supabase_db_connection_string" {
+  type      = string
+  sensitive = true
+  default   = ""
+}
+
 variable "clickhouse_connection_string" {
   type      = string
   sensitive = true
@@ -44,6 +50,11 @@ variable "supabase_jwt_secrets" {
   sensitive = true
 }
 
+variable "enable_auth_user_sync_background_worker" {
+  type    = bool
+  default = false
+}
+
 variable "otel_collector_grpc_port" {
   type    = number
   default = 4317

iac/provider-gcp/Makefile

@@ -76,6 +76,8 @@ tf_vars := \
 	$(call tfvar, LOKI_BOOT_DISK_TYPE) \
 	$(call tfvar, LOKI_USE_V13_SCHEMA_FROM) \
 	$(call tfvar, DASHBOARD_API_COUNT) \
+	$(call tfvar, SUPABASE_DB_CONNECTION_STRING) \
+	$(call tfvar, ENABLE_AUTH_USER_SYNC_BACKGROUND_WORKER) \
 	$(call tfvar, DEFAULT_PERSISTENT_VOLUME_TYPE) \
 	$(call tfvar, PERSISTENT_VOLUME_TYPES) \
 	$(call tfvar, DB_MAX_OPEN_CONNECTIONS) \

iac/provider-gcp/main.tf

@@ -275,7 +275,9 @@ module "nomad" {
   otel_collector_resources_cpu_count = var.otel_collector_resources_cpu_count
 
   # Dashboard API
-  dashboard_api_count = var.dashboard_api_count
+  dashboard_api_count                     = var.dashboard_api_count
+  supabase_db_connection_string           = var.supabase_db_connection_string
+  enable_auth_user_sync_background_worker = var.enable_auth_user_sync_background_worker
 
   # Docker reverse proxy
   docker_reverse_proxy_port                = var.docker_reverse_proxy_port

iac/provider-gcp/nomad/main.tf

@@ -130,14 +130,16 @@ module "dashboard_api" {
 
   image = data.google_artifact_registry_docker_image.dashboard_api_image[0].self_link
 
-  postgres_connection_string             = data.google_secret_manager_secret_version.postgres_connection_string.secret_data
-  auth_db_connection_string              = data.google_secret_manager_secret_version.postgres_connection_string.secret_data
-  auth_db_read_replica_connection_string = trimspace(data.google_secret_manager_secret_version.postgres_read_replica_connection_string.secret_data)
-  clickhouse_connection_string           = local.clickhouse_connection_string
-  supabase_jwt_secrets                   = trimspace(data.google_secret_manager_secret_version.supabase_jwt_secrets.secret_data)
-  redis_url                              = local.redis_url
-  redis_cluster_url                      = local.redis_cluster_url
-  redis_tls_ca_base64                    = trimspace(data.google_secret_manager_secret_version.redis_tls_ca_base64.secret_data)
+  postgres_connection_string              = data.google_secret_manager_secret_version.postgres_connection_string.secret_data
+  auth_db_connection_string               = data.google_secret_manager_secret_version.postgres_connection_string.secret_data
+  auth_db_read_replica_connection_string  = trimspace(data.google_secret_manager_secret_version.postgres_read_replica_connection_string.secret_data)
+  supabase_db_connection_string           = var.supabase_db_connection_string
+  clickhouse_connection_string            = local.clickhouse_connection_string
+  supabase_jwt_secrets                    = trimspace(data.google_secret_manager_secret_version.supabase_jwt_secrets.secret_data)
+  redis_url                               = local.redis_url
+  redis_cluster_url                       = local.redis_cluster_url
+  redis_tls_ca_base64                     = trimspace(data.google_secret_manager_secret_version.redis_tls_ca_base64.secret_data)
+  enable_auth_user_sync_background_worker = var.enable_auth_user_sync_background_worker
 
   otel_collector_grpc_port = var.otel_collector_grpc_port
   logs_proxy_port          = var.logs_proxy_port

iac/provider-gcp/nomad/variables.tf

@@ -454,6 +454,17 @@ variable "dashboard_api_count" {
   default = 0
 }
 
+variable "supabase_db_connection_string" {
+  type      = string
+  default   = ""
+  sensitive = true
+}
+
+variable "enable_auth_user_sync_background_worker" {
+  type    = bool
+  default = false
+}
+
 variable "volume_token_issuer" {
   type = string
 }

iac/provider-gcp/variables.tf

@@ -230,6 +230,17 @@ variable "dashboard_api_count" {
   default = 0
 }
 
+variable "supabase_db_connection_string" {
+  type      = string
+  default   = ""
+  sensitive = true
+}
+
+variable "enable_auth_user_sync_background_worker" {
+  type    = bool
+  default = false
+}
+
 variable "docker_reverse_proxy_port" {
   type = object({
     name        = string

packages/api/go.mod

@@ -35,7 +35,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/grafana/loki/v3 v3.6.4
 	github.com/hashicorp/nomad/api v0.0.0-20251216171439-1dee0671280e
-	github.com/jackc/pgx/v5 v5.7.5
+	github.com/jackc/pgx/v5 v5.9.1
 	github.com/launchdarkly/go-sdk-common/v3 v3.3.0
 	github.com/launchdarkly/go-server-sdk/v7 v7.13.0
 	github.com/oapi-codegen/gin-middleware v1.0.2
@@ -386,7 +386,7 @@ require (
 	golang.org/x/crypto v0.49.0 // indirect
 	golang.org/x/exp v0.0.0-20260212183809-81e46e3db34a // indirect
 	golang.org/x/image v0.38.0 // indirect
-	golang.org/x/mod v0.33.0 // indirect
+	golang.org/x/mod v0.34.0 // indirect
 	golang.org/x/oauth2 v0.34.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
 	golang.org/x/text v0.35.0 // indirect