Merge branch 'main' into feature/dashboard-plans

# Conflicts: # AGENTS.md
2026-09-09 16:04:41 +00:00 · 2026-05-30 09:49:12 +00:00
parent 7f830b976f 9027086d77
commit 44c837d8ac
54 changed files with 1709 additions and 213 deletions
@@ -29,6 +29,7 @@ Other CI-touching rules:

 Commit hygiene:

+- when instructed to make a commit, use the subject format `feat: one line explanation`
 - commit messages on this repo do not include `Co-Authored-By:` or other AI/agent attribution footers. Keep messages to subject + body explaining the why. If a commit slips through with an attribution footer, rewrite it before opening or updating a PR.

 ### Verification: what to run, what to skip
@@ -64,6 +65,21 @@ Infra runs in docker; the Go services and frontends run natively on the host for

 Prefer native `make backend` over rebuilding the docker backend image: docker rebuilds are slow because the image bakes in the migrations and the compiled binary, so a one-line change means a full image build + container recreate. The native targets skip all of that. The dockerized hot-reload flow (`make app`) and prod-image smoke test (`make up`) remain available when you specifically need containers.

+Dashboard realtime:
+
+- dashboard experiences should be realtime by default. When emails arrive, contacts are added, records change, or any dashboard-visible feature updates, the dashboard should reflect it live without requiring a manual refresh
+- aim for a responsive, Discord-like product feel: presence, counts, lists, detail panes, notifications, and workflow state should stay current across every dashboard feature where live updates are meaningful
+- when changing dashboard behavior, it is acceptable to safely change the API structure if a better solution exists. Before making an API shape change, ask the user how they want to handle it, especially when the current API may already be published or backwards compatibility might require a new API version
+
+Public API quality bar:
+
+- treat customer-facing API changes as contract changes. Prefer additive changes inside a version, and use a new API version for incompatible behavior once an endpoint is published
+- every API-key-capable route must have an explicit API permission gate and, for JWT callers, the matching organization permission gate
+- side-effectful POST/PATCH/PUT/DELETE endpoints should support `Idempotency-Key` or have a documented reason why retries are naturally safe
+- error responses should include stable machine-readable `code` and `request_id` fields in addition to human-readable text
+- list endpoints should use consistent `data` plus `pagination` shapes with opaque cursors; invalid cursors or limits should return `400` instead of being ignored
+- webhook endpoints must stay HMAC-signed, HTTPS by default, and protected against obvious SSRF targets. Only development/self-hosted environments should opt into unsafe webhook URLs
+
 ## System Shape

 - `cmd/backend`: API and business orchestration
