Align WA registration flow and proxy lease handling

.env.example

@@ -16,3 +16,8 @@ WA_APP_REDIS_URL=
 
 # Optional upstream proxies (leave empty for direct outbound)
 WA_COMMON_PROXY=
+
+# Optional registration proxy-runtime lease. Modes: optional, disabled, required.
+WA_REGISTRATION_PROXY_LEASE_MODE=optional
+PROXY_RUNTIME_API_BASE_URL=
+PROXY_RUNTIME_SERVICE_AUTH_TOKEN=

README.md

@@ -42,6 +42,9 @@ docker compose up -d
 - `WA_APP_PG_DSN`：可选 PostgreSQL DSN；为空时使用内置 SQLite 持久化。
 - `WA_APP_REDIS_URL`：可选 Redis URL；为空时使用内置 SQLite 运行态存储。
 - `WA_COMMON_PROXY`：系统默认 WA 出站代理；账号未配置代理策略且阶段代理为空时使用，仍为空则直连。
+- `WA_REGISTRATION_PROXY_LEASE_MODE`：注册链路 proxy-runtime 租约模式；`optional` 会失败回退，`disabled` 关闭租约，`required` 强制租约成功。
+- `PROXY_RUNTIME_API_BASE_URL`：可选 proxy-runtime API 地址；留空时不请求动态租约。
+- `PROXY_RUNTIME_SERVICE_AUTH_TOKEN`：可选 proxy-runtime 服务令牌；示例文件只保留空占位。
 
 PostgreSQL 和 Redis 都是可选组件。需要启用时，在 `docker-compose.yml` 中取消对应服务注释，并在 `.env` 中填写 `WA_APP_PG_DSN` / `WA_APP_REDIS_URL`。
 

cmd/wa-app-service/dashboard_http.go

@@ -267,7 +267,7 @@ func (s *dashboardHTTP) handleAccountOTPMessages(w http.ResponseWriter, r *http.
 		WaAccountId:            q.Get("wa_account_id"),
 		Limit:                  int32(positiveInt(q.Get("limit"), 20)),
 		Cursor:                 q.Get("cursor"),
-		IncludeSensitiveValues: true,
+		IncludeSensitiveValues: queryBool(q.Get("include_sensitive_values"), true),
 	})
 	if err != nil {
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "load WA OTP history failed"})
@@ -847,6 +847,17 @@ func positiveInt(value string, fallback int) int {
 	return parsed
 }
 
