diff --git a/internal/environment/fingerprint_update_test.go b/internal/environment/fingerprint_update_test.go index 85ee74b..6183473 100644 --- a/internal/environment/fingerprint_update_test.go +++ b/internal/environment/fingerprint_update_test.go @@ -19,7 +19,7 @@ func seedFingerprintAccount(t *testing.T, ctx context.Context, accountID string) databaseURL = isolatedDatabaseURL(t, databaseURL) store := openFullyMigratedHub(t, ctx, databaseURL) t.Cleanup(func() { _ = store.Close() }) - if _, err := store.CreateGateway(ctx, "gw-fp", "http://127.0.0.1:8081", ""); err != nil { + if _, err := store.CreateGateway(ctx, "gw-fp"); err != nil { t.Fatal(err) } if _, err := store.db.ExecContext(ctx, ` diff --git a/internal/environment/gateway_task.go b/internal/environment/gateway_task.go new file mode 100644 index 0000000..452107a --- /dev/null +++ b/internal/environment/gateway_task.go @@ -0,0 +1,202 @@ +package environment + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "time" +) + +// GatewayTaskStatus 是任务生命周期；dispatched 后连接中断标记 failed，不自动重发。 +const ( + TaskPending = "pending" + TaskDispatched = "dispatched" + TaskSucceeded = "succeeded" + TaskFailed = "failed" +) + +// GatewayTask 是一次平台→网关通信的持久化记录，全量任务即执行记录。 +type GatewayTask struct { + ID int64 `json:"id"` + GatewayID int64 `json:"gateway_id"` + Method string `json:"method"` + Path string `json:"path"` + Payload json.RawMessage `json:"payload"` + Status string `json:"status"` + ResponseStatus int `json:"response_status"` + Error string `json:"error"` + ResponseBytes int64 `json:"response_bytes"` + TimeoutSeconds int `json:"timeout_seconds"` + CreatedAt time.Time `json:"created_at"` + DispatchedAt *time.Time `json:"dispatched_at"` + FinishedAt *time.Time `json:"finished_at"` +} + +func validGatewayMethod(method string) bool { + switch method { + case "GET", "POST", "PUT", "DELETE": + return true + } + return false +} + +const gatewayTaskColumns = `id, gateway_id, method, path, payload, status, response_status, error, response_bytes, timeout_seconds, created_at, dispatched_at, finished_at` + +func scanGatewayTask(row rowScanner) (GatewayTask, error) { + var task GatewayTask + var dispatched, finished sql.NullTime + if err := row.Scan(&task.ID, &task.GatewayID, &task.Method, &task.Path, &task.Payload, &task.Status, + &task.ResponseStatus, &task.Error, &task.ResponseBytes, &task.TimeoutSeconds, + &task.CreatedAt, &dispatched, &finished); err != nil { + return GatewayTask{}, rowError(err) + } + if dispatched.Valid { + task.DispatchedAt = &dispatched.Time + } + if finished.Valid { + task.FinishedAt = &finished.Time + } + return task, nil +} + +// CreateGatewayTask 入队一个任务；payload 必须是合法 JSON。 +func (s *Store) CreateGatewayTask(ctx context.Context, gatewayID int64, method, path string, payload []byte, timeout time.Duration) (GatewayTask, error) { + if gatewayID <= 0 || !validGatewayMethod(method) || path == "" || len(path) > 512 || !json.Valid(payload) { + return GatewayTask{}, ErrInvalid + } + seconds := int(timeout / time.Second) + if seconds < 1 || seconds > 600 { + return GatewayTask{}, ErrInvalid + } + if len(payload) == 0 { + payload = []byte(`{}`) + } + return scanGatewayTask(s.db.QueryRowContext(ctx, ` + INSERT INTO gateway_task (gateway_id, method, path, payload, timeout_seconds) + VALUES ($1, $2, $3, $4, $5) + RETURNING `+gatewayTaskColumns, gatewayID, method, path, payload, seconds)) +} + +// DispatchGatewayTask 抢占一个待下发任务；仅当仍为 pending 时成功，保证单次下发。 +func (s *Store) DispatchGatewayTask(ctx context.Context, taskID int64) (bool, error) { + result, err := s.db.ExecContext(ctx, ` + UPDATE gateway_task SET status = 'dispatched', dispatched_at = now() + WHERE id = $1 AND status = 'pending'`, taskID) + if err != nil { + return false, err + } + count, err := result.RowsAffected() + if err != nil { + return false, err + } + return count == 1, nil +} + +// FinishGatewayTask 记录任务终态；dispatched 中任务由结果更新，pending 任务允许直接终态（网关离线），重复投递幂等忽略。 +func (s *Store) FinishGatewayTask(ctx context.Context, taskID int64, status string, responseStatus int, taskError string, responseBytes int64) (bool, error) { + if status != TaskSucceeded && status != TaskFailed { + return false, ErrInvalid + } + if len(taskError) > 2048 { + taskError = taskError[:2048] + } + result, err := s.db.ExecContext(ctx, ` + UPDATE gateway_task SET status = $2, response_status = $3, error = $4, response_bytes = $5, finished_at = now() + WHERE id = $1 AND status IN ('dispatched', 'pending')`, taskID, status, responseStatus, taskError, responseBytes) + if err != nil { + return false, err + } + count, err := result.RowsAffected() + if err != nil { + return false, err + } + return count == 1, nil +} + +// InterruptDispatchedTasks 连接断开时把该网关在途任务标记失败（结果未确认，不自动重发）。 +func (s *Store) InterruptDispatchedTasks(ctx context.Context, gatewayID int64, reason string) error { + if len(reason) > 2048 { + reason = reason[:2048] + } + _, err := s.db.ExecContext(ctx, ` + UPDATE gateway_task SET status = 'failed', error = $2, finished_at = now() + WHERE gateway_id = $1 AND status = 'dispatched'`, gatewayID, reason) + return err +} + +// ResetGatewayConnections 平台启动时清空内存会话遗留的在线标记。 +func (s *Store) ResetGatewayConnections(ctx context.Context) error { + _, err := s.db.ExecContext(ctx, `UPDATE gateway SET online = false, version = '' WHERE online`) + return err +} + +// NextPendingGatewayTasks 读取某网关按 id 序的待下发任务（补发队列）。 +func (s *Store) NextPendingGatewayTasks(ctx context.Context, gatewayID int64, limit int) ([]GatewayTask, error) { + if gatewayID <= 0 || limit <= 0 { + return nil, ErrInvalid + } + rows, err := s.db.QueryContext(ctx, ` + SELECT `+gatewayTaskColumns+` FROM gateway_task + WHERE gateway_id = $1 AND status = 'pending' ORDER BY id LIMIT $2`, gatewayID, limit) + if err != nil { + return nil, err + } + defer rows.Close() + tasks := []GatewayTask{} + for rows.Next() { + task, err := scanGatewayTask(rows) + if err != nil { + return nil, err + } + tasks = append(tasks, task) + } + return tasks, rows.Err() +} + +type GatewayTaskPage struct { + Data []GatewayTask `json:"data"` + Page int `json:"page"` + PageSize int `json:"page_size"` + Total int `json:"total"` +} + +// ListGatewayTasksPage 分页读取网关执行记录，最新创建在前。 +func (s *Store) ListGatewayTasksPage(ctx context.Context, gatewayID int64, page, pageSize int) (GatewayTaskPage, error) { + if gatewayID <= 0 { + return GatewayTaskPage{}, ErrInvalid + } + if page < 1 || pageSize < 1 || pageSize > 100 { + return GatewayTaskPage{}, ErrInvalid + } + result := GatewayTaskPage{Data: []GatewayTask{}, Page: page, PageSize: pageSize} + if err := s.db.QueryRowContext(ctx, `SELECT count(*) FROM gateway_task WHERE gateway_id = $1`, gatewayID).Scan(&result.Total); err != nil { + return result, err + } + rows, err := s.db.QueryContext(ctx, ` + SELECT `+gatewayTaskColumns+` FROM gateway_task WHERE gateway_id = $1 + ORDER BY id DESC LIMIT $2 OFFSET $3`, gatewayID, pageSize, (page-1)*pageSize) + if err != nil { + return result, err + } + defer rows.Close() + for rows.Next() { + task, err := scanGatewayTask(rows) + if err != nil { + return result, err + } + result.Data = append(result.Data, task) + } + return result, rows.Err() +} + +// CountPendingGatewayTasks 返回各网关待执行任务数（队列状态展示）。 +func (s *Store) CountPendingGatewayTasks(ctx context.Context, gatewayID int64) (int, error) { + var count int + if err := s.db.QueryRowContext(ctx, `SELECT count(*) FROM gateway_task WHERE gateway_id = $1 AND status = 'pending'`, gatewayID).Scan(&count); err != nil { + return 0, err + } + return count, nil +} + +var ErrGatewayOffline = errors.New("网关离线，无法执行任务") diff --git a/internal/environment/memory_schema_upgrade_test.go b/internal/environment/memory_schema_upgrade_test.go index daf11a6..a1bffd5 100644 --- a/internal/environment/memory_schema_upgrade_test.go +++ b/internal/environment/memory_schema_upgrade_test.go @@ -14,7 +14,7 @@ func TestEnvironmentMemorySchemaUpgradeAndRepeatedStartup(t *testing.T) { ctx := context.Background() url := isolatedDatabaseURL(t, databaseURL) s := openFullyMigratedHub(t, ctx, url) - if _, err := s.CreateGateway(ctx, "memory-upgrade", "http://127.0.0.1:8090", ""); err != nil { + if _, err := s.CreateGateway(ctx, "memory-upgrade"); err != nil { t.Fatal(err) } before, err := s.CreateStandaloneEnv(ctx, "memory-upgrade", Fingerprint{Timezone: "Asia/Shanghai"}, 4096) diff --git a/internal/environment/migrations/1058_gateway_outbound_channel.sql b/internal/environment/migrations/1058_gateway_outbound_channel.sql new file mode 100644 index 0000000..18d7b1a --- /dev/null +++ b/internal/environment/migrations/1058_gateway_outbound_channel.sql @@ -0,0 +1,35 @@ +-- 网关通信方向反转为出站连接：网关配置平台地址与 access_key 后主动连接平台， +-- 全部业务通信复用该连接；旧 endpoint 直连与定时探活字段废弃。 +-- 既有网关需要重新配置：旧 endpoint/token 不再保留兼容路径。 +ALTER TABLE gateway + DROP COLUMN endpoint, + DROP COLUMN health_status, + DROP COLUMN last_check_reason, + DROP COLUMN last_checked_at, + ADD COLUMN access_key text NOT NULL, + ADD COLUMN online boolean NOT NULL DEFAULT false, + ADD COLUMN last_seen_at timestamptz, + ADD COLUMN version text NOT NULL DEFAULT ''; + +-- 既有网关记录补一个一次性访问密钥（随机 48 位十六进制），新网关由平台生成。 +UPDATE gateway SET access_key = encode(sha256(('creator-hub-gateway-' || id::text || random()::text)::bytea), 'hex') +WHERE access_key = ''; + +-- 统一任务模型：所有平台→网关通信都是持久化任务；全量任务即执行记录。 +CREATE TABLE gateway_task ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + gateway_id bigint NOT NULL REFERENCES gateway(id) ON DELETE CASCADE, + method text NOT NULL CHECK (method IN ('GET','POST','PUT','DELETE')), + path text NOT NULL CHECK (length(path) <= 512), + payload jsonb NOT NULL DEFAULT '{}'::jsonb, + status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','dispatched','succeeded','failed')), + response_status int, + error text NOT NULL DEFAULT '' CHECK (length(error) <= 2048), + response_bytes bigint NOT NULL DEFAULT 0, + timeout_seconds int NOT NULL DEFAULT 30 CHECK (timeout_seconds BETWEEN 1 AND 600), + created_at timestamptz NOT NULL DEFAULT now(), + dispatched_at timestamptz, + finished_at timestamptz +); +CREATE INDEX gateway_task_gateway_dispatch_idx ON gateway_task (gateway_id, status, id); +CREATE INDEX gateway_task_finished_idx ON gateway_task (finished_at); diff --git a/internal/environment/standalone_order_test.go b/internal/environment/standalone_order_test.go index 89ff5e3..ce7d84a 100644 --- a/internal/environment/standalone_order_test.go +++ b/internal/environment/standalone_order_test.go @@ -15,7 +15,7 @@ func TestPendingEnvironmentDatabaseIDOrder(t *testing.T) { ctx := context.Background() store := openFullyMigratedHub(t, ctx, isolatedDatabaseURL(t, databaseURL)) t.Cleanup(func() { _ = store.Close() }) - if _, err := store.CreateGateway(ctx, "order-gateway", "http://127.0.0.1:8090", "order-gateway-test-token"); err != nil { + if _, err := store.CreateGateway(ctx, "order-gateway"); err != nil { t.Fatal(err) } first, err := store.CreateStandaloneEnv(ctx, "order-gateway", Fingerprint{}, 0) diff --git a/internal/environment/store.go b/internal/environment/store.go index ebd4a47..b5fe680 100644 --- a/internal/environment/store.go +++ b/internal/environment/store.go @@ -212,6 +212,9 @@ var migration1056 string //go:embed migrations/1057_openai_settings.sql var migration1057 string +//go:embed migrations/1058_gateway_outbound_channel.sql +var migration1058 string + var ( ErrConflict = errors.New("resource conflicts with existing state") ErrInvalid = errors.New("invalid environment input") @@ -232,17 +235,17 @@ type Store struct { lockAdmission chan struct{} } -// Gateway 是平台注册的 native browser gateway 节点；Token 由平台生成，明文存储供页面复制（开发阶段约定）。 -// 健康字段由控制面后台定时探活写入，只保留最近一次结果，不保留历史。 +// Gateway 是平台注册的 native browser gateway 节点；access_key 由平台生成，明文存储供页面复制。 +// 在线状态由出站 WS 会话维护：连接建立置 true、断开置 false，心跳超时在查询时动态判离。 type Gateway struct { - Name string `json:"name"` - Endpoint string `json:"endpoint"` - Token string `json:"token"` - HealthStatus string `json:"health_status"` - LastCheckReason string `json:"last_check_reason"` - LastCheckedAt *time.Time `json:"last_checked_at"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID int64 `json:"id"` + Name string `json:"name"` + AccessKey string `json:"access_key"` + Online bool `json:"online"` + LastSeenAt *time.Time `json:"last_seen_at"` + Version string `json:"version"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } // Env 是一个浏览器环境；浏览器安装和默认运行时由 gateway 宿主机配置。 @@ -354,7 +357,7 @@ func (s *Store) migrate(ctx context.Context) error { {1029, migration1029}, {1030, migration1030}, {1031, migration1031}, {1032, migration1032}, {1033, migration1033}, {1034, migration1034}, {1035, migration1035}, {1036, migration1036}, {1037, migration1037}, {1038, migration1038}, {1039, migration1039}, {1040, migration1040}, {1041, migration1041}, {1042, migration1042}, {1043, migration1043}, {1044, migration1044}, {1045, migration1045}, {1046, migration1046}, - {43, migration043}, {44, migration044}, {1047, migration1047}, {1048, migration1048}, {1049, migration1049}, {1050, migration1050}, {1051, migration1051}, {1052, migration1052}, {1053, migration1053}, {1054, migration1054}, {1055, migration1055}, {1056, migration1056}, {1057, migration1057}} { + {43, migration043}, {44, migration044}, {1047, migration1047}, {1048, migration1048}, {1049, migration1049}, {1050, migration1050}, {1051, migration1051}, {1052, migration1052}, {1053, migration1053}, {1054, migration1054}, {1055, migration1055}, {1056, migration1056}, {1057, migration1057}, {1058, migration1058}} { var applied bool if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = $1)`, migration.version).Scan(&applied); err != nil { return errors.New("read environment schema migration state") @@ -377,59 +380,61 @@ func (s *Store) migrate(ctx context.Context) error { func scanGateway(row rowScanner) (Gateway, error) { var gateway Gateway - var checked sql.NullTime - if err := row.Scan(&gateway.Name, &gateway.Endpoint, &gateway.Token, &gateway.HealthStatus, - &gateway.LastCheckReason, &checked, &gateway.CreatedAt, &gateway.UpdatedAt); err != nil { + var seen sql.NullTime + if err := row.Scan(&gateway.ID, &gateway.Name, &gateway.AccessKey, &gateway.Online, + &seen, &gateway.Version, &gateway.CreatedAt, &gateway.UpdatedAt); err != nil { return Gateway{}, rowError(err) } - if checked.Valid { - gateway.LastCheckedAt = &checked.Time + if seen.Valid { + gateway.LastSeenAt = &seen.Time + } + // 心跳超时即离线：连接进程异常退出时 online 可能残留，查询以最近上报时间为准。 + if gateway.Online && (gateway.LastSeenAt == nil || time.Since(*gateway.LastSeenAt) > 45*time.Second) { + gateway.Online = false } return gateway, nil } -func (s *Store) CreateGateway(ctx context.Context, name, endpoint, token string) (Gateway, error) { - name, endpoint, token = strings.TrimSpace(name), strings.TrimSpace(endpoint), strings.TrimSpace(token) - if !gatewayNamePattern.MatchString(name) || !validHTTPURL(endpoint) { - return Gateway{}, ErrInvalid - } - // 未显式提供 token 时由平台生成；提供则与网关进程的 GATEWAY_TOKEN 保持一致。 - if token != "" { - if !tokenPattern.MatchString(token) { - return Gateway{}, ErrInvalid - } - } else { - token = newToken() - } - result, err := scanGateway(s.db.QueryRowContext(ctx, ` - INSERT INTO gateway (name, endpoint, token) VALUES ($1, $2, $3) - RETURNING name, endpoint, token, health_status, last_check_reason, last_checked_at, created_at, updated_at`, name, endpoint, token)) - if err != nil { - return Gateway{}, err - } - return result, nil -} +const gatewayColumns = `id, name, access_key, online, last_seen_at, version, created_at, updated_at` -// UpdateGateway 修改网关名称、Endpoint 和令牌。名称变更由数据库外键 ON UPDATE CASCADE -// 原子同步 browser_env 引用；空令牌表示保留当前令牌，避免只改地址时意外轮换凭证。 -func (s *Store) UpdateGateway(ctx context.Context, currentName, name, endpoint, token string) (Gateway, error) { - currentName, name = strings.TrimSpace(currentName), strings.TrimSpace(name) - endpoint, token = strings.TrimSpace(endpoint), strings.TrimSpace(token) - if !gatewayNamePattern.MatchString(currentName) || !gatewayNamePattern.MatchString(name) || !validHTTPURL(endpoint) || - (token != "" && !tokenPattern.MatchString(token)) { +func (s *Store) CreateGateway(ctx context.Context, name string) (Gateway, error) { + name = strings.TrimSpace(name) + if !gatewayNamePattern.MatchString(name) { return Gateway{}, ErrInvalid } return scanGateway(s.db.QueryRowContext(ctx, ` - UPDATE gateway SET name = $1, endpoint = $2, - token = CASE WHEN $3 = '' THEN token ELSE $3 END, updated_at = now() - WHERE name = $4 - RETURNING name, endpoint, token, health_status, last_check_reason, last_checked_at, created_at, updated_at`, name, endpoint, token, currentName)) + INSERT INTO gateway (name, access_key) VALUES ($1, $2) + RETURNING `+gatewayColumns, name, newToken())) +} + +// UpdateGateway 修改网关名称。名称变更由数据库外键 ON UPDATE CASCADE +// 原子同步 browser_env 引用。 +func (s *Store) UpdateGateway(ctx context.Context, currentName, name string) (Gateway, error) { + currentName, name = strings.TrimSpace(currentName), strings.TrimSpace(name) + if !gatewayNamePattern.MatchString(currentName) || !gatewayNamePattern.MatchString(name) { + return Gateway{}, ErrInvalid + } + return scanGateway(s.db.QueryRowContext(ctx, ` + UPDATE gateway SET name = $1, updated_at = now() + WHERE name = $2 + RETURNING `+gatewayColumns, name, currentName)) +} + +// ResetGatewayKey 轮换访问密钥；旧 key 立即失效，网关必须重新配置。 +func (s *Store) ResetGatewayKey(ctx context.Context, name string) (Gateway, error) { + name = strings.TrimSpace(name) + if !gatewayNamePattern.MatchString(name) { + return Gateway{}, ErrInvalid + } + return scanGateway(s.db.QueryRowContext(ctx, ` + UPDATE gateway SET access_key = $2, updated_at = now() + WHERE name = $1 + RETURNING `+gatewayColumns, name, newToken())) } func (s *Store) ListGateways(ctx context.Context) ([]Gateway, error) { rows, err := s.db.QueryContext(ctx, ` - SELECT name, endpoint, token, health_status, last_check_reason, last_checked_at, created_at, updated_at - FROM gateway ORDER BY created_at, name`) + SELECT `+gatewayColumns+` FROM gateway ORDER BY created_at, name`) if err != nil { return nil, errors.New("read gateways") } @@ -450,19 +455,32 @@ func (s *Store) GetGateway(ctx context.Context, name string) (Gateway, error) { return Gateway{}, ErrInvalid } return scanGateway(s.db.QueryRowContext(ctx, ` - SELECT name, endpoint, token, health_status, last_check_reason, last_checked_at, created_at, updated_at - FROM gateway WHERE name = $1`, name)) + SELECT `+gatewayColumns+` FROM gateway WHERE name = $1`, name)) } -// RecordGatewayCheck 覆盖写入最近一次探活结果；不保留历史。status 只允许 healthy/unhealthy。 -func (s *Store) RecordGatewayCheck(ctx context.Context, name, status, reason string) (Gateway, error) { - if !gatewayNamePattern.MatchString(name) || (status != "healthy" && status != "unhealthy") { +// TouchGatewaySession 心跳节流落库：刷新 last_seen_at 与网关上报版本。 +func (s *Store) TouchGatewaySession(ctx context.Context, gatewayID int64, version string) error { + _, err := s.db.ExecContext(ctx, ` + UPDATE gateway SET last_seen_at = now(), + version = CASE WHEN $2 = '' THEN version ELSE $2 END, online = true + WHERE id = $1`, gatewayID, version) + return err +} + +// SetGatewayOnline 由会话生命周期维护在线标记；查询时另按 last_seen_at 兼动判定。 +func (s *Store) SetGatewayOnline(ctx context.Context, gatewayID int64, online bool) error { + _, err := s.db.ExecContext(ctx, `UPDATE gateway SET online = $2, updated_at = now() WHERE id = $1`, gatewayID, online) + return err +} + +// GetGatewayByAccessKey 用访问密钥解析接入会话的网关身份。 +func (s *Store) GetGatewayByAccessKey(ctx context.Context, accessKey string) (Gateway, error) { + authorization := strings.TrimPrefix(accessKey, "Bearer ") + if authorization == "" { return Gateway{}, ErrInvalid } return scanGateway(s.db.QueryRowContext(ctx, ` - UPDATE gateway SET health_status = $2, last_check_reason = $3, last_checked_at = now(), updated_at = now() - WHERE name = $1 - RETURNING name, endpoint, token, health_status, last_check_reason, last_checked_at, created_at, updated_at`, name, status, reason)) + SELECT `+gatewayColumns+` FROM gateway WHERE access_key = $1`, authorization)) } func (s *Store) DeleteGateway(ctx context.Context, name string) error { diff --git a/internal/environment/store_test.go b/internal/environment/store_test.go index 07e1030..6fc4023 100644 --- a/internal/environment/store_test.go +++ b/internal/environment/store_test.go @@ -230,28 +230,18 @@ func TestFingerprintValidateRejectsUnsupportedValues(t *testing.T) { func TestStoreValidationRejectsInvalidInputsBeforePersistence(t *testing.T) { store := &Store{} ctx := context.Background() - if _, err := store.CreateGateway(ctx, "bad name!", "http://gw:8081", ""); !errors.Is(err, ErrInvalid) { + if _, err := store.CreateGateway(ctx, "bad name!"); !errors.Is(err, ErrInvalid) { t.Fatalf("expected invalid gateway name, got %v", err) } - if _, err := store.CreateGateway(ctx, "gw-1", "ftp://gw:8081", ""); !errors.Is(err, ErrInvalid) { - t.Fatalf("expected invalid gateway endpoint, got %v", err) - } - if _, err := store.CreateGateway(ctx, "gw-1", "http://gw:8081", "short-token"); !errors.Is(err, ErrInvalid) { - t.Fatalf("expected invalid gateway token, got %v", err) - } for _, test := range []struct { name string currentName string newName string - endpoint string - token string }{ - {name: "current name", currentName: "bad name!", newName: "gw-2", endpoint: "http://gw:8081"}, - {name: "new name", currentName: "gw-1", newName: "bad name!", endpoint: "http://gw:8081"}, - {name: "endpoint", currentName: "gw-1", newName: "gw-2", endpoint: "ftp://gw:8081"}, - {name: "token", currentName: "gw-1", newName: "gw-2", endpoint: "http://gw:8081", token: "short-token"}, + {name: "current name", currentName: "bad name!", newName: "gw-2"}, + {name: "new name", currentName: "gw-1", newName: "bad name!"}, } { - if _, err := store.UpdateGateway(ctx, test.currentName, test.newName, test.endpoint, test.token); !errors.Is(err, ErrInvalid) { + if _, err := store.UpdateGateway(ctx, test.currentName, test.newName); !errors.Is(err, ErrInvalid) { t.Fatalf("expected invalid gateway update %s, got %v", test.name, err) } } @@ -290,7 +280,7 @@ func TestFingerprintSeedIsGloballyUnique(t *testing.T) { if _, err := store.db.ExecContext(ctx, `TRUNCATE browser_env, social_account, gateway CASCADE`); err != nil { t.Fatal(err) } - if _, err := store.CreateGateway(ctx, "gw-seed", "http://127.0.0.1:8081", "unit-test-gateway-token"); err != nil { + if _, err := store.CreateGateway(ctx, "gw-seed"); err != nil { t.Fatal(err) } if _, err := store.db.ExecContext(ctx, ` @@ -340,18 +330,18 @@ func TestHubWorkflow(t *testing.T) { t.Fatal(err) } - gateway, err := store.CreateGateway(ctx, "gw-main", "http://127.0.0.1:8081", "") + gateway, err := store.CreateGateway(ctx, "gw-main") if err != nil { t.Fatal(err) } - if len(gateway.Token) != 48 { - t.Fatalf("platform must assign a 48-hex-char token: %q", gateway.Token) + if len(gateway.AccessKey) != 48 { + t.Fatalf("platform must assign a 48-hex-char access key: %q", gateway.AccessKey) } - custom, err := store.CreateGateway(ctx, "gw-custom", "http://127.0.0.3:8081", "operator-provided-token-1234") - if err != nil || custom.Token != "operator-provided-token-1234" { - t.Fatalf("explicit token must be honored: %#v %v", custom, err) + custom, err := store.CreateGateway(ctx, "gw-custom") + if err != nil || custom.AccessKey == "" || custom.AccessKey == gateway.AccessKey { + t.Fatalf("each gateway must receive a distinct generated key: %#v %v", custom, err) } - if _, err := store.CreateGateway(ctx, "gw-main", "http://127.0.0.2:8081", ""); !errors.Is(err, ErrConflict) { + if _, err := store.CreateGateway(ctx, "gw-main"); !errors.Is(err, ErrConflict) { t.Fatalf("expected duplicate gateway conflict, got %v", err) } @@ -386,21 +376,21 @@ func TestHubWorkflow(t *testing.T) { if listed[0].Name != "店铺一号" || listed[0].Fingerprint.Seed != 1001 || listed[0].Fingerprint.Timezone != "Asia/Shanghai" { t.Fatalf("fingerprint must round trip through jsonb: %#v", listed[0]) } - updatedGateway, err := store.UpdateGateway(ctx, "gw-main", "gw-renamed", "http://127.0.0.4:8081", "") - if err != nil || updatedGateway.Name != "gw-renamed" || updatedGateway.Endpoint != "http://127.0.0.4:8081" || updatedGateway.Token != gateway.Token { - t.Fatalf("gateway update did not preserve the token: %#v err=%v", updatedGateway, err) + updatedGateway, err := store.UpdateGateway(ctx, "gw-main", "gw-renamed") + if err != nil || updatedGateway.Name != "gw-renamed" || updatedGateway.AccessKey != gateway.AccessKey { + t.Fatalf("gateway update did not preserve the access key: %#v err=%v", updatedGateway, err) } renamedEnv, err := store.GetEnv(ctx, "shop-01") if err != nil || renamedEnv.Gateway != "gw-renamed" { t.Fatalf("gateway rename did not cascade to environment: %#v err=%v", renamedEnv, err) } - if _, err := store.UpdateGateway(ctx, "gw-renamed", "gw-custom", "http://127.0.0.4:8081", ""); !errors.Is(err, ErrConflict) { + if _, err := store.UpdateGateway(ctx, "gw-renamed", "gw-custom"); !errors.Is(err, ErrConflict) { t.Fatalf("expected gateway rename conflict, got %v", err) } - if _, err := store.UpdateGateway(ctx, "missing", "gw-missing", "http://127.0.0.5:8081", ""); !errors.Is(err, ErrNotFound) { + if _, err := store.UpdateGateway(ctx, "missing", "gw-missing"); !errors.Is(err, ErrNotFound) { t.Fatalf("expected missing gateway on update, got %v", err) } - if _, err := store.UpdateGateway(ctx, "gw-renamed", "gw-main", "http://127.0.0.1:8081", ""); err != nil { + if _, err := store.UpdateGateway(ctx, "gw-renamed", "gw-main"); err != nil { t.Fatalf("restore gateway name after cascade check: %v", err) } if _, err := store.GetEnv(ctx, "ghost"); !errors.Is(err, ErrNotFound) { @@ -442,7 +432,7 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) { VALUES ('account-a', 'os_keyring', 'creatorhub/account-a', 'mock', 'account-a')`); err != nil { t.Fatal(err) } - if _, err := store.CreateGateway(ctx, "gw-main", "http://127.0.0.1:8081", "unit-test-gateway-token"); err != nil { + if _, err := store.CreateGateway(ctx, "gw-main"); err != nil { t.Fatal(err) } @@ -697,7 +687,7 @@ func TestRuntimeCleanupFenceAndBoundCreateReadiness(t *testing.T) { VALUES ('fence-a', 'os_keyring', 'creatorhub/fence-a', 'mock', 'fence-a')`); err != nil { t.Fatal(err) } - if _, err := store.CreateGateway(ctx, "gw-fence", "http://127.0.0.1:8081", "unit-test-gateway-token"); err != nil { + if _, err := store.CreateGateway(ctx, "gw-fence"); err != nil { t.Fatal(err) } exit, err := store.CreateNetworkExit(ctx, NetworkExit{Protocol: "http", Host: "proxy.example", Port: 8080}) @@ -810,7 +800,7 @@ func TestNetworkExitLifecycleStoreOperations(t *testing.T) { } } -func TestGatewayHealthCheckPersistence(t *testing.T) { +func TestGatewaySessionPersistence(t *testing.T) { databaseURL := os.Getenv("CREATORHUB_POSTGRES_TEST_URL") if databaseURL == "" { t.Skip("set CREATORHUB_POSTGRES_TEST_URL to run PostgreSQL integration coverage") @@ -819,52 +809,65 @@ func TestGatewayHealthCheckPersistence(t *testing.T) { store := openFullyMigratedHub(t, ctx, isolatedDatabaseURL(t, databaseURL)) t.Cleanup(func() { _ = store.Close() }) - gateway, err := store.CreateGateway(ctx, "gw-health", "http://127.0.0.1:8081", "") + gateway, err := store.CreateGateway(ctx, "gw-health") if err != nil { t.Fatal(err) } - if gateway.HealthStatus != "unchecked" || gateway.LastCheckReason != "" || gateway.LastCheckedAt != nil { - t.Fatalf("new gateway must start unchecked: %#v", gateway) + if gateway.Online || gateway.LastSeenAt != nil || gateway.Version != "" { + t.Fatalf("new gateway must start offline: %#v", gateway) + } + if len(gateway.AccessKey) < 16 { + t.Fatalf("new gateway must have a generated access key: %#v", gateway) } - checked, err := store.RecordGatewayCheck(ctx, "gw-health", "healthy", "") - if err != nil { + if err := store.TouchGatewaySession(ctx, gateway.ID, "2026.10.0"); err != nil { t.Fatal(err) } - if checked.HealthStatus != "healthy" || checked.LastCheckReason != "" || checked.LastCheckedAt == nil { - t.Fatalf("healthy check was not persisted: %#v", checked) - } - - failed, err := store.RecordGatewayCheck(ctx, "gw-health", "unhealthy", `Get "http://127.0.0.1:8081/healthz": connection refused`) - if err != nil { + if err := store.SetGatewayOnline(ctx, gateway.ID, true); err != nil { t.Fatal(err) } - if failed.HealthStatus != "unhealthy" || failed.LastCheckReason == "" || failed.LastCheckedAt == nil { - t.Fatalf("unhealthy check was not persisted: %#v", failed) - } - if !failed.LastCheckedAt.After(*checked.LastCheckedAt) && failed.LastCheckedAt.Equal(*checked.LastCheckedAt) { - t.Fatalf("last_checked_at must advance: %v", failed.LastCheckedAt) - } - reread, err := store.GetGateway(ctx, "gw-health") - if err != nil || reread.HealthStatus != "unhealthy" || reread.LastCheckReason != failed.LastCheckReason { - t.Fatalf("health fields must round trip through get/list: %#v err=%v", reread, err) + if err != nil || !reread.Online || reread.LastSeenAt == nil || reread.Version != "2026.10.0" { + t.Fatalf("heartbeat must persist online state: %#v err=%v", reread, err) + } + if reread.AccessKey != gateway.AccessKey { + t.Fatalf("heartbeat must not rotate access key: %#v", reread) + } + + byKey, err := store.GetGatewayByAccessKey(ctx, "Bearer "+gateway.AccessKey) + if err != nil || byKey.ID != gateway.ID { + t.Fatalf("access key lookup failed: %#v err=%v", byKey, err) + } + if _, err := store.GetGatewayByAccessKey(ctx, "Bearer wrong-key"); !errors.Is(err, ErrNotFound) { + t.Fatalf("invalid access key must be rejected, got %v", err) + } + + // 离线标记由会话关闭写入；查询层再按心跳超时兜底判定。 + if err := store.SetGatewayOnline(ctx, gateway.ID, false); err != nil { + t.Fatal(err) } listed, err := store.ListGateways(ctx) - if err != nil || len(listed) != 1 || listed[0].HealthStatus != "unhealthy" { - t.Fatalf("health fields must round trip through list: list=%#v err=%v", listed, err) + if err != nil || len(listed) != 1 || listed[0].Online { + t.Fatalf("offline state must round trip through list: list=%#v err=%v", listed, err) } - if _, err := store.RecordGatewayCheck(ctx, "gw-health", "degraded", ""); !errors.Is(err, ErrInvalid) { - t.Fatalf("unknown status must be rejected, got %v", err) + rotated, err := store.ResetGatewayKey(ctx, "gw-health") + if err != nil || rotated.AccessKey == "" || rotated.AccessKey == gateway.AccessKey { + t.Fatalf("access key rotation failed: %#v err=%v", rotated, err) } - if _, err := store.RecordGatewayCheck(ctx, "missing", "healthy", ""); !errors.Is(err, ErrNotFound) { - t.Fatalf("missing gateway must surface not found, got %v", err) + if _, err := store.GetGatewayByAccessKey(ctx, "Bearer "+gateway.AccessKey); !errors.Is(err, ErrNotFound) { + t.Fatalf("old access key must stop resolving after rotation, got %v", err) } - // 改名/改地址不触碰健康字段（由探活负责覆盖）。 - renamed, err := store.UpdateGateway(ctx, "gw-health", "gw-health-2", "http://127.0.0.2:8081", "") - if err != nil || renamed.HealthStatus != "unhealthy" || renamed.LastCheckReason == "" { - t.Fatalf("update must preserve health fields: %#v err=%v", renamed, err) + // 平台启动重置：清空进程崩溃遗留的 online 标记。 + if err := store.SetGatewayOnline(ctx, gateway.ID, true); err != nil { + t.Fatal(err) + } + if err := store.ResetGatewayConnections(ctx); err != nil { + t.Fatal(err) + } + final, err := store.GetGateway(ctx, "gw-health") + if err != nil || final.Online { + t.Fatalf("startup reset must clear online flags: %#v err=%v", final, err) } }