feat: add worker enrollment install

cmd/backend/main.go

@@ -581,18 +581,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"),
 		)

cmd/worker/main.go

@@ -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.

internal/api/handler/admin_workers_ssh.go

@@ -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)

internal/api/handler/worker_enrollment.go

@@ -0,0 +1,87 @@
+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"
+)
+
+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"
+}

internal/api/routes.go

@@ -38,6 +38,11 @@ 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.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.

internal/app/worker_orchestrator/orchestrator.go

@@ -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,26 @@ func (o *Orchestrator) Install(ctx context.Context, workerID uuid.UUID) error {
 	return o.repo.UpdateInstallState(ctx, workerID, models.WorkerInstallStateInstalled, "")
 }
 
+// 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 +605,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 +615,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
 }
 

internal/repository/pg_worker_heartbeat.go

@@ -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]

internal/repository/pg_worker_ssh.go

@@ -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

scripts/install-worker.sh

@@ -24,6 +24,11 @@
 #   curl -fsSL https://get.warmbly.com/worker | sudo bash -s -- \
 #     --env-file /root/worker.env
 #
+# Or with a one-time enrollment token from the dashboard:
+#
+#   curl -fsSL https://get.warmbly.com/worker | 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.
 
@@ -43,6 +48,8 @@ CONTAINER_NAME="warmbly-worker"
 ACTION="install"
 INTERACTIVE=1
 SUPPLIED_ENV_FILE=""
+ENROLL_TOKEN=""
+API_BASE="${WARMBLY_API_BASE:-https://api.warmbly.com}"
 
 # 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 +118,9 @@ 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})
 
   --kafka <bootstrap>          Kafka bootstrap servers (host:port[,host:port])
   --kafka-user <user>
@@ -151,6 +161,8 @@ 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 ;;
 
     --kafka)           CFG[KAFKA_BOOTSTRAP_SERVERS]="$2"; shift 2 ;;
     --kafka-user)      CFG[KAFKA_SASL_USERNAME]="$2"; shift 2 ;;
@@ -315,6 +327,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 +382,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"
@@ -499,6 +560,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"