+func queryBool(value string, fallback bool) bool {
+	switch strings.ToLower(strings.TrimSpace(value)) {
+	case "true", "1", "yes", "on":
+		return true
+	case "false", "0", "no", "off":
+		return false
+	default:
+		return fallback
+	}
+}
+
 func methodNotAllowed(w http.ResponseWriter, allowed string) {
 	w.Header().Set("Allow", allowed)
 	writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"})

cmd/wa-app-service/main.go

@@ -57,6 +57,8 @@ func main() {
 	}
 	service := app.NewServer(store, runtime, engine, clock, ids)
 	service.SetCommonProxyURL(cfg.CommonProxy)
+	service.SetRegistrationProxyLeaseMode(cfg.RegistrationProxyLeaseMode)
+	service.SetProxyRuntimeLeaseClient(cfg.ProxyRuntimeAPI, cfg.ProxyRuntimeToken)
 	authConfig := newDashboardAuthConfig(cfg.DashboardAuthPass)
 	grpcListenAddr := configValue(cfg.GRPCListenAddr, defaultGRPCListenAddr)
 	dashboardHTTPAddr := configValue(cfg.DashboardHTTPAddr, defaultDashboardHTTPAddr)

docker-compose.yml

@@ -11,6 +11,9 @@ services:
       WA_APP_PG_DSN: ${WA_APP_PG_DSN:-}
       WA_APP_REDIS_URL: ${WA_APP_REDIS_URL:-}
       WA_COMMON_PROXY: ${WA_COMMON_PROXY:-}
+      WA_REGISTRATION_PROXY_LEASE_MODE: ${WA_REGISTRATION_PROXY_LEASE_MODE:-optional}
+      PROXY_RUNTIME_API_BASE_URL: ${PROXY_RUNTIME_API_BASE_URL:-}
+      PROXY_RUNTIME_SERVICE_AUTH_TOKEN: ${PROXY_RUNTIME_SERVICE_AUTH_TOKEN:-}
       WA_APP_DATA_DIR: ${WA_APP_DATA_DIR:-/var/lib/wa-app}
     ports:
       - "127.0.0.1:50091:50091"

internal/app/action_gateway.go

@@ -18,13 +18,15 @@ import (
 )
 
 const transientStateTTL = 30 * time.Minute
+const registrationAttemptStateTTL = 26 * time.Hour
 const registrationOTPWaitDefaultTTL = 20 * time.Minute
 
 type registrationOTPWait struct {
-	WAAccountID           string `json:"wa_account_id"`
-	VerificationRequestID string `json:"verification_request_id"`
-	ResumeURL             string `json:"resume_url"`
-	CreatedAtUnix         int64  `json:"created_at_unix"`
+	WAAccountID           string                 `json:"wa_account_id"`
+	VerificationRequestID string                 `json:"verification_request_id"`
+	ResumeURL             string                 `json:"resume_url"`
+	CreatedAtUnix         int64                  `json:"created_at_unix"`
+	ProxyLease            registrationProxyLease `json:"proxy_lease,omitempty"`
 }
 
 type actionGateway struct{ server *Server }
@@ -61,6 +63,10 @@ func (g *actionGateway) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		result, err = g.resumeOTP(r.Context(), payload)
 	case "registration/submit-otp":
 		result, err = g.submitOTP(r.Context(), payload)
+	case "registration/account-transfer/refresh":
+		result, err = g.refreshAccountTransferChallenge(r.Context(), payload)
+	case "registration/account-transfer/poll":
+		result, err = g.pollAccountTransferRegistration(r.Context(), payload)
 	case "registration/cleanup-failed-account":
 		result, err = g.cleanupFailedRegistration(r.Context(), payload)
 	case "registration/persist-login-state":
@@ -168,7 +174,7 @@ func (g *actionGateway) requestSMSOTP(ctx context.Context, payload map[string]an
 	if reason := directRegistrationMethodUnsupportedReason(method); reason != "" {
 		return registrationMethodUnsupportedMap(method, reason), nil
 	}
-	runner, route, managedRoute, err := g.registrationRequestRunner(ctx, payload)
+	runner, route, managedRoute, proxyLease, err := g.registrationRequestRunner(ctx, payload)
 	if err != nil {
 		return nil, err
 	}
@@ -182,20 +188,42 @@ func (g *actionGateway) requestSMSOTP(ctx context.Context, payload map[string]an
 	}, runner)
 	runner.CloseIdleConnections()
 	if err != nil {
+		g.releaseRegistrationProxyLease(context.Background(), proxyLease)
 		return nil, err
 	}
 	if resp.GetError() != nil {
+		g.releaseRegistrationProxyLease(context.Background(), proxyLease)
 		return map[string]any{"success": false, "error": protoMap(resp.GetError()), "error_message": resp.GetError().GetMessage()}, nil
 	}
 	record := resp.GetVerificationRequest()
+	success := record.GetStatus() == waappv1.VerificationRequestStatus_VERIFICATION_REQUEST_STATUS_SENT || record.GetStatus() == waappv1.VerificationRequestStatus_VERIFICATION_REQUEST_STATUS_WAITING
+	if !success {
+		g.releaseRegistrationProxyLease(context.Background(), proxyLease)
+	}
+	if success && validRegistrationProxyLease(proxyLease) {
+		wait := registrationOTPWait{
+			WAAccountID:           record.GetWaAccountId(),
+			VerificationRequestID: record.GetVerificationRequestId(),
+			CreatedAtUnix:         time.Now().UTC().Unix(),
+			ProxyLease:            proxyLease,
+		}
+		if err := g.saveRegistrationOTPWait(ctx, wait, registrationOTPWaitDefaultTTL); err != nil {
+			g.releaseRegistrationProxyLease(context.Background(), proxyLease)
+			return nil, err
+		}
+	}
 	response := map[string]any{
-		"success":                 record.GetStatus() == waappv1.VerificationRequestStatus_VERIFICATION_REQUEST_STATUS_SENT || record.GetStatus() == waappv1.VerificationRequestStatus_VERIFICATION_REQUEST_STATUS_WAITING,
+		"success":                 success,
 		"status":                  record.GetStatus().String(),
 		"verification_request_id": record.GetVerificationRequestId(),
 		"verification_request":    protoMap(record),
 		"method_statuses":         protoMethodStatusMaps(record.GetMethodStatuses()),
 		"proxy":                   registrationProxyRouteMap(route, managedRoute),
 	}
+	if challenge := resp.GetAccountTransferChallenge(); challenge != nil {
+		response["account_transfer_challenge"] = protoMap(challenge)
+		response["registration_phase"] = "ACCOUNT_TRANSFER_WAITING"
+	}
 	if seconds := durationSeconds(record.GetRetryAfter()); seconds > 0 {
 		response["retry_after_seconds"] = seconds
 	}
@@ -207,6 +235,11 @@ func (g *actionGateway) awaitOTP(ctx context.Context, payload map[string]any) (m
 	if err != nil {
 		return nil, err
 	}
+	if wait.ProxyLease.LeaseID == "" && g.server.registrationProxyLeaseEnabled() {
+		if existing, err := g.loadRegistrationOTPWait(ctx, wait.WAAccountID, wait.VerificationRequestID); err == nil {
+			wait.ProxyLease = existing.ProxyLease
+		}
+	}
 	if err := g.saveRegistrationOTPWait(ctx, wait, ttl); err != nil {
 		return nil, err
 	}
@@ -231,6 +264,7 @@ func (g *actionGateway) resumeOTP(ctx context.Context, payload map[string]any) (
 		if err := postRegistrationOTPResume(ctx, wait, code); err != nil {
 			return nil, err
 		}
+		g.releaseRegistrationProxyLease(context.Background(), wait.ProxyLease)
 		_ = g.deleteRegistrationOTPWait(ctx, wait)
 		return map[string]any{"success": true, "wa_account_id": wait.WAAccountID, "verification_request_id": wait.VerificationRequestID}, nil
 	}
@@ -357,7 +391,7 @@ func registrationOTPWaitAccountKey(waAccountIDValue string) string {
 }
 
 func (g *actionGateway) submitOTP(ctx context.Context, payload map[string]any) (map[string]any, error) {
-	runner, route, managedRoute, err := g.registrationSubmitRunner(ctx, payload)
+	runner, route, managedRoute, proxyLease, err := g.registrationSubmitRunner(ctx, payload)
 	if err != nil {
 		return nil, err
 	}
@@ -374,6 +408,13 @@ func (g *actionGateway) submitOTP(ctx context.Context, payload map[string]any) (
 		return map[string]any{"success": false, "error": protoMap(resp.GetError()), "error_message": resp.GetError().GetMessage(), "registration": protoMap(resp.GetRegistration())}, nil
 	}
 	success := resp.GetRegistration().GetStatus() == waappv1.RegistrationStatus_REGISTRATION_STATUS_REGISTERED && resp.GetLoginState().GetStatus() == waappv1.LoginStateStatus_LOGIN_STATE_STATUS_ACTIVE
+	if success {
+		g.releaseRegistrationProxyLease(context.Background(), proxyLease)
+		_ = g.deleteRegistrationOTPWait(ctx, registrationOTPWait{
+			WAAccountID:           resp.GetRegistration().GetWaAccountId(),
+			VerificationRequestID: resp.GetRegistration().GetVerificationRequestId(),
+		})
+	}
 	return map[string]any{
 		"success":      success,
 		"status":       resp.GetRegistration().GetStatus().String(),
@@ -383,15 +424,99 @@ func (g *actionGateway) submitOTP(ctx context.Context, payload map[string]any) (
 	}, nil
 }
 
+func (g *actionGateway) refreshAccountTransferChallenge(ctx context.Context, payload map[string]any) (map[string]any, error) {
+	resp, err := g.server.RefreshAccountTransferChallenge(ctx, &waappv1.RefreshAccountTransferChallengeRequest{
+		Context:               actionContext(payload),
+		VerificationRequestId: textField(payload, "verification_request_id"),
+	})
+	if err != nil {
+		return nil, err
+	}
+	if resp.GetError() != nil {
+		return map[string]any{"success": false, "error": protoMap(resp.GetError()), "error_message": resp.GetError().GetMessage()}, nil
+	}
+	return map[string]any{
+		"success":                    true,
+		"registration_phase":         "ACCOUNT_TRANSFER_WAITING",
+		"account_transfer_challenge": protoMap(resp.GetAccountTransferChallenge()),
+	}, nil
+}
+
+func (g *actionGateway) pollAccountTransferRegistration(ctx context.Context, payload map[string]any) (map[string]any, error) {
+	attempts := int(numberField(payload, "max_attempts"))
+	if attempts <= 0 {
+		attempts = 1
+	}
+	if attempts > 100 {
+		attempts = 100
+	}
+	interval := time.Duration(numberField(payload, "interval_seconds")) * time.Second
+	var result map[string]any
+	for attempt := 0; attempt < attempts; attempt++ {
+		if attempt > 0 && interval > 0 {
+			timer := time.NewTimer(interval)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return nil, ctx.Err()
+			case <-timer.C:
+			}
+		}
+		submitPayload := cloneActionPayload(payload)
+		submitPayload["code"] = ""
+		resultValue, err := g.submitOTP(ctx, submitPayload)
+		if err != nil {
+			return nil, err
+		}
+		result = resultValue
+		if boolField(result, "success") {
+			_ = g.deleteRegistrationOTPWait(ctx, registrationOTPWait{WAAccountID: textField(payload, "wa_account_id"), VerificationRequestID: textField(payload, "verification_request_id")})
+			result["attempts"] = attempt + 1
+			return result, nil
+		}
+		if !accountTransferPollRetryable(result) {
+			result["attempts"] = attempt + 1
+			return result, nil
+		}
+	}
+	if result == nil {
+		result = map[string]any{"success": false}
+	}
+	result["registration_phase"] = "ACCOUNT_TRANSFER_WAITING"
+	result["attempts"] = attempts
+	return result, nil
+}
+
+func accountTransferPollRetryable(result map[string]any) bool {
+	if result == nil {
+		return true
+	}
+	if textField(result, "status") == waappv1.RegistrationStatus_REGISTRATION_STATUS_SUBMITTED.String() {
+		return true
+	}
+	errorMap := objectField(result, "error")
+	if boolField(errorMap, "retryable") {
+		return true
+	}
+	message := strings.ToLower(firstNonEmpty(textField(result, "error_message"), textField(errorMap, "message")))
+	return strings.Contains(message, "pending") || strings.Contains(message, "temporarily") || strings.Contains(message, "too_recent")
+}
+
 func (g *actionGateway) cleanupFailedRegistration(ctx context.Context, payload map[string]any) (map[string]any, error) {
 	reqCtx := actionContext(payload)
 	accountID := cleanupWAAccountID(payload)
 	verificationRequestID := cleanupVerificationRequestID(payload)
 	if verificationRequestID != "" || accountID != "" {
-		_ = g.deleteRegistrationOTPWait(ctx, registrationOTPWait{
-			WAAccountID:           accountID,
-			VerificationRequestID: verificationRequestID,
-		})
+		wait, err := g.loadRegistrationOTPWait(ctx, accountID, verificationRequestID)
+		if err == nil {
+			g.releaseRegistrationProxyLease(context.Background(), wait.ProxyLease)
+			_ = g.deleteRegistrationOTPWait(ctx, wait)
+		} else {
+			_ = g.deleteRegistrationOTPWait(ctx, registrationOTPWait{
+				WAAccountID:           accountID,
+				VerificationRequestID: verificationRequestID,
+			})
+		}
 	}
 	if accountID == "" {
 		return map[string]any{"success": true, "deleted": false, "reason": "missing_wa_account_id"}, nil
@@ -554,7 +679,7 @@ func (s *Server) ensureDefaultProtocolProfile(ctx context.Context) (*waappv1.Pro
 			waappv1.ProtocolCapability_PROTOCOL_CAPABILITY_MESSAGE_SESSION,
 			waappv1.ProtocolCapability_PROTOCOL_CAPABILITY_ACCOUNT_SETTINGS,
 		},
-		RegistrationFlows: []waappv1.RegistrationFlowKind{waappv1.RegistrationFlowKind_REGISTRATION_FLOW_KIND_NEW_ACCOUNT},
+		RegistrationFlows: []waappv1.RegistrationFlowKind{waappv1.RegistrationFlowKind_REGISTRATION_FLOW_KIND_NEW_ACCOUNT, waappv1.RegistrationFlowKind_REGISTRATION_FLOW_KIND_EXISTING_ACCOUNT},
 		MessageTransports: []waappv1.MessageTransportKind{waappv1.MessageTransportKind_MESSAGE_TRANSPORT_KIND_LONG_CONNECTION},
 		DiscoveredAt:      timestamppb.New(now),
 		Audit:             &waappv1.AuditStamp{CreatedAt: timestamppb.New(now), UpdatedAt: timestamppb.New(now)},
@@ -577,10 +702,10 @@ func (g *actionGateway) nativeEngineForPayload(payload map[string]any) (*NativeE
 	return engine.WithProxyURL(proxyURL)
 }
 
-func (g *actionGateway) registrationRequestRunner(ctx context.Context, payload map[string]any) (*NativeEngine, WAProxyRoute, bool, error) {
+func (g *actionGateway) registrationRequestRunner(ctx context.Context, payload map[string]any) (*NativeEngine, WAProxyRoute, bool, registrationProxyLease, error) {
 	engine, err := g.nativeEngine()
 	if err != nil {
-		return nil, WAProxyRoute{}, false, err
+		return nil, WAProxyRoute{}, false, registrationProxyLease{}, err
 	}
 	route, useProxy, err := g.server.resolveWAProxyRoute(ctx, waProxyResolveRequest{
 		Stage:       waProxyStageRegistration,
@@ -589,35 +714,51 @@ func (g *actionGateway) registrationRequestRunner(ctx context.Context, payload m
 		CountryCode: proxyCountryCodeFromPayload(payload),
 	})
 	if err != nil {
-		return nil, WAProxyRoute{}, false, err
+		return nil, WAProxyRoute{}, false, registrationProxyLease{}, err
 	}
 	if !useProxy {
-		return engine, route, false, nil
+		return engine, route, false, registrationProxyLease{}, nil
+	}
+	lease, leasedRoute, err := g.acquireRegistrationProxyLease(ctx, payload, route, registrationOTPWaitDefaultTTL)
+	if err != nil {
+		return nil, WAProxyRoute{}, false, registrationProxyLease{}, err
+	}
+	if validRegistrationProxyLease(lease) {
+		route = leasedRoute
 	}
 	proxied, err := engine.WithProxyURL(route.ProxyURL)
 	if err != nil {
-		return nil, WAProxyRoute{}, false, err
+		g.releaseRegistrationProxyLease(context.Background(), lease)
+		return nil, WAProxyRoute{}, false, registrationProxyLease{}, err
 	}
-	return proxied, route, true, nil
+	return proxied, route, true, lease, nil
 }
 
-func (g *actionGateway) registrationSubmitRunner(ctx context.Context, payload map[string]any) (*NativeEngine, WAProxyRoute, bool, error) {
+func (g *actionGateway) registrationSubmitRunner(ctx context.Context, payload map[string]any) (*NativeEngine, WAProxyRoute, bool, registrationProxyLease, error) {
 	engine, err := g.nativeEngine()
 	if err != nil {
-		return nil, WAProxyRoute{}, false, err
+		return nil, WAProxyRoute{}, false, registrationProxyLease{}, err
+	}
+	if wait, err := g.loadRegistrationOTPWait(ctx, textField(payload, "wa_account_id"), textField(payload, "verification_request_id")); err == nil && g.server.registrationProxyLeaseEnabled() && validRegistrationProxyLease(wait.ProxyLease) {
+		route := proxyRuntimeLeaseRoute(wait.ProxyLease, WAProxyRoute{Source: waProxySourceSystemCommon, PolicyMode: waProxyModeCommon})
+		proxied, err := engine.WithProxyURL(route.ProxyURL)
+		if err != nil {
+			return nil, WAProxyRoute{}, false, registrationProxyLease{}, err
+		}
+		return proxied, route, true, wait.ProxyLease, nil
 	}
 	route, useProxy, err := g.registrationSubmitProxyRoute(ctx, payload)
 	if err != nil {
-		return nil, WAProxyRoute{}, false, err
+		return nil, WAProxyRoute{}, false, registrationProxyLease{}, err
 	}
 	if !useProxy {
-		return engine, route, false, nil
+		return engine, route, false, registrationProxyLease{}, nil
 	}
 	proxied, err := engine.WithProxyURL(route.ProxyURL)
 	if err != nil {
-		return nil, WAProxyRoute{}, false, err
+		return nil, WAProxyRoute{}, false, registrationProxyLease{}, err
 	}
-	return proxied, route, true, nil
+	return proxied, route, true, registrationProxyLease{}, nil
 }
 
 func (g *actionGateway) registrationSubmitProxyRoute(ctx context.Context, payload map[string]any) (WAProxyRoute, bool, error) {