@@ -36,6 +36,7 @@ import (
 	"github.com/warmbly/warmbly/internal/app/feature"
 	"github.com/warmbly/warmbly/internal/app/fleet"
 	"github.com/warmbly/warmbly/internal/app/group"
+	idempotencyapp "github.com/warmbly/warmbly/internal/app/idempotency"
 	"github.com/warmbly/warmbly/internal/app/integration"
 	"github.com/warmbly/warmbly/internal/app/organization"
 	"github.com/warmbly/warmbly/internal/app/ratelimit"
@@ -116,6 +117,7 @@ func main() {
 	var categoryService group.GroupService
 	var crmService crm.CRMService
 	var apiKeyService apikey.APIKeyService
+	var idempotencyService idempotencyapp.Service

 	// New services for trial, feature gates, and worker assignment
 	var trialService trial.TrialService
@@ -458,6 +460,7 @@ func main() {
 		organizationRepoForHandler = organizationRepository
 		taskRepository := repository.NewTaskRepository(primaryDB.Pool)
 		apiKeyRepository := repository.NewAPIKeyRepository(primaryDB)
+		idempotencyService = idempotencyapp.NewService(primaryDB.Pool)
 		crmRepository := repository.NewCRMRepository(primaryDB.Pool)
 		advancedRepository := repository.NewAdvancedOutreachRepository(primaryDB.Pool)
 		templateRepository := repository.NewTemplateRepository(primaryDB.Pool)
@@ -594,18 +597,23 @@ func main() {
 			credentialsRepository,
 			cipherService,
 			worker_orchestrator.WorkerEnvConfig{
-				AppEnv:               os.Getenv("APP_ENV"),
-				WorkerImage:          getenvDefault("WORKER_IMAGE", "ghcr.io/warmbly/worker:latest"),
-				KafkaBootstrap:       os.Getenv("KAFKA_BOOTSTRAP_SERVERS"),
-				KafkaSASLUsername:    os.Getenv("KAFKA_SASL_USERNAME"),
-				KafkaSASLPassword:    os.Getenv("KAFKA_SASL_PASSWORD"),
-				SchemaRegistryURL:    os.Getenv("SCHEMA_REGISTRY_URL"),
-				SchemaRegistryKey:    os.Getenv("SCHEMA_REGISTRY_KEY"),
-				SchemaRegistrySecret: os.Getenv("SCHEMA_REGISTRY_SECRET"),
-				RedisURL:             os.Getenv("REDIS"),
-				AWSRegion:            os.Getenv("AWS_REGION"),
-				AWSAccessKeyID:       os.Getenv("WORKER_AWS_ACCESS_KEY_ID"),
-				AWSSecretAccessKey:   os.Getenv("WORKER_AWS_SECRET_ACCESS_KEY"),
+				AppEnv:                   os.Getenv("APP_ENV"),
+				WorkerImage:              getenvDefault("WORKER_IMAGE", "ghcr.io/warmbly/worker:latest"),
+				KafkaBootstrap:           os.Getenv("KAFKA_BOOTSTRAP_SERVERS"),
+				KafkaSASLUsername:        os.Getenv("KAFKA_SASL_USERNAME"),
+				KafkaSASLPassword:        os.Getenv("KAFKA_SASL_PASSWORD"),
+				SchemaRegistryURL:        os.Getenv("SCHEMA_REGISTRY_URL"),
+				SchemaRegistryKey:        os.Getenv("SCHEMA_REGISTRY_KEY"),
+				SchemaRegistrySecret:     os.Getenv("SCHEMA_REGISTRY_SECRET"),
+				RedisURL:                 os.Getenv("REDIS"),
+				AWSRegion:                os.Getenv("AWS_REGION"),
+				AWSAccessKeyID:           os.Getenv("WORKER_AWS_ACCESS_KEY_ID"),
+				AWSSecretAccessKey:       os.Getenv("WORKER_AWS_SECRET_ACCESS_KEY"),
+				EncryptedKeysBackendURL:  os.Getenv("ENCRYPTED_KEYS_BACKEND_URL"),
+				EncryptedKeysWorkerToken: os.Getenv("INTERNAL_API_TOKEN"),
+				EventBusProvider:         os.Getenv("EVENTBUS_PROVIDER"),
+				NATSURL:                  os.Getenv("NATS_URL"),
+				CodecProvider:            os.Getenv("CODEC_PROVIDER"),
 			},
 			getenvDefault("WORKER_INSTALLER_PATH", "/app/scripts/install-worker.sh"),
 		)
@@ -639,6 +647,7 @@ func main() {
 			cache,
 			&oauth2Cfg.InboxAuthorization,
 			workerAssignmentService,
+			streamingPublisher,
 		)
 		// Fan out email-account lifecycle events to customer webhooks.
 		emailService.WireWebhooks(webhookService)
@@ -655,7 +664,7 @@ func main() {
 		rateLimitRepository := repository.NewRateLimitRepository(primaryDB)
 		rateLimitService = ratelimit.NewService(cache, rateLimitRepository)
 		sequenceService = sequence.NewService(sequenceRepostory)
-		contactService = contact.NewService(contactRepostory, subscriptionRepository, planRepository)
+		contactService = contact.NewService(contactRepostory, subscriptionRepository, planRepository, streamingPublisher)
 		apiKeyService = apikey.NewService(cache, apiKeyRepository)
 		crmService = crm.NewService(crmRepository)
 		socketService = socket.NewService(cache, tokenService)
@@ -845,6 +854,7 @@ func main() {
 	m := &middleware.Handler{
 		TokenService:        tokenService,
 		APIKeyService:       apiKeyService,
+		IdempotencyService:  idempotencyService,
 		OrganizationService: organizationService,
 	}

@@ -1,10 +1,14 @@
 package main

 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"log"
+	"net/http"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
 	"time"

@@ -164,6 +168,7 @@ func main() {
 	// the rolling 1m counters into a WorkerHealth event, publishes via the
 	// event bus so the consumer can write a row into worker_health_samples.
 	go workerService.Heartbeat(ctx)
+	go runInternalHeartbeat(ctx, workerID, bindIP)
 	go workerService.RunHealth(ctx, 30*time.Second)

 	// Graceful shutdown
@@ -183,6 +188,60 @@ func main() {
 	log.Println("Worker stopped")
 }

+func runInternalHeartbeat(ctx context.Context, workerID uuid.UUID, bindIP string) {
+	baseURL := strings.TrimRight(os.Getenv("ENCRYPTED_KEYS_BACKEND_URL"), "/")
+	token := os.Getenv("ENCRYPTED_KEYS_WORKER_TOKEN")
+	if baseURL == "" || token == "" {
+		return
+	}
+	reportedIP := os.Getenv("WORKER_PUBLIC_IP")
+	if reportedIP == "" && bindIP != "default route" {
+		reportedIP = bindIP
+	}
+	if reportedIP == "" {
+		reportedIP = "unknown"
+	}
+
+	client := &http.Client{Timeout: 10 * time.Second}
+	send := func() {
+		payload := map[string]string{
+			"worker_id":   workerID.String(),
+			"bind_ip":     reportedIP,
+			"tier":        os.Getenv("WORKER_TIER"),
+			"egress_kind": os.Getenv("WORKER_EGRESS_KIND"),
+		}
+		body, _ := json.Marshal(payload)
+		req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/api/v1/internal/worker/heartbeat", bytes.NewReader(body))
+		if err != nil {
+			log.Println("failed to build internal heartbeat:", err)
+			return
+		}
+		req.Header.Set("Authorization", "Bearer "+token)
+		req.Header.Set("Content-Type", "application/json")
+		resp, err := client.Do(req)
+		if err != nil {
+			log.Println("failed internal heartbeat:", err)
+			return
+		}
+		defer resp.Body.Close()
+		if resp.StatusCode >= 300 {
+			log.Println("internal heartbeat returned status", resp.StatusCode)
+		}
+	}
+
+	send()
+	ticker := time.NewTicker(90 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			send()
+		}
+	}
+}
+
 // uuidNamespaceURL is the RFC 4122 URL namespace, matching the value used by
 // scripts/install-worker.sh when deriving the per-IP worker ID. Keep these in
 // sync: the installer and the worker must agree on the derivation.
@@ -61,14 +61,21 @@ Add a worker from the admin dashboard:

 1. Provision a VPS, note its public IP + root user
 2. Admin → Workers → Add Worker
-3. Paste the generated SSH public key into the VPS's `~/.ssh/authorized_keys`
-4. Click Test, then Install
+3. Copy the generated enrollment command
+4. Run it on the VPS as root

-The backend SSHes in, uploads `scripts/install-worker.sh`, configures systemd, and starts the worker container. From then on, all lifecycle operations (restart, update, system updates, reboot, rotate keys, logs, uninstall) happen from the dashboard.
+The installer is served by the backend at `/worker-install.sh`. It exchanges the one-time token for worker config, writes `/etc/warmbly/worker.env`, configures systemd, enables a daily randomized self-update timer, and starts the worker container. The worker then heartbeats back to the backend and marks itself installed.
+
+The older SSH-managed path is still supported: paste the generated SSH public key into the VPS's `~/.ssh/authorized_keys`, then click Test and Install. From then on, lifecycle operations (restart, update, system updates, reboot, rotate keys, logs, uninstall) can happen from the dashboard.

 Manual install on the VPS is also supported:

 ```bash
+curl -fsSL https://api.example.com/worker-install.sh | sudo bash -s -- \
+  --enroll wmenroll_... \
+  --api-base https://api.example.com
+
+# or fully manual:
 sudo bash scripts/install-worker.sh \
  --kafka kafka.example.com:9092 \
  --schema-registry https://schema.example.com \
@@ -52,11 +52,13 @@ curl -X GET "https://api.warmbly.com/api-keys/permissions" \
    { "name": "BULK_CAMPAIGNS", "value": 1024, "description": "Bulk campaign operations", "category": "bulk" },
    { "name": "REALTIME_SUBSCRIBE", "value": 2048, "description": "Subscribe to realtime events", "category": "special" },
    { "name": "WEBHOOKS", "value": 4096, "description": "Manage webhook endpoints", "category": "special" },
-    { "name": "API_KEYS", "value": 8192, "description": "Create and manage API keys", "category": "special" }
+    { "name": "API_KEYS", "value": 8192, "description": "Create and manage API keys", "category": "special" },
+    { "name": "INTEGRATIONS", "value": 1048576, "description": "Connect and manage third-party integrations", "category": "special" },
+    { "name": "WARMUP_ROUTING", "value": 2097152, "description": "Manage warmup routing rules", "category": "special" }
  ],
  "presets": {
    "read_only": 688159,
-    "full_access": 1048575
+    "full_access": 4194303
  }
 }
 ```
@@ -75,7 +77,7 @@ curl -X GET "https://api.warmbly.com/api-keys/permissions" \
 | Preset | Value | Description |
 |--------|-------|-------------|
 | `read_only` | 688159 | All read permissions across emails, campaigns, contacts, unibox, analytics, templates, CRM, audit logs |
-| `full_access` | 1048575 | Every defined permission bit |
+| `full_access` | 4194303 | Every defined permission bit |

 ## Working with Permissions

@@ -29,6 +29,8 @@ curl -X GET "https://api.warmbly.com/api-keys" \
  -H "Content-Type: application/json"
 ```

+For mutation retries, include an `Idempotency-Key` header with a unique value per logical operation. Warmbly stores completed mutation responses for 24 hours per organization and key, then replays matching retries instead of performing the operation again.
+
 ## Key Security Best Practices

 <Callout type="warn" title="Keep Your Keys Secret">
@@ -116,6 +116,15 @@ When an endpoint says "JWT permission: X / API permission: Y", the dual-auth mid
 | POST | `/deliverability/events` | `WRITE_CAMPAIGNS` |
 | GET | `/tasks/dlq` | `SEND_CAMPAIGNS` |
 | POST | `/tasks/dlq/:id/replay` | `SEND_CAMPAIGNS` |
+| GET/POST/PATCH/DELETE | `/webhooks[/:id]` | `WEBHOOKS` |
+| POST | `/webhooks/:id/rotate-secret` | `WEBHOOKS` |
+| GET | `/webhooks/:id/deliveries` | `WEBHOOKS` |
+| GET/POST/DELETE | `/integrations/*` | `INTEGRATIONS` |
+| GET/POST/PATCH/DELETE | `/warmup/routing[/:id]` | `WARMUP_ROUTING` |
+
+### Retry safety
+
+Mutating API requests may include an `Idempotency-Key` header. Warmbly stores the completed response for 24 hours per organization and key. Reusing the same key with the same method, route, query, and body replays the original response with `X-Idempotent-Replayed: true`; reusing the key with a different request returns `409 Conflict`.

 ### Reference data

@@ -14,10 +14,14 @@ All errors follow this structure:
 ```json
 {
  "error": "Error Type",
-  "message": "Human-readable description of what went wrong."
+  "message": "Human-readable description of what went wrong.",
+  "code": "machine_readable_code",
+  "request_id": "req_or_uuid_for_support"
 }
 ```

+`error` and `message` are for people. Client logic should use `code`, HTTP status, and endpoint-specific fields such as `retry_after`. Include `request_id` when contacting support.
+
 ## HTTP Status Codes

 ### Client Errors (4xx)
@@ -56,7 +60,9 @@ Returned when the request cannot be processed due to invalid syntax.
 ```json
 {
  "error": "Bad Request",
-  "message": "invalid request body"
+  "message": "invalid request body",
+  "code": "bad_request",
+  "request_id": "4bbbd1b2-8f86-47dd-8a7f-9476501ad20e"
 }
 ```

@@ -1,6 +1,6 @@
 ---
 title: Permissions Reference
-description: Complete reference for the 20 API permissions available in Warmbly.
+description: Complete reference for the 22 API permissions available in Warmbly.
 ---

 # Permissions Reference
@@ -23,7 +23,7 @@ Warmbly uses a bitmask system for API permissions. Each permission is one bit in
 | `BULK_CONTACTS` | 9 | 512 | bulk | Bulk import/export/delete contacts |
 | `BULK_CAMPAIGNS` | 10 | 1024 | bulk | Bulk campaign operations |
 | `REALTIME_SUBSCRIBE` | 11 | 2048 | special | Subscribe to realtime events |
-| `WEBHOOKS` | 12 | 4096 | special | Manage webhook endpoints (reserved) |
+| `WEBHOOKS` | 12 | 4096 | special | Manage webhook endpoints |
 | `API_KEYS` | 13 | 8192 | special | Create and manage API keys |
 | `SEND_CAMPAIGNS` | 14 | 16384 | write | Start and stop campaigns (sends real mail) |
 | `READ_TEMPLATES` | 15 | 32768 | read | View reply templates |
@@ -31,6 +31,8 @@ Warmbly uses a bitmask system for API permissions. Each permission is one bit in
 | `READ_CRM` | 17 | 131072 | read | View pipelines, deals, and CRM tasks |
 | `WRITE_CRM` | 18 | 262144 | write | Create and modify pipelines, deals, CRM tasks |
 | `READ_AUDIT_LOGS` | 19 | 524288 | read | View organization audit logs |
+| `INTEGRATIONS` | 20 | 1048576 | special | Connect and manage third-party integrations |
+| `WARMUP_ROUTING` | 21 | 2097152 | special | Manage warmup routing rules |

 `SEND_CAMPAIGNS` is intentionally separate from `WRITE_CAMPAIGNS`: editing a campaign draft and starting one that actually transmits mail are different blast radii, so a key can be granted the first without the second.

@@ -50,7 +52,7 @@ High-volume operations. These can touch large numbers of rows in a single reques

 ### Special

-Realtime subscriptions, webhooks, self-service key management. Grant individually.
+Realtime subscriptions, webhooks, integrations, warmup routing, and self-service key management. Grant individually.

 ## Preset Combinations

@@ -67,12 +69,12 @@ READ_EMAILS | READ_CAMPAIGNS | READ_CONTACTS | READ_UNIBOX | READ_ANALYTICS
 = 688159
 ```

-### Full Access — 1048575
+### Full Access — 4194303

-All 20 permissions:
+All 22 permissions:

 ```
-(1 << 20) - 1 = 1048575
+(1 << 22) - 1 = 4194303
 ```

 ## Working with Bitmasks
@@ -101,7 +103,7 @@ hasPermission(16387, 64);    // false — WRITE_CAMPAIGNS missing

 ### Rejecting unknown bits

-`POST /api-keys` rejects any request whose `permissions` field has bits outside the known set, so a stale client can't accidentally request a future permission. The current mask of valid bits is `1048575`.
+`POST /api-keys` rejects any request whose `permissions` field has bits outside the known set, so a stale client can't accidentally request a future permission. The current mask of valid bits is `4194303`.

 ## Common Permission Sets

@@ -107,7 +107,7 @@ func (h *Handler) AdminCreateWorker(c *gin.Context) {
 			errx.JSON(c, errx.New(errx.Internal, "failed to generate enrollment token"))
 			return
 		}
-		enrollToken = hex.EncodeToString(raw)
+		enrollToken = "wmenroll_" + hex.EncodeToString(raw)
 		sum := sha256.Sum256([]byte(enrollToken))
 		enrollHash = hex.EncodeToString(sum[:])
 		exp := time.Now().Add(2 * time.Hour)
@@ -18,7 +18,7 @@ func (h *Handler) EmailsSearch(c *gin.Context) {
 	tag := c.Query("tag")
 	limit := c.Query("limit")

-	resp, err := h.EmailService.Search(c.Request.Context(), userID, query, cursor, tag, limit)
+	resp, err := h.EmailService.Search(c.Request.Context(), userID, query, cursor, tag, limit, middleware.GetAPIKeyAllowedEmailAccounts(c))
 	if err != nil {
 		errx.Handle(c, err)
 		return
@@ -10,13 +10,16 @@ type Handler struct {
 	service group.GroupService
 }

-func New(r *gin.RouterGroup, service group.GroupService, name string) {
+func New(r *gin.RouterGroup, service group.GroupService, name string, middleware ...gin.HandlerFunc) {
 	h := &Handler{
 		name:    name,
 		service: service,
 	}

 	g := r.Group("/" + name)
+	if len(middleware) > 0 {
+		g.Use(middleware...)
+	}
 	{
 		g.POST("", h.Create)
 		g.PATCH("/:gid", h.Update)
@@ -164,6 +164,6 @@ func requireOrgID(c *gin.Context) (uuid.UUID, bool) {
 	if orgID := middleware.GetOrganizationID(c); orgID != nil {
 		return *orgID, true
 	}
-	c.JSON(http.StatusForbidden, gin.H{"error": "organization context required"})
+	errx.JSON(c, errx.New(errx.Forbidden, "organization context required"))
 	return uuid.Nil, false
 }
@@ -7,6 +7,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"

+	"github.com/warmbly/warmbly/internal/errx"
 	"github.com/warmbly/warmbly/internal/models"
 )

@@ -29,7 +30,7 @@ func (h *Handler) ListWebhookEndpoints(c *gin.Context) {
 	}
 	endpoints, err := h.WebhookService.ListEndpoints(c.Request.Context(), orgID)
 	if err != nil {
-		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list endpoints"})
+		errx.JSON(c, errx.New(errx.Internal, "failed to list endpoints"))
 		return
 	}
 	if endpoints == nil {
@@ -51,7 +52,7 @@ func (h *Handler) CreateWebhookEndpoint(c *gin.Context) {
 	}
 	var p webhookEndpointPayload
 	if err := c.ShouldBindJSON(&p); err != nil {
-		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload"})
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid payload"))
 		return
 	}
 	enabled := true
@@ -60,7 +61,7 @@ func (h *Handler) CreateWebhookEndpoint(c *gin.Context) {
 	}
 	endpoint, err := h.WebhookService.CreateEndpoint(c.Request.Context(), orgID, p.URL, p.Description, p.EventTypes, enabled)
 	if err != nil {
-		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		errx.JSON(c, errx.New(errx.BadRequest, err.Error()))
 		return
 	}
 	c.JSON(http.StatusCreated, endpoint)
@@ -75,12 +76,12 @@ func (h *Handler) UpdateWebhookEndpoint(c *gin.Context) {
 	}
 	endpointID, err := uuid.Parse(c.Param("id"))
 	if err != nil {
-		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid endpoint id"})
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid endpoint id"))
 		return
 	}
 	var p webhookEndpointPayload
 	if err := c.ShouldBindJSON(&p); err != nil {
-		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload"})
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid payload"))
 		return
 	}
 	enabled := true
@@ -89,7 +90,7 @@ func (h *Handler) UpdateWebhookEndpoint(c *gin.Context) {
 	}
 	endpoint, err := h.WebhookService.UpdateEndpoint(c.Request.Context(), orgID, endpointID, p.URL, p.Description, p.EventTypes, enabled)
 	if err != nil {
-		c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
+		errx.JSON(c, errx.New(errx.BadRequest, err.Error()))
 		return
 	}
 	c.JSON(http.StatusOK, endpoint)
@@ -104,11 +105,11 @@ func (h *Handler) DeleteWebhookEndpoint(c *gin.Context) {
 	}
 	endpointID, err := uuid.Parse(c.Param("id"))
 	if err != nil {
-		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid endpoint id"})
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid endpoint id"))
 		return
 	}
 	if err := h.WebhookService.DeleteEndpoint(c.Request.Context(), orgID, endpointID); err != nil {
-		c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
+		errx.JSON(c, errx.New(errx.NotFound, err.Error()))
 		return
 	}
 	c.Status(http.StatusNoContent)
@@ -124,12 +125,12 @@ func (h *Handler) RotateWebhookSecret(c *gin.Context) {
 	}
 	endpointID, err := uuid.Parse(c.Param("id"))
 	if err != nil {
-		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid endpoint id"})
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid endpoint id"))
 		return
 	}
 	secret, err := h.WebhookService.RotateSecret(c.Request.Context(), orgID, endpointID)
 	if err != nil {
-		c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
+		errx.JSON(c, errx.New(errx.NotFound, err.Error()))
 		return
 	}
 	c.JSON(http.StatusOK, gin.H{"secret": secret})
@@ -144,18 +145,21 @@ func (h *Handler) ListWebhookDeliveries(c *gin.Context) {
 	}
 	endpointID, err := uuid.Parse(c.Param("id"))
 	if err != nil {
-		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid endpoint id"})
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid endpoint id"))
 		return
 	}
 	limit := 50
 	if raw := c.Query("limit"); raw != "" {
-		if n, err := strconv.Atoi(raw); err == nil && n > 0 {
-			limit = n
+		n, err := strconv.Atoi(raw)
+		if err != nil || n <= 0 || n > 200 {
+			errx.JSON(c, errx.New(errx.BadRequest, "limit must be between 1 and 200"))
+			return
 		}
+		limit = n
 	}
 	deliveries, err := h.WebhookService.ListDeliveries(c.Request.Context(), orgID, endpointID, limit)
 	if err != nil {
-		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list deliveries"})
+		errx.JSON(c, errx.New(errx.Internal, "failed to list deliveries"))
 		return
 	}
 	if deliveries == nil {
@@ -0,0 +1,102 @@
+package handler
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"net/http"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+func (h *Handler) ServeWorkerInstaller(c *gin.Context) {
+	if h.WorkerOrchestrator == nil {
+		errx.JSON(c, errx.New(errx.ServiceUnavailable, "worker installer is not configured"))
+		return
+	}
+	script, err := h.WorkerOrchestrator.InstallerScript()
+	if err != nil {
+		errx.JSON(c, errx.New(errx.Internal, "failed to load worker installer"))
+		return
+	}
+	c.Header("Content-Type", "text/x-shellscript; charset=utf-8")
+	c.Header("Cache-Control", "no-cache")
+	c.Data(http.StatusOK, "text/x-shellscript; charset=utf-8", script)
+}
+
+type workerEnrollmentRequest struct {
+	Token    string `json:"token"`
+	PublicIP string `json:"public_ip,omitempty"`
+}
+
+// EnrollWorker exchanges a one-time enrollment token for a complete worker
+// dotenv file. It is intentionally public: the high-entropy one-time token is
+// the credential, and it is consumed atomically before secrets are returned.
+func (h *Handler) EnrollWorker(c *gin.Context) {
+	if h.WorkerRepo == nil || h.WorkerOrchestrator == nil {
+		errx.JSON(c, errx.New(errx.ServiceUnavailable, "worker enrollment is not configured"))
+		return
+	}
+
+	var req workerEnrollmentRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid request body"))
+		return
+	}
+	req.Token = strings.TrimSpace(req.Token)
+	if req.Token == "" {
+		errx.JSON(c, errx.New(errx.BadRequest, "enrollment token is required"))
+		return
+	}
+
+	sum := sha256.Sum256([]byte(req.Token))
+	worker, err := h.WorkerRepo.ConsumeEnrollmentToken(c.Request.Context(), hex.EncodeToString(sum[:]))
+	if err != nil {
+		errx.JSON(c, errx.New(errx.Internal, "failed to consume enrollment token"))
+		return
+	}
+	if worker == nil {
+		errx.JSON(c, errx.New(errx.Unauthorized, "enrollment token is invalid or expired"))
+		return
+	}
+
+	ip := strings.TrimSpace(req.PublicIP)
+	if ip == "" {
+		ip = c.ClientIP()
+	}
+	if ip != "" {
+		_ = h.WorkerRepo.RecordEnrolledIP(c.Request.Context(), worker.ID, ip)
+	}
+
+	envFile, _, err := h.WorkerOrchestrator.RenderEnrollmentEnv(c.Request.Context(), worker.ID)
+	if err != nil {
+		errx.JSON(c, errx.New(errx.Internal, "failed to render worker config"))
+		return
+	}
+	envFile += "WORKER_TIER=" + workerTierLabel(worker) + "\n"
+	if ip != "" {
+		envFile += "WORKER_PUBLIC_IP=" + ip + "\n"
+	}
+	if worker.EgressKind != "" {
+		envFile += "WORKER_EGRESS_KIND=" + string(worker.EgressKind) + "\n"
+	}
+
+	c.Header("Content-Type", "text/plain; charset=utf-8")
+	c.Header("Cache-Control", "no-store")
+	c.String(http.StatusOK, envFile)
+}
+
+func workerTierLabel(w *models.Worker) string {
+	if w == nil {
+		return "shared_premium"
+	}
+	if w.WorkerType == models.WorkerTypeDedicated {
+		return "dedicated"
+	}
+	if w.FreeTier {
+		return "shared_free"
+	}
+	return "shared_premium"
+}
@@ -13,12 +13,13 @@ import (
 )

 const (
-	APIKeyIDKey          = "api_key_id"
-	APIKeyPermissionsKey = "api_key_permissions"
-	APIKeyUserIDKey      = "api_key_user_id"
-	AuthTypeKey          = "auth_type"
-	AuthTypeJWT          = "jwt"
-	AuthTypeAPIKey       = "api_key"
+	APIKeyIDKey                   = "api_key_id"
+	APIKeyPermissionsKey          = "api_key_permissions"
+	APIKeyAllowedEmailAccountsKey = "api_key_allowed_email_accounts"
+	APIKeyUserIDKey               = "api_key_user_id"
+	AuthTypeKey                   = "auth_type"
+	AuthTypeJWT                   = "jwt"
+	AuthTypeAPIKey                = "api_key"
 )

 // APIKeyMiddleware accepts only API key auth ("Bearer wmbly_..."). Reserved
@@ -96,6 +97,8 @@ func (h *Handler) validateAPIKey(c *gin.Context, rawKey string) {
 		c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{
 			"error":       "rate_limit_exceeded",
 			"message":     fmt.Sprintf("API key exceeded %d requests per minute", limit),
+			"code":        "rate_limit_exceeded",
+			"request_id":  c.GetString(RequestIDContextKey),
 			"retry_after": retryAfter,
 		})
 		return
@@ -104,6 +107,7 @@ func (h *Handler) validateAPIKey(c *gin.Context, rawKey string) {
 	c.Set(AuthTypeKey, AuthTypeAPIKey)
 	c.Set(APIKeyIDKey, key.ID.String())
 	c.Set(APIKeyPermissionsKey, key.Permissions)
+	c.Set(APIKeyAllowedEmailAccountsKey, key.AllowedEmailAccounts)
 	c.Set(UserIDKey, key.UserID.String())
 	c.Set(OrganizationIDKey, key.OrganizationID)

@@ -213,6 +217,41 @@ func (h *Handler) RequireAccess(orgPerm models.OrganizationPermission, apiPerm u
 	}
 }

+// RequireAPIKeyEmailAccountParam enforces an API key's optional
+// allowed_email_accounts allowlist against a route parameter. JWT callers and
+// unrestricted API keys pass through.
+func RequireAPIKeyEmailAccountParam(param string) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		if c.GetString(AuthTypeKey) != AuthTypeAPIKey {
+			c.Next()
+			return
+		}
+
+		allowed := GetAPIKeyAllowedEmailAccounts(c)
+		if len(allowed) == 0 {
+			c.Next()
+			return
+		}
+
+		accountID, err := uuid.Parse(c.Param(param))
+		if err != nil {
+			errx.Handle(c, errx.ErrUuid)
+			c.Abort()
+			return
+		}
+
+		for _, id := range allowed {
+			if id == accountID {
+				c.Next()
+				return
+			}
+		}
+
+		errx.Handle(c, errx.New(errx.Forbidden, "email account is not allowed for this API key"))
+		c.Abort()
+	}
+}
+
 // GetAuthType returns "jwt" or "api_key" (empty if unauthenticated).
 func GetAuthType(c *gin.Context) string {
 	return c.GetString(AuthTypeKey)
@@ -245,3 +284,17 @@ func GetAPIKeyPermissions(c *gin.Context) uint64 {
 	}
 	return permissions
 }
+
+// GetAPIKeyAllowedEmailAccounts returns the optional email-account allowlist
+// attached to the authenticating API key. Empty means unrestricted.
+func GetAPIKeyAllowedEmailAccounts(c *gin.Context) []uuid.UUID {
+	value, exists := c.Get(APIKeyAllowedEmailAccountsKey)
+	if !exists {
+		return nil
+	}
+	ids, ok := value.([]uuid.UUID)
+	if !ok {
+		return nil
+	}
+	return ids
+}
@@ -6,6 +6,7 @@ import (
 	"testing"

 	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
 	"github.com/warmbly/warmbly/internal/models"
 )

@@ -13,6 +14,51 @@ func init() {
 	gin.SetMode(gin.TestMode)
 }

+func TestRequireAPIKeyEmailAccountParam(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	allowedID := uuid.New()
+	deniedID := uuid.New()
+
+	tests := []struct {
+		name       string
+		authType   string
+		allowlist  []uuid.UUID
+		pathID     uuid.UUID
+		wantStatus int
+	}{
+		{"jwt bypasses allowlist", AuthTypeJWT, []uuid.UUID{allowedID}, deniedID, http.StatusOK},
+		{"unrestricted key bypasses allowlist", AuthTypeAPIKey, nil, deniedID, http.StatusOK},
+		{"allowed key passes", AuthTypeAPIKey, []uuid.UUID{allowedID}, allowedID, http.StatusOK},
+		{"denied key fails", AuthTypeAPIKey, []uuid.UUID{allowedID}, deniedID, http.StatusForbidden},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := gin.New()
+			r.Use(RequestIDMiddleware())
+			r.GET("/emails/:id",
+				func(c *gin.Context) {
+					c.Set(AuthTypeKey, tt.authType)
+					c.Set(APIKeyAllowedEmailAccountsKey, tt.allowlist)
+					c.Next()
+				},
+				RequireAPIKeyEmailAccountParam("id"),
+				func(c *gin.Context) {
+					c.Status(http.StatusOK)
+				},
+			)
+
+			req := httptest.NewRequest(http.MethodGet, "/emails/"+tt.pathID.String(), nil)
+			rec := httptest.NewRecorder()
+			r.ServeHTTP(rec, req)
+
+			if rec.Code != tt.wantStatus {
+				t.Fatalf("status = %d, want %d; body=%s", rec.Code, tt.wantStatus, rec.Body.String())
+			}
+		})
+	}
+}
+
 func TestRequireAPIPermission(t *testing.T) {
 	tests := []struct {
 		name       string
@@ -2,6 +2,7 @@ package middleware

 import (
 	"github.com/warmbly/warmbly/internal/app/apikey"
+	"github.com/warmbly/warmbly/internal/app/idempotency"
 	"github.com/warmbly/warmbly/internal/app/organization"
 	"github.com/warmbly/warmbly/internal/app/ratelimit"
 	"github.com/warmbly/warmbly/internal/app/token"
@@ -10,6 +11,7 @@ import (
 type Handler struct {
 	TokenService        token.TokenService
 	APIKeyService       apikey.APIKeyService
+	IdempotencyService  idempotency.Service
 	RateLimitService    ratelimit.RateLimitService
 	OrganizationService organization.OrganizationService
 }
@@ -0,0 +1,158 @@
+package middleware
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
+	"io"
+	"net/http"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/warmbly/warmbly/internal/app/idempotency"
+	"github.com/warmbly/warmbly/internal/errx"
+)
+
+const (
+	IdempotencyKeyHeader      = "Idempotency-Key"
+	IdempotencyReplayedHeader = "X-Idempotent-Replayed"
+)
+
+// IdempotencyMiddleware implements Stripe-style retry safety for mutating API
+// requests. It is opt-in per request via Idempotency-Key and scoped by
+// organization, so the same key cannot collide across tenants.
+func (h *Handler) IdempotencyMiddleware() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		key := strings.TrimSpace(c.GetHeader(IdempotencyKeyHeader))
+		if key == "" || !isIdempotentMethod(c.Request.Method) {
+			c.Next()
+			return
+		}
+		if h.IdempotencyService == nil {
+			errx.Handle(c, errx.New(errx.ServiceUnavailable, "idempotency service is not available"))
+			c.Abort()
+			return
+		}
+		if !validIdempotencyKey(key) {
+			errx.Handle(c, errx.New(errx.BadRequest, "Idempotency-Key must be 1-255 visible ASCII characters"))
+			c.Abort()
+			return
+		}
+
+		orgID := GetOrganizationID(c)
+		if orgID == nil {
+			errx.Handle(c, errx.New(errx.BadRequest, "Idempotency-Key requires an organization context"))
+			c.Abort()
+			return
+		}
+
+		body, err := readAndRestoreBody(c)
+		if err != nil {
+			errx.Handle(c, errx.New(errx.BadRequest, "failed to read request body"))
+			c.Abort()
+			return
+		}
+
+		path := c.FullPath()
+		if path == "" {
+			path = c.Request.URL.Path
+		}
+		requestHash := hashRequest(c.Request.Method, path, c.Request.URL.RawQuery, body)
+		record, state, xerr := h.IdempotencyService.Begin(c.Request.Context(), *orgID, key, c.Request.Method, path, requestHash)
+		if xerr != nil {
+			errx.Handle(c, xerr)
+			c.Abort()
+			return
+		}
+
+		switch state {
+		case idempotency.StateReplay:
+			c.Header(IdempotencyReplayedHeader, "true")
+			if record.ContentType != nil && *record.ContentType != "" {
+				c.Header("Content-Type", *record.ContentType)
+			}
+			c.Data(record.StatusCode, c.Writer.Header().Get("Content-Type"), record.ResponseBody)
+			c.Abort()
+			return
+		case idempotency.StateProcessing:
+			errx.Handle(c, errx.New(errx.Conflict, "an identical request is still processing"))
+			c.Abort()
+			return
+		case idempotency.StateConflict:
+			errx.Handle(c, errx.New(errx.Conflict, "Idempotency-Key was already used with a different request"))
+			c.Abort()
+			return
+		}
+
+		capture := &captureResponseWriter{ResponseWriter: c.Writer}
+		c.Writer = capture
+		c.Next()
+
+		status := capture.Status()
+		if status == 0 {
+			status = http.StatusOK
+		}
+		contentType := capture.Header().Get("Content-Type")
+		_ = h.IdempotencyService.Complete(c.Request.Context(), record.ID, status, capture.body.Bytes(), contentType)
+	}
+}
+
+func isIdempotentMethod(method string) bool {
+	switch method {
+	case http.MethodPost, http.MethodPut, http.MethodPatch, http.MethodDelete:
+		return true
+	default:
+		return false
+	}
+}
+
+func validIdempotencyKey(key string) bool {
+	if key == "" || len(key) > 255 {
+		return false
+	}
+	for _, r := range key {
+		if r < 33 || r > 126 {
+			return false
+		}
+	}
+	return true
+}
+
+func readAndRestoreBody(c *gin.Context) ([]byte, error) {
+	if c.Request.Body == nil {
+		return nil, nil
+	}
+	body, err := io.ReadAll(c.Request.Body)
+	if err != nil {
+		return nil, err
+	}
+	c.Request.Body = io.NopCloser(bytes.NewReader(body))
+	return body, nil
+}
+
+func hashRequest(method, path, rawQuery string, body []byte) string {
+	h := sha256.New()
+	h.Write([]byte(method))
+	h.Write([]byte{0})
+	h.Write([]byte(path))
+	h.Write([]byte{0})
+	h.Write([]byte(rawQuery))
+	h.Write([]byte{0})
+	h.Write(body)
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+type captureResponseWriter struct {
+	gin.ResponseWriter
+	body bytes.Buffer
+}
+
+func (w *captureResponseWriter) Write(data []byte) (int, error) {
+	w.body.Write(data)
+	return w.ResponseWriter.Write(data)
+}
+
+func (w *captureResponseWriter) WriteString(data string) (int, error) {
+	w.body.WriteString(data)
+	return w.ResponseWriter.WriteString(data)
+}
@@ -0,0 +1,108 @@
+package middleware
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/app/idempotency"
+	"github.com/warmbly/warmbly/internal/errx"
+)
+
+type fakeIdempotencyService struct {
+	record *idempotency.Record
+	state  idempotency.State
+	body   []byte
+}
+
+func (s *fakeIdempotencyService) Begin(ctx context.Context, orgID uuid.UUID, key, method, path, requestHash string) (*idempotency.Record, idempotency.State, *errx.Error) {
+	if s.record == nil {
+		s.record = &idempotency.Record{ID: uuid.New(), StatusCode: http.StatusCreated}
+	}
+	return s.record, s.state, nil
+}
+
+func (s *fakeIdempotencyService) Complete(ctx context.Context, recordID uuid.UUID, statusCode int, responseBody []byte, contentType string) *errx.Error {
+	s.record.StatusCode = statusCode
+	s.body = append([]byte(nil), responseBody...)
+	return nil
+}
+
+func TestIdempotencyMiddlewareStoresResponse(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	orgID := uuid.New()
+	svc := &fakeIdempotencyService{state: idempotency.StateStarted}
+	h := &Handler{IdempotencyService: svc}
+
+	r := gin.New()
+	r.Use(RequestIDMiddleware())
+	r.POST("/contacts",
+		func(c *gin.Context) {
+			c.Set(OrganizationIDKey, orgID)
+			c.Next()
+		},
+		h.IdempotencyMiddleware(),
+		func(c *gin.Context) {
+			c.JSON(http.StatusCreated, gin.H{"id": "contact_123"})
+		},
+	)
+
+	req := httptest.NewRequest(http.MethodPost, "/contacts", nil)
+	req.Header.Set(IdempotencyKeyHeader, "idem_123")
+	rec := httptest.NewRecorder()
+	r.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("status = %d, want %d", rec.Code, http.StatusCreated)
+	}
+	if string(svc.body) == "" {
+		t.Fatal("expected response body to be stored")
+	}
+}
+
+func TestIdempotencyMiddlewareReplaysResponse(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	orgID := uuid.New()
+	contentType := "application/json; charset=utf-8"
+	svc := &fakeIdempotencyService{
+		state: idempotency.StateReplay,
+		record: &idempotency.Record{
+			ID:           uuid.New(),
+			StatusCode:   http.StatusCreated,
+			ResponseBody: []byte(`{"id":"contact_123"}`),
+			ContentType:  &contentType,
+		},
+	}
+	h := &Handler{IdempotencyService: svc}
+
+	r := gin.New()
+	r.Use(RequestIDMiddleware())
+	r.POST("/contacts",
+		func(c *gin.Context) {
+			c.Set(OrganizationIDKey, orgID)
+			c.Next()
+		},
+		h.IdempotencyMiddleware(),
+		func(c *gin.Context) {
+			c.JSON(http.StatusTeapot, gin.H{"unexpected": true})
+		},
+	)
+
+	req := httptest.NewRequest(http.MethodPost, "/contacts", nil)
+	req.Header.Set(IdempotencyKeyHeader, "idem_123")
+	rec := httptest.NewRecorder()
+	r.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("status = %d, want %d", rec.Code, http.StatusCreated)
+	}
+	if rec.Header().Get(IdempotencyReplayedHeader) != "true" {
+		t.Fatalf("missing replay header")
+	}
+	if got := rec.Body.String(); got != `{"id":"contact_123"}` {
+		t.Fatalf("body = %q", got)
+	}
+}
@@ -0,0 +1,47 @@
+package middleware
+
+import (
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+)
+
+const (
+	RequestIDContextKey = "request_id"
+	RequestIDHeader     = "X-Request-Id"
+)
+
+// RequestIDMiddleware attaches a stable request ID to every request and
+// response. Clients may provide one for cross-system tracing; invalid or
+// oversized values are replaced so logs and responses stay safe.
+func RequestIDMiddleware() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		requestID := sanitizeRequestID(c.GetHeader(RequestIDHeader))
+		if requestID == "" {
+			requestID = uuid.NewString()
+		}
+
+		c.Set(RequestIDContextKey, requestID)
+		c.Header(RequestIDHeader, requestID)
+		c.Next()
+	}
+}
+
+func sanitizeRequestID(value string) string {
+	value = strings.TrimSpace(value)
+	if value == "" || len(value) > 128 {
+		return ""
+	}
+	for _, r := range value {
+		switch {
+		case r >= 'a' && r <= 'z':
+		case r >= 'A' && r <= 'Z':
+		case r >= '0' && r <= '9':
+		case r == '-' || r == '_' || r == '.' || r == ':':
+		default:
+			return ""
+		}
+	}
+	return value
+}
@@ -0,0 +1,55 @@
+package middleware
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+)
+
+func TestRequestIDMiddlewareUsesClientRequestID(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	r := gin.New()
+	r.Use(RequestIDMiddleware())
+	r.GET("/x", func(c *gin.Context) {
+		c.String(http.StatusOK, c.GetString(RequestIDContextKey))
+	})
+
+	req := httptest.NewRequest(http.MethodGet, "/x", nil)
+	req.Header.Set(RequestIDHeader, "client-trace_123")
+	rec := httptest.NewRecorder()
+	r.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want %d", rec.Code, http.StatusOK)
+	}
+	if got := rec.Header().Get(RequestIDHeader); got != "client-trace_123" {
+		t.Fatalf("response request id = %q", got)
+	}
+	if got := rec.Body.String(); got != "client-trace_123" {
+		t.Fatalf("context request id = %q", got)
+	}
+}
+
+func TestRequestIDMiddlewareReplacesUnsafeRequestID(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	r := gin.New()
+	r.Use(RequestIDMiddleware())
+	r.GET("/x", func(c *gin.Context) {
+		c.String(http.StatusOK, c.GetString(RequestIDContextKey))
+	})
+
+	req := httptest.NewRequest(http.MethodGet, "/x", nil)
+	req.Header.Set(RequestIDHeader, "bad/request/id")
+	rec := httptest.NewRecorder()
+	r.ServeHTTP(rec, req)
+
+	got := rec.Header().Get(RequestIDHeader)
+	if got == "" || got == "bad/request/id" {
+		t.Fatalf("response request id = %q", got)
+	}
+	if got != rec.Body.String() {
+		t.Fatalf("header request id %q does not match context %q", got, rec.Body.String())
+	}
+}
@@ -22,6 +22,7 @@ func Run(
 	gin.SetMode(ginMode)

 	r := gin.Default()
+	r.Use(middleware.RequestIDMiddleware())

 	r.GET("/health", func(c *gin.Context) {
 		c.JSON(200, gin.H{"status": "ok"})
@@ -37,6 +38,12 @@ func Run(
 	r.POST("/api/v1/integrations/inbound/calendly/:secret", h.InboundCalendly)
 	r.POST("/api/v1/integrations/inbound/cal-com/:secret", h.InboundCalCom)

+	// Public worker enrollment. The one-time enrollment token is the
+	// credential; successful exchange returns a dotenv file for the installer
+	// and consumes the token.
+	r.GET("/worker-install.sh", h.ServeWorkerInstaller)
+	r.POST("/api/v1/workers/enroll", h.EnrollWorker)
+
 	// Public OAuth-bouncer pages used by the mailbox onboarding popup.
 	// The provider redirects here; the page postMessages the code/state
 	// back to the SPA opener which then calls /emails/onboarding/oauth/finish.
@@ -61,10 +68,23 @@ func Run(
 	}

 	corsConfig := cors.Config{
-		AllowMethods:  []string{"POST", "GET", "PATCH", "OPTIONS", "DELETE"},
-		AllowHeaders:  []string{"Origin", "Content-Type", "Authorization"},
-		ExposeHeaders: []string{"Content-Length"},
-		MaxAge:        12 * time.Hour,
+		AllowMethods: []string{"POST", "GET", "PUT", "PATCH", "OPTIONS", "DELETE"},
+		AllowHeaders: []string{
+			"Origin",
+			"Content-Type",
+			"Authorization",
+			"Idempotency-Key",
+			"X-Request-Id",
+		},
+		ExposeHeaders: []string{
+			"Content-Length",
+			"X-Request-Id",
+			"X-RateLimit-Limit",
+			"X-RateLimit-Remaining",
+			"X-RateLimit-Policy",
+			"Retry-After",
+		},
+		MaxAge: 12 * time.Hour,
 	}
 	switch {
 	case len(allowedOrigins) == 0 && ginMode != gin.ReleaseMode:
@@ -127,17 +147,17 @@ func Run(
 	// auth types; APIKeyUsageMiddleware records one log row per API-key
 	// request (JWT requests are skipped).
 	protected := r.Group("")
-	protected.Use(m.CombinedAuthMiddleware(), m.APIKeyUsageMiddleware())
+	protected.Use(m.CombinedAuthMiddleware(), m.APIKeyUsageMiddleware(), m.IdempotencyMiddleware())
 	{
 		emails := protected.Group("/emails")
 		emails.Use(m.RateLimitMiddleware(models.RateLimitWrite))
 		{
 			emails.GET("", m.RequireAccess(models.PermViewCampaigns, models.APIPermReadEmails), h.EmailsSearch)
-			emails.GET("/:id", m.RequireAccess(models.PermViewCampaigns, models.APIPermReadEmails), h.GetEmail)
-			emails.PATCH("/:id", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), h.UpdateEmail)
-			emails.PATCH("/:id/track", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), h.UpdateEmailTrackingDomain)
-			emails.DELETE("/:id", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), h.DeleteEmail)
-			emails.POST("/:id/send", m.RequireOrganization(), m.RequireAccess(models.PermSendCampaigns, models.APIPermSendCampaigns), h.SendEmailFromAccount)
+			emails.GET("/:id", m.RequireAccess(models.PermViewCampaigns, models.APIPermReadEmails), middleware.RequireAPIKeyEmailAccountParam("id"), h.GetEmail)
+			emails.PATCH("/:id", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), middleware.RequireAPIKeyEmailAccountParam("id"), h.UpdateEmail)
+			emails.PATCH("/:id/track", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), middleware.RequireAPIKeyEmailAccountParam("id"), h.UpdateEmailTrackingDomain)
+			emails.DELETE("/:id", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), middleware.RequireAPIKeyEmailAccountParam("id"), h.DeleteEmail)
+			emails.POST("/:id/send", m.RequireOrganization(), m.RequireAccess(models.PermSendCampaigns, models.APIPermSendCampaigns), middleware.RequireAPIKeyEmailAccountParam("id"), h.SendEmailFromAccount)
 		}

 		// Email onboarding is JWT-only — it writes user-encrypted refresh
@@ -217,12 +237,11 @@ func Run(
 			contacts.GET("/:id/deals", m.RequireAccess(models.PermViewContacts, models.APIPermReadCRM), h.GetDealsByContact)
 		}

-		// Group endpoints (folders / tags / categories) don't yet have
-		// dedicated permission bits — gate them on the broadest read scope
-		// for now so an API key needs at least one collection permission.
-		grouph.New(protected, h.FolderService, "folders")
-		grouph.New(protected, h.TagService, "tags")
-		grouph.New(protected, h.CategoryService, "categories")
+		// Group endpoints map to the resources they organize: campaign
+		// folders, email-account tags, and contact categories.
+		grouph.New(protected, h.FolderService, "folders", m.RequireAccess(models.PermManageCampaigns, models.APIPermWriteCampaigns))
+		grouph.New(protected, h.TagService, "tags", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails))
+		grouph.New(protected, h.CategoryService, "categories", m.RequireAccess(models.PermManageContacts, models.APIPermWriteContacts))

 		unibox := protected.Group("/unibox")
 		unibox.Use(m.RateLimitMiddleware(models.RateLimitRead))
@@ -312,7 +331,7 @@ func Run(

 		// Customer-facing webhooks (org-scoped).
 		webhooks := protected.Group("/webhooks")
-		webhooks.Use(m.RequireOrganization(), m.RateLimitMiddleware(models.RateLimitWrite))
+		webhooks.Use(m.RequireOrganization(), m.RequireAccess(models.PermManageSettings, models.APIPermWebhooks), m.RateLimitMiddleware(models.RateLimitWrite))
 		{
 			webhooks.GET("", h.ListWebhookEndpoints)
 			webhooks.POST("", h.CreateWebhookEndpoint)
@@ -326,7 +345,7 @@ func Run(
 		// "available integrations" list; connections are this org's live
 		// state for each provider.
 		integrations := protected.Group("/integrations")
-		integrations.Use(m.RequireOrganization(), m.RateLimitMiddleware(models.RateLimitWrite))
+		integrations.Use(m.RequireOrganization(), m.RequireAccess(models.PermManageSettings, models.APIPermIntegrations), m.RateLimitMiddleware(models.RateLimitWrite))
 		{
 			integrations.GET("/catalog", h.ListIntegrationCatalog)
 			integrations.GET("/connections", h.ListIntegrationConnections)
@@ -339,7 +358,7 @@ func Run(
 		// preferences for premium-pool partner selection — e.g. send
 		// to Gmail recipients only from Google-classified senders.
 		warmupRouting := protected.Group("/warmup/routing")
-		warmupRouting.Use(m.RequireOrganization(), m.RateLimitMiddleware(models.RateLimitWrite))
+		warmupRouting.Use(m.RequireOrganization(), m.RequireAccess(models.PermManageSettings, models.APIPermWarmupRouting), m.RateLimitMiddleware(models.RateLimitWrite))
 		{
 			warmupRouting.GET("", h.ListWarmupRoutingRules)
 			warmupRouting.POST("", h.CreateWarmupRoutingRule)
@@ -213,7 +213,7 @@ func (s *analyticsService) GetAccountStatus(ctx context.Context, userID, account

 func (s *analyticsService) GetAllAccountStatuses(ctx context.Context, userID uuid.UUID) ([]models.EmailAccountStatus, *errx.Error) {
 	// Get all email accounts for user
-	emailsResult, xerr := s.emailRepo.Search(ctx, userID.String(), "", nil, nil, 1000)
+	emailsResult, xerr := s.emailRepo.Search(ctx, userID.String(), "", nil, nil, 1000, nil)
 	if xerr != nil {
 		return nil, xerr
 	}
@@ -51,7 +51,7 @@ func (s *JobsService) HandleFlagsAdd(ctx context.Context, e *models.JobEventFlag
 		return nil
 	}

-	return s.UniboxRepository.UpdateEntry(
+	if err := s.UniboxRepository.UpdateEntry(
 		ctx,
 		e.UserID,
 		e.EmailID,
@@ -59,7 +59,12 @@ func (s *JobsService) HandleFlagsAdd(ctx context.Context, e *models.JobEventFlag
 		&repository.UpdateUniboxEntry{
 			Flags: email.Flags,
 		},
-	)
+	); err != nil {
+		return err
+	}
+
+	s.publishEmailUpdated(ctx, e.UserID, email)
+	return nil
 }

 func warmupTokenFromFlags(flags []string) string {
@@ -106,7 +111,7 @@ func (s *JobsService) HandleFlagsRemove(ctx context.Context, e *models.JobEventF
 		return nil
 	}

-	return s.UniboxRepository.UpdateEntry(
+	if err := s.UniboxRepository.UpdateEntry(
 		ctx,
 		e.UserID,
 		e.EmailID,
@@ -114,5 +119,11 @@ func (s *JobsService) HandleFlagsRemove(ctx context.Context, e *models.JobEventF
 		&repository.UpdateUniboxEntry{
 			Flags: newFlags,
 		},
-	)
+	); err != nil {
+		return err
+	}
+
+	email.Flags = newFlags
+	s.publishEmailUpdated(ctx, e.UserID, email)
+	return nil
 }
@@ -9,6 +9,7 @@ import (

 	"github.com/google/uuid"
 	"github.com/warmbly/warmbly/internal/config"
+	"github.com/warmbly/warmbly/internal/infrastructure/pubsub"
 	"github.com/warmbly/warmbly/internal/models"
 )

@@ -36,6 +37,9 @@ func (s *JobsService) HandleNewEmail(ctx context.Context, e *models.JobEventNewE
 		CaptureError(e.UserID, e.Message.EmailID, err)
 		return err
 	}
+	if s.StreamingPublisher != nil && e.Message != nil {
+		s.StreamingPublisher.PublishEmailReceived(ctx, emailInboxEvent(e.UserID, e.Message))
+	}

 	// Advanced reply-intent automation is best-effort and should not block inbox ingest.
 	if s.AdvancedService != nil {
@@ -45,6 +49,25 @@ func (s *JobsService) HandleNewEmail(ctx context.Context, e *models.JobEventNewE
 	return nil
 }

+func (s *JobsService) publishEmailUpdated(ctx context.Context, userID uuid.UUID, message *models.EmailMessageStoreData) {
+	if s.StreamingPublisher == nil || message == nil {
+		return
+	}
+	s.StreamingPublisher.PublishEmailUpdated(ctx, emailInboxEvent(userID, message))
+}
+
+func emailInboxEvent(userID uuid.UUID, message *models.EmailMessageStoreData) *pubsub.EmailInboxEvent {
+	return &pubsub.EmailInboxEvent{
+		BaseEvent:      pubsub.BaseEvent{UserID: userID.String()},
+		EmailAccountID: message.EmailID.String(),
+		MessageID:      message.ID.String(),
+		ThreadID:       message.ThreadID,
+		Subject:        message.Subject,
+		From:           strings.Join(message.FromAddr, ", "),
+		Preview:        message.Snippet,
+	}
+}
+
 // extractHeaderValue extracts a custom header value from the email message
 // Checks InReplyTo field encoding or direct header access
 func extractHeaderValue(msg *models.EmailMessageStoreData, headerName string) string {
@@ -31,5 +31,14 @@ func (s *JobsService) HandleUpdateEmail(ctx context.Context, e *models.JobEventE
 		updateData.ModSeq = &e.ModSeq
 	}

-	return s.UniboxRepository.UpdateEntry(ctx, e.UserID, e.EmailID, e.ID, &updateData)
+	if err := s.UniboxRepository.UpdateEntry(ctx, e.UserID, e.EmailID, e.ID, &updateData); err != nil {
+		return err
+	}
+
+	email.Flags = e.Flags
+	email.UID = e.UID
+	email.Mailbox = e.Mailbox
+	email.ModSeq = e.ModSeq
+	s.publishEmailUpdated(ctx, e.UserID, email)
+	return nil
 }
@@ -31,7 +31,13 @@ func (s *contactService) Add(ctx context.Context, userID string, contacts []mode
 		}
 	}

-	return s.contactRepository.Add(ctx, userID, contacts)
+	created, xerr := s.contactRepository.Add(ctx, userID, contacts)
+	if xerr != nil {
+		return nil, xerr
+	}
+
+	s.publishContactsReload(ctx, userID, "contacts:add")
+	return created, nil
 }

 func (s *contactService) Search(ctx context.Context, userID, cursor, category, limit string, filters models.SearchContacts) (*models.ContactsResult, *errx.Error) {
@@ -53,19 +59,41 @@ func (s *contactService) Search(ctx context.Context, userID, cursor, category, l
 }

 func (s *contactService) BulkUpdate(ctx context.Context, userID string, data *models.BulkEditContactsData) ([]models.Contact, *errx.Error) {
-	return s.contactRepository.BulkUpdate(ctx, userID, data)
+	updated, xerr := s.contactRepository.BulkUpdate(ctx, userID, data)
+	if xerr != nil {
+		return nil, xerr
+	}
+
+	s.publishContactsReload(ctx, userID, "contacts:bulk_update")
+	return updated, nil
 }

 func (s *contactService) Update(ctx context.Context, userID, contactID string, data *models.UpdateContact) (*models.Contact, *errx.Error) {
-	return s.contactRepository.Update(ctx, userID, contactID, data)
+	updated, xerr := s.contactRepository.Update(ctx, userID, contactID, data)
+	if xerr != nil {
+		return nil, xerr
+	}
+
+	s.publishContactsReload(ctx, userID, "contacts:update:"+contactID)
+	return updated, nil
 }

 func (s *contactService) BulkDelete(ctx context.Context, userID string, contactIDs []string) *errx.Error {
-	return s.contactRepository.BulkDelete(ctx, userID, contactIDs)
+	if xerr := s.contactRepository.BulkDelete(ctx, userID, contactIDs); xerr != nil {
+		return xerr
+	}
+
+	s.publishContactsReload(ctx, userID, "contacts:bulk_delete")
+	return nil
 }

 func (s *contactService) Delete(ctx context.Context, userID string, contactID string) *errx.Error {
-	return s.contactRepository.Delete(ctx, userID, contactID)
+	if xerr := s.contactRepository.Delete(ctx, userID, contactID); xerr != nil {
+		return xerr
+	}
+
+	s.publishContactsReload(ctx, userID, "contacts:delete:"+contactID)
+	return nil
 }

 func (s *contactService) GetDetail(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID) (*models.ContactDetail, *errx.Error) {
@@ -296,6 +296,9 @@ func (s *contactService) ImportCommit(
 	}

 	res.EndedAt = time.Now().UTC()
+	if res.Imported > 0 || res.Updated > 0 {
+		s.publishContactsReload(ctx, userID, "contacts:import")
+	}
 	return res, nil
 }

@@ -7,6 +7,7 @@ import (

 	"github.com/google/uuid"
 	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/infrastructure/pubsub"
 	"github.com/warmbly/warmbly/internal/models"
 	"github.com/warmbly/warmbly/internal/repository"
 )
@@ -47,19 +48,34 @@ type ContactService interface {
 }

 type contactService struct {
-	contactRepository repository.ContactRepository
-	subRepo           repository.SubscriptionRepository
-	planRepo          repository.PlanRepository
+	contactRepository  repository.ContactRepository
+	subRepo            repository.SubscriptionRepository
+	planRepo           repository.PlanRepository
+	streamingPublisher *pubsub.StreamingPublisher
 }

 func NewService(
 	contactRepository repository.ContactRepository,
 	subRepo repository.SubscriptionRepository,
 	planRepo repository.PlanRepository,
+	streamingPublisher ...*pubsub.StreamingPublisher,
 ) ContactService {
+	var publisher *pubsub.StreamingPublisher
+	if len(streamingPublisher) > 0 {
+		publisher = streamingPublisher[0]
+	}
+
 	return &contactService{
-		contactRepository: contactRepository,
-		subRepo:           subRepo,
-		planRepo:          planRepo,
+		contactRepository:  contactRepository,
+		subRepo:            subRepo,
+		planRepo:           planRepo,
+		streamingPublisher: publisher,
 	}
 }
+
+func (s *contactService) publishContactsReload(ctx context.Context, userID string, operationID string) {
+	if s.streamingPublisher == nil {
+		return
+	}
+	s.streamingPublisher.PublishContactsReload(ctx, userID, operationID)
+}
@@ -3,12 +3,14 @@ package email
 import (
 	"context"

+	"github.com/google/uuid"
 	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/infrastructure/pubsub"
 	"github.com/warmbly/warmbly/internal/models"
 	"github.com/warmbly/warmbly/internal/utils/validate"
 )

-func (s *emailService) Search(ctx context.Context, userID, search, cursor, tag, limit string) (*models.EmailsResult, *errx.Error) {
+func (s *emailService) Search(ctx context.Context, userID, search, cursor, tag, limit string, allowedAccountIDs []uuid.UUID) (*models.EmailsResult, *errx.Error) {
 	cursorId, err := validate.Uuid(cursor)
 	if err != nil {
 		return nil, err
@@ -27,7 +29,7 @@ func (s *emailService) Search(ctx context.Context, userID, search, cursor, tag,
 		return nil, err
 	}

-	return s.emailRepository.Search(ctx, userID, search, cursorId, tagId, limitN)
+	return s.emailRepository.Search(ctx, userID, search, cursorId, tagId, limitN, allowedAccountIDs)
 }

 func (s *emailService) Get(ctx context.Context, userID, emailAccountID string) (*models.Email, *errx.Error) {
@@ -41,6 +43,7 @@ func (s *emailService) Update(ctx context.Context, userID, emailAccountID string
 	}

 	s.syncWarmupPoolMembership(ctx, account)
+	s.publishAccountEvent(ctx, pubsub.EventAccountSynced, account)
 	return account, nil
 }

@@ -59,6 +62,7 @@ func (s *emailService) Delete(ctx context.Context, userID, emailAccountID string
 	}

 	s.removeFromAllWarmupPools(ctx, account)
+	s.publishAccountEvent(ctx, pubsub.EventAccountDisconnected, account)

 	if s.webhookService != nil && account != nil && account.OrganizationID != nil {
 		_, _ = s.webhookService.Dispatch(ctx, *account.OrganizationID, models.WebhookEventEmailAccountRemoved, map[string]any{
@@ -14,6 +14,7 @@ import (
 	"github.com/warmbly/warmbly/internal/app/dailythrottle"
 	"github.com/warmbly/warmbly/internal/config"
 	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/infrastructure/pubsub"
 	"github.com/warmbly/warmbly/internal/models"
 	"github.com/warmbly/warmbly/internal/pkg/crypt"
 	"golang.org/x/oauth2"
@@ -161,6 +162,7 @@ func (s *emailService) OAuthFinish(ctx context.Context, userID, code, state stri
 		ExpiresAt:      tok.Expiry,
 	})
 	if xerr == nil && acc != nil {
+		s.publishAccountEvent(ctx, pubsub.EventAccountConnected, acc)
 		s.dispatchAccountConnected(ctx, sess.OrganizationID, acc)
 	}
 	return acc, xerr
@@ -222,6 +224,7 @@ func (s *emailService) OnboardSMTPIMAP(ctx context.Context, userID string, orgID
 		}
 	}

+	s.publishAccountEvent(ctx, pubsub.EventAccountConnected, acc)
 	s.dispatchAccountConnected(ctx, orgID, acc)
 	return acc, nil
 }
@@ -15,12 +15,13 @@ import (
 	"github.com/warmbly/warmbly/internal/events"
 	"github.com/warmbly/warmbly/internal/infrastructure/cache"
 	"github.com/warmbly/warmbly/internal/infrastructure/kafka"
+	"github.com/warmbly/warmbly/internal/infrastructure/pubsub"
 	"github.com/warmbly/warmbly/internal/models"
 	"github.com/warmbly/warmbly/internal/repository"
 )

 type EmailService interface {
-	Search(ctx context.Context, userID, search, cursor, tag, limit string) (*models.EmailsResult, *errx.Error)
+	Search(ctx context.Context, userID, search, cursor, tag, limit string, allowedAccountIDs []uuid.UUID) (*models.EmailsResult, *errx.Error)
 	Get(ctx context.Context, userID, emailAccountID string) (*models.Email, *errx.Error)
 	Update(ctx context.Context, userID, emailAccountID string, udata *models.UpdateEmail) (*models.Email, *errx.Error)
 	UpdateTrackingDomain(ctx context.Context, userID, emailAccountID, domain string) *errx.Error
@@ -38,16 +39,17 @@ type EmailService interface {
 }

 type emailService struct {
-	emailRepository  repository.EmailRepository
-	cipherService    cipher.CipherService
-	featureGate      feature.FeatureGateService
-	warmupService    warmupapp.Service
-	publisher        events.Publisher
-	producer         *kafka.Producer
-	r                *cache.Cache
-	oauthInbox       *config.Oauth2Inbox
-	workerAssignment worker.WorkerAssignmentService
-	throttle         dailythrottle.Service
+	emailRepository    repository.EmailRepository
+	cipherService      cipher.CipherService
+	featureGate        feature.FeatureGateService
+	warmupService      warmupapp.Service
+	publisher          events.Publisher
+	streamingPublisher *pubsub.StreamingPublisher
+	producer           *kafka.Producer
+	r                  *cache.Cache
+	oauthInbox         *config.Oauth2Inbox
+	workerAssignment   worker.WorkerAssignmentService
+	throttle           dailythrottle.Service
 	// webhookService is optional. When non-nil, account lifecycle events
 	// (email_account.connected, email_account.removed) are dispatched to
 	// subscribed customer webhooks.
@@ -74,13 +76,20 @@ func NewService(
 	featureGate feature.FeatureGateService,
 	warmupService warmupapp.Service,
 	publisher events.Publisher,
+	streamingPublisher ...*pubsub.StreamingPublisher,
 ) EmailService {
+	var realtime *pubsub.StreamingPublisher
+	if len(streamingPublisher) > 0 {
+		realtime = streamingPublisher[0]
+	}
+
 	return &emailService{
-		emailRepository: emailRepository,
-		cipherService:   cipherService,
-		featureGate:     featureGate,
-		warmupService:   warmupService,
-		publisher:       publisher,
+		emailRepository:    emailRepository,
+		cipherService:      cipherService,
+		featureGate:        featureGate,
+		warmupService:      warmupService,
+		publisher:          publisher,
+		streamingPublisher: realtime,
 	}
 }

@@ -94,16 +103,40 @@ func NewServiceWithKafka(
 	r *cache.Cache,
 	oauthInbox *config.Oauth2Inbox,
 	workerAssignment worker.WorkerAssignmentService,
+	streamingPublisher ...*pubsub.StreamingPublisher,
 ) EmailService {
+	var realtime *pubsub.StreamingPublisher
+	if len(streamingPublisher) > 0 {
+		realtime = streamingPublisher[0]
+	}
+
 	return &emailService{
-		emailRepository:  emailRepository,
-		cipherService:    cipherService,
-		featureGate:      featureGate,
-		warmupService:    warmupService,
-		publisher:        publisher,
-		producer:         producer,
-		r:                r,
-		oauthInbox:       oauthInbox,
-		workerAssignment: workerAssignment,
+		emailRepository:    emailRepository,
+		cipherService:      cipherService,
+		featureGate:        featureGate,
+		warmupService:      warmupService,
+		publisher:          publisher,
+		streamingPublisher: realtime,
+		producer:           producer,
+		r:                  r,
+		oauthInbox:         oauthInbox,
+		workerAssignment:   workerAssignment,
 	}
 }
+
+func (s *emailService) publishAccountEvent(ctx context.Context, eventType pubsub.EventType, account *models.Email) {
+	if s.streamingPublisher == nil || account == nil {
+		return
+	}
+
+	s.streamingPublisher.PublishAccountEvent(ctx, &pubsub.AccountEvent{
+		BaseEvent: pubsub.BaseEvent{
+			EventType: eventType,
+			UserID:    account.UserID,
+		},
+		EmailAccountID: account.ID.String(),
+		Email:          account.Email,
+		Provider:       account.Provider,
+		Status:         account.Status,
+	})
+}
@@ -0,0 +1,129 @@
+package idempotency
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/warmbly/warmbly/internal/errx"
+)
+
+const ttl = 24 * time.Hour
+
+type State string
+
+const (
+	StateStarted    State = "started"
+	StateReplay     State = "replay"
+	StateProcessing State = "processing"
+	StateConflict   State = "conflict"
+)
+
+type Record struct {
+	ID           uuid.UUID
+	Method       string
+	Path         string
+	RequestHash  string
+	Status       string
+	StatusCode   int
+	ResponseBody []byte
+	ContentType  *string
+}
+
+type Service interface {
+	Begin(ctx context.Context, orgID uuid.UUID, key, method, path, requestHash string) (*Record, State, *errx.Error)
+	Complete(ctx context.Context, recordID uuid.UUID, statusCode int, responseBody []byte, contentType string) *errx.Error
+}
+
+type service struct {
+	db *pgxpool.Pool
+}
+
+func NewService(db *pgxpool.Pool) Service {
+	return &service{db: db}
+}
+
+func (s *service) Begin(ctx context.Context, orgID uuid.UUID, key, method, path, requestHash string) (*Record, State, *errx.Error) {
+	if s == nil || s.db == nil {
+		return nil, "", errx.New(errx.ServiceUnavailable, "idempotency service is not available")
+	}
+
+	_, _ = s.db.Exec(ctx, `
+		DELETE FROM api_idempotency_keys
+		WHERE organization_id = $1 AND key = $2 AND expires_at < now()
+	`, orgID, key)
+
+	var id uuid.UUID
+	err := s.db.QueryRow(ctx, `
+		INSERT INTO api_idempotency_keys (
+			organization_id, key, method, path, request_hash, status, expires_at
+		)
+		VALUES ($1, $2, $3, $4, $5, 'processing', now() + ($6::integer * interval '1 second'))
+		ON CONFLICT (organization_id, key) DO NOTHING
+		RETURNING id
+	`, orgID, key, method, path, requestHash, int(ttl.Seconds())).Scan(&id)
+	if err == nil {
+		return &Record{ID: id, Method: method, Path: path, RequestHash: requestHash, Status: "processing"}, StateStarted, nil
+	}
+	if err != pgx.ErrNoRows {
+		return nil, "", errx.InternalError()
+	}
+
+	record, xerr := s.get(ctx, orgID, key)
+	if xerr != nil {
+		return nil, "", xerr
+	}
+	if record.Method != method || record.Path != path || record.RequestHash != requestHash {
+		return record, StateConflict, nil
+	}
+	if record.Status == "completed" {
+		return record, StateReplay, nil
+	}
+	return record, StateProcessing, nil
+}
+
+func (s *service) Complete(ctx context.Context, recordID uuid.UUID, statusCode int, responseBody []byte, contentType string) *errx.Error {
+	if s == nil || s.db == nil {
+		return errx.New(errx.ServiceUnavailable, "idempotency service is not available")
+	}
+	_, err := s.db.Exec(ctx, `
+		UPDATE api_idempotency_keys
+		SET status = 'completed',
+		    status_code = $2,
+		    response_body = $3,
+		    content_type = NULLIF($4, ''),
+		    updated_at = now()
+		WHERE id = $1
+	`, recordID, statusCode, responseBody, contentType)
+	if err != nil {
+		return errx.InternalError()
+	}
+	return nil
+}
+
+func (s *service) get(ctx context.Context, orgID uuid.UUID, key string) (*Record, *errx.Error) {
+	var record Record
+	err := s.db.QueryRow(ctx, `
+		SELECT id, method, path, request_hash, status, COALESCE(status_code, 0), COALESCE(response_body, ''::bytea), content_type
+		FROM api_idempotency_keys
+		WHERE organization_id = $1 AND key = $2
+	`, orgID, key).Scan(
+		&record.ID,
+		&record.Method,
+		&record.Path,
+		&record.RequestHash,
+		&record.Status,
+		&record.StatusCode,
+		&record.ResponseBody,
+		&record.ContentType,
+	)
+	if err == pgx.ErrNoRows {
+		return nil, errx.ErrNotFound
+	}
+	if err != nil {
+		return nil, errx.InternalError()
+	}
+	return &record, nil
+}
@@ -14,8 +14,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
 	"time"

@@ -186,8 +188,9 @@ func generateSecret() (string, error) {
 }

 // validateURL keeps malformed entries and obvious SSRF targets out of the
-// table. We do not enforce HTTPS at insert time because internal-network
-// integrations and ngrok-style local tests legitimately use http://.
+// table. Public webhook endpoints must use HTTPS and route to public hosts.
+// Local/self-hosted development can set WARMBLY_ALLOW_UNSAFE_WEBHOOK_URLS=true
+// to permit HTTP and private targets.
 func validateURL(raw string) error {
 	raw = strings.TrimSpace(raw)
 	if raw == "" {
@@ -197,15 +200,35 @@ func validateURL(raw string) error {
 	if err != nil {
 		return fmt.Errorf("invalid url: %w", err)
 	}
-	if u.Scheme != "http" && u.Scheme != "https" {
-		return fmt.Errorf("url scheme must be http or https")
+	allowUnsafe := strings.EqualFold(os.Getenv("WARMBLY_ALLOW_UNSAFE_WEBHOOK_URLS"), "true")
+	if u.Scheme != "https" && !(allowUnsafe && u.Scheme == "http") {
+		return fmt.Errorf("url scheme must be https")
 	}
 	if u.Host == "" {
 		return fmt.Errorf("url must have a host")
 	}
+	if !allowUnsafe && isPrivateWebhookHost(u.Hostname()) {
+		return fmt.Errorf("url host must be publicly routable")
+	}
 	return nil
 }

+func isPrivateWebhookHost(host string) bool {
+	host = strings.Trim(strings.ToLower(host), "[]")
+	if host == "" || host == "localhost" || strings.HasSuffix(host, ".localhost") {
+		return true
+	}
+	ip := net.ParseIP(host)
+	if ip == nil {
+		return false
+	}
+	return ip.IsLoopback() ||
+		ip.IsPrivate() ||
+		ip.IsLinkLocalUnicast() ||
+		ip.IsLinkLocalMulticast() ||
+		ip.IsUnspecified()
+}
+
 func validateEventTypes(eventTypes []string) error {
 	for _, t := range eventTypes {
 		if !models.IsValidWebhookEventType(t) {
@@ -73,7 +73,10 @@ func TestValidateURL_RejectsBadInput(t *testing.T) {
 		"javascript:alert(1)":        true,
 		"http://":                    true,
 		"https://example.com/hook":   false,
-		"http://localhost:3000/hook": false,
+		"http://localhost:3000/hook": true,
+		"https://localhost/hook":     true,
+		"https://127.0.0.1/hook":     true,
+		"https://10.0.0.10/hook":     true,
 	}
 	for input, wantErr := range cases {
 		err := validateURL(input)
@@ -85,3 +88,11 @@ func TestValidateURL_RejectsBadInput(t *testing.T) {
 		}
 	}
 }
+
+func TestValidateURL_AllowsUnsafeLocalDevelopmentWhenEnabled(t *testing.T) {
+	t.Setenv("WARMBLY_ALLOW_UNSAFE_WEBHOOK_URLS", "true")
+
+	if err := validateURL("http://localhost:3000/hook"); err != nil {
+		t.Fatalf("expected unsafe local URL to be accepted in development mode: %v", err)
+	}
+}
@@ -57,6 +57,13 @@ type WorkerEnvConfig struct {
 	AWSRegion          string
 	AWSAccessKeyID     string
 	AWSSecretAccessKey string
+
+	EncryptedKeysBackendURL  string
+	EncryptedKeysWorkerToken string
+
+	EventBusProvider string
+	NATSURL          string
+	CodecProvider    string
 }

 type Orchestrator struct {
@@ -143,6 +150,30 @@ func (o *Orchestrator) Install(ctx context.Context, workerID uuid.UUID) error {
 	return o.repo.UpdateInstallState(ctx, workerID, models.WorkerInstallStateInstalled, "")
 }

+func (o *Orchestrator) InstallerScript() ([]byte, error) {
+	return os.ReadFile(o.installerPath)
+}
+
+// RenderEnrollmentEnv returns a complete dotenv payload for the one-command
+// enrollment flow. The token exchange authenticates the caller; this method
+// only renders the config the installer writes to disk.
+func (o *Orchestrator) RenderEnrollmentEnv(ctx context.Context, workerID uuid.UUID) (string, string, error) {
+	envContent, image, err := o.renderEnvFile(ctx, workerID)
+	if err != nil {
+		return "", "", err
+	}
+	var b strings.Builder
+	b.WriteString("# Warmbly worker enrollment config\n")
+	b.WriteString("WORKER_ID=")
+	b.WriteString(workerID.String())
+	b.WriteString("\n")
+	b.WriteString("WARMBLY_WORKER_IMAGE=")
+	b.WriteString(image)
+	b.WriteString("\n")
+	b.WriteString(envContent)
+	return b.String(), image, nil
+}
+
 // ApplyConfig re-writes /etc/warmbly/worker.env from the worker's current
 // profile + AWS creds and restarts the service. Cheaper than Install — does
 // not touch Docker or the installer script. Use after a credential change.
@@ -578,6 +609,9 @@ func (o *Orchestrator) renderEnvFile(ctx context.Context, workerID uuid.UUID) (e
 	write("AWS_REGION", env.AWSRegion)
 	write("AWS_ACCESS_KEY_ID", env.AWSAccessKeyID)
 	write("AWS_SECRET_ACCESS_KEY", env.AWSSecretAccessKey)
+	write("ENCRYPTED_KEYS_PROVIDER", "http")
+	write("ENCRYPTED_KEYS_BACKEND_URL", env.EncryptedKeysBackendURL)
+	write("ENCRYPTED_KEYS_WORKER_TOKEN", env.EncryptedKeysWorkerToken)
 	write("KAFKA_BOOTSTRAP_SERVERS", env.KafkaBootstrap)
 	write("KAFKA_SASL_USERNAME", env.KafkaSASLUsername)
 	write("KAFKA_SASL_PASSWORD", env.KafkaSASLPassword)
@@ -585,6 +619,9 @@ func (o *Orchestrator) renderEnvFile(ctx context.Context, workerID uuid.UUID) (e
 	write("SCHEMA_REGISTRY_KEY", env.SchemaRegistryKey)
 	write("SCHEMA_REGISTRY_SECRET", env.SchemaRegistrySecret)
 	write("REDIS", env.RedisURL)
+	write("EVENTBUS_PROVIDER", env.EventBusProvider)
+	write("NATS_URL", env.NATSURL)
+	write("CODEC_PROVIDER", env.CodecProvider)
 	return b.String(), image, nil
 }

@@ -42,3 +42,16 @@ var codeToString = map[Code]string{
 	NotImplemented:     "Not Implemented",
 	ServiceUnavailable: "Service Unavailable",
 }
+
+var codeToIdentifier = map[Code]string{
+	BadRequest:         "bad_request",
+	Unauthorized:       "unauthorized",
+	Forbidden:          "forbidden",
+	NotFound:           "not_found",
+	Conflict:           "conflict",
+	Unprocessable:      "unprocessable",
+	TooManyRequests:    "rate_limit_exceeded",
+	Internal:           "internal_error",
+	NotImplemented:     "not_implemented",
+	ServiceUnavailable: "service_unavailable",
+}
@@ -35,8 +35,10 @@ var (

 // --- Gin handler helper ---
 type response struct {
-	Error   string `json:"error"`
-	Message string `json:"message"`
+	Error     string `json:"error"`
+	Message   string `json:"message"`
+	Code      string `json:"code"`
+	RequestID string `json:"request_id,omitempty"`
 }

 func InternalError() *Error {
@@ -50,8 +52,10 @@ func Handle(c *gin.Context, err error) {
 		httpCode := codeToHTTP[bizErr.Code]
 		httpError := codeToString[bizErr.Code]
 		c.JSON(httpCode, response{
-			Error:   httpError,
-			Message: bizErr.Message,
+			Error:     httpError,
+			Message:   bizErr.Message,
+			Code:      codeToIdentifier[bizErr.Code],
+			RequestID: c.GetString("request_id"),
 		})
 		return
 	}
@@ -65,7 +69,9 @@ func JSON(c *gin.Context, err *Error) {
 	httpCode := codeToHTTP[err.Code]
 	httpError := codeToString[err.Code]
 	c.JSON(httpCode, response{
-		Error:   httpError,
-		Message: err.Message,
+		Error:     httpError,
+		Message:   err.Message,
+		Code:      codeToIdentifier[err.Code],
+		RequestID: c.GetString("request_id"),
 	})
 }
@@ -0,0 +1,37 @@
+package errx
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+)
+
+func TestJSONIncludesStableCodeAndRequestID(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Set("request_id", "req_test_123")
+
+	JSON(c, New(BadRequest, "invalid cursor"))
+
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want %d", rec.Code, http.StatusBadRequest)
+	}
+
+	var body response
+	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	if body.Code != "bad_request" {
+		t.Fatalf("code = %q", body.Code)
+	}
+	if body.RequestID != "req_test_123" {
+		t.Fatalf("request_id = %q", body.RequestID)
+	}
+	if body.Error != "Bad Request" || body.Message != "invalid cursor" {
+		t.Fatalf("unexpected body: %+v", body)
+	}
+}
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS api_idempotency_keys;
@@ -0,0 +1,19 @@
+CREATE TABLE IF NOT EXISTS api_idempotency_keys (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    organization_id UUID NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
+    key TEXT NOT NULL,
+    method TEXT NOT NULL,
+    path TEXT NOT NULL,
+    request_hash TEXT NOT NULL,
+    status TEXT NOT NULL CHECK (status IN ('processing', 'completed')),
+    status_code INTEGER,
+    response_body BYTEA,
+    content_type TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    expires_at TIMESTAMPTZ NOT NULL,
+    UNIQUE (organization_id, key)
+);
+
+CREATE INDEX IF NOT EXISTS idx_api_idempotency_keys_expires
+    ON api_idempotency_keys (expires_at);
@@ -0,0 +1,7 @@
+DROP INDEX IF EXISTS idx_webhook_deliveries_endpoint_event;
+
+DROP INDEX IF EXISTS idx_webhook_deliveries_due;
+
+CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_due
+    ON webhook_deliveries (next_attempt_at)
+    WHERE status IN ('pending', 'retry');
@@ -0,0 +1,14 @@
+DROP INDEX IF EXISTS idx_webhook_deliveries_due;
+
+CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_due
+    ON webhook_deliveries (next_attempt_at)
+    WHERE status = 'pending';
+
+DELETE FROM webhook_deliveries a
+USING webhook_deliveries b
+WHERE a.endpoint_id = b.endpoint_id
+  AND a.event_id = b.event_id
+  AND a.ctid < b.ctid;
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_webhook_deliveries_endpoint_event
+    ON webhook_deliveries (endpoint_id, event_id);
@@ -184,6 +184,46 @@ func (p *StreamingPublisher) PublishEmailReceived(ctx context.Context, event *Em
 	}
 }

+// PublishEmailUpdated notifies user that an inbox row changed.
+func (p *StreamingPublisher) PublishEmailUpdated(ctx context.Context, event *EmailInboxEvent) {
+	if p.client == nil {
+		return
+	}
+
+	event.EventType = EventEmailUpdated
+	event.Timestamp = time.Now()
+
+	attrs := map[string]string{
+		"user_id":    event.UserID,
+		"email_id":   event.EmailAccountID,
+		"event_type": string(EventEmailUpdated),
+	}
+
+	if err := p.client.Publish(ctx, TopicEmailInbox, event, attrs); err != nil {
+		// Log error but don't fail
+	}
+}
+
+// PublishEmailDeleted notifies user that an inbox row was removed.
+func (p *StreamingPublisher) PublishEmailDeleted(ctx context.Context, event *EmailInboxEvent) {
+	if p.client == nil {
+		return
+	}
+
+	event.EventType = EventEmailDeleted
+	event.Timestamp = time.Now()
+
+	attrs := map[string]string{
+		"user_id":    event.UserID,
+		"email_id":   event.EmailAccountID,
+		"event_type": string(EventEmailDeleted),
+	}
+
+	if err := p.client.Publish(ctx, TopicEmailInbox, event, attrs); err != nil {
+		// Log error but don't fail
+	}
+}
+
 // PublishContactsReload signals frontend to reload contacts
 func (p *StreamingPublisher) PublishContactsReload(ctx context.Context, userID, operationID string) {
 	if p.client == nil {
@@ -43,6 +43,10 @@ const (

 	// Audit trail
 	APIPermReadAuditLogs
+
+	// Organization operations
+	APIPermIntegrations  // Connect and manage third-party integrations
+	APIPermWarmupRouting // Manage warmup routing rules
 )

 // AllAPIPermissionsMask is the OR of every defined permission bit.
@@ -58,7 +62,8 @@ const AllAPIPermissionsMask uint64 = APIPermReadEmails | APIPermReadCampaigns |
 	APIPermSendCampaigns |
 	APIPermReadTemplates | APIPermWriteTemplates |
 	APIPermReadCRM | APIPermWriteCRM |
-	APIPermReadAuditLogs
+	APIPermReadAuditLogs |
+	APIPermIntegrations | APIPermWarmupRouting

 // Preset permission sets surfaced via GET /api-keys/permissions so a
 // caller can grant a sane default without picking bits by hand.
@@ -73,7 +78,8 @@ var (
 		APIPermBulkContacts | APIPermBulkCampaigns |
 		APIPermSendCampaigns |
 		APIPermWriteTemplates | APIPermWriteCRM |
-		APIPermRealtimeSubscribe | APIPermWebhooks | APIPermAPIKeys
+		APIPermRealtimeSubscribe | APIPermWebhooks | APIPermAPIKeys |
+		APIPermIntegrations | APIPermWarmupRouting
 )

 type APIPermission struct {
@@ -104,6 +110,8 @@ var AllAPIPermissions = []APIPermission{
 	{"REALTIME_SUBSCRIBE", APIPermRealtimeSubscribe, "Subscribe to realtime events", "special"},
 	{"WEBHOOKS", APIPermWebhooks, "Manage webhook endpoints", "special"},
 	{"API_KEYS", APIPermAPIKeys, "Create and manage API keys", "special"},
+	{"INTEGRATIONS", APIPermIntegrations, "Connect and manage third-party integrations", "special"},
+	{"WARMUP_ROUTING", APIPermWarmupRouting, "Manage warmup routing rules", "special"},
 }

 // HasAPIPermission reports whether the bitmask grants every bit in `required`.
@@ -40,7 +40,7 @@ type OAuthCredentials struct {
 }

 type EmailRepository interface {
-	Search(ctx context.Context, userID, search string, cursor, tag *string, limit int32) (*models.EmailsResult, *errx.Error)
+	Search(ctx context.Context, userID, search string, cursor, tag *string, limit int32, allowedAccountIDs []uuid.UUID) (*models.EmailsResult, *errx.Error)
 	Get(ctx context.Context, userID, emailAccountID string) (*models.Email, *errx.Error)
 	GetByID(ctx context.Context, emailAccountID uuid.UUID) (*models.Email, *errx.Error)
 	GetByTags(ctx context.Context, userID string, tags []string) ([]models.Email, *errx.Error)
@@ -350,7 +350,7 @@ func (r *emailRepository) NewSMTPIMAPAccount(ctx context.Context, userID string,
 	}, nil
 }

-func (r *emailRepository) Search(ctx context.Context, userID, search string, cursor, tag *string, limit int32) (*models.EmailsResult, *errx.Error) {
+func (r *emailRepository) Search(ctx context.Context, userID, search string, cursor, tag *string, limit int32, allowedAccountIDs []uuid.UUID) (*models.EmailsResult, *errx.Error) {
 	tx, err := r.DB.Begin(ctx)
 	if err != nil {
 		db.CaptureError(err, "", nil, "begin")
@@ -384,17 +384,23 @@ func (r *emailRepository) Search(ctx context.Context, userID, search string, cur
 		 AND ($4::uuid IS NULL OR EXISTS (
 		  SELECT 1 FROM email_tags cf WHERE cf.email_id = ea.id AND cf.tag_id = $4
 		 ))
+		 AND ($6::uuid[] IS NULL OR ea.id = ANY($6::uuid[]))
 		GROUP BY ea.id
 		ORDER BY ea.created_at DESC, ea.id DESC
 		LIMIT $5
 	`

+	var allowedAccountParam any
+	if len(allowedAccountIDs) > 0 {
+		allowedAccountParam = allowedAccountIDs
+	}
 	params := []any{
 		userID,
 		cursor,
 		"%" + search + "%",
 		tag,
 		limit + 1,
+		allowedAccountParam,
 	}

 	rows, err := tx.Query(ctx, query, params...)
@@ -441,12 +447,14 @@ func (r *emailRepository) Search(ctx context.Context, userID, search string, cur
 			  AND ($3::uuid IS NULL OR EXISTS (
 				SELECT 1 FROM email_tags cf WHERE cf.email_id = ea.id AND cf.tag_id = $3
 			  ))
+			  AND ($4::uuid[] IS NULL OR ea.id = ANY($4::uuid[]))
 		`

 		params = []any{
 			userID,
 			"%" + search + "%",
 			tag,
+			allowedAccountParam,
 		}

 		var tmp int64
@@ -24,6 +24,13 @@ func (r *workerRepository) UpsertOnHeartbeat(ctx context.Context, id uuid.UUID,
 		VALUES ($1, $2, $3, TRUE, $4, $5, $6, 'healthy', 0)
 		ON CONFLICT (id) DO UPDATE
 		   SET ip_addr = EXCLUDED.ip_addr,
+		       active = TRUE,
+		       install_state = CASE
+		         WHEN workers.install_state IN ('pending', 'provisioning', 'error') THEN 'installed'::worker_install_state
+		         ELSE workers.install_state
+		       END,
+		       last_seen_at = now(),
+		       last_error = NULL,
 		       updated_at = now()
 	`
 	name := "auto-registered-" + id.String()[:8]
@@ -255,7 +255,11 @@ func (r *workerRepository) ListWorkersByProfile(ctx context.Context, profileID u

 func (r *workerRepository) RecordEnrolledIP(ctx context.Context, id uuid.UUID, ip string) error {
 	_, err := r.db.Exec(ctx, `
-		UPDATE workers SET ip_addr = $2, ssh_host = COALESCE(NULLIF(ssh_host,''), $2), updated_at = NOW()
+		UPDATE workers
+		SET ip_addr = $2,
+		    ssh_host = COALESCE(NULLIF(ssh_host,''), $2),
+		    install_state = 'provisioning',
+		    updated_at = NOW()
 		WHERE id = $1
 	`, id, ip)
 	return err
@@ -86,6 +86,14 @@ defmodule Realtime.CloudPubSub.Subscriber do
      Logger.debug("Broadcast #{event_type} to #{topic}")
    end

+    org_id = event["org_id"] || event["organization_id"]
+
+    if org_id do
+      topic = "org:#{org_id}"
+      Phoenix.PubSub.broadcast(Realtime.PubSub, topic, {:pubsub_event, event})
+      Logger.debug("Broadcast #{event_type} to #{topic}")
+    end
+
    # Broadcast to entity-specific channels
    broadcast_to_entity_channels(event)
  end
@@ -13,7 +13,7 @@
 #
 # Quick start:
 #
-#   curl -fsSL https://get.warmbly.com/worker | sudo bash -s -- \
+#   curl -fsSL https://api.warmbly.com/worker-install.sh | sudo bash -s -- \
 #     --kafka kafka.warmbly.com:9092 \
 #     --schema-registry https://schema.warmbly.com \
 #     --redis redis://cache.warmbly.com:6379 \
@@ -21,9 +21,14 @@
 #
 # Or with a pre-built env file:
 #
-#   curl -fsSL https://get.warmbly.com/worker | sudo bash -s -- \
+#   curl -fsSL https://api.warmbly.com/worker-install.sh | sudo bash -s -- \
 #     --env-file /root/worker.env
 #
+# Or with a one-time enrollment token from the dashboard:
+#
+#   curl -fsSL https://api.warmbly.com/worker-install.sh | sudo bash -s -- \
+#     --enroll wmenroll_...
+#
 # Re-running is safe: existing env values are preserved unless overridden,
 # and the worker ID will resolve to the same value as long as the IP is stable.

@@ -39,10 +44,16 @@ UNIT_FILE="/etc/systemd/system/warmbly-worker.service"
 TEMPLATE_UNIT_FILE="/etc/systemd/system/warmbly-worker@.service"
 INSTANCES_DIR="${CONFIG_DIR}/instances"
 CONTAINER_NAME="warmbly-worker"
+INSTALLER_BIN="/usr/local/bin/warmbly-worker-installer"
+AUTO_UPDATE_UNIT_FILE="/etc/systemd/system/warmbly-worker-auto-update.service"
+AUTO_UPDATE_TIMER_FILE="/etc/systemd/system/warmbly-worker-auto-update.timer"

 ACTION="install"
 INTERACTIVE=1
 SUPPLIED_ENV_FILE=""
+ENROLL_TOKEN=""
+API_BASE="${WARMBLY_API_BASE:-https://api.warmbly.com}"
+AUTO_UPDATE=1

 # Comma-separated list of IPv4 addresses for multi-IP install. When non-empty,
 # the installer drops one templated systemd unit per IP, each bound to that IP
@@ -111,6 +122,10 @@ Configuration flags:
  --tier <shared|dedicated>    Worker tier label (default: shared)
  --image <ref>                Docker image (default: ${IMAGE})
  --env-file <path>            Use this env file verbatim, skip prompts
+  --enroll <token>             Exchange a one-time dashboard enrollment token
+                               for worker config and install without prompts
+  --api-base <url>             API base for --enroll (default: ${API_BASE})
+  --no-auto-update             Do not install the systemd auto-update timer

  --kafka <bootstrap>          Kafka bootstrap servers (host:port[,host:port])
  --kafka-user <user>
@@ -151,6 +166,9 @@ while [[ $# -gt 0 ]]; do
    --tier)            CFG[WORKER_TIER]="$2"; shift 2 ;;
    --image)           IMAGE="$2"; shift 2 ;;
    --env-file)        SUPPLIED_ENV_FILE="$2"; shift 2 ;;
+    --enroll)          ENROLL_TOKEN="$2"; INTERACTIVE=0; shift 2 ;;
+    --api-base)        API_BASE="$2"; shift 2 ;;
+    --no-auto-update)  AUTO_UPDATE=0; shift ;;

    --kafka)           CFG[KAFKA_BOOTSTRAP_SERVERS]="$2"; shift 2 ;;
    --kafka-user)      CFG[KAFKA_SASL_USERNAME]="$2"; shift 2 ;;
@@ -315,6 +333,54 @@ merge_existing_env() {
  done < "$ENV_FILE"
 }

+env_file_value() {
+  local file="$1" key="$2"
+  grep -E "^${key}=" "$file" 2>/dev/null | head -1 | cut -d= -f2-
+}
+
+fetch_enrollment_env() {
+  [[ -n "$ENROLL_TOKEN" ]] || return 1
+  command -v curl >/dev/null 2>&1 || die "curl is required for --enroll"
+
+  local ip="${IP_OVERRIDE}"
+  if [[ -z "$ip" ]]; then
+    ip="$(detect_public_ip || true)"
+  fi
+
+  local tmp; tmp="$(mktemp)"
+  local body
+  body="{\"token\":\"${ENROLL_TOKEN}\""
+  if [[ -n "$ip" ]]; then
+    body+=",\"public_ip\":\"${ip}\""
+  fi
+  body+="}"
+
+  log "exchanging enrollment token at ${API_BASE}"
+  curl -fsS \
+    -X POST \
+    -H "Content-Type: application/json" \
+    -H "Accept: text/plain" \
+    --data "$body" \
+    "${API_BASE%/}/api/v1/workers/enroll" > "$tmp" || {
+      rm -f "$tmp"
+      die "enrollment failed"
+    }
+
+  local worker_id image
+  worker_id="$(env_file_value "$tmp" WORKER_ID)"
+  image="$(env_file_value "$tmp" WARMBLY_WORKER_IMAGE)"
+  [[ -n "$worker_id" ]] || die "enrollment response did not include WORKER_ID"
+
+  install -d -m 0700 "$CONFIG_DIR"
+  install -m 0600 "$tmp" "$ENV_FILE"
+  rm -f "$tmp"
+
+  WORKER_ID_OVERRIDE="$worker_id"
+  [[ -n "$image" ]] && IMAGE="$image"
+  ok "enrollment config installed"
+  return 0
+}
+
 write_env_file() {
  install -d -m 0700 "$CONFIG_DIR"
  local tmp; tmp="$(mktemp)"
@@ -322,10 +388,11 @@ write_env_file() {
    echo "# Warmbly worker config — managed by install-worker.sh"
    echo "# $(date -u +%Y-%m-%dT%H:%M:%SZ)"
    for key in APP_ENV AWS_CONFIG_ENABLED AWS_REGION AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY \
+               ENCRYPTED_KEYS_PROVIDER ENCRYPTED_KEYS_BACKEND_URL ENCRYPTED_KEYS_WORKER_TOKEN \
               KAFKA_BOOTSTRAP_SERVERS KAFKA_SASL_USERNAME KAFKA_SASL_PASSWORD \
               SCHEMA_REGISTRY_URL SCHEMA_REGISTRY_KEY SCHEMA_REGISTRY_SECRET \
-               REDIS WORKER_TIER; do
-      printf '%s=%s\n' "$key" "${CFG[$key]}"
+               REDIS EVENTBUS_PROVIDER NATS_URL CODEC_PROVIDER WORKER_TIER WORKER_PUBLIC_IP WORKER_EGRESS_KIND; do
+      printf '%s=%s\n' "$key" "${CFG[$key]:-}"
    done
    for kv in "${EXTRA_ENVS[@]}"; do
      printf '%s\n' "$kv"
@@ -336,6 +403,61 @@ write_env_file() {
  ok "wrote $ENV_FILE"
 }

+resolve_update_image() {
+  local configured=""
+  if [[ -f "$ENV_FILE" ]]; then
+    configured="$(env_file_value "$ENV_FILE" WARMBLY_WORKER_IMAGE || true)"
+  fi
+  if [[ -n "$configured" ]]; then
+    IMAGE="$configured"
+  fi
+}
+
+install_auto_update_timer() {
+  [[ "$AUTO_UPDATE" -eq 1 ]] || return 0
+  command -v curl >/dev/null 2>&1 || return 0
+
+  cat > "$INSTALLER_BIN" <<EOF
+#!/usr/bin/env bash
+set -euo pipefail
+tmp="\$(mktemp)"
+curl -fsSL "${API_BASE%/}/worker-install.sh" -o "\$tmp"
+install -m 0755 "\$tmp" "$INSTALLER_BIN"
+rm -f "\$tmp"
+exec "$INSTALLER_BIN" --update --api-base "${API_BASE%/}"
+EOF
+  chmod 0755 "$INSTALLER_BIN"
+
+  cat > "$AUTO_UPDATE_UNIT_FILE" <<EOF
+[Unit]
+Description=Warmbly worker auto-update
+After=network-online.target docker.service
+Wants=network-online.target docker.service
+
+[Service]
+Type=oneshot
+ExecStart=$INSTALLER_BIN
+EOF
+
+  cat > "$AUTO_UPDATE_TIMER_FILE" <<EOF
+[Unit]
+Description=Run Warmbly worker auto-update
+
+[Timer]
+OnCalendar=*-*-* 04:00:00
+RandomizedDelaySec=3600
+Persistent=true
+
+[Install]
+WantedBy=timers.target
+EOF
+
+  chmod 0644 "$AUTO_UPDATE_UNIT_FILE" "$AUTO_UPDATE_TIMER_FILE"
+  systemctl daemon-reload
+  systemctl enable --now warmbly-worker-auto-update.timer >/dev/null 2>&1 || true
+  ok "auto-update timer enabled"
+}
+
 resolve_worker_id() {
  local ip="${IP_OVERRIDE}"
  local id
@@ -499,6 +621,10 @@ list_installed_instances() {
 # ---------- actions ----------

 prepare_common_env() {
+  if fetch_enrollment_env; then
+    return
+  fi
+
  if [[ -n "$SUPPLIED_ENV_FILE" ]]; then
    [[ -f "$SUPPLIED_ENV_FILE" ]] || die "env file not found: $SUPPLIED_ENV_FILE"
    install -d -m 0700 "$CONFIG_DIR"
@@ -546,6 +672,7 @@ do_install() {

  systemctl enable warmbly-worker.service >/dev/null 2>&1 || true
  systemctl restart warmbly-worker.service
+  install_auto_update_timer
  ok "warmbly-worker started"

  sleep 2
@@ -633,6 +760,7 @@ do_install_multi() {
    systemctl enable "warmbly-worker@${inst}.service" >/dev/null 2>&1 || true
    systemctl restart "warmbly-worker@${inst}.service"
  done
+  install_auto_update_timer
  ok "started ${count} worker instances"

  sleep 2
@@ -674,6 +802,7 @@ EOF

 do_update() {
  require_root
+  resolve_update_image

  local -a instances=()
  while IFS= read -r line; do
@@ -753,6 +882,9 @@ do_uninstall() {
    docker rm -f "$CONTAINER_NAME" >/dev/null 2>&1 || true
  fi

+  systemctl daemon-reload
+  systemctl disable --now warmbly-worker-auto-update.timer 2>/dev/null || true
+  rm -f "$AUTO_UPDATE_UNIT_FILE" "$AUTO_UPDATE_TIMER_FILE" "$INSTALLER_BIN"
  systemctl daemon-reload
  ok "worker removed (config preserved in $CONFIG_DIR)"
 }
@@ -27,6 +27,7 @@ import {
 } from "@/lib/api/client/app/admin/workers";
 import { assignWorkerProfile, listWorkerProfiles } from "@/lib/api/client/app/admin/credentials";
 import type { CreateWorkerResponse } from "@/lib/api/models/app/admin/Worker";
+import { API_URL } from "@/lib/information";

 type Purpose = "shared" | "dedicated" | "risky";

@@ -57,7 +58,7 @@ const initialState: WizardState = {
    risk_pool: "clean",
    dedicated_user_id: "",
    dedicated_subscription_id: "",
-    auto_install: true,
+    auto_install: false,
 };

 const purposeDefaults: Record<Purpose, Partial<WizardState>> = {
@@ -123,6 +124,7 @@ export default function AdminAddWorkerWizard() {
                ssh_host: state.ssh_host,
                ssh_port: state.ssh_port,
                ssh_user: state.ssh_user,
+                generate_enrollment_token: true,
            });
            setResult(created);
            append("✓ worker row created");
@@ -134,7 +136,13 @@ export default function AdminAddWorkerWizard() {
                append("✓ profile assigned");
            }

-            // Stop here unless admin wants to install now too.
+            if (created.enrollment_token) {
+                append("ready — run the enrollment command on the VPS");
+                setRunning(false);
+                return;
+            }
+
+            // Stop here unless admin wants the older SSH install path too.
            if (!state.auto_install) {
                append("ready — paste the SSH key into the VPS, then click Test → Install on the detail page");
                setRunning(false);
@@ -193,7 +201,7 @@ export default function AdminAddWorkerWizard() {
    const canNext = (() => {
        switch (step) {
            case 1: return true;
-            case 2: return state.ssh_host && state.ssh_port > 0 && preflight?.ok;
+            case 2: return state.ssh_host && state.ssh_port > 0;
            case 3: return state.name.length > 0;
            case 4: return state.purpose !== "dedicated" || (state.dedicated_user_id && state.dedicated_subscription_id);
            default: return true;
@@ -265,10 +273,10 @@ export default function AdminAddWorkerWizard() {
            {/* Step 2 */}
            {step === 2 && (
                <>
-                    <StepHeader n={2} title="How do we reach the VPS?" />
+                    <StepHeader n={2} title="Where will this worker run?" />
                    <p className="text-slate-500 text-sm mb-3">
-                        We'll check reachability before creating any database rows, so a typo here
-                        won't leave an orphan worker behind.
+                        This IP is used for the worker record and the one-command enrollment config.
+                        SSH reachability is only needed if you use the older dashboard-driven install path.
                    </p>
                    <div className="grid grid-cols-1 sm:grid-cols-3 gap-3">
                        <div className="col-span-2">
@@ -438,21 +446,13 @@ export default function AdminAddWorkerWizard() {
                                    )}
                                </dl>
                            </div>
-                            <label className="flex items-center gap-2 text-sm text-slate-600 mb-3">
-                                <input
-                                    type="checkbox"
-                                    checked={state.auto_install}
-                                    onChange={(e) => setState((s) => ({ ...s, auto_install: e.target.checked }))}
-                                />
-                                Install immediately after creating (recommended)
-                            </label>
                            {err && <p className="text-red-600 text-sm mb-2">{err}</p>}
                            <button
                                onClick={activate}
                                disabled={running}
                                className="bg-blue-600 hover:bg-blue-700 text-white px-4 py-2 rounded-lg text-sm font-medium disabled:opacity-50"
                            >
-                                {running ? "Creating…" : "Create worker"}
+                                {running ? "Creating…" : "Create enrollment command"}
                            </button>
                        </>
                    ) : (
@@ -624,6 +624,9 @@ function PostCreatePanel({
    onInstall: () => void;
 }) {
    const [pasted, setPasted] = useState(false);
+    const enrollCommand = result.enrollment_token
+        ? `curl -fsSL ${API_URL}/worker-install.sh | sudo bash -s -- --enroll ${result.enrollment_token} --api-base ${API_URL}`
+        : "";

    return (
        <div>
@@ -631,6 +634,26 @@ function PostCreatePanel({
                ✓ Worker created. ID: <span className="font-mono">{result.id}</span>
            </div>

+            {result.enrollment_token && (
+                <div className="mb-4">
+                    <div className={lbl}>Run on the VPS</div>
+                    <div className="bg-slate-900 text-slate-100 p-3 rounded text-xs font-mono overflow-auto whitespace-pre-wrap">
+                        {enrollCommand}
+                    </div>
+                    <div className="flex items-center justify-between mt-1">
+                        <p className="text-slate-500 text-xs">
+                            Token expires in {Math.round((result.enrollment_token_ttl_seconds ?? 7200) / 60)} minutes and is consumed after first use.
+                        </p>
+                        <button
+                            onClick={() => navigator.clipboard?.writeText(enrollCommand)}
+                            className="text-xs text-blue-600 hover:underline"
+                        >
+                            Copy
+                        </button>
+                    </div>
+                </div>
+            )}
+
            <div className="mb-4">
                <div className={lbl}>SSH public key</div>
                <textarea
@@ -654,14 +677,16 @@ function PostCreatePanel({
                </div>
            </div>

-            <div className="mb-4 bg-slate-900 text-slate-100 p-3 rounded text-xs font-mono overflow-auto">
-                <div className="text-slate-400 mb-1">Or run this one-liner on the VPS:</div>
-                ssh root@{result.ssh_host ?? "&lt;host&gt;"} {"\\"}
-                {"\n"}
-                {"  "}'mkdir -p ~/.ssh &amp;&amp; chmod 700 ~/.ssh &amp;&amp; echo "{result.ssh_public_key}" &gt;&gt; ~/.ssh/authorized_keys'
-            </div>
+            {!result.enrollment_token && (
+                <div className="mb-4 bg-slate-900 text-slate-100 p-3 rounded text-xs font-mono overflow-auto">
+                    <div className="text-slate-400 mb-1">Or run this one-liner on the VPS:</div>
+                    ssh root@{result.ssh_host ?? "&lt;host&gt;"} {"\\"}
+                    {"\n"}
+                    {"  "}'mkdir -p ~/.ssh &amp;&amp; chmod 700 ~/.ssh &amp;&amp; echo "{result.ssh_public_key}" &gt;&gt; ~/.ssh/authorized_keys'
+                </div>
+            )}

-            {autoInstall && (
+            {autoInstall && !result.enrollment_token && (
                <>
                    <label className="flex items-center gap-2 text-sm text-slate-600 mb-3">
                        <input
@@ -4,15 +4,12 @@ import { cn } from '@/lib/utils'
 export function ConnectionIndicator() {
  const { status, quality } = useConnectionStatus()

-  if (status === 'connected' && quality === 'good') {
-    return null
-  }
-
  return (
    <div className="flex items-center gap-2">
      <span
        className={cn(
          'w-2 h-2 rounded-full',
+          status === 'connected' && quality === 'good' && 'bg-emerald-500 animate-pulse',
          status === 'connected' && quality === 'degraded' && 'bg-amber-500',
          status === 'connected' && quality === 'poor' && 'bg-red-500',
          status === 'connecting' && 'bg-amber-500',
@@ -20,6 +17,7 @@ export function ConnectionIndicator() {
        )}
      />
      <span className="text-xs text-muted-foreground hidden sm:inline">
+        {status === 'connected' && quality === 'good' && 'Live'}
        {status === 'connecting' && 'Reconnecting...'}
        {status === 'disconnected' && 'Disconnected'}
        {status === 'connected' && quality === 'degraded' && 'Slow connection'}
@@ -1,4 +1,5 @@
-import { useEffect } from 'react'
+import { useCallback, useEffect } from 'react'
+import { useQueryClient, type QueryKey } from '@tanstack/react-query'
 import { useSocket } from './context/socket'
 import { useAppStore } from '@/stores'
 import { useUserProfile } from './context/user'
@@ -6,76 +7,204 @@ import { useUserProfile } from './context/user'
 export function useRealtimeEvents() {
  const { isConnected, subscribeToChannel } = useSocket()
  const { user } = useUserProfile()
+  const queryClient = useQueryClient()
  const currentOrg = useAppStore((s) => s.currentOrganization)

  const updateCampaign = useAppStore((s) => s.updateCampaign)
-  const addUniboxEmail = useAppStore((s) => s.addUniboxEmail)
  const incrementUnseenCount = useAppStore((s) => s.incrementUnseenCount)
  const updateDeal = useAppStore((s) => s.updateDeal)
  const setSubscription = useAppStore((s) => s.setSubscription)

-  // User channel events
-  useEffect(() => {
-    if (!isConnected || !user?.email) return
+  const invalidate = useCallback(
+    (queryKeys: QueryKey[]) => {
+      for (const queryKey of queryKeys) {
+        void queryClient.invalidateQueries({ queryKey })
+      }
+    },
+    [queryClient],
+  )

-    const topic = `user:${user.email}`
-    const unsubs: (() => void)[] = []
+  const handleRealtimeEvent = useCallback(
+    (payload: Record<string, unknown>) => {
+      const rawEvent = String(
+        payload.event_type ?? payload.type ?? payload._event ?? '',
+      )
+      const event = rawEvent.replace(/[.:\s-]+/g, '_').toUpperCase()
+      if (!event) return

-    // New email received
-    unsubs.push(
-      subscribeToChannel(topic, 'new_email', (payload) => {
-        addUniboxEmail(payload as any)
+      const getString = (key: string) => {
+        const value = payload[key]
+        return typeof value === 'string' && value.length > 0 ? value : null
+      }
+      const includes = (...needles: string[]) =>
+        needles.some((needle) => event.includes(needle))
+
+      const campaignId = getString('campaign_id')
+      const contactId = getString('contact_id')
+      const dealId = getString('deal_id')
+      const threadId = getString('thread_id')
+      const emailId = getString('email_id') ?? getString('message_id')
+
+      if (includes('EMAIL_RECEIVED', 'NEW_EMAIL', 'INBOX_NEW')) {
        incrementUnseenCount()
-      })
-    )
+        invalidate([
+          ['unibox'],
+          ['analytics'],
+          ['emails', 'list'],
+        ])
+        if (threadId) invalidate([['unibox', 'thread', threadId]])
+        if (emailId) invalidate([['unibox', 'email', emailId]])
+        return
+      }

-    // Campaign status changed
-    unsubs.push(
-      subscribeToChannel(topic, 'campaign_status_changed', (payload) => {
-        const { campaign_id, ...updates } = payload as any
-        if (campaign_id) {
-          updateCampaign(campaign_id, updates)
+      if (includes('EMAIL_UPDATED', 'EMAIL_DELETED', 'INBOX_UPDATE')) {
+        invalidate([['unibox'], ['analytics']])
+        if (threadId) invalidate([['unibox', 'thread', threadId]])
+        if (emailId) invalidate([['unibox', 'email', emailId]])
+        return
+      }
+
+      if (includes('CONTACT')) {
+        invalidate([
+          ['contacts'],
+          ['campaigns', 'list'],
+          ['analytics'],
+          ['organizations', 'limits'],
+        ])
+        if (contactId) invalidate([['contacts', contactId]])
+        return
+      }
+
+      if (
+        includes(
+          'CAMPAIGN',
+          'EMAIL_SENT',
+          'EMAIL_OPENED',
+          'EMAIL_CLICKED',
+          'EMAIL_REPLIED',
+          'EMAIL_BOUNCED',
+          'TASK_PROGRESS',
+        )
+      ) {
+        if (campaignId) {
+          const status = getString('status')
+          updateCampaign(campaignId, status ? { status } : {})
+          invalidate([
+            ['campaigns', campaignId],
+            ['campaigns', campaignId, 'logs'],
+            ['analytics', 'campaigns', campaignId],
+            ['analytics', 'campaigns', campaignId, 'daily'],
+            ['analytics', 'campaigns', campaignId, 'hourly'],
+          ])
        }
-      })
-    )
+        invalidate([
+          ['campaigns', 'list'],
+          ['analytics'],
+          ['contacts'],
+        ])
+        if (contactId) invalidate([['contacts', contactId]])
+        return
+      }

-    // Subscription changed
-    unsubs.push(
-      subscribeToChannel(topic, 'subscription_changed', (payload) => {
+      if (includes('ACCOUNT', 'EMAIL_STATUS', 'EMAIL_ERROR', 'WARMUP')) {
+        invalidate([
+          ['emails', 'list'],
+          ['analytics', 'accounts'],
+          ['analytics', 'warmup'],
+          ['analytics', 'dashboard'],
+        ])
+        return
+      }
+
+      if (includes('DEAL')) {
+        if (dealId) updateDeal(dealId, payload as any)
+        invalidate([['crm', 'deals'], ['crm', 'pipelines'], ['contacts']])
+        return
+      }
+
+      if (includes('PIPELINE', 'STAGE')) {
+        invalidate([['crm', 'pipelines'], ['crm', 'deals']])
+        return
+      }
+
+      if (includes('CRM_TASK', 'TASK')) {
+        invalidate([['crm', 'tasks'], ['crm', 'deals']])
+        return
+      }
+
+      if (includes('SUBSCRIPTION', 'PLAN', 'BILLING', 'LIMIT')) {
        setSubscription(payload as any)
-      })
-    )
+        invalidate([
+          ['subscription'],
+          ['organizations', 'current'],
+          ['organizations', 'limits'],
+          ['auth', 'me'],
+        ])
+        return
+      }

-    return () => unsubs.forEach((fn) => fn())
-  }, [isConnected, user?.email, subscribeToChannel, addUniboxEmail, incrementUnseenCount, updateCampaign, setSubscription])
+      if (includes('MEMBER', 'INVITATION', 'ORGANIZATION', 'SETTINGS')) {
+        invalidate([
+          ['organizations'],
+          ['organizations', 'current'],
+          ['organizations', 'invitations'],
+          ['auth', 'me'],
+        ])
+        return
+      }
+
+      if (includes('API_KEY')) {
+        invalidate([['api-keys']])
+        return
+      }
+
+      if (includes('TEMPLATE')) {
+        invalidate([['templates']])
+        return
+      }
+
+      if (includes('AUDIT')) {
+        invalidate([['audit']])
+        return
+      }
+
+      if (includes('DANGER', 'DELETION')) {
+        invalidate([
+          ['dangerzone'],
+          ['auth', 'me'],
+          ['organizations', 'current'],
+        ])
+        return
+      }
+
+      invalidate([
+        ['analytics', 'dashboard'],
+        ['auth', 'me'],
+      ])
+    },
+    [
+      incrementUnseenCount,
+      invalidate,
+      setSubscription,
+      updateCampaign,
+      updateDeal,
+    ],
+  )
+
+  // User channel events. Topic uses the user UUID; the realtime server
+  // rejects email-address topics and only authorizes `user:{sub}`.
+  useEffect(() => {
+    if (!isConnected || !user?.id) return
+
+    const topic = `user:${user.id}`
+    return subscribeToChannel(topic, '*', handleRealtimeEvent)
+  }, [isConnected, user?.id, subscribeToChannel, handleRealtimeEvent])

  // Org channel events
  useEffect(() => {
    if (!isConnected || !currentOrg?.id) return

    const topic = `org:${currentOrg.id}`
-    const unsubs: (() => void)[] = []
-
-    // Deal updated
-    unsubs.push(
-      subscribeToChannel(topic, 'deal_updated', (payload) => {
-        const { deal_id, ...updates } = payload as any
-        if (deal_id) {
-          updateDeal(deal_id, updates)
-        }
-      })
-    )
-
-    // Campaign events
-    unsubs.push(
-      subscribeToChannel(topic, 'email_sent', (payload) => {
-        const { campaign_id } = payload as any
-        if (campaign_id) {
-          updateCampaign(campaign_id, {})
-        }
-      })
-    )
-
-    return () => unsubs.forEach((fn) => fn())
-  }, [isConnected, currentOrg?.id, subscribeToChannel, updateDeal, updateCampaign])
+    return subscribeToChannel(topic, '*', handleRealtimeEvent)
+  }, [isConnected, currentOrg?.id, subscribeToChannel, handleRealtimeEvent])
 }