Files

2073 lines
80 KiB
Go

package api
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"regexp"
"strings"
"time"
"git.ipao.vip/rogee/creator-hub/internal/controlplane/gatewayclient"
hub "git.ipao.vip/rogee/creator-hub/internal/environment"
"github.com/gofiber/fiber/v3"
"github.com/sirupsen/logrus"
)
// HubStore 是控制面编排所需的存储能力;生产实现为 *hub.Store,测试使用内存桩。
type HubStore interface {
LockResources(ctx context.Context, aliases, exitIDs []string, _ ...[]string) (func(), error)
CreateGateway(ctx context.Context, name, endpoint, token string) (hub.Gateway, error)
UpdateGateway(ctx context.Context, currentName, name, endpoint, token string) (hub.Gateway, error)
ListGateways(ctx context.Context) ([]hub.Gateway, error)
GetGateway(ctx context.Context, name string) (hub.Gateway, error)
DeleteGateway(ctx context.Context, name string) error
CreateEnv(ctx context.Context, env hub.Env) error
ListEnvs(ctx context.Context) ([]hub.Env, error)
GetEnv(ctx context.Context, alias string) (hub.Env, error)
CreateNetworkExit(ctx context.Context, exit hub.NetworkExit) (hub.NetworkExit, error)
ListNetworkExits(ctx context.Context) ([]hub.NetworkExit, error)
GetNetworkExit(ctx context.Context, id string) (hub.NetworkExit, error)
GetNetworkExitAccess(ctx context.Context, id string) (hub.NetworkExitAccess, error)
RecordNetworkExitCheck(ctx context.Context, id string, observation hub.ExitObservation, failureReason string) (hub.NetworkExit, string, error)
DisableNetworkExit(ctx context.Context, id string) (hub.NetworkExit, error)
CreateBoundEnv(ctx context.Context, env hub.Env, accountID, exitID string) (hub.EnvironmentContext, bool, error)
GetEnvironmentContext(ctx context.Context, alias string) (hub.EnvironmentContext, error)
GetEnvironmentContextForAccount(ctx context.Context, accountID string) (hub.EnvironmentContext, error)
ValidateEnvironmentRebind(ctx context.Context, alias, exitID string, expectedBindingVersion int64) error
RebindEnvironment(ctx context.Context, alias, exitID, runtimeID string, expectedBindingVersion int64, networkID ...string) (hub.EnvironmentContext, error)
ActivateRuntime(ctx context.Context, alias, runtimeID string, bindingVersion int64, exitID string, networkID ...string) (hub.EnvironmentContext, error)
ReleaseRuntime(ctx context.Context, environment hub.EnvironmentContext) error
SetRuntimeCleanupPending(ctx context.Context, environment hub.EnvironmentContext, pending bool) error
AppendEnvironmentAction(ctx context.Context, eventType string, action hub.EnvironmentAction) error
}
type networkExitAdminStore interface {
UpdateNetworkExit(context.Context, string, hub.NetworkExit) (hub.NetworkExit, error)
EnableNetworkExit(context.Context, string) (hub.NetworkExit, error)
DeleteNetworkExit(context.Context, string) error
}
type RuntimeStopStore interface {
LockResources(ctx context.Context, aliases, exitIDs []string, _ ...[]string) (func(), error)
GetEnvironmentContextForAccount(ctx context.Context, accountID string) (hub.EnvironmentContext, error)
GetGateway(ctx context.Context, name string) (hub.Gateway, error)
ReleaseRuntime(ctx context.Context, environment hub.EnvironmentContext) error
SetRuntimeCleanupPending(ctx context.Context, environment hub.EnvironmentContext, pending bool) error
AppendEnvironmentAction(ctx context.Context, eventType string, action hub.EnvironmentAction) error
}
type runtimeCleanupStore interface {
SetRuntimeCleanupPending(ctx context.Context, environment hub.EnvironmentContext, pending bool) error
}
const (
gatewayLongTimeout = 11 * time.Minute // 覆盖网关侧最长 10 分钟的镜像拉取
gatewayBrowserOperationTimeout = 90 * time.Second
gatewayReconcileDelay = 100 * time.Millisecond
gatewayReconcileAttempts = 10
missingRuntimeID = "runtime-not-found"
)
var (
// Runtime and network identifiers are opaque, gateway-issued generations.
// Their exact value is fenced and persisted; only control characters and
// oversized values are rejected here.
gatewayGenerationIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$`)
gatewayRuntimeIDPattern = gatewayGenerationIDPattern
gatewayNetworkIDPattern = gatewayGenerationIDPattern
)
const (
defaultGatewayResponseLimit = 1 << 20
largeGatewayResponseLimit = 16 << 20
)
// gatewayCall 调用某个网关的 /v1 路由;ok 为 false 时 status/body 携带网关错误。
func gatewayCall(ctx context.Context, target hub.Gateway, method, path string, body any, timeout time.Duration) (status int, responseBody []byte, err error) {
return gatewayclient.Call(ctx, target, method, path, body, timeout)
}
func gatewayCallWithLimit(ctx context.Context, target hub.Gateway, method, path string, body any, timeout time.Duration, responseLimit int) (status int, responseBody []byte, err error) {
return gatewayclient.CallWithLimit(ctx, target, method, path, body, timeout, responseLimit)
}
func gatewayCreatePayload(environment hub.EnvironmentContext, _ string, networkExit gatewayNetworkExit) map[string]any {
fingerprint := environment.Fingerprint
fingerprint.ProxyServer = ""
fingerprint.DisableNonProxiedUDP = false
profileID := environment.AccountID
if profileID == "" {
profileID = environment.Alias
}
payload := map[string]any{
"alias": environment.Alias,
"name": environment.Name,
"profile_id": profileID,
"cmd": append(fingerprint.Args(), "about:blank"),
"binding_version": environment.BindingVersion,
"network_exit_id": environment.Exit.ID,
"network_exit": networkExit,
}
return payload
}
func gatewayCreatePayloadForAccount(environment hub.EnvironmentContext, browserPath string, networkExit gatewayNetworkExit) map[string]any {
payload := gatewayCreatePayload(environment, browserPath, networkExit)
if !accountRunnable(environment) {
payload["network_exit_id"], payload["network_exit"], payload["stopped"] = "", gatewayNetworkExit{}, true
}
return payload
}
func gatewayProxyPayload(environment hub.EnvironmentContext, runtimeID, networkID string, networkExit gatewayNetworkExit) map[string]any {
return map[string]any{
"binding_version": environment.BindingVersion,
"runtime_id": runtimeID,
"network_id": networkID,
"network_exit_id": environment.Exit.ID,
"network_exit": networkExit,
}
}
func gatewayGenerationPayload(environment hub.EnvironmentContext) map[string]any {
return map[string]any{"binding_version": environment.BindingVersion, "runtime_id": environment.RuntimeID, "network_id": environment.RuntimeNetworkID, "network_exit_id": environment.Exit.ID}
}
func gatewayCleanupGenerationPayload(environment hub.EnvironmentContext) map[string]any {
bindingVersion, runtimeID, networkID := environment.RuntimeCleanupBindingVersion, environment.RuntimeCleanupRuntimeID, environment.RuntimeCleanupNetworkID
if bindingVersion == 0 {
bindingVersion, runtimeID, networkID = environment.BindingVersion, environment.RuntimeID, environment.RuntimeNetworkID
}
return map[string]any{"binding_version": bindingVersion, "runtime_id": runtimeID, "network_id": networkID}
}
func runtimeCleanupGeneration(environment hub.EnvironmentContext, bindingVersion int64, runtimeID string, networkIDs ...string) hub.EnvironmentContext {
environment.RuntimeCleanupBindingVersion = bindingVersion
environment.RuntimeCleanupInstanceID = environment.RuntimeInstanceID
environment.RuntimeCleanupRuntimeID = runtimeID
environment.RuntimeCleanupNetworkID = environment.RuntimeNetworkID
if len(networkIDs) == 1 {
environment.RuntimeCleanupNetworkID = networkIDs[0]
}
return environment
}
func gatewayNetworkID(body []byte) string {
var envelope struct {
NetworkID string `json:"network_id"`
}
if json.Unmarshal(body, &envelope) == nil && gatewayNetworkIDPattern.MatchString(envelope.NetworkID) {
return envelope.NetworkID
}
return ""
}
func reconcileGatewayCreate(ctx context.Context, store runtimeCleanupStore, target hub.Gateway,
environment hub.EnvironmentContext, status int, callErr error, body []byte) error {
// 网关 4xx 拒绝且未产生 network 代:创建在 gateway 侧任何 runtime 变更之前就被确定性拒绝,
// 无需清理 fence;其余情况(5xx、断连、无效 201)结果未知,仍走 fail-closed。
deterministicRejection := callErr == nil && status >= 400 && status < 500
logEntry := logrus.WithFields(logrus.Fields{
"service": "control-plane",
"event_type": "runtime_create_reconciliation",
"alias": environment.Alias,
"gateway": target.Name,
"binding_version": environment.BindingVersion,
"gateway_status": status,
"network_id_known": gatewayNetworkID(body) != "",
"deterministic": deterministicRejection,
})
if callErr != nil {
logEntry = logEntry.WithError(callErr)
}
if deterministicRejection {
logEntry.Info("gateway rejected runtime create before native runtime side effects")
} else {
logEntry.Warn("gateway runtime create result requires reconciliation")
}
return reconcileGatewayCreateGeneration(ctx, store, target, environment, environment.BindingVersion,
gatewayNetworkID(body), deterministicRejection)
}
func reconcileGatewayCreateGeneration(ctx context.Context, store runtimeCleanupStore, target hub.Gateway,
environment hub.EnvironmentContext, bindingVersion int64, networkID string, deterministicRejection bool) error {
if networkID == "" {
if deterministicRejection {
// 网关明确拒绝:创建未发生、无副作用,不写 pending,让调用方透传真实失败原因,
// 绑定保持可重试(否则会把可判定的失败写成永远无法自动清理的“未知代”毒状态)。
return nil
}
pending := runtimeCleanupGeneration(environment, bindingVersion, missingRuntimeID, "")
return errors.Join(hub.ErrConflict, store.SetRuntimeCleanupPending(ctx, pending, true))
}
runtime, found, err := reconcileGatewayRuntime(ctx, target, environment.Alias)
if err != nil {
runtimeID := missingRuntimeID
if bindingVersion == environment.BindingVersion && environment.RuntimeID != "" {
runtimeID = environment.RuntimeID
}
pending := runtimeCleanupGeneration(environment, bindingVersion, runtimeID, networkID)
return errors.Join(err, store.SetRuntimeCleanupPending(ctx, pending, true))
}
if found && (runtime.BindingVersion != bindingVersion || runtime.NetworkID != networkID) {
pending := runtimeCleanupGeneration(environment, bindingVersion, missingRuntimeID, networkID)
return errors.Join(hub.ErrConflict, store.SetRuntimeCleanupPending(ctx, pending, true))
}
if found {
environment = runtimeCleanupGeneration(environment, bindingVersion, runtime.ID, networkID)
} else {
environment = runtimeCleanupGeneration(environment, bindingVersion, missingRuntimeID, networkID)
}
_, cleanupErr := removeGatewayRuntime(ctx, store, target, environment)
return cleanupErr
}
// gatewayFailure 将网关错误转换为对调用方可读的失败;4xx 保留原状态,其余一律 502。
type gatewayFailure struct {
status int
message string
}
func (e gatewayFailure) Error() string { return e.message }
func gatewayRejected(status int, body []byte) error {
if status >= 400 && status < 500 {
return gatewayFailure{status: status, message: "gateway rejected: " + errorFromBody(body, status)}
}
return gatewayFailure{status: http.StatusBadGateway, message: fmt.Sprintf("gateway call failed with status %d", status)}
}
func gatewayUnreachable(err error) error {
return gatewayFailure{status: http.StatusBadGateway, message: fmt.Sprintf("gateway unreachable: %v", err)}
}
func reconcileGatewayRuntime(ctx context.Context, target hub.Gateway, alias string) (runtimeStatus, bool, error) {
var lastErr error
for attempt := 0; attempt < gatewayReconcileAttempts; attempt++ {
if attempt > 0 {
timer := time.NewTimer(gatewayReconcileDelay)
select {
case <-ctx.Done():
timer.Stop()
return runtimeStatus{}, false, ctx.Err()
case <-timer.C:
}
}
status, body, err := gatewayCall(ctx, target, http.MethodGet, "/v1/browsers", nil, 30*time.Second)
if err != nil {
lastErr = err
continue
}
if status != http.StatusOK {
lastErr = gatewayRejected(status, body)
continue
}
browsers, parseErr := parseGatewayBrowserList(body)
if parseErr != nil {
lastErr = parseErr
continue
}
lastErr = nil
for _, browser := range browsers {
if browser.Alias == alias {
return browser, true, nil
}
}
}
return runtimeStatus{}, false, lastErr
}
func parseGatewayBrowserList(body []byte) ([]runtimeStatus, error) {
var entries []json.RawMessage
if err := json.Unmarshal(body, &entries); err != nil || entries == nil {
return nil, errors.New("gateway returned an invalid browser list")
}
browsers := make([]runtimeStatus, 0, len(entries))
aliases := make(map[string]bool, len(entries))
for _, entry := range entries {
var browser *runtimeStatus
if err := json.Unmarshal(entry, &browser); err != nil || browser == nil || browser.ID == "" || browser.Alias == "" || browser.State == "" ||
browser.BindingVersion < 0 || aliases[browser.Alias] {
return nil, errors.New("gateway returned an invalid browser list")
}
aliases[browser.Alias] = true
browsers = append(browsers, *browser)
}
return browsers, nil
}
func errorFromBody(body []byte, status int) string {
var envelope struct {
Error string `json:"error"`
}
if json.Unmarshal(body, &envelope) == nil && envelope.Error != "" {
return envelope.Error
}
return http.StatusText(status)
}
type runtimeStatus struct {
ID string `json:"id"`
Alias string `json:"alias"`
Name string `json:"name"`
State string `json:"state"`
Status string `json:"status"`
Endpoint string `json:"endpoint"`
BindingVersion int64 `json:"binding_version"`
NetworkExitID string `json:"network_exit_id"`
NetworkID string `json:"network_id"`
NodeID string `json:"node_id"`
CleanupState string `json:"cleanup_state"`
CleanupError string `json:"cleanup_error"`
ProxyReady bool `json:"proxy_ready"`
}
type envView struct {
hub.Env
State string `json:"state"`
Status string `json:"status"`
RuntimeID string `json:"runtime_id"`
Endpoint string `json:"endpoint"`
AccountID string `json:"account_id"`
AccountStatus string `json:"account_status"`
AuthorizationStatus string `json:"authorization_status"`
NetworkExitID string `json:"network_exit_id"`
NetworkExitHealth string `json:"network_exit_health"`
BindingVersion int64 `json:"binding_version"`
RuntimeInstanceID string `json:"runtime_instance_id"`
RuntimeNodeID string `json:"runtime_node_id,omitempty"`
RuntimeStatus string `json:"runtime_status,omitempty"`
RuntimeCleanupState string `json:"runtime_cleanup_state,omitempty"`
RuntimeCleanupError string `json:"runtime_cleanup_error,omitempty"`
GatewayReachable bool `json:"gateway_reachable"`
ScheduleStatus string `json:"schedule_status"`
ScheduleBlockReason string `json:"schedule_block_reason,omitempty"`
RecoveryRequired bool `json:"recovery_required"`
CleanupPending bool `json:"cleanup_pending"`
}
func environmentScheduleReadiness(environment hub.EnvironmentContext) (string, string) {
switch {
case environment.AuthorizationStatus != "authorized":
return "blocked", "account_revoked"
case environment.AccountStatus != "active":
return "blocked", "account_paused"
case environment.Exit.ID != "" && environment.Exit.HealthStatus != "healthy":
return "blocked", "network_exit_unhealthy"
case environment.RuntimeCleanupPending:
return "blocked", "runtime_stop_pending"
case environment.RuntimeInstanceID == "":
return "blocked", "runtime_missing"
default:
return "ready", ""
}
}
func runtimeMatchesBinding(runtime runtimeStatus, environment hub.EnvironmentContext) bool {
if !validCreatedRuntime(runtime, environment, runtime.State == "running") || runtime.BindingVersion != environment.BindingVersion ||
(runtime.NetworkExitID != environment.Exit.ID && (runtime.State == "running" || runtime.NetworkExitID != "")) {
return false
}
return true
}
func validCreatedRuntime(created runtimeStatus, environment hub.EnvironmentContext, running bool) bool {
if !gatewayRuntimeIDPattern.MatchString(created.ID) ||
(created.NetworkID != "" && !gatewayNetworkIDPattern.MatchString(created.NetworkID)) {
return false
}
if !running {
return true
}
if created.NetworkID == "" {
return false
}
if environment.RuntimeInstanceID != "" || environment.RuntimeID != "" {
return environment.RuntimeInstanceID != "" && environment.RuntimeID == created.ID &&
environment.RuntimeNetworkID != "" && created.NetworkID == environment.RuntimeNetworkID
}
return environment.RuntimeNetworkID == "" || created.NetworkID == environment.RuntimeNetworkID
}
func accountRunnable(environment hub.EnvironmentContext) bool {
return environment.AccountStatus == "active" && environment.AuthorizationStatus == "authorized"
}
func releaseRuntime(ctx context.Context, store RuntimeStopStore, environment hub.EnvironmentContext) error {
err := store.ReleaseRuntime(ctx, environment)
if errors.Is(err, hub.ErrConflict) {
return nil
}
return err
}
func releaseRuntimeWithReconcileAudit(ctx context.Context, store HubStore, environment hub.EnvironmentContext) error {
action := actionForEnvironment("reconcile", environment)
if err := store.AppendEnvironmentAction(ctx, "environment_action_requested", action); err != nil {
return err
}
releaseErr := releaseRuntime(ctx, store, environment)
action.Outcome, action.ReasonCode = "succeeded", "runtime_released"
if releaseErr != nil {
action.Outcome, action.ReasonCode = "failed", "runtime_release_failed"
}
if auditErr := store.AppendEnvironmentAction(ctx, "environment_action_finished", action); auditErr != nil {
return errors.Join(releaseErr, auditErr)
}
return releaseErr
}
func registerHub(app *fiber.App, store HubStore) {
registerHubWithNetwork(app, store, defaultNetworkExitProbe(), nil)
}
func registerHubWithNetwork(app *fiber.App, store HubStore, probe NetworkExitProbe, resolve func(hub.NetworkExitAccess) (string, error)) {
app.Get("/api/browsers", listBrowsers(store))
app.Get("/api/browsers/:alias", getBrowser(store))
app.Post("/api/browsers", lockBrowserCreate(store, createBrowser(store, probe, resolve)))
app.Post("/api/browsers/:alias/:action", lockBrowserAlias(store, browserAction(store, probe, resolve)))
app.Delete("/api/browsers/:alias", lockBrowserAlias(store, deleteBrowser(store)))
app.Get("/api/network-exits", listNetworkExits(store))
app.Get("/api/network-exits/:id", getNetworkExit(store))
app.Post("/api/network-exits", createNetworkExit(store))
app.Post("/api/network-exits/:id/check", checkNetworkExit(store, probe))
app.Post("/api/network-exits/:id/disable", disableNetworkExit(store, probe, resolve))
app.Put("/api/network-exits/:id", func(c fiber.Ctx) error {
admin, ok := store.(networkExitAdminStore)
if !ok {
return hubError(c, hub.ErrConflict)
}
input := hub.NetworkExit{}
if err := decodeHubJSON(c, &input); err != nil {
return hubError(c, err)
}
exit, err := admin.UpdateNetworkExit(c.Context(), c.Params("id"), input)
if err != nil {
return hubError(c, err)
}
return c.JSON(exit)
})
app.Post("/api/network-exits/:id/enable", func(c fiber.Ctx) error {
admin, ok := store.(networkExitAdminStore)
if !ok {
return hubError(c, hub.ErrConflict)
}
exit, err := admin.EnableNetworkExit(c.Context(), c.Params("id"))
if err != nil {
return hubError(c, err)
}
return c.JSON(exit)
})
app.Delete("/api/network-exits/:id", func(c fiber.Ctx) error {
admin, ok := store.(networkExitAdminStore)
if !ok {
return hubError(c, hub.ErrConflict)
}
if err := admin.DeleteNetworkExit(c.Context(), c.Params("id")); err != nil {
return hubError(c, err)
}
return c.SendStatus(fiber.StatusNoContent)
})
app.Get("/api/gateways", func(c fiber.Ctx) error {
gateways, err := store.ListGateways(c.Context())
if err != nil {
return hubError(c, err)
}
return c.JSON(gateways)
})
app.Post("/api/gateways", func(c fiber.Ctx) error {
input := struct {
Name string `json:"name"`
Endpoint string `json:"endpoint"`
Token string `json:"token"`
}{}
if err := decodeHubJSON(c, &input); err != nil {
return hubError(c, err)
}
gateway, err := store.CreateGateway(c.Context(), input.Name, input.Endpoint, input.Token)
if err != nil {
return hubError(c, err)
}
return c.Status(fiber.StatusCreated).JSON(gateway)
})
app.Put("/api/gateways/:name", func(c fiber.Ctx) error {
input := struct {
Name string `json:"name"`
Endpoint string `json:"endpoint"`
Token string `json:"token"`
}{}
if err := decodeHubJSON(c, &input); err != nil {
return hubError(c, err)
}
gateway, err := store.UpdateGateway(c.Context(), c.Params("name"), input.Name, input.Endpoint, input.Token)
if err != nil {
return hubError(c, err)
}
return c.JSON(gateway)
})
app.Delete("/api/gateways/:name", func(c fiber.Ctx) error {
if err := store.DeleteGateway(c.Context(), c.Params("name")); err != nil {
return hubError(c, err)
}
return c.SendStatus(fiber.StatusNoContent)
})
}
func getBrowser(store HubStore) fiber.Handler {
return func(c fiber.Ctx) error {
environment, err := store.GetEnvironmentContext(c.Context(), c.Params("alias"))
if err != nil {
return hubError(c, err)
}
view := envView{
Env: environment.Env, State: "recorded", Status: "已记录运行实例",
AccountID: environment.AccountID, AccountStatus: environment.AccountStatus,
AuthorizationStatus: environment.AuthorizationStatus, NetworkExitID: environment.Exit.ID,
NetworkExitHealth: environment.Exit.HealthStatus, BindingVersion: environment.BindingVersion,
RuntimeInstanceID: environment.RuntimeInstanceID, RuntimeID: environment.RuntimeID, RuntimeNodeID: environment.RuntimeNodeID,
ScheduleStatus: "blocked", ScheduleBlockReason: "binding_missing",
RecoveryRequired: true, CleanupPending: environment.RuntimeCleanupPending,
}
view.ScheduleStatus, view.ScheduleBlockReason = environmentScheduleReadiness(environment)
if environment.RuntimeInstanceID == "" {
view.State, view.Status = "missing", "未记录运行实例"
}
runtimes, gatewayRead, gatewayErrors := gatewayRuntimeSnapshot(c.Context(), store, []hub.Env{environment.Env})
if gatewayErrors[environment.Gateway] != nil {
view.State, view.Status, view.GatewayReachable = "gateway_unreachable", "网关不可达", false
} else if gatewayRead[environment.Gateway] {
view.GatewayReachable = true
if runtime, found := runtimes[environment.Gateway][environment.Alias]; found {
view.RuntimeID, view.RuntimeStatus, view.RuntimeNodeID = runtime.ID, runtime.Status, runtime.NodeID
view.State, view.Status = runtime.State, runtime.Status
view.RuntimeCleanupState, view.RuntimeCleanupError = runtime.CleanupState, runtime.CleanupError
view.CleanupPending = view.CleanupPending || runtime.CleanupState == "pending"
view.RecoveryRequired = view.RecoveryRequired || runtime.State != "running" || runtime.CleanupState == "pending"
if runtime.State == "running" && runtime.Status == "ready" {
view.Endpoint = runtime.Endpoint
}
} else if view.RuntimeInstanceID != "" {
view.State, view.Status = "runtime_missing", "网关中未找到运行实例"
}
}
return c.JSON(view)
}
}
func lockBrowserAlias(store HubStore, handler fiber.Handler) fiber.Handler {
return func(c fiber.Ctx) error {
requestedExitID := ""
if c.Params("action") == "rebind" {
var input struct {
NetworkExitID string `json:"network_exit_id"`
}
if json.Unmarshal(c.Body(), &input) == nil {
if hub.ValidNetworkExitID(input.NetworkExitID) {
requestedExitID = input.NetworkExitID
}
}
}
alias := c.Params("alias")
unlock, err := lockAliasResources(c.Context(), store, alias, requestedExitID)
if err != nil {
return hubError(c, err)
}
defer unlock()
return handler(c)
}
}
func lockBrowserCreate(store HubStore, handler fiber.Handler) fiber.Handler {
return func(c fiber.Ctx) error {
var input struct {
Alias string `json:"alias"`
AccountID string `json:"account_id"`
NetworkExitID string `json:"network_exit_id"`
}
if json.Unmarshal(c.Body(), &input) != nil || input.Alias == "" {
return handler(c)
}
exitIDs := []string(nil)
if hub.ValidNetworkExitID(input.NetworkExitID) {
exitIDs = []string{input.NetworkExitID}
}
unlock, err := store.LockResources(c.Context(), nonEmpty(input.Alias, input.AccountID), exitIDs)
if err != nil {
return hubError(c, err)
}
defer unlock()
return handler(c)
}
}
func lockEnvironmentResources(ctx context.Context, store HubStore, envs []hub.Env) (func(), error) {
aliases := make([]string, 0, len(envs))
seen := make(map[string]bool, len(envs))
for _, env := range envs {
if !seen[env.Alias] {
seen[env.Alias] = true
aliases = append(aliases, env.Alias)
}
}
for {
before, err := environmentResourceMap(ctx, store, aliases)
if err != nil {
return nil, err
}
exitIDs := environmentResourceValues(before)
unlock, err := store.LockResources(ctx, aliases, exitIDs)
if err != nil {
return nil, err
}
after, err := environmentResourceMap(ctx, store, aliases)
if err != nil {
unlock()
return nil, err
}
if equalEnvironmentResourceMaps(before, after) {
return unlock, nil
}
unlock()
}
}
type environmentResource struct {
alias string
exitID string
}
type resourceLockStore interface {
LockResources(ctx context.Context, aliases, exitIDs []string, _ ...[]string) (func(), error)
GetEnvironmentContext(ctx context.Context, alias string) (hub.EnvironmentContext, error)
}
func lockAliasResources(ctx context.Context, store resourceLockStore, alias, requestedExitID string) (func(), error) {
for {
before, beforeFound, err := environmentResources(ctx, store, alias)
if err != nil {
return nil, err
}
unlock, err := store.LockResources(ctx, []string{alias}, nonEmpty(before.exitID, requestedExitID))
if err != nil {
return nil, err
}
after, afterFound, err := environmentResources(ctx, store, alias)
if err != nil {
unlock()
return nil, err
}
if beforeFound == afterFound && before == after {
return unlock, nil
}
unlock()
}
}
func lockAccountResources(ctx context.Context, store RuntimeStopStore, accountID string) (func(), error) {
if store == nil {
return func() {}, nil
}
for {
before, beforeFound, err := accountEnvironmentResources(ctx, store, accountID)
if err != nil {
return nil, err
}
unlock, err := store.LockResources(ctx, nonEmpty(accountID, before.alias), nonEmpty(before.exitID))
if err != nil {
return nil, err
}
after, afterFound, err := accountEnvironmentResources(ctx, store, accountID)
if err != nil {
unlock()
return nil, err
}
if beforeFound == afterFound && before == after {
return unlock, nil
}
unlock()
}
}
func accountEnvironmentResources(ctx context.Context, store RuntimeStopStore, accountID string) (environmentResource, bool, error) {
environment, err := store.GetEnvironmentContextForAccount(ctx, accountID)
if errors.Is(err, hub.ErrNotFound) {
return environmentResource{}, false, nil
}
if err != nil {
return environmentResource{}, false, err
}
return environmentResource{alias: environment.Alias, exitID: environment.Exit.ID}, true, nil
}
func environmentResources(ctx context.Context, store resourceLockStore, alias string) (environmentResource, bool, error) {
environment, err := store.GetEnvironmentContext(ctx, alias)
if errors.Is(err, hub.ErrNotFound) {
return environmentResource{}, false, nil
}
if err != nil {
return environmentResource{}, false, err
}
return environmentResource{alias: environment.Alias, exitID: environment.Exit.ID}, true, nil
}
func environmentResourceMap(ctx context.Context, store HubStore, aliases []string) (map[string]environmentResource, error) {
result := make(map[string]environmentResource, len(aliases))
for _, alias := range aliases {
resources, found, err := environmentResources(ctx, store, alias)
if err != nil {
return nil, err
}
if found {
result[alias] = resources
}
}
return result, nil
}
func nonEmpty(values ...string) []string {
result := make([]string, 0, len(values))
seen := make(map[string]bool, len(values))
for _, value := range values {
if value != "" && !seen[value] {
seen[value] = true
result = append(result, value)
}
}
return result
}
func environmentResourceValues(values map[string]environmentResource) []string {
exitIDs := make([]string, 0, len(values))
for _, value := range values {
exitIDs = append(exitIDs, value.exitID)
}
return nonEmpty(exitIDs...)
}
func equalEnvironmentResourceMaps(left, right map[string]environmentResource) bool {
if len(left) != len(right) {
return false
}
for key, value := range left {
other, ok := right[key]
if !ok || other != value {
return false
}
}
return true
}
func listNetworkExits(store HubStore) fiber.Handler {
return func(c fiber.Ctx) error {
exits, err := store.ListNetworkExits(c.Context())
if err != nil {
return hubError(c, err)
}
return c.JSON(exits)
}
}
func getNetworkExit(store HubStore) fiber.Handler {
return func(c fiber.Ctx) error {
exit, err := store.GetNetworkExit(c.Context(), c.Params("id"))
if err != nil {
return hubError(c, err)
}
return c.JSON(exit)
}
}
func createNetworkExit(store HubStore) fiber.Handler {
return func(c fiber.Ctx) error {
var input struct {
Protocol string `json:"protocol"`
Host string `json:"host"`
Port int `json:"port"`
Username string `json:"username"`
Password string `json:"password"`
ExpectedPublicIP string `json:"expected_public_ip"`
ExpectedRegion string `json:"expected_region"`
}
if err := decodeHubJSON(c, &input); err != nil {
return hubError(c, err)
}
exit, err := store.CreateNetworkExit(c.Context(), hub.NetworkExit{
Protocol: input.Protocol, Host: input.Host, Port: input.Port,
Username: input.Username, Password: input.Password,
ExpectedPublicIP: input.ExpectedPublicIP, ExpectedRegion: input.ExpectedRegion,
})
if err != nil {
return hubError(c, err)
}
return c.Status(fiber.StatusCreated).JSON(exit)
}
}
func checkNetworkExit(store HubStore, probe NetworkExitProbe) fiber.Handler {
return func(c fiber.Ctx) error {
unlock, err := store.LockResources(c.Context(), nil, []string{c.Params("id")}, nil)
if err != nil {
return hubError(c, err)
}
defer unlock()
access, err := store.GetNetworkExitAccess(c.Context(), c.Params("id"))
if err != nil {
return hubError(c, err)
}
if access.HealthStatus == "disabled" {
return hubError(c, hub.ErrConflict)
}
observation, failureReason := probe.Check(c.Context(), access)
exit, reason, err := store.RecordNetworkExitCheck(c.Context(), access.ID, observation, failureReason)
if err != nil {
return hubError(c, err)
}
return c.JSON(struct {
hub.NetworkExit
ReasonCode string `json:"reason_code"`
}{exit, reason})
}
}
func disableNetworkExit(store HubStore, probe NetworkExitProbe, resolve func(hub.NetworkExitAccess) (string, error)) fiber.Handler {
return func(c fiber.Ctx) error {
unlock, err := store.LockResources(c.Context(), nil, []string{c.Params("id")}, nil)
if err != nil {
return hubError(c, err)
}
exit, err := store.DisableNetworkExit(c.Context(), c.Params("id"))
unlock()
if err != nil {
return hubError(c, err)
}
if err := reconcileRuntimeLeasesUnlocked(c.Context(), store, probe, resolve); err != nil {
return hubError(c, err)
}
return c.JSON(exit)
}
}
func verifyNetworkExit(ctx context.Context, store HubStore, probe NetworkExitProbe, id string) (hub.NetworkExitAccess, string, error) {
access, err := store.GetNetworkExitAccess(ctx, id)
if err != nil {
return hub.NetworkExitAccess{}, "exit_unavailable", err
}
if access.HealthStatus == "disabled" {
return hub.NetworkExitAccess{}, "exit_disabled", hub.ErrConflict
}
observation, failureReason := probe.Check(ctx, access)
exit, reason, err := store.RecordNetworkExitCheck(ctx, id, observation, failureReason)
if err != nil {
return hub.NetworkExitAccess{}, reason, err
}
if exit.HealthStatus != "healthy" {
return hub.NetworkExitAccess{}, reason, hub.ErrConflict
}
access, err = store.GetNetworkExitAccess(ctx, id)
return access, reason, err
}
func listBrowsers(store HubStore) fiber.Handler {
return func(c fiber.Ctx) error {
envs, err := store.ListEnvs(c.Context())
if err != nil {
return hubError(c, err)
}
runtimes, gatewayRead, gatewayErrors := gatewayRuntimeSnapshot(c.Context(), store, envs)
views := make([]envView, 0, len(envs))
for _, env := range envs {
view := envView{
Env: env, State: "missing", Status: "未记录运行实例",
ScheduleStatus: "blocked", ScheduleBlockReason: "binding_missing",
RecoveryRequired: true, GatewayReachable: false,
}
if environment, contextErr := store.GetEnvironmentContext(c.Context(), env.Alias); contextErr == nil {
view.AccountID, view.AccountStatus, view.AuthorizationStatus = environment.AccountID, environment.AccountStatus, environment.AuthorizationStatus
view.NetworkExitID, view.NetworkExitHealth, view.BindingVersion = environment.Exit.ID, environment.Exit.HealthStatus, environment.BindingVersion
view.RuntimeInstanceID = environment.RuntimeInstanceID
view.RuntimeID = environment.RuntimeID
view.RuntimeNodeID = environment.RuntimeNodeID
view.ScheduleStatus, view.ScheduleBlockReason = environmentScheduleReadiness(environment)
view.CleanupPending = environment.RuntimeCleanupPending
view.RecoveryRequired = (environment.Exit.ID != "" && environment.Exit.HealthStatus != "healthy") || environment.RuntimeCleanupPending
if environment.RuntimeCleanupPending {
view.State, view.Status = "cleanup_pending", "运行资源待清理"
} else if environment.RuntimeInstanceID != "" {
view.State, view.Status = "recorded", "已记录运行实例"
}
} else if !errors.Is(contextErr, hub.ErrNotFound) {
return hubError(c, contextErr)
}
if gatewayErrors[env.Gateway] != nil {
view.State, view.Status = "gateway_unreachable", "网关不可达"
view.RecoveryRequired = true
} else if gatewayRead[env.Gateway] {
view.GatewayReachable = true
if runtime, found := runtimes[env.Gateway][env.Alias]; found {
view.RuntimeID, view.RuntimeStatus, view.RuntimeNodeID = runtime.ID, runtime.Status, runtime.NodeID
view.State, view.Status = runtime.State, runtime.Status
view.RuntimeCleanupState, view.RuntimeCleanupError = runtime.CleanupState, runtime.CleanupError
view.CleanupPending = view.CleanupPending || runtime.CleanupState == "pending"
view.RecoveryRequired = view.RecoveryRequired || runtime.State != "running" || runtime.CleanupState == "pending"
if runtime.State == "running" && runtime.Status == "ready" {
view.Endpoint = runtime.Endpoint
}
} else if view.RuntimeInstanceID != "" {
view.State, view.Status = "runtime_missing", "网关中未找到运行实例"
view.RecoveryRequired = true
}
}
views = append(views, view)
}
return c.JSON(views)
}
}
func reconcileRuntimeLeases(ctx context.Context, store HubStore, probe NetworkExitProbe, resolve func(hub.NetworkExitAccess) (string, error)) error {
return reconcileRuntimeLeasesUnlocked(ctx, store, probe, resolve)
}
func reconcileRuntimeLeasesUnlocked(ctx context.Context, store HubStore, probe NetworkExitProbe, resolve func(hub.NetworkExitAccess) (string, error)) error {
envs, err := store.ListEnvs(ctx)
if err != nil {
return err
}
unlock, err := lockEnvironmentResources(ctx, store, envs)
if err != nil {
return err
}
defer unlock()
runtimes, gatewayRead, gatewayErrors := gatewayRuntimeSnapshot(ctx, store, envs)
return reconcileRuntimeSnapshot(ctx, store, probe, resolve, envs, runtimes, gatewayRead, gatewayErrors)
}
func gatewayRuntimeSnapshot(ctx context.Context, store HubStore, envs []hub.Env) (map[string]map[string]runtimeStatus, map[string]bool, map[string]error) {
gateways := map[string]bool{}
runtimes := map[string]map[string]runtimeStatus{}
gatewayRead := map[string]bool{}
gatewayErrors := map[string]error{}
for _, env := range envs {
if gateways[env.Gateway] {
continue
}
gateways[env.Gateway] = true
gateway, err := store.GetGateway(ctx, env.Gateway)
if err != nil {
gatewayErrors[env.Gateway] = gatewayUnreachable(err)
continue
}
status, body, callErr := gatewayCall(ctx, gateway, http.MethodGet, "/v1/browsers", nil, 30*time.Second)
if callErr != nil {
gatewayErrors[env.Gateway] = gatewayUnreachable(callErr)
continue
}
if status != http.StatusOK {
gatewayErrors[env.Gateway] = gatewayRejected(status, body)
continue
}
list, parseErr := parseGatewayBrowserList(body)
if parseErr != nil {
gatewayErrors[env.Gateway] = gatewayFailure{status: http.StatusBadGateway, message: "gateway returned an invalid browser list"}
continue
}
byAlias := map[string]runtimeStatus{}
for _, runtime := range list {
byAlias[runtime.Alias] = runtime
}
runtimes[env.Gateway] = byAlias
gatewayRead[env.Gateway] = true
}
return runtimes, gatewayRead, gatewayErrors
}
func reconcileRuntimeSnapshot(ctx context.Context, store HubStore, probe NetworkExitProbe, resolve func(hub.NetworkExitAccess) (string, error), envs []hub.Env,
runtimes map[string]map[string]runtimeStatus, gatewayRead map[string]bool, gatewayErrors map[string]error) error {
var firstErr error
for _, env := range envs {
if readErr := gatewayErrors[env.Gateway]; readErr != nil {
if firstErr == nil {
firstErr = readErr
}
continue
}
environment, err := store.GetEnvironmentContext(ctx, env.Alias)
if errors.Is(err, hub.ErrNotFound) {
continue
}
if err != nil {
return err
}
if environment.RuntimeCleanupPending {
target, targetErr := store.GetGateway(ctx, environment.Gateway)
if targetErr != nil {
return targetErr
}
if _, cleanupErr := removeGatewayRuntime(ctx, store, target, environment); cleanupErr != nil {
return cleanupErr
}
continue
}
runtime, found := runtimes[env.Gateway][env.Alias]
if !accountRunnable(environment) {
if found && runtime.State == "running" {
if err := stopEnvironmentRuntime(ctx, store, environment); err != nil {
return err
}
} else if gatewayRead[env.Gateway] && environment.RuntimeInstanceID != "" {
if err := releaseRuntimeWithReconcileAudit(ctx, store, environment); err != nil {
return err
}
}
continue
}
if found && runtime.State == "running" {
auditRecovery := !runtimeMatchesBinding(runtime, environment) || !runtime.ProxyReady
action := actionForEnvironment("reconcile", environment)
if auditRecovery {
if err := store.AppendEnvironmentAction(ctx, "environment_action_requested", action); err != nil {
return err
}
}
ready, restoreErr := restoreOrRebuildRuntime(ctx, store, probe, resolve, environment, runtime)
if auditRecovery {
current, contextErr := store.GetEnvironmentContext(ctx, environment.Alias)
if contextErr == nil {
action.RuntimeInstanceID, action.BindingVersion, action.NetworkExitID =
current.RuntimeInstanceID, current.BindingVersion, current.Exit.ID
} else {
action.RuntimeInstanceID = ""
action.BindingVersion = 0
action.NetworkExitID = ""
restoreErr = errors.Join(restoreErr, contextErr)
}
action.Outcome, action.ReasonCode = "succeeded", "runtime_reconciled"
if !ready {
action.Outcome, action.ReasonCode = "failed", "runtime_unavailable"
}
if restoreErr != nil {
action.Outcome, action.ReasonCode = runtimeRecoveryFailure(ctx, store, environment.Alias, restoreErr)
}
if err := store.AppendEnvironmentAction(ctx, "environment_action_finished", action); err != nil {
return errors.Join(restoreErr, err)
}
}
if restoreErr != nil {
return restoreErr
}
} else if gatewayRead[env.Gateway] && environment.RuntimeInstanceID != "" {
if err := releaseRuntimeWithReconcileAudit(ctx, store, environment); err != nil {
return err
}
}
}
return firstErr
}
func runtimeRecoveryFailure(ctx context.Context, store HubStore, alias string, err error) (string, string) {
var gatewayErr gatewayFailure
if errors.As(err, &gatewayErr) {
return "unknown", "gateway_result_unknown"
}
if environment, contextErr := store.GetEnvironmentContext(ctx, alias); contextErr == nil && environment.RuntimeCleanupPending {
return "unknown", "cleanup_result_unknown"
}
return "failed", "runtime_persistence_failed"
}
func restoreOrRebuildRuntime(ctx context.Context, store HubStore, probe NetworkExitProbe,
_ func(hub.NetworkExitAccess) (string, error), environment hub.EnvironmentContext, runtime runtimeStatus) (bool, error) {
if environment.RuntimeCleanupPending {
target, err := store.GetGateway(ctx, environment.Gateway)
if err != nil {
return false, err
}
_, err = removeGatewayRuntime(ctx, store, target, environment)
return false, err
}
var access hub.NetworkExitAccess
var err error
if environment.Exit.ID != "" {
access, _, err = verifyNetworkExit(ctx, store, probe, environment.Exit.ID)
if err != nil {
return false, discardRuntime(ctx, store, environment)
}
}
target, err := store.GetGateway(ctx, environment.Gateway)
if err != nil {
return false, err
}
if runtimeMatchesBinding(runtime, environment) && runtime.ProxyReady {
_, err := activateGatewayRuntime(ctx, store, target, environment, runtime.ID, runtime.NetworkID, runtime.NodeID)
return err == nil, err
}
networkExit := gatewayNetworkExit{}
if environment.Exit.ID != "" {
networkExit = gatewayNetworkExitFor(access)
}
if runtimeMatchesBinding(runtime, environment) {
if environment.Exit.ID == "" {
runtime.ProxyReady = true
} else {
status, _, callErr := gatewayCall(ctx, target, http.MethodPost, "/v1/browsers/"+environment.Alias+"/proxy",
gatewayProxyPayload(environment, runtime.ID, runtime.NetworkID, networkExit), 30*time.Second)
if callErr == nil && status == http.StatusNoContent {
runtime.ProxyReady = true
}
}
if runtime.ProxyReady {
_, err := activateGatewayRuntime(ctx, store, target, environment, runtime.ID, runtime.NetworkID, runtime.NodeID)
return err == nil, err
}
}
_, removeErr := removeGatewayRuntime(ctx, store, target, environment)
if removeErr != nil {
return false, removeErr
}
environment, err = store.GetEnvironmentContext(ctx, environment.Alias)
if err != nil {
return false, err
}
browserPath := ""
status, body, callErr := gatewayCall(ctx, target, http.MethodPost, "/v1/browsers",
gatewayCreatePayloadForAccount(environment, browserPath, networkExit), gatewayLongTimeout)
if callErr != nil {
return false, errors.Join(gatewayUnreachable(callErr), reconcileGatewayCreate(ctx, store, target, environment, status, callErr, body))
}
if status != http.StatusCreated {
return false, errors.Join(gatewayRejected(status, body), reconcileGatewayCreate(ctx, store, target, environment, status, callErr, body))
}
var created runtimeStatus
if json.Unmarshal(body, &created) != nil || !validCreatedRuntime(created, environment, true) {
return false, errors.Join(errors.New("gateway returned an invalid runtime generation"),
reconcileGatewayCreate(ctx, store, target, environment, status, callErr, body))
}
_, err = activateGatewayRuntime(ctx, store, target, environment, created.ID, created.NetworkID, created.NodeID)
return err == nil, err
}
type runtimeNodeStore interface {
SetRuntimeNode(context.Context, string, string, string) error
}
func activateGatewayRuntime(ctx context.Context, store HubStore, target hub.Gateway, environment hub.EnvironmentContext, runtimeID, networkID string, nodeIDs ...string) (hub.EnvironmentContext, error) {
if !validCreatedRuntime(runtimeStatus{ID: runtimeID, NetworkID: networkID}, environment, true) {
return hub.EnvironmentContext{}, hub.ErrConflict
}
current, err := store.ActivateRuntime(ctx, environment.Alias, runtimeID, environment.BindingVersion, environment.Exit.ID, networkID)
if err == nil && len(nodeIDs) > 0 && nodeIDs[0] != "" {
if recorder, ok := store.(runtimeNodeStore); ok {
if nodeErr := recorder.SetRuntimeNode(ctx, environment.Alias, runtimeID, nodeIDs[0]); nodeErr != nil {
err = nodeErr
} else {
current.RuntimeNodeID = nodeIDs[0]
}
}
}
if err == nil {
return current, nil
}
// A scheduler heartbeat can win the activation race after the action has
// created the runtime. If it persisted this exact generation, keep the
// runtime and treat the activation as successful instead of deleting the
// winner from the gateway.
if errors.Is(err, hub.ErrConflict) {
if reconciled, readErr := store.GetEnvironmentContext(ctx, environment.Alias); readErr == nil &&
reconciled.BindingVersion == environment.BindingVersion &&
reconciled.RuntimeID == runtimeID &&
reconciled.RuntimeNetworkID == networkID &&
reconciled.RuntimeInstanceID != "" {
return reconciled, nil
}
}
cleanup := environment
cleanup.RuntimeID, cleanup.RuntimeNetworkID = runtimeID, networkID
_, cleanupErr := removeGatewayRuntime(ctx, store, target, cleanup)
return hub.EnvironmentContext{}, errors.Join(err, cleanupErr)
}
func discardRuntime(ctx context.Context, store HubStore, environment hub.EnvironmentContext) error {
target, targetErr := store.GetGateway(ctx, environment.Gateway)
if targetErr != nil {
return targetErr
}
_, err := removeGatewayRuntime(ctx, store, target, environment)
return err
}
func createBrowser(store HubStore, probe NetworkExitProbe, resolve func(hub.NetworkExitAccess) (string, error)) fiber.Handler {
return func(c fiber.Ctx) error {
input := struct {
Alias string `json:"alias"`
Name string `json:"name"`
Gateway string `json:"gateway"`
Fingerprint hub.Fingerprint `json:"fingerprint"`
AccountID string `json:"account_id"`
NetworkExitID string `json:"network_exit_id"`
}{}
if err := decodeHubJSON(c, &input); err != nil {
return hubError(c, err)
}
if input.Fingerprint.ProxyServer != "" {
return hubError(c, hub.ErrInvalid)
}
input.Fingerprint.DisableNonProxiedUDP = false
env := hub.Env{Alias: input.Alias, Name: input.Name, Gateway: input.Gateway, Fingerprint: input.Fingerprint}
if err := env.Fingerprint.Validate(); err != nil {
return c.Status(fiber.StatusBadRequest).JSON(map[string]string{"error": err.Error()})
}
environment, created, err := store.CreateBoundEnv(c.Context(), env, input.AccountID, input.NetworkExitID)
if err != nil {
return hubError(c, err)
}
action := actionForEnvironment("create", environment)
if err := store.AppendEnvironmentAction(c.Context(), "environment_action_requested", action); err != nil {
return hubError(c, err)
}
finish := func(outcome, reason string, current hub.EnvironmentContext) error {
action.Outcome, action.ReasonCode, action.RuntimeInstanceID = outcome, reason, current.RuntimeInstanceID
action.BindingVersion, action.NetworkExitID = current.BindingVersion, current.Exit.ID
return store.AppendEnvironmentAction(c.Context(), "environment_action_finished", action)
}
var access hub.NetworkExitAccess
if input.NetworkExitID != "" {
var reason string
access, reason, err = verifyNetworkExit(c.Context(), store, probe, input.NetworkExitID)
if err != nil {
_ = finish("failed", reason, environment)
return hubError(c, err)
}
}
gateway, err := store.GetGateway(c.Context(), env.Gateway)
if err != nil {
_ = finish("failed", "gateway_unavailable", environment)
return hubError(c, err)
}
if environment.RuntimeCleanupPending {
if _, cleanupErr := removeGatewayRuntime(c.Context(), store, gateway, environment); cleanupErr != nil {
_ = finish("unknown", "cleanup_result_unknown", environment)
return hubError(c, cleanupErr)
}
environment, err = store.GetEnvironmentContext(c.Context(), env.Alias)
if err != nil {
return hubError(c, err)
}
}
running := accountRunnable(environment)
browserPath := ""
networkExit := gatewayNetworkExit{}
if input.NetworkExitID != "" {
networkExit = gatewayNetworkExitFor(access)
}
if !created {
runtime, found, reconcileErr := reconcileGatewayRuntime(c.Context(), gateway, env.Alias)
if reconcileErr != nil {
_ = finish("unknown", "gateway_result_unknown", environment)
return hubError(c, gatewayUnreachable(reconcileErr))
}
if found && !running {
if !runtimeMatchesBinding(runtime, environment) {
_ = finish("failed", "runtime_unavailable", environment)
return hubError(c, hub.ErrConflict)
}
if runtime.State == "running" {
stopped := environment
stopped.RuntimeID, stopped.RuntimeNetworkID = runtime.ID, runtime.NetworkID
if err := stopEnvironmentRuntime(c.Context(), store, stopped); err != nil {
_ = finish("unknown", "cleanup_result_unknown", environment)
return hubError(c, err)
}
}
if err := finish("succeeded", "environment_reused", environment); err != nil {
return hubError(c, err)
}
return c.JSON(map[string]string{"alias": env.Alias})
}
if found && runtime.State == "running" {
ready, restoreErr := restoreOrRebuildRuntime(c.Context(), store, probe, resolve, environment, runtime)
if restoreErr != nil {
outcome, reason := runtimeRecoveryFailure(c.Context(), store, environment.Alias, restoreErr)
_ = finish(outcome, reason, environment)
return hubError(c, restoreErr)
}
if !ready {
_ = finish("failed", "runtime_unavailable", environment)
return c.Status(fiber.StatusConflict).JSON(map[string]string{"error": "environment exists but runtime is not running"})
}
environment, err = store.GetEnvironmentContext(c.Context(), env.Alias)
if err != nil {
return hubError(c, err)
}
if err := finish("succeeded", "environment_reused", environment); err != nil {
return hubError(c, err)
}
return c.JSON(map[string]string{"alias": env.Alias})
}
}
status, body, callErr := gatewayCall(c.Context(), gateway, http.MethodPost, "/v1/browsers",
gatewayCreatePayloadForAccount(environment, browserPath, networkExit), gatewayLongTimeout)
if callErr != nil || status != http.StatusCreated {
createErr := gatewayRejected(status, body)
if callErr != nil {
createErr = gatewayUnreachable(callErr)
}
reconcileErr := reconcileGatewayCreate(c.Context(), store, gateway, environment, status, callErr, body)
if reconcileErr != nil {
_ = finish("unknown", "gateway_result_unknown", environment)
return hubError(c, errors.Join(createErr, reconcileErr))
}
_ = finish("failed", "gateway_create_failed", environment)
return hubError(c, createErr)
}
var createdRuntime runtimeStatus
if json.Unmarshal(body, &createdRuntime) != nil || !validCreatedRuntime(createdRuntime, environment, running) {
reconcileErr := reconcileGatewayCreate(c.Context(), store, gateway, environment, status, callErr, body)
_ = finish("unknown", "gateway_result_unknown", environment)
return hubError(c, errors.Join(gatewayFailure{status: http.StatusBadGateway,
message: "gateway create result unknown; environment retained for reconciliation"}, reconcileErr))
}
if !running {
if err := finish("succeeded", "environment_created", environment); err != nil {
return hubError(c, err)
}
return c.Status(fiber.StatusCreated).JSON(map[string]string{"alias": env.Alias})
}
environment, err = activateGatewayRuntime(c.Context(), store, gateway, environment, createdRuntime.ID, createdRuntime.NetworkID)
if err != nil {
_ = finish("failed", "runtime_persistence_failed", environment)
return hubError(c, err)
}
if err := finish("succeeded", "environment_created", environment); err != nil {
return hubError(c, err)
}
return c.Status(fiber.StatusCreated).JSON(map[string]string{"alias": env.Alias})
}
}
func browserAction(store HubStore, probe NetworkExitProbe, resolve func(hub.NetworkExitAccess) (string, error)) fiber.Handler {
return func(c fiber.Ctx) error {
switch c.Params("action") {
case "start", "stop":
return lifecycleAction(store, probe, resolve, c)
case "rebind":
return rebindBrowser(store, probe, resolve, c)
default:
return hubError(c, hub.ErrInvalid)
}
}
}
func lifecycleAction(store HubStore, probe NetworkExitProbe, resolve func(hub.NetworkExitAccess) (string, error), c fiber.Ctx) error {
environment, err := store.GetEnvironmentContext(c.Context(), c.Params("alias"))
if err != nil {
return hubError(c, err)
}
actionName := strings.Clone(c.Params("action"))
if actionName == "start" {
if !accountRunnable(environment) {
return hubError(c, hub.ErrConflict)
}
action := actionForEnvironment(actionName, environment)
if err := store.AppendEnvironmentAction(c.Context(), "environment_action_requested", action); err != nil {
return hubError(c, err)
}
finish := func(outcome, reason string, current hub.EnvironmentContext) error {
action.Outcome, action.ReasonCode, action.RuntimeInstanceID = outcome, reason, current.RuntimeInstanceID
action.BindingVersion, action.NetworkExitID = current.BindingVersion, current.Exit.ID
return store.AppendEnvironmentAction(c.Context(), "environment_action_finished", action)
}
return startBrowser(store, probe, resolve, c, environment, finish)
}
if err := stopEnvironmentRuntime(c.Context(), store, environment); err != nil {
return hubError(c, err)
}
return c.SendStatus(fiber.StatusNoContent)
}
func stopAccountRuntime(ctx context.Context, store RuntimeStopStore, accountID string) error {
environment, err := store.GetEnvironmentContextForAccount(ctx, accountID)
if errors.Is(err, hub.ErrNotFound) {
return nil
}
if err != nil {
return err
}
return stopEnvironmentRuntime(ctx, store, environment)
}
func stopEnvironmentRuntime(ctx context.Context, store RuntimeStopStore, environment hub.EnvironmentContext) error {
action := actionForEnvironment("stop", environment)
if err := store.AppendEnvironmentAction(ctx, "environment_action_requested", action); err != nil {
return err
}
finish := func(outcome, reason string) error {
action.Outcome, action.ReasonCode = outcome, reason
return store.AppendEnvironmentAction(ctx, "environment_action_finished", action)
}
gateway, err := store.GetGateway(ctx, environment.Gateway)
if err != nil {
_ = finish("failed", "gateway_unavailable")
return err
}
if environment.RuntimeCleanupPending {
if _, cleanupErr := removeGatewayRuntime(ctx, store, gateway, environment); cleanupErr != nil {
_ = finish("unknown", "cleanup_result_unknown")
return cleanupErr
}
return finish("succeeded", "environment_stopped")
}
if environment.RuntimeID == "" {
runtime, found, reconcileErr := reconcileGatewayRuntime(ctx, gateway, environment.Alias)
if reconcileErr != nil {
_ = finish("unknown", "gateway_result_unknown")
return gatewayUnreachable(reconcileErr)
}
if !found {
return finish("succeeded", "environment_stopped")
}
if runtime.BindingVersion != environment.BindingVersion {
_ = finish("failed", "runtime_generation_conflict")
return hub.ErrConflict
}
environment = runtimeCleanupGeneration(environment, environment.BindingVersion, runtime.ID, runtime.NetworkID)
} else {
environment = runtimeCleanupGeneration(environment, environment.BindingVersion, environment.RuntimeID)
}
if err := store.SetRuntimeCleanupPending(ctx, environment, true); err != nil {
_ = finish("failed", "runtime_release_failed")
return err
}
status, body, callErr := gatewayCall(ctx, gateway, http.MethodPost, "/v1/browsers/"+environment.Alias+"/stop",
gatewayCleanupGenerationPayload(environment), 30*time.Second)
if callErr != nil || status >= http.StatusInternalServerError {
runtime, found, reconcileErr := reconcileGatewayRuntime(ctx, gateway, environment.Alias)
if reconcileErr != nil {
_ = finish("unknown", "gateway_result_unknown")
if callErr != nil {
return gatewayUnreachable(callErr)
}
return gatewayRejected(status, body)
}
if !found || runtime.State != "running" {
if err := store.SetRuntimeCleanupPending(ctx, environment, false); err != nil {
_ = finish("failed", "runtime_release_failed")
return err
}
return finish("succeeded", "gateway_reconciled")
}
_ = finish("failed", "gateway_action_failed")
if callErr != nil {
return gatewayUnreachable(callErr)
}
return gatewayRejected(status, body)
}
if status != http.StatusNoContent && status != http.StatusNotModified && status != http.StatusNotFound {
_ = finish("failed", "gateway_rejected")
return gatewayRejected(status, body)
}
if err := store.SetRuntimeCleanupPending(ctx, environment, false); err != nil {
_ = finish("failed", "runtime_release_failed")
return err
}
return finish("succeeded", "environment_stopped")
}
func startBrowser(store HubStore, probe NetworkExitProbe, _ func(hub.NetworkExitAccess) (string, error), c fiber.Ctx,
environment hub.EnvironmentContext, finish func(string, string, hub.EnvironmentContext) error) error {
if err := startBrowserRuntime(c.Context(), store, probe, environment, finish); err != nil {
return hubError(c, err)
}
return c.SendStatus(fiber.StatusNoContent)
}
func startBrowserRuntime(ctx context.Context, store HubStore, probe NetworkExitProbe,
environment hub.EnvironmentContext, finish func(string, string, hub.EnvironmentContext) error) error {
if !accountRunnable(environment) {
return hub.ErrConflict
}
var access hub.NetworkExitAccess
var err error
if environment.Exit.ID != "" {
var reason string
access, reason, err = verifyNetworkExit(ctx, store, probe, environment.Exit.ID)
if err != nil {
if cleanupErr := discardRuntime(ctx, store, environment); cleanupErr != nil {
_ = finish("unknown", "cleanup_result_unknown", environment)
return cleanupErr
}
_ = finish("failed", reason, environment)
return err
}
}
gateway, err := store.GetGateway(ctx, environment.Gateway)
if err != nil {
_ = finish("failed", "gateway_unavailable", environment)
return err
}
if environment.RuntimeCleanupPending {
if _, cleanupErr := removeGatewayRuntime(ctx, store, gateway, environment); cleanupErr != nil {
_ = finish("unknown", "cleanup_result_unknown", environment)
return cleanupErr
}
environment, err = store.GetEnvironmentContext(ctx, environment.Alias)
if err != nil {
return err
}
if !accountRunnable(environment) {
return hub.ErrConflict
}
}
browserPath := ""
networkExit := gatewayNetworkExit{}
if environment.Exit.ID != "" {
networkExit = gatewayNetworkExitFor(access)
}
runtime, found, err := reconcileGatewayRuntime(ctx, gateway, environment.Alias)
if err != nil {
_ = finish("unknown", "gateway_result_unknown", environment)
return gatewayUnreachable(err)
}
if found && runtime.State == "running" && runtime.ProxyReady && runtimeMatchesBinding(runtime, environment) {
environment, err = activateGatewayRuntime(ctx, store, gateway, environment, runtime.ID, runtime.NetworkID, runtime.NodeID)
if err != nil {
_ = finish("failed", "runtime_persistence_failed", environment)
return err
}
if err := finish("succeeded", "gateway_reconciled", environment); err != nil {
return err
}
return nil
}
if found {
cleanup := environment
cleanup.RuntimeID, cleanup.RuntimeNetworkID = runtime.ID, runtime.NetworkID
if _, removeErr := removeGatewayRuntime(ctx, store, gateway, cleanup); removeErr != nil {
_ = finish("unknown", "gateway_result_unknown", environment)
return removeErr
}
} else if err := releaseRuntime(ctx, store, environment); err != nil {
_ = finish("failed", "runtime_release_failed", environment)
return err
}
status, body, callErr := gatewayCall(ctx, gateway, http.MethodPost, "/v1/browsers",
gatewayCreatePayload(environment, browserPath, networkExit), gatewayLongTimeout)
if callErr != nil || status != http.StatusCreated {
reconcileErr := reconcileGatewayCreate(ctx, store, gateway, environment, status, callErr, body)
if reconcileErr != nil {
_ = finish("unknown", "gateway_result_unknown", environment)
return errors.Join(gatewayFailure{status: http.StatusBadGateway, message: "gateway start result unknown; retry to reconcile"}, reconcileErr)
}
_ = finish("failed", "gateway_create_failed", environment)
if callErr != nil {
return gatewayUnreachable(callErr)
}
return gatewayRejected(status, body)
}
var created runtimeStatus
if json.Unmarshal(body, &created) != nil || !validCreatedRuntime(created, environment, true) {
reconcileErr := reconcileGatewayCreate(ctx, store, gateway, environment, status, callErr, body)
_ = finish("unknown", "gateway_result_unknown", environment)
return errors.Join(gatewayFailure{status: http.StatusBadGateway, message: "gateway start result unknown; retry to reconcile"}, reconcileErr)
}
environment, err = activateGatewayRuntime(ctx, store, gateway, environment, created.ID, created.NetworkID, created.NodeID)
if err != nil {
_ = finish("failed", "runtime_persistence_failed", environment)
return err
}
if err := finish("succeeded", "environment_started", environment); err != nil {
return err
}
return nil
}
type runtimeCreateSpec struct {
browserPath string
networkExit gatewayNetworkExit
}
func prepareRuntimeCreate(_ context.Context, _ HubStore, _ func(hub.NetworkExitAccess) (string, error),
environment hub.EnvironmentContext, access hub.NetworkExitAccess) (runtimeCreateSpec, error) {
networkExit := gatewayNetworkExit{}
if environment.Exit.ID != "" {
networkExit = gatewayNetworkExitFor(access)
}
return runtimeCreateSpec{networkExit: networkExit}, nil
}
func createGatewayRuntime(ctx context.Context, target hub.Gateway, environment hub.EnvironmentContext, spec runtimeCreateSpec) (runtimeStatus, error) {
status, body, callErr := gatewayCall(ctx, target, http.MethodPost, "/v1/browsers",
gatewayCreatePayloadForAccount(environment, spec.browserPath, spec.networkExit), gatewayLongTimeout)
unknown := runtimeStatus{NetworkID: gatewayNetworkID(body)}
if callErr == nil && status == http.StatusCreated {
var created runtimeStatus
if json.Unmarshal(body, &created) == nil && validCreatedRuntime(created, environment, accountRunnable(environment)) {
created.Alias, created.BindingVersion = environment.Alias, environment.BindingVersion
if accountRunnable(environment) {
created.State, created.NetworkExitID, created.ProxyReady = "running", environment.Exit.ID, true
} else {
created.State = "exited"
}
return created, nil
}
}
if callErr != nil {
return unknown, gatewayUnreachable(callErr)
}
if status != http.StatusCreated {
return unknown, gatewayRejected(status, body)
}
return unknown, errors.New("gateway returned an invalid runtime generation")
}
func createStoppedGatewayRuntime(ctx context.Context, target hub.Gateway, environment hub.EnvironmentContext, browserPath string) error {
payload := gatewayCreatePayload(environment, browserPath, gatewayNetworkExit{})
payload["network_exit_id"], payload["stopped"] = "", true
status, body, callErr := gatewayCall(ctx, target, http.MethodPost, "/v1/browsers", payload, gatewayLongTimeout)
if callErr == nil && status == http.StatusCreated {
var created runtimeStatus
if json.Unmarshal(body, &created) == nil && validCreatedRuntime(created, environment, false) {
return nil
}
return errors.New("gateway returned an invalid stopped runtime generation")
}
if callErr != nil {
return gatewayUnreachable(callErr)
}
if status != http.StatusCreated {
return gatewayRejected(status, body)
}
return errors.New("gateway did not preserve the stopped runtime")
}
func removeGatewayRuntime(ctx context.Context, store runtimeCleanupStore, target hub.Gateway, environment hub.EnvironmentContext) (bool, error) {
removed := environment.RuntimeCleanupPending
if environment.RuntimeCleanupPending {
if environment.RuntimeCleanupRuntimeID == missingRuntimeID && environment.RuntimeCleanupNetworkID == "" {
// 未知代 pending:创建结果未知时登记的 runtime-not-found 没有实物 ID 可供清理。
// 按别名反查网关:无容器则该代无实物,直接解除;有同版本容器则采用其实物代继续清理;
// 版本不符说明别名下已是别的代,维持人工对账。
if environment.RuntimeCleanupBindingVersion < 1 {
return false, hub.ErrReconcileRequired
}
discovered, found, discoverErr := reconcileGatewayRuntime(ctx, target, environment.Alias)
if discoverErr != nil {
return false, discoverErr
}
if !found {
return true, store.SetRuntimeCleanupPending(ctx, environment, false)
}
if discovered.BindingVersion != environment.RuntimeCleanupBindingVersion {
return false, hub.ErrReconcileRequired
}
if err := store.SetRuntimeCleanupPending(ctx, environment, false); err != nil {
return false, err
}
environment = runtimeCleanupGeneration(environment, environment.RuntimeCleanupBindingVersion, discovered.ID, discovered.NetworkID)
if err := store.SetRuntimeCleanupPending(ctx, environment, true); err != nil {
return false, err
}
} else if environment.RuntimeCleanupBindingVersion < 1 || environment.RuntimeCleanupRuntimeID == "" {
return false, hub.ErrReconcileRequired
}
}
if !environment.RuntimeCleanupPending {
runtime, found, err := reconcileGatewayRuntime(ctx, target, environment.Alias)
if err != nil {
return false, err
}
bindingVersion, runtimeID, networkID := environment.BindingVersion, environment.RuntimeID, environment.RuntimeNetworkID
if environment.RuntimeCleanupBindingVersion > 0 {
bindingVersion, runtimeID, networkID = environment.RuntimeCleanupBindingVersion,
environment.RuntimeCleanupRuntimeID, environment.RuntimeCleanupNetworkID
}
if found {
if runtime.BindingVersion != bindingVersion || runtimeID == missingRuntimeID ||
(runtimeID != "" && runtimeID != runtime.ID) || (runtimeID == "" && networkID == "" && runtime.State == "running") {
return false, hub.ErrConflict
}
runtimeID = runtime.ID
if networkID != "" {
if runtime.NetworkID != "" && runtime.NetworkID != networkID {
return false, hub.ErrConflict
}
} else {
networkID = runtime.NetworkID
}
} else if environment.RuntimeID == "" {
runtimeID = missingRuntimeID
}
environment = runtimeCleanupGeneration(environment, bindingVersion, runtimeID, networkID)
}
if !environment.RuntimeCleanupPending {
if err := store.SetRuntimeCleanupPending(ctx, environment, true); err != nil {
return false, err
}
}
for attempt := 0; attempt < 2; attempt++ {
status, body, callErr := gatewayCall(ctx, target, http.MethodDelete, "/v1/browsers/"+environment.Alias,
gatewayCleanupGenerationPayload(environment), 30*time.Second)
if callErr == nil && (status == http.StatusNoContent || status == http.StatusNotFound) {
return true, store.SetRuntimeCleanupPending(ctx, environment, false)
}
if callErr == nil && status == http.StatusAccepted {
removed = true
continue
}
_, found, reconcileErr := reconcileGatewayRuntime(ctx, target, environment.Alias)
if reconcileErr != nil {
if callErr != nil {
return removed, errors.Join(gatewayUnreachable(callErr), gatewayUnreachable(reconcileErr))
}
return removed, errors.Join(gatewayRejected(status, body), gatewayUnreachable(reconcileErr))
}
if found {
if callErr != nil {
return removed, gatewayUnreachable(callErr)
}
return removed, gatewayRejected(status, body)
}
removed = true
if callErr != nil {
return true, gatewayUnreachable(callErr)
}
return true, gatewayRejected(status, body)
}
if removed {
return true, gatewayFailure{status: http.StatusBadGateway, message: "gateway removed the runtime but network cleanup is pending"}
}
return false, gatewayFailure{status: http.StatusBadGateway, message: "gateway runtime cleanup is pending"}
}
func restoreRebindRuntime(ctx context.Context, store HubStore, resolve func(hub.NetworkExitAccess) (string, error),
target hub.Gateway, environment hub.EnvironmentContext, previous runtimeStatus, found bool, prepared *runtimeCreateSpec) (bool, error) {
if !found {
return true, releaseRuntime(ctx, store, environment)
}
if environment.Exit.ID == "" {
if err := releaseRuntime(ctx, store, environment); err != nil {
return false, err
}
if previous.State == "running" {
return false, nil
}
if err := createStoppedGatewayRuntime(ctx, target, environment, ""); err != nil {
_, cleanupErr := removeGatewayRuntime(ctx, store, target, environment)
return false, errors.Join(err, cleanupErr)
}
return true, nil
}
spec := runtimeCreateSpec{}
if prepared != nil {
spec = *prepared
} else {
access, err := store.GetNetworkExitAccess(ctx, environment.Exit.ID)
if err != nil {
return false, err
}
spec, err = prepareRuntimeCreate(ctx, store, resolve, environment, access)
if err != nil {
return false, err
}
}
if err := releaseRuntime(ctx, store, environment); err != nil {
return false, err
}
cleanupContext, err := store.GetEnvironmentContext(ctx, environment.Alias)
if err != nil {
return false, err
}
created, err := createGatewayRuntime(ctx, target, environment, spec)
if err != nil {
cleanupErr := reconcileGatewayCreateGeneration(ctx, store, target, cleanupContext, environment.BindingVersion, created.NetworkID, false)
return false, errors.Join(err, cleanupErr)
}
if previous.State == "running" && accountRunnable(environment) {
_, err = activateGatewayRuntime(ctx, store, target, environment, created.ID, created.NetworkID)
} else if accountRunnable(environment) {
stopped := environment
stopped.RuntimeID, stopped.RuntimeNetworkID = created.ID, created.NetworkID
status, body, callErr := gatewayCall(ctx, target, http.MethodPost, "/v1/browsers/"+environment.Alias+"/stop",
gatewayGenerationPayload(stopped), 30*time.Second)
if callErr != nil {
err = gatewayUnreachable(callErr)
} else if status != http.StatusNoContent {
err = gatewayRejected(status, body)
}
}
if err != nil {
_, cleanupErr := removeGatewayRuntime(ctx, store, target, environment)
return false, errors.Join(err, cleanupErr)
}
return true, nil
}
func rebindRecoveryError(ready bool, err error) error {
if err != nil {
return err
}
if !ready {
return gatewayFailure{status: http.StatusBadGateway, message: "rebind recovery is incomplete; retry reconciliation"}
}
return nil
}
func runtimeCreateSpecMatches(_ context.Context, _ HubStore, current, previous hub.EnvironmentContext, prepared *runtimeCreateSpec) bool {
return prepared != nil && current.BindingVersion == previous.BindingVersion && current.Exit.ID == previous.Exit.ID
}
func rebindBrowser(store HubStore, probe NetworkExitProbe, resolve func(hub.NetworkExitAccess) (string, error), c fiber.Ctx) error {
var input struct {
NetworkExitID string `json:"network_exit_id"`
}
if err := decodeHubJSON(c, &input); err != nil {
return hubError(c, err)
}
before, err := store.GetEnvironmentContext(c.Context(), c.Params("alias"))
if err != nil {
return hubError(c, err)
}
action := actionForEnvironment("rebind", before)
if err := store.AppendEnvironmentAction(c.Context(), "environment_action_requested", action); err != nil {
return hubError(c, err)
}
finish := func(outcome, reason string, current hub.EnvironmentContext) error {
action.Outcome, action.ReasonCode = outcome, reason
action.NetworkExitID, action.BindingVersion, action.RuntimeInstanceID = current.Exit.ID, current.BindingVersion, current.RuntimeInstanceID
return store.AppendEnvironmentAction(c.Context(), "environment_action_finished", action)
}
if !hub.ValidNetworkExitID(input.NetworkExitID) {
if err := finish("failed", "rebind_input_rejected", before); err != nil {
return hubError(c, err)
}
return hubError(c, hub.ErrInvalid)
}
access, reason, err := verifyNetworkExit(c.Context(), store, probe, input.NetworkExitID)
if err != nil {
_ = finish("failed", reason, before)
return hubError(c, err)
}
target, err := store.GetGateway(c.Context(), before.Gateway)
if err != nil {
_ = finish("failed", "gateway_unavailable", before)
return hubError(c, err)
}
if before.RuntimeCleanupPending {
if _, cleanupErr := removeGatewayRuntime(c.Context(), store, target, before); cleanupErr != nil {
_ = finish("unknown", "cleanup_result_unknown", before)
return hubError(c, cleanupErr)
}
before, err = store.GetEnvironmentContext(c.Context(), before.Alias)
if err != nil {
return hubError(c, err)
}
}
runtime, found, err := reconcileGatewayRuntime(c.Context(), target, before.Alias)
if err != nil {
_ = finish("unknown", "gateway_result_unknown", before)
return hubError(c, gatewayUnreachable(err))
}
wasRunning := found && runtime.State == "running"
if err := store.ValidateEnvironmentRebind(c.Context(), before.Alias, input.NetworkExitID, before.BindingVersion); err != nil {
_ = finish("failed", "rebind_not_allowed", before)
return hubError(c, err)
}
var previousSpec *runtimeCreateSpec
if found && before.Exit.ID == "" {
previousSpec = &runtimeCreateSpec{}
} else if found {
previousAccess, accessErr := store.GetNetworkExitAccess(c.Context(), before.Exit.ID)
if accessErr != nil {
_ = finish("failed", "exit_unavailable", before)
return hubError(c, accessErr)
}
prepared, prepareErr := prepareRuntimeCreate(c.Context(), store, resolve, before, previousAccess)
if prepareErr != nil {
_ = finish("failed", "runtime_prepare_failed", before)
return hubError(c, prepareErr)
}
previousSpec = &prepared
}
candidate := before
candidate.Exit, candidate.BindingVersion = access.NetworkExit, before.BindingVersion+1
candidate.RuntimeInstanceID, candidate.RuntimeID = "", ""
var nextSpec runtimeCreateSpec
if wasRunning {
nextSpec, err = prepareRuntimeCreate(c.Context(), store, resolve, candidate, access)
if err != nil {
_ = finish("failed", "runtime_prepare_failed", before)
return hubError(c, err)
}
}
if found {
_, removeErr := removeGatewayRuntime(c.Context(), store, target, before)
if removeErr != nil {
_ = finish("unknown", "gateway_result_unknown", before)
return hubError(c, removeErr)
}
}
var candidateRuntime runtimeStatus
if wasRunning {
candidateRuntime, err = createGatewayRuntime(c.Context(), target, candidate, nextSpec)
if err != nil {
cleanupContext, contextErr := store.GetEnvironmentContext(c.Context(), before.Alias)
if contextErr != nil {
_ = finish("unknown", "gateway_result_unknown", before)
return hubError(c, errors.Join(err, contextErr))
}
cleanupErr := reconcileGatewayCreateGeneration(c.Context(), store, target, cleanupContext, candidate.BindingVersion, candidateRuntime.NetworkID, false)
if cleanupErr != nil {
_ = finish("unknown", "gateway_result_unknown", before)
return hubError(c, errors.Join(err, cleanupErr))
}
current, contextErr := store.GetEnvironmentContext(c.Context(), before.Alias)
ready := false
if contextErr == nil {
ready, contextErr = restoreRebindRuntime(c.Context(), store, resolve, target, current, runtime, found, previousSpec)
}
contextErr = rebindRecoveryError(ready, contextErr)
if contextErr != nil {
_ = finish("unknown", "rebind_recovery_failed", before)
return hubError(c, errors.Join(err, contextErr))
}
_ = finish("failed", "gateway_create_failed", before)
return hubError(c, err)
}
}
runtimeID := ""
if accountRunnable(candidate) {
runtimeID = candidateRuntime.ID
}
after, err := store.RebindEnvironment(c.Context(), before.Alias, input.NetworkExitID, runtimeID, before.BindingVersion, candidateRuntime.NetworkID)
if err != nil {
if candidateRuntime.ID != "" {
cleanup, contextErr := store.GetEnvironmentContext(c.Context(), before.Alias)
if contextErr != nil || cleanup.BindingVersion != before.BindingVersion || cleanup.Exit.ID != before.Exit.ID {
_ = finish("unknown", "rebind_recovery_failed", before)
return hubError(c, errors.Join(err, contextErr, hub.ErrConflict))
}
cleanup.RuntimeID, cleanup.RuntimeNetworkID = candidateRuntime.ID, candidateRuntime.NetworkID
_, cleanupErr := removeGatewayRuntime(c.Context(), store, target, cleanup)
if cleanupErr != nil {
_ = finish("unknown", "rebind_recovery_failed", before)
return hubError(c, errors.Join(err, cleanupErr))
}
}
if found {
current, contextErr := store.GetEnvironmentContext(c.Context(), before.Alias)
ready := false
if contextErr == nil {
prepared := previousSpec
if !runtimeCreateSpecMatches(c.Context(), store, current, before, previousSpec) {
prepared = nil
}
ready, contextErr = restoreRebindRuntime(c.Context(), store, resolve, target, current, runtime, true, prepared)
}
contextErr = rebindRecoveryError(ready, contextErr)
if contextErr != nil {
_ = finish("unknown", "rebind_recovery_failed", before)
return hubError(c, errors.Join(err, contextErr))
}
}
_ = finish("failed", "rebind_not_allowed", before)
return hubError(c, err)
}
if err := finish("succeeded", "environment_rebound", after); err != nil {
return hubError(c, err)
}
return c.JSON(after)
}
func deleteBrowser(store HubStore) fiber.Handler {
return func(c fiber.Ctx) error {
environment, err := store.GetEnvironmentContext(c.Context(), c.Params("alias"))
if err != nil {
return hubError(c, err)
}
action := actionForEnvironment("recycle", environment)
if err := store.AppendEnvironmentAction(c.Context(), "environment_action_requested", action); err != nil {
return hubError(c, err)
}
gateway, err := store.GetGateway(c.Context(), environment.Gateway)
if err != nil {
action.Outcome, action.ReasonCode = "failed", "gateway_unavailable"
_ = store.AppendEnvironmentAction(c.Context(), "environment_action_finished", action)
return hubError(c, err)
}
if _, removeErr := removeGatewayRuntime(c.Context(), store, gateway, environment); removeErr != nil {
action.Outcome, action.ReasonCode = "unknown", "cleanup_result_unknown"
_ = store.AppendEnvironmentAction(c.Context(), "environment_action_finished", action)
return hubError(c, removeErr)
}
action.Outcome, action.ReasonCode = "succeeded", "environment_recycled"
if err := store.AppendEnvironmentAction(c.Context(), "environment_action_finished", action); err != nil {
return hubError(c, err)
}
return c.SendStatus(fiber.StatusNoContent)
}
}
func actionForEnvironment(actionName string, environment hub.EnvironmentContext) hub.EnvironmentAction {
return hub.EnvironmentAction{
OperationID: hub.NewOperationID(), Action: actionName, ReasonCode: "action_requested",
AccountID: environment.AccountID, BrowserEnvAlias: environment.Alias, NetworkExitID: environment.Exit.ID,
RuntimeInstanceID: environment.RuntimeInstanceID, BindingVersion: environment.BindingVersion,
}
}
func decodeHubJSON(c fiber.Ctx, destination any) error {
decoder := json.NewDecoder(bytes.NewReader(c.Body()))
decoder.DisallowUnknownFields()
if err := decoder.Decode(destination); err != nil {
return hub.ErrInvalid
}
if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
return hub.ErrInvalid
}
return nil
}
func hubError(c fiber.Ctx, err error) error {
var failure gatewayFailure
if errors.As(err, &failure) {
return c.Status(failure.status).JSON(map[string]string{"error": failure.message})
}
status := fiber.StatusInternalServerError
message := "hub operation failed"
switch {
case errors.Is(err, hub.ErrInvalid):
status, message = fiber.StatusBadRequest, hub.ErrInvalid.Error()
case errors.Is(err, hub.ErrReconcileRequired):
status, message = fiber.StatusConflict, hub.ErrReconcileRequired.Error()
case errors.Is(err, hub.ErrConflict):
status, message = fiber.StatusConflict, hub.ErrConflict.Error()
case errors.Is(err, hub.ErrNotFound):
status, message = fiber.StatusNotFound, hub.ErrNotFound.Error()
default:
return c.Status(status).JSON(map[string]string{"error": err.Error()})
}
return c.Status(status).JSON(map[string]string{"error": message})
}
// RegisterEnvironments exposes environment routes to the grouped API package.
func RegisterEnvironments(app *fiber.App, store *hub.Store) {
registerHub(app, store)
}
// ReconcileRuntimeLeases performs one native runtime lease reconciliation pass.
func ReconcileRuntimeLeases(ctx context.Context, store *hub.Store) error {
return reconcileRuntimeLeases(ctx, store, defaultNetworkExitProbe(), nil)
}
// RegisterHubWithNetwork exposes environment route registration with explicit network dependencies for tests and composition.
func RegisterHubWithNetwork(app *fiber.App, store HubStore, probe NetworkExitProbe, resolve func(hub.NetworkExitAccess) (string, error)) {
registerHubWithNetwork(app, store, probe, resolve)
}
// hubStore remains the package-local name used by lifecycle test doubles; HubStore is its composition boundary.
type hubStore = HubStore