diff --git a/.env.example b/.env.example index cac55fe..f50a8c8 100644 --- a/.env.example +++ b/.env.example @@ -1,3 +1,5 @@ DOCKER_GID=989 CREATORHUB_PORT=8082 GATEWAY_TOKEN=dev-creatorhub-gateway-token +CONTROL_PLANE_USERNAME= +CONTROL_PLANE_PASSWORD= diff --git a/README.md b/README.md index be6944b..3324cc4 100644 --- a/README.md +++ b/README.md @@ -14,10 +14,14 @@ 需要 Docker Compose： ```bash +export CONTROL_PLANE_USERNAME=creatorhub +export CONTROL_PLANE_PASSWORD="$(openssl rand -hex 24)" DOCKER_GID=$(stat -c %g /var/run/docker.sock) docker compose up --build ``` -打开 ；局域网内用宿主机 IP 访问同一端口。首次使用：在「网关管理」用 Compose 里的 `GATEWAY_TOKEN` 注册 `http://docker-gateway:8081`，在「镜像版本」添加可用的指纹浏览器镜像引用，即可创建环境；网关会在镜像缺失时自动拉取。架构、API 契约、失败语义和 `docker.sock` 风险边界见 +两个控制面凭据变量均为必填；Compose 不提供默认凭据。开发环境可复制 `.env.example` 后显式填写本地值。 + +打开 ，使用 `CONTROL_PLANE_USERNAME` / `CONTROL_PLANE_PASSWORD` 登录；局域网内用宿主机 IP 访问同一端口。首次使用：在「网关管理」用 Compose 里的 `GATEWAY_TOKEN` 注册 `http://docker-gateway:8081`，在「镜像版本」添加可用的指纹浏览器镜像引用，即可创建环境；网关会在镜像缺失时自动拉取。架构、API 契约、失败语义和 `docker.sock` 风险边界见 [《浏览器容器控制面》](docs/architecture/container-control.md)。 最小验证： diff --git a/cmd/control-plane/hub_test.go b/cmd/control-plane/hub_test.go index 8918b55..d768025 100644 --- a/cmd/control-plane/hub_test.go +++ b/cmd/control-plane/hub_test.go @@ -782,13 +782,17 @@ func newTestAppWithNetwork(t *testing.T, store *memoryStore, gateway *fakeGatewa return app } -func do(app *fiber.App, method, path, body string) *httptest.ResponseRecorder { +func do(app *fiber.App, method, path, body string, credentials ...string) *httptest.ResponseRecorder { response := httptest.NewRecorder() var reader io.Reader if body != "" { reader = strings.NewReader(body) } - adaptor.FiberApp(app).ServeHTTP(response, httptest.NewRequest(method, path, reader)) + request := httptest.NewRequest(method, path, reader) + if len(credentials) == 2 { + request.SetBasicAuth(credentials[0], credentials[1]) + } + adaptor.FiberApp(app).ServeHTTP(response, request) return response } diff --git a/cmd/control-plane/main.go b/cmd/control-plane/main.go index f931e46..fa79629 100644 --- a/cmd/control-plane/main.go +++ b/cmd/control-plane/main.go @@ -2,7 +2,11 @@ package main import ( "context" + "crypto/sha256" + "crypto/subtle" + "encoding/base64" "errors" + "io" "net" "net/url" "os" @@ -15,6 +19,7 @@ import ( "git.ipao.vip/rogee/creator-hub/internal/hub" "git.ipao.vip/rogee/creator-hub/internal/phasea" + "git.ipao.vip/rogee/creator-hub/internal/taskstate" "github.com/gofiber/fiber/v3" "github.com/sirupsen/logrus" "github.com/spf13/cobra" @@ -22,10 +27,9 @@ import ( ) type config struct { - listenAddr string - webDir string - databaseURL string - logLevel logrus.Level + listenAddr, webDir, databaseURL string + username, password string + logLevel logrus.Level } func main() { @@ -65,6 +69,9 @@ func newCommand() *cobra.Command { return err } defer hubStore.Close() + notify := newAttentionNotifier(os.Stderr) + phaseAStore.SetTaskNotifier(notify) + hubStore.SetTaskNotifier(notify) logStartup(cfg) heartbeatContext, stopHeartbeat := context.WithCancel(command.Context()) heartbeatDone := make(chan struct{}) @@ -72,7 +79,7 @@ func newCommand() *cobra.Command { defer close(heartbeatDone) runtimeLeaseHeartbeat(heartbeatContext, hubStore) }() - listenErr := newHandlerWithStores(cfg.webDir, phaseAStore, hubStore).Listen(cfg.listenAddr, fiber.ListenConfig{ + listenErr := newHandlerWithStores(cfg.webDir, cfg.username, cfg.password, phaseAStore, hubStore).Listen(cfg.listenAddr, fiber.ListenConfig{ GracefulContext: command.Context(), DisableStartupMessage: true, }) @@ -84,6 +91,21 @@ func newCommand() *cobra.Command { return command } +func newAttentionNotifier(output io.Writer) taskstate.Notifier { + logger := logrus.New() + logger.SetFormatter(&logrus.JSONFormatter{}) + logger.SetOutput(output) + return func(transition taskstate.Transition) { + if transition.State != "policy_hold" && transition.State != "needs_confirmation" { + return + } + logger.WithFields(logrus.Fields{ + "service": "control-plane", "notification_channel": "structured_log", "event_type": transition.State, + "reason_code": transition.ReasonCode, "account_id": transition.AccountID, "task_id": transition.TaskID, + }).Warn("operator attention required") + } +} + func runtimeLeaseHeartbeat(ctx context.Context, store hubStore) { ticker := time.NewTicker(20 * time.Second) defer ticker.Stop() @@ -116,6 +138,8 @@ func loadConfig() (config, error) { _ = v.BindEnv("web_dir", "WEB_DIR") _ = v.BindEnv("database_url", "DATABASE_URL") _ = v.BindEnv("log_level", "LOG_LEVEL") + _ = v.BindEnv("username", "CONTROL_PLANE_USERNAME") + _ = v.BindEnv("password", "CONTROL_PLANE_PASSWORD") level, err := logrus.ParseLevel(v.GetString("log_level")) if err != nil { @@ -125,6 +149,8 @@ func loadConfig() (config, error) { listenAddr: strings.TrimSpace(v.GetString("listen_addr")), webDir: strings.TrimSpace(v.GetString("web_dir")), databaseURL: strings.TrimSpace(v.GetString("database_url")), + username: strings.TrimSpace(v.GetString("username")), + password: v.GetString("password"), logLevel: level, } if cfg.listenAddr == "" { @@ -136,6 +162,12 @@ func loadConfig() (config, error) { if cfg.webDir == "" { return config{}, errors.New("WEB_DIR must not be empty") } + if cfg.username == "" || strings.Contains(cfg.username, ":") { + return config{}, errors.New("CONTROL_PLANE_USERNAME must be non-empty and contain no colon") + } + if len(cfg.password) < 16 { + return config{}, errors.New("CONTROL_PLANE_PASSWORD must contain at least 16 bytes") + } databaseURL, err := url.Parse(cfg.databaseURL) if err != nil || databaseURL.Host == "" || databaseURL.Path == "" || databaseURL.Path == "/" || (databaseURL.Scheme != "postgres" && databaseURL.Scheme != "postgresql") { @@ -156,11 +188,11 @@ func validateListenAddr(addr string) error { return nil } -func newHandler(webDirectory string) *fiber.App { - return newHandlerWithStores(webDirectory, nil, nil) +func newHandler(webDirectory, username, password string) *fiber.App { + return newHandlerWithStores(webDirectory, username, password, nil, nil) } -func newHandlerWithStores(webDirectory string, phaseAStore *phasea.Store, hubStore *hub.Store) *fiber.App { +func newHandlerWithStores(webDirectory, username, password string, phaseAStore *phasea.Store, hubStore *hub.Store) *fiber.App { app := fiber.New(fiber.Config{ AppName: "CreatorHub control plane", BodyLimit: 1 << 20, @@ -172,6 +204,7 @@ func newHandlerWithStores(webDirectory string, phaseAStore *phasea.Store, hubSto c.Status(fiber.StatusNoContent) return nil }) + app.Use(authenticate(username, password)) if hubStore != nil { registerHub(app, hubStore) } @@ -182,6 +215,23 @@ func newHandlerWithStores(webDirectory string, phaseAStore *phasea.Store, hubSto return app } +func authenticate(username, password string) fiber.Handler { + wantUser, wantPassword := sha256.Sum256([]byte(username)), sha256.Sum256([]byte(password)) + return func(c fiber.Ctx) error { + encoded, ok := strings.CutPrefix(c.Get(fiber.HeaderAuthorization), "Basic ") + decoded, err := base64.StdEncoding.DecodeString(encoded) + user, suppliedPassword, found := strings.Cut(string(decoded), ":") + gotUser, gotPassword := sha256.Sum256([]byte(user)), sha256.Sum256([]byte(suppliedPassword)) + userOK := subtle.ConstantTimeCompare(gotUser[:], wantUser[:]) + passwordOK := subtle.ConstantTimeCompare(gotPassword[:], wantPassword[:]) + if !ok || err != nil || !found || userOK != 1 || passwordOK != 1 { + c.Set(fiber.HeaderWWWAuthenticate, `Basic realm="CreatorHub", charset="UTF-8"`) + return c.Status(fiber.StatusUnauthorized).JSON(map[string]string{"error": "authentication required"}) + } + return c.Next() + } +} + func spaHandler(directory string) fiber.Handler { return func(c fiber.Ctx) error { path, err := url.PathUnescape(strings.TrimPrefix(c.Path(), "/")) diff --git a/cmd/control-plane/main_test.go b/cmd/control-plane/main_test.go index 7a8ed2b..454229b 100644 --- a/cmd/control-plane/main_test.go +++ b/cmd/control-plane/main_test.go @@ -20,6 +20,7 @@ import ( "git.ipao.vip/rogee/creator-hub/internal/hub" "git.ipao.vip/rogee/creator-hub/internal/phasea" + "git.ipao.vip/rogee/creator-hub/internal/taskstate" "github.com/gofiber/fiber/v3" "github.com/sirupsen/logrus" "github.com/spf13/cobra" @@ -67,12 +68,116 @@ func TestExecuteContextStopsOnSIGTERM(t *testing.T) { } func TestLoadConfigRejectsInvalidDatabase(t *testing.T) { + t.Setenv("CONTROL_PLANE_USERNAME", "operator") + t.Setenv("CONTROL_PLANE_PASSWORD", "unit-test-password") t.Setenv("DATABASE_URL", "file:///tmp/creatorhub.db") if _, err := loadConfig(); err == nil { t.Fatal("expected non-Postgres database URL to be rejected") } } +func TestLoadConfigRequiresControlPlaneCredentials(t *testing.T) { + t.Setenv("CONTROL_PLANE_USERNAME", "") + t.Setenv("CONTROL_PLANE_PASSWORD", "") + if _, err := loadConfig(); err == nil { + t.Fatal("expected missing control-plane credentials to be rejected") + } +} + +func TestControlPlaneAuthentication(t *testing.T) { + logger := logrus.StandardLogger() + previousOutput := logger.Out + var logs bytes.Buffer + logrus.SetOutput(&logs) + t.Cleanup(func() { logrus.SetOutput(previousOutput) }) + + directory := t.TempDir() + if err := os.WriteFile(filepath.Join(directory, "index.html"), []byte("index"), 0o600); err != nil { + t.Fatal(err) + } + app := newHandler(directory, "operator", "unit-test-password") + + health, err := app.Test(httptest.NewRequest(http.MethodGet, "/healthz", nil)) + if err != nil || health.StatusCode != http.StatusNoContent { + t.Fatalf("health check must remain public: status=%d err=%v", health.StatusCode, err) + } + health.Body.Close() + + for _, path := range []string{ + "/", "/api/phase-a/accounts", "/api/browsers", "/api/network-exits", "/api/phase-a/tasks", "/api/phase-a/audit", + } { + request := httptest.NewRequest(http.MethodGet, path, nil) + response, err := app.Test(request) + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != http.StatusUnauthorized || response.Header.Get("WWW-Authenticate") == "" { + t.Fatalf("GET %s was not protected: status=%d", path, response.StatusCode) + } + } + + for name, test := range map[string]struct { + user, password string + want int + }{ + "valid": {"operator", "unit-test-password", http.StatusOK}, + "wrong user": {"other", "unit-test-password", http.StatusUnauthorized}, + "wrong password": {"operator", "credential-must-not-be-logged", http.StatusUnauthorized}, + } { + t.Run(name, func(t *testing.T) { + request := httptest.NewRequest(http.MethodGet, "/", nil) + request.SetBasicAuth(test.user, test.password) + response, err := app.Test(request) + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != test.want { + t.Fatalf("status=%d want=%d", response.StatusCode, test.want) + } + }) + } + if strings.Contains(logs.String(), "credential-must-not-be-logged") { + t.Fatalf("authentication credential reached logs: %s", logs.String()) + } +} + +func TestOperatorNotificationFiltersAndRedacts(t *testing.T) { + previousLevel := logrus.GetLevel() + t.Cleanup(func() { logrus.SetLevel(previousLevel) }) + var output bytes.Buffer + notify := newAttentionNotifier(&output) + + notify(taskstate.Transition{State: "succeeded", ReasonCode: "task_succeeded", AccountID: "account-a", TaskID: "task-a"}) + if output.Len() != 0 { + t.Fatalf("successful task emitted a notification: %s", output.String()) + } + + for _, level := range []logrus.Level{logrus.ErrorLevel, logrus.FatalLevel, logrus.PanicLevel} { + for _, test := range []struct{ state, reason string }{ + {"policy_hold", "account_paused"}, {"needs_confirmation", "exit_unhealthy"}, + } { + output.Reset() + logrus.SetLevel(level) + notify(taskstate.Transition{State: test.state, ReasonCode: test.reason, AccountID: "account-a", TaskID: "task-a"}) + var entry map[string]any + if err := json.Unmarshal(output.Bytes(), &entry); err != nil { + t.Fatal(err) + } + if entry["event_type"] != test.state || entry["reason_code"] != test.reason || + entry["notification_channel"] != "structured_log" || entry["account_id"] != "account-a" || entry["task_id"] != "task-a" { + t.Fatalf("unexpected notification at LOG_LEVEL=%s: %#v", level, entry) + } + for _, forbidden := range []string{"password", "authorization", "credential", "token", "secret"} { + if strings.Contains(strings.ToLower(output.String()), forbidden) { + t.Fatalf("notification contained sensitive field %q: %s", forbidden, output.String()) + } + } + } + } +} + func TestPhaseAAccountRequestRejectsSecretsAndUnknownFields(t *testing.T) { for name, body := range map[string]string{ "raw password": `{"id":"account-a","platform":"mock","platform_account_key":"a","authorization_kind":"owned","credential_reference":{"id":"credential-a","provider":"os_keyring","key":"creatorhub/account-a"},"password":"secret"}`, @@ -116,17 +221,38 @@ func TestPhaseAAccountHTTPWorkflowRedactsSecrets(t *testing.T) { if err != nil { t.Fatal(err) } - if err := hubStore.Close(); err != nil { + t.Cleanup(func() { _ = hubStore.Close() }) + + webDirectory := t.TempDir() + if err := os.WriteFile(filepath.Join(webDirectory, "index.html"), []byte("index"), 0o600); err != nil { t.Fatal(err) } - + protectedApp := newHandlerWithStores(webDirectory, "operator", "unit-test-password", store, hubStore) + for _, route := range []struct{ method, path string }{ + {http.MethodGet, "/api/phase-a/accounts"}, {http.MethodPost, "/api/phase-a/accounts"}, + {http.MethodPost, "/api/phase-a/accounts/account-http/pause"}, {http.MethodGet, "/api/phase-a/tasks"}, + {http.MethodPost, "/api/phase-a/mock/execute"}, {http.MethodGet, "/api/phase-a/audit"}, + {http.MethodGet, "/api/browsers"}, {http.MethodPost, "/api/browsers/account-http/start"}, + {http.MethodDelete, "/api/browsers/account-http"}, {http.MethodPut, "/api/browser-images/1"}, + {http.MethodPost, "/api/network-exits/exit-http/check"}, {http.MethodPost, "/api/network-exits/exit-http/disable"}, + } { + if response := do(protectedApp, route.method, route.path, ""); response.Code != http.StatusUnauthorized { + t.Fatalf("unauthenticated %s %s returned %d", route.method, route.path, response.Code) + } + } + for _, path := range []string{"/api/phase-a/accounts", "/api/browsers", "/api/phase-a/audit"} { + if response := do(protectedApp, http.MethodGet, path, "", "operator", "unit-test-password"); response.Code != http.StatusOK { + t.Fatalf("authenticated GET %s returned %d: %s", path, response.Code, response.Body.String()) + } + } app := fiber.New() + app.Use(authenticate("operator", "unit-test-password")) registerPhaseA(app, store, nil) credentialKey := "creatorhub/phase-http-secret" accountBody := `{"id":"account-http","platform":"mock","platform_account_key":"phase-http","authorization_kind":"owned","credential_reference":{"id":"credential-http","provider":"os_keyring","key":"` + credentialKey + `"}}` request := func(method, path, body string, wantStatus int) *httptest.ResponseRecorder { t.Helper() - response := do(app, method, path, body) + response := do(app, method, path, body, "operator", "unit-test-password") if response.Code != wantStatus { t.Fatalf("%s %s returned %d, want %d: %s", method, path, response.Code, wantStatus, response.Body.String()) } @@ -204,7 +330,7 @@ func TestSPAServesFileAndIndexFallback(t *testing.T) { if err := os.WriteFile(filepath.Join(directory, "..", "outside-secret.txt"), []byte("secret"), 0o600); err != nil { t.Fatal(err) } - app := newHandler(directory) + app := newHandler(directory, "operator", "unit-test-password") for path, want := range map[string]string{ "/app.js": "asset", "/hello%20world.js": "space", @@ -212,7 +338,9 @@ func TestSPAServesFileAndIndexFallback(t *testing.T) { "/%2e%2e%2foutside-secret.txt": "index", "/dashboard": "index", } { - response, err := app.Test(httptest.NewRequest(http.MethodGet, path, nil)) + request := httptest.NewRequest(http.MethodGet, path, nil) + request.SetBasicAuth("operator", "unit-test-password") + response, err := app.Test(request) if err != nil { t.Fatal(err) } diff --git a/compose.yaml b/compose.yaml index 5a56ec5..e7827c7 100644 --- a/compose.yaml +++ b/compose.yaml @@ -3,6 +3,8 @@ services: build: . environment: DATABASE_URL: postgres://creatorhub@postgres/creatorhub?sslmode=disable + CONTROL_PLANE_USERNAME: ${CONTROL_PLANE_USERNAME:?required} + CONTROL_PLANE_PASSWORD: ${CONTROL_PLANE_PASSWORD:?required} ports: - "${CREATORHUB_PORT:-8080}:8080" read_only: true diff --git a/docs/deployment.md b/docs/deployment.md index 7258997..74049ca 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -1,6 +1,6 @@ # CreatorHub 部署 -本文档适用于当前阶段 A：在一台 Linux 主机上通过 Docker Compose 部署。开发阶段不设访问限制，安全由部署者自行把控。 +本文档适用于当前阶段 A：在一台 Linux 主机上通过 Docker Compose 部署。控制面使用单用户 HTTP Basic Auth；当前不提供 RBAC 或多租户隔离。 ## 部署内容 @@ -29,6 +29,8 @@ test -S /var/run/docker.sock docker info >/dev/null docker compose version +: "${CONTROL_PLANE_USERNAME:?set CONTROL_PLANE_USERNAME or fill .env}" +: "${CONTROL_PLANE_PASSWORD:?set CONTROL_PLANE_PASSWORD or fill .env}" docker compose config --quiet ``` @@ -40,6 +42,8 @@ docker compose config --quiet export DOCKER_GID="$(stat -c '%g' /var/run/docker.sock)" export CREATORHUB_PORT=8080 export GATEWAY_TOKEN="$(openssl rand -hex 24)" # 亦可在 .env 中设置 +export CONTROL_PLANE_USERNAME=creatorhub +export CONTROL_PLANE_PASSWORD="$(openssl rand -hex 24)" docker compose config --quiet docker compose up --detach --build @@ -64,6 +68,7 @@ curl --fail --silent --show-error \ curl --fail --silent --show-error \ --retry 30 --retry-delay 2 --retry-connrefused \ + --user "${CONTROL_PLANE_USERNAME}:${CONTROL_PLANE_PASSWORD}" \ "http://127.0.0.1:${CREATORHUB_PORT}/api/browsers" >/dev/null docker compose exec -T postgres \ @@ -91,8 +96,10 @@ Compose 部署时通常只需设置以下宿主机变量： | `CREATORHUB_PORT` | `8080` | 控制面宿主机端口，局域网可访问 | | `DOCKER_GID` | `999` | Docker socket 的宿主机组 ID；必须按实际值设置 | | `GATEWAY_TOKEN` | `dev-creatorhub-gateway-token` | 网关与控制面共享的 Bearer 令牌；生产须改为随机值，并同步填入网关注册表单 | +| `CONTROL_PLANE_USERNAME` | 无（必填） | 控制面唯一用户；不能包含冒号 | +| `CONTROL_PLANE_PASSWORD` | 无（必填） | 控制面密码，至少 16 字节；使用随机值 | -服务本身支持并校验以下环境变量；`compose.yaml` 已提供当前部署所需的值： +服务本身支持并校验以下环境变量；`compose.yaml` 会在控制面凭据缺失或为空时拒绝渲染： | 服务 | 变量 | 当前 Compose 值 | | --- | --- | --- | @@ -100,6 +107,8 @@ Compose 部署时通常只需设置以下宿主机变量： | `creator-hub` | `WEB_DIR` | 镜像内固定为 `/app/web` | | `creator-hub` | `DATABASE_URL` | `postgres://creatorhub@postgres/creatorhub?sslmode=disable` | | `creator-hub` | `LOG_LEVEL` | 默认 `info` | +| `creator-hub` | `CONTROL_PLANE_USERNAME` | 必填；HTTP Basic Auth 用户名 | +| `creator-hub` | `CONTROL_PLANE_PASSWORD` | 必填且至少 16 字节；不会写入日志或响应 | | `docker-gateway` | `LISTEN_ADDR` | 默认 `:8081` | | `docker-gateway` | `DOCKER_SOCKET` | 默认值和 Compose 挂载均固定为 `/var/run/docker.sock`；不能只覆盖环境变量 | | `docker-gateway` | `BROWSER_NETWORK` | `creatorhub_browser` | @@ -108,6 +117,16 @@ Compose 部署时通常只需设置以下宿主机变量： 不要把凭据写入仓库或 Compose 文件。 +### P0-lite 停机通知 + +当前只使用 `creator-hub` 的专用 Logrus JSON logger 作为通知渠道；它固定输出警告，不继承业务 `LOG_LEVEL`。选择它是因为 Compose 已可靠收集服务日志，不需要新增外部账号、凭据、网络重试或通知依赖。仅 `policy_hold` 和 `needs_confirmation` 会产生 `operator attention required`，字段限定为 `event_type`、`reason_code`、账号/任务 ID；不会包含请求头、Secret 引用或凭据值。运维可用下列命令接入现有日志采集或人工查看： + +```bash +docker compose logs creator-hub | grep 'operator attention required' +``` + +该渠道是单实例 P0-lite 能力，不保证外部送达、升级或确认回执；只有出现明确的多渠道/送达需求时才增加 webhook 或消息平台。 + ## 更新与回滚 更新前记录当前版本并备份数据库： diff --git a/internal/hub/environment.go b/internal/hub/environment.go index c13d127..0be2e13 100644 --- a/internal/hub/environment.go +++ b/internal/hub/environment.go @@ -11,6 +11,8 @@ import ( "regexp" "strings" "time" + + "git.ipao.vip/rogee/creator-hub/internal/taskstate" ) var exitIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$`) @@ -260,12 +262,14 @@ func (s *Store) RecordNetworkExitCheck(ctx context.Context, id string, observati WHERE id = $1`, id, observation.PublicIP, observation.Region, status, reason, version); err != nil { return NetworkExit{}, "persistence_failed", errors.New("record network exit check") } + var transitions []taskstate.Transition if changed { - if err := invalidateAccountsForExit(ctx, tx, id); err != nil { + transitions, err = invalidateAccountsForExit(ctx, tx, id) + if err != nil { return NetworkExit{}, "persistence_failed", err } } - if err := commitHub(tx); err != nil { + if err := s.commitTransitions(tx, transitions); err != nil { return NetworkExit{}, "persistence_failed", err } exit, err := s.GetNetworkExit(ctx, id) @@ -301,6 +305,7 @@ func (s *Store) DisableNetworkExit(ctx context.Context, id string) (NetworkExit, if err := tx.QueryRowContext(ctx, `SELECT health_status FROM network_exit WHERE id = $1 FOR UPDATE`, id).Scan(&oldStatus); err != nil { return NetworkExit{}, rowError(err) } + var transitions []taskstate.Transition if oldStatus != "disabled" { if _, err := tx.ExecContext(ctx, ` UPDATE network_exit SET health_status = 'disabled', last_check_reason = 'exit_disabled', @@ -308,18 +313,19 @@ func (s *Store) DisableNetworkExit(ctx context.Context, id string) (NetworkExit, WHERE id = $1`, id); err != nil { return NetworkExit{}, errors.New("disable network exit") } - if err := invalidateAccountsForExit(ctx, tx, id); err != nil { + transitions, err = invalidateAccountsForExit(ctx, tx, id) + if err != nil { return NetworkExit{}, err } } - if err := commitHub(tx); err != nil { + if err := s.commitTransitions(tx, transitions); err != nil { return NetworkExit{}, err } return s.GetNetworkExit(ctx, id) } -func invalidateAccountsForExit(ctx context.Context, tx *sql.Tx, exitID string) error { - if _, err := tx.ExecContext(ctx, ` +func invalidateAccountsForExit(ctx context.Context, tx *sql.Tx, exitID string) ([]taskstate.Transition, error) { + rows, err := tx.QueryContext(ctx, ` WITH changed AS ( UPDATE social_account account SET status = 'paused', paused_at = COALESCE(paused_at, now()), version = account.version + 1, updated_at = now() @@ -332,14 +338,18 @@ func invalidateAccountsForExit(ctx context.Context, tx *sql.Tx, exitID string) e hold_reason = 'exit_unhealthy', verification_result = NULL, verified_at = NULL, verified_by = NULL, lease_owner = NULL, lease_until = NULL, updated_at = now() FROM changed WHERE task.account_id = changed.id AND task.state IN ('queued', 'executing') - RETURNING task.current_attempt_id, task.state + RETURNING task.current_attempt_id, task.state, task.hold_reason, task.account_id, task.id + ), finished AS ( + UPDATE execution_attempt attempt SET finished_at = now(), outcome = 'uncertain' + FROM held WHERE held.state = 'needs_confirmation' AND attempt.id = held.current_attempt_id + AND attempt.finished_at IS NULL + RETURNING attempt.id ) - UPDATE execution_attempt attempt SET finished_at = now(), outcome = 'uncertain' - FROM held WHERE held.state = 'needs_confirmation' AND attempt.id = held.current_attempt_id - AND attempt.finished_at IS NULL`, exitID); err != nil { - return errors.New("invalidate network exit accounts") + SELECT state, hold_reason, account_id, id FROM held`, exitID) + if err != nil { + return nil, errors.New("invalidate network exit accounts") } - return nil + return scanTransitions(rows, "read invalidated network exit tasks") } func (s *Store) CreateBoundEnv(ctx context.Context, env Env, accountID, exitID string) (EnvironmentContext, bool, error) { diff --git a/internal/hub/migration_test.go b/internal/hub/migration_test.go index 0ce811c..e61346c 100644 --- a/internal/hub/migration_test.go +++ b/internal/hub/migration_test.go @@ -7,10 +7,12 @@ import ( "fmt" "net/url" "os" + "slices" "testing" "time" "git.ipao.vip/rogee/creator-hub/internal/phasea" + "git.ipao.vip/rogee/creator-hub/internal/taskstate" ) func TestUnifiedAccountMigration(t *testing.T) { @@ -294,6 +296,8 @@ func TestUnifiedAccountMigration(t *testing.T) { if err != nil { t.Fatal(err) } + var notifications []taskstate.Transition + store.SetTaskNotifier(func(transition taskstate.Transition) { notifications = append(notifications, transition) }) if err := store.UpgradeEnv(ctx, "mapped", "2"); err != nil { t.Fatal(err) } @@ -304,6 +308,14 @@ func TestUnifiedAccountMigration(t *testing.T) { assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'upgrade-task' AND state = 'policy_hold' AND hold_reason = 'binding_version_changed'`, 1) assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'upgrade-executing' AND state = 'needs_confirmation' AND hold_reason = 'task_result_uncertain' AND lease_owner IS NULL`, 1) assertDatabaseCount(t, db, `SELECT count(*) FROM execution_attempt WHERE id = 'upgrade-attempt' AND outcome = 'uncertain' AND finished_at IS NOT NULL`, 1) + for _, want := range []taskstate.Transition{ + {State: "policy_hold", ReasonCode: "binding_version_changed", AccountID: "mapped", TaskID: "upgrade-task"}, + {State: "needs_confirmation", ReasonCode: "task_result_uncertain", AccountID: "mapped", TaskID: "upgrade-executing"}, + } { + if !slices.Contains(notifications, want) { + t.Fatalf("missing upgrade transition %+v in %+v", want, notifications) + } + } phaseAStore, err = phasea.Open(ctx, testURL) if err != nil { t.Fatal(err) diff --git a/internal/hub/store.go b/internal/hub/store.go index ecb4e97..e0a5413 100644 --- a/internal/hub/store.go +++ b/internal/hub/store.go @@ -15,6 +15,7 @@ import ( "time" "unicode/utf8" + "git.ipao.vip/rogee/creator-hub/internal/taskstate" "github.com/jackc/pgx/v5/pgconn" _ "github.com/jackc/pgx/v5/stdlib" ) @@ -69,7 +70,10 @@ var ( imageRefPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:/@-]{0,300}$`) ) -type Store struct{ db *sql.DB } +type Store struct { + db *sql.DB + notify taskstate.Notifier +} // Gateway 是平台注册的 docker-gateway 实例；Token 由平台生成，明文存储供页面复制（开发阶段约定）。 type Gateway struct { @@ -122,6 +126,17 @@ func Open(ctx context.Context, databaseURL string) (*Store, error) { func (s *Store) Close() error { return s.db.Close() } +func (s *Store) SetTaskNotifier(notify taskstate.Notifier) { s.notify = notify } + +func (s *Store) notifyTransitions(transitions []taskstate.Transition) { + if s.notify == nil { + return + } + for _, transition := range transitions { + s.notify(transition) + } +} + func (s *Store) migrate(ctx context.Context) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { @@ -393,7 +408,7 @@ func (s *Store) UpgradeEnv(ctx context.Context, alias, version string) error { WHERE browser_env_alias = $1`, alias); err != nil { return errors.New("version upgraded environment binding") } - if _, err := tx.ExecContext(ctx, ` + rows, err := tx.QueryContext(ctx, ` WITH changed AS ( UPDATE social_account account SET status = 'paused', paused_at = now(), version = account.version + 1, updated_at = now() @@ -407,14 +422,22 @@ func (s *Store) UpgradeEnv(ctx context.Context, alias, version string) error { verification_result = NULL, verified_at = NULL, verified_by = NULL, lease_owner = NULL, lease_until = NULL, updated_at = now() FROM changed WHERE task.account_id = changed.id AND task.state IN ('queued', 'executing') - RETURNING task.current_attempt_id, task.state + RETURNING task.current_attempt_id, task.state, task.hold_reason, task.account_id, task.id + ), finished AS ( + UPDATE execution_attempt attempt SET finished_at = now(), outcome = 'uncertain' + FROM held WHERE held.state = 'needs_confirmation' AND attempt.id = held.current_attempt_id + AND attempt.finished_at IS NULL + RETURNING attempt.id ) - UPDATE execution_attempt attempt SET finished_at = now(), outcome = 'uncertain' - FROM held WHERE held.state = 'needs_confirmation' AND attempt.id = held.current_attempt_id - AND attempt.finished_at IS NULL`, alias); err != nil { + SELECT state, hold_reason, account_id, id FROM held`, alias) + if err != nil { return errors.New("invalidate upgraded environment account") } - return commitHub(tx) + transitions, err := scanTransitions(rows, "read upgraded environment tasks") + if err != nil { + return err + } + return s.commitTransitions(tx, transitions) } func (s *Store) DeleteEnv(ctx context.Context, alias string) error { @@ -474,6 +497,30 @@ func commitHub(tx *sql.Tx) error { return nil } +func (s *Store) commitTransitions(tx *sql.Tx, transitions []taskstate.Transition) error { + if err := commitHub(tx); err != nil { + return err + } + s.notifyTransitions(transitions) + return nil +} + +func scanTransitions(rows *sql.Rows, failure string) ([]taskstate.Transition, error) { + defer rows.Close() + var transitions []taskstate.Transition + for rows.Next() { + var transition taskstate.Transition + if err := rows.Scan(&transition.State, &transition.ReasonCode, &transition.AccountID, &transition.TaskID); err != nil { + return nil, errors.New(failure) + } + transitions = append(transitions, transition) + } + if err := rows.Err(); err != nil { + return nil, errors.New(failure) + } + return transitions, nil +} + func rowError(err error) error { if errors.Is(err, sql.ErrNoRows) { return ErrNotFound diff --git a/internal/hub/store_test.go b/internal/hub/store_test.go index 4e7e8d6..bb83a79 100644 --- a/internal/hub/store_test.go +++ b/internal/hub/store_test.go @@ -6,9 +6,12 @@ import ( "errors" "os" "reflect" + "slices" "strings" "testing" "time" + + "git.ipao.vip/rogee/creator-hub/internal/taskstate" ) func TestFingerprintArgsFollowUpstreamCommandLineContract(t *testing.T) { @@ -250,6 +253,8 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) { ctx := context.Background() store := openFullyMigratedHub(t, ctx, isolatedDatabaseURL(t, databaseURL)) t.Cleanup(func() { _ = store.Close() }) + var notifications []taskstate.Transition + store.SetTaskNotifier(func(transition taskstate.Transition) { notifications = append(notifications, transition) }) if _, err := store.db.ExecContext(ctx, `TRUNCATE audit_event, runtime_instance, environment_binding, network_exit, social_account, credential_reference, browser_env, browser_image, gateway CASCADE`); err != nil { t.Fatal(err) @@ -532,6 +537,42 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) { assertDatabaseCount(t, store.db, `SELECT count(*) FROM operation_task WHERE id = 'exit-hold-queued' AND state = 'policy_hold' AND hold_reason = 'exit_unhealthy'`, 1) assertDatabaseCount(t, store.db, `SELECT count(*) FROM operation_task WHERE id = 'exit-hold-executing' AND state = 'needs_confirmation' AND hold_reason = 'exit_unhealthy' AND lease_owner IS NULL`, 1) assertDatabaseCount(t, store.db, `SELECT count(*) FROM execution_attempt WHERE id = 'exit-hold-attempt' AND outcome = 'uncertain' AND finished_at IS NOT NULL`, 1) + for _, want := range []taskstate.Transition{ + {State: "policy_hold", ReasonCode: "exit_unhealthy", AccountID: "account-a", TaskID: "exit-hold-queued"}, + {State: "needs_confirmation", ReasonCode: "exit_unhealthy", AccountID: "account-a", TaskID: "exit-hold-executing"}, + } { + if !slices.Contains(notifications, want) { + t.Fatalf("missing network exit transition %+v in %+v", want, notifications) + } + } + if err := store.db.QueryRowContext(ctx, `SELECT version FROM social_account WHERE id = 'account-a'`).Scan(&accountVersion); err != nil { + t.Fatal(err) + } + if _, err := store.db.ExecContext(ctx, `INSERT INTO content_draft (id, account_id, version, content) VALUES ('exit-disable-draft', 'account-a', 1, 'test')`); err != nil { + t.Fatal(err) + } + if _, err := store.db.ExecContext(ctx, `INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version) + VALUES ('exit-disable-confirmation', 'account-a', $1, 'exit-disable-draft', 1, 1)`, accountVersion); err != nil { + t.Fatal(err) + } + if _, err := store.db.ExecContext(ctx, `INSERT INTO operation_task + (id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version, state, lease_owner, lease_until) + VALUES + ('exit-disable-queued', 'exit-disable-queued-key', 'account-a', $1, 'exit-disable-draft', 1, 'exit-disable-confirmation', 1, 'queued', NULL, NULL), + ('exit-disable-executing', 'exit-disable-executing-key', 'account-a', $1, 'exit-disable-draft', 1, 'exit-disable-confirmation', 1, 'executing', 'worker-disabled', now() + interval '1 minute')`, accountVersion); err != nil { + t.Fatal(err) + } + if _, err := store.DisableNetworkExit(ctx, newGeneration.Exit.ID); err != nil { + t.Fatal(err) + } + for _, want := range []taskstate.Transition{ + {State: "policy_hold", ReasonCode: "exit_unhealthy", AccountID: "account-a", TaskID: "exit-disable-queued"}, + {State: "needs_confirmation", ReasonCode: "exit_unhealthy", AccountID: "account-a", TaskID: "exit-disable-executing"}, + } { + if !slices.Contains(notifications, want) { + t.Fatalf("missing disabled exit transition %+v in %+v", want, notifications) + } + } var auditText string if err := store.db.QueryRowContext(ctx, `SELECT string_agg(row_to_json(event)::text, '') FROM audit_event event`).Scan(&auditText); err != nil { t.Fatal(err) diff --git a/internal/phasea/store.go b/internal/phasea/store.go index 106b896..c1a5d53 100644 --- a/internal/phasea/store.go +++ b/internal/phasea/store.go @@ -14,6 +14,7 @@ import ( "strings" "time" + "git.ipao.vip/rogee/creator-hub/internal/taskstate" "github.com/jackc/pgx/v5/pgconn" _ "github.com/jackc/pgx/v5/stdlib" ) @@ -32,7 +33,10 @@ var ( eventPattern = regexp.MustCompile(`^[a-z0-9_]{1,64}$`) ) -type Store struct{ db *sql.DB } +type Store struct { + db *sql.DB + notify taskstate.Notifier +} type Account struct { ID string `json:"id"` @@ -207,6 +211,17 @@ func Open(ctx context.Context, databaseURL string) (*Store, error) { func (s *Store) Close() error { return s.db.Close() } +func (s *Store) SetTaskNotifier(notify taskstate.Notifier) { s.notify = notify } + +func (s *Store) notifyTransitions(transitions []taskstate.Transition) { + if s.notify == nil { + return + } + for _, transition := range transitions { + s.notify(transition) + } +} + func (s *Store) migrate(ctx context.Context) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { @@ -985,19 +1000,18 @@ func (s *Store) disableAccount(ctx context.Context, accountID string, revoke boo if err != nil { return err } - if unchanged { - return commit(tx) + if !unchanged { + eventType := "account_paused" + if revoke { + eventType = "account_revoked" + } + if err := appendAudit(ctx, tx, eventType, reason, accountID, "", 0, "", "", map[string]any{ + "account_version": version, "tasks_held": len(held), "attempts_interrupted": len(interrupted), + }); err != nil { + return err + } } - eventType := "account_paused" - if revoke { - eventType = "account_revoked" - } - if err := appendAudit(ctx, tx, eventType, reason, accountID, "", 0, "", "", map[string]any{ - "account_version": version, "tasks_held": held, "attempts_interrupted": interrupted, - }); err != nil { - return err - } - return commit(tx) + return s.commitTransitions(tx, append(held, interrupted...)) } func (s *Store) ResumeAccount(ctx context.Context, accountID string) error { @@ -1049,19 +1063,19 @@ func (s *Store) ResumeAccount(ctx context.Context, accountID string) error { return commit(tx) } -func holdQueuedTasks(ctx context.Context, tx *sql.Tx, accountID, reason string) (int64, error) { - result, err := tx.ExecContext(ctx, ` +func holdQueuedTasks(ctx context.Context, tx *sql.Tx, accountID, reason string) ([]taskstate.Transition, error) { + rows, err := tx.QueryContext(ctx, ` UPDATE operation_task SET state = 'policy_hold', hold_reason = $2, verification_result = NULL, verified_at = NULL, verified_by = NULL, updated_at = now() - WHERE account_id = $1 AND state = 'queued'`, accountID, reason) + WHERE account_id = $1 AND state = 'queued' + RETURNING state, hold_reason, account_id, id`, accountID, reason) if err != nil { - return 0, errors.New("hold queued account tasks") + return nil, errors.New("hold queued account tasks") } - held, _ := result.RowsAffected() - return held, nil + return scanTransitions(rows, "read held account tasks") } -func interruptExecutingTasks(ctx context.Context, tx *sql.Tx, accountID, reason string) (int64, error) { +func interruptExecutingTasks(ctx context.Context, tx *sql.Tx, accountID, reason string) ([]taskstate.Transition, error) { rows, err := tx.QueryContext(ctx, ` UPDATE operation_task SET state = 'needs_confirmation', hold_reason = $2, verification_result = NULL, verified_at = NULL, verified_by = NULL, @@ -1069,7 +1083,7 @@ func interruptExecutingTasks(ctx context.Context, tx *sql.Tx, accountID, reason WHERE account_id = $1 AND state = 'executing' RETURNING id, current_attempt_id, confirmation_id, confirmation_version`, accountID, reason) if err != nil { - return 0, errors.New("interrupt executing account tasks") + return nil, errors.New("interrupt executing account tasks") } type interruptedTask struct { taskID, attemptID, confirmationID string @@ -1082,28 +1096,32 @@ func interruptExecutingTasks(ctx context.Context, tx *sql.Tx, accountID, reason var confirmationVersion sql.NullInt64 if err := rows.Scan(&task.taskID, &attemptID, &confirmationID, &confirmationVersion); err != nil { rows.Close() - return 0, errors.New("read interrupted account task") + return nil, errors.New("read interrupted account task") } task.attemptID, task.confirmationID, task.confirmationVersion = attemptID.String, confirmationID.String, confirmationVersion.Int64 tasks = append(tasks, task) } if err := rows.Close(); err != nil { - return 0, errors.New("close interrupted account task rows") + return nil, errors.New("close interrupted account task rows") } + transitions := make([]taskstate.Transition, 0, len(tasks)) for _, task := range tasks { if task.attemptID != "" { if _, err := tx.ExecContext(ctx, ` UPDATE execution_attempt SET finished_at = now(), outcome = 'uncertain' WHERE id = $1 AND finished_at IS NULL`, task.attemptID); err != nil { - return 0, errors.New("mark interrupted attempt uncertain") + return nil, errors.New("mark interrupted attempt uncertain") } } if err := appendAudit(ctx, tx, "task_interrupted", reason, accountID, task.confirmationID, task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil { - return 0, err + return nil, err } + transitions = append(transitions, taskstate.Transition{ + State: "needs_confirmation", ReasonCode: reason, AccountID: accountID, TaskID: task.taskID, + }) } - return int64(len(tasks)), nil + return transitions, nil } func (s *Store) VerifyTask(ctx context.Context, taskID, result string) error { @@ -1300,7 +1318,7 @@ func (s *Store) CancelTask(ctx context.Context, taskID string) error { return errors.New("begin cancel transaction") } defer tx.Rollback() - var accountID, state string + var accountID, state, holdReason string var attemptID sql.NullString var confirmationID sql.NullString var confirmationVersion sql.NullInt64 @@ -1310,8 +1328,8 @@ func (s *Store) CancelTask(ctx context.Context, taskID string) error { verification_result = NULL, verified_at = NULL, verified_by = NULL, lease_owner = NULL, lease_until = NULL, updated_at = now() WHERE id = $1 AND state IN ('queued', 'executing', 'needs_confirmation', 'policy_hold') - RETURNING account_id, state, current_attempt_id, confirmation_id, confirmation_version`, taskID).Scan( - &accountID, &state, &attemptID, &confirmationID, &confirmationVersion) + RETURNING account_id, state, COALESCE(hold_reason, ''), current_attempt_id, confirmation_id, confirmation_version`, taskID).Scan( + &accountID, &state, &holdReason, &attemptID, &confirmationID, &confirmationVersion) if err != nil { return rowError(err) } @@ -1325,7 +1343,7 @@ func (s *Store) CancelTask(ctx context.Context, taskID string) error { if err := appendAudit(ctx, tx, "task_cancel_requested", "task_cancel_requested", accountID, confirmationID.String, confirmationVersion.Int64, attemptID.String, taskID, map[string]string{"state": state}); err != nil { return err } - return commit(tx) + return s.commitTransitions(tx, []taskstate.Transition{{State: state, ReasonCode: holdReason, AccountID: accountID, TaskID: taskID}}) } func (s *Store) ExecuteMock(ctx context.Context, workerID, outcome string) (Execution, error) { @@ -1345,12 +1363,15 @@ func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) { return Execution{}, errors.New("begin claim transaction") } defer tx.Rollback() - if err := quarantineExpired(ctx, tx); err != nil { + expired, err := quarantineExpired(ctx, tx) + if err != nil { return Execution{}, err } - if err := quarantineInvalid(ctx, tx); err != nil { + invalid, err := quarantineInvalid(ctx, tx) + if err != nil { return Execution{}, err } + transitions := append(expired, invalid...) var execution Execution err = tx.QueryRowContext(ctx, ` WITH candidate AS ( @@ -1381,7 +1402,7 @@ func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) { RETURNING t.id, t.account_id, t.confirmation_id, t.confirmation_version`, workerID).Scan( &execution.TaskID, &execution.AccountID, &execution.ConfirmationID, &execution.ConfirmationVersion) if errors.Is(err, sql.ErrNoRows) { - if err := commit(tx); err != nil { + if err := s.commitTransitions(tx, transitions); err != nil { return Execution{}, err } return Execution{}, nil @@ -1401,7 +1422,7 @@ func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) { if err := appendAudit(ctx, tx, "task_claimed", "task_claimed", execution.AccountID, execution.ConfirmationID, execution.ConfirmationVersion, execution.AttemptID, execution.TaskID, map[string]string{"worker_id": workerID}); err != nil { return Execution{}, err } - if err := commit(tx); err != nil { + if err := s.commitTransitions(tx, transitions); err != nil { return Execution{}, err } return execution, nil @@ -1498,14 +1519,16 @@ func (s *Store) complete(ctx context.Context, workerID string, execution Executi if err := appendAudit(ctx, tx, "task_finished", reason, execution.AccountID, execution.ConfirmationID, execution.ConfirmationVersion, execution.AttemptID, execution.TaskID, map[string]string{"state": state}); err != nil { return Execution{}, err } - if err := commit(tx); err != nil { + if err := s.commitTransitions(tx, []taskstate.Transition{{ + State: state, ReasonCode: holdReason, AccountID: execution.AccountID, TaskID: execution.TaskID, + }}); err != nil { return Execution{}, err } execution.State = state return execution, nil } -func quarantineExpired(ctx context.Context, tx *sql.Tx) error { +func quarantineExpired(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition, error) { rows, err := tx.QueryContext(ctx, ` UPDATE operation_task SET state = 'needs_confirmation', hold_reason = 'execution_lease_expired', verification_result = NULL, verified_at = NULL, verified_by = NULL, @@ -1513,7 +1536,7 @@ func quarantineExpired(ctx context.Context, tx *sql.Tx) error { WHERE state = 'executing' AND lease_until < now() RETURNING id, account_id, current_attempt_id, confirmation_id, confirmation_version`) if err != nil { - return errors.New("quarantine expired tasks") + return nil, errors.New("quarantine expired tasks") } type expiredTask struct { taskID, accountID, attemptID, confirmationID string @@ -1527,27 +1550,31 @@ func quarantineExpired(ctx context.Context, tx *sql.Tx) error { var confirmationVersion sql.NullInt64 if err := rows.Scan(&taskID, &accountID, &attemptID, &confirmationID, &confirmationVersion); err != nil { rows.Close() - return errors.New("read expired task") + return nil, errors.New("read expired task") } expired = append(expired, expiredTask{taskID, accountID, attemptID.String, confirmationID.String, confirmationVersion.Int64}) } if err := rows.Close(); err != nil { - return errors.New("close expired task rows") + return nil, errors.New("close expired task rows") } + transitions := make([]taskstate.Transition, 0, len(expired)) for _, task := range expired { if task.attemptID != "" { if _, err := tx.ExecContext(ctx, `UPDATE execution_attempt SET finished_at = now(), outcome = 'uncertain' WHERE id = $1 AND finished_at IS NULL`, task.attemptID); err != nil { - return errors.New("mark expired attempt uncertain") + return nil, errors.New("mark expired attempt uncertain") } } if err := appendAudit(ctx, tx, "task_lease_expired", "execution_lease_expired", task.accountID, task.confirmationID, task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil { - return err + return nil, err } + transitions = append(transitions, taskstate.Transition{ + State: "needs_confirmation", ReasonCode: "execution_lease_expired", AccountID: task.accountID, TaskID: task.taskID, + }) } - return nil + return transitions, nil } -func quarantineInvalid(ctx context.Context, tx *sql.Tx) error { +func quarantineInvalid(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition, error) { rows, err := tx.QueryContext(ctx, ` WITH invalid AS ( SELECT t.id, @@ -1606,7 +1633,7 @@ func quarantineInvalid(ctx context.Context, tx *sql.Tx) error { FROM invalid WHERE task.id = invalid.id RETURNING task.id, task.account_id, task.confirmation_id, task.confirmation_version, task.state, invalid.reason_code`) if err != nil { - return errors.New("quarantine invalid tasks") + return nil, errors.New("quarantine invalid tasks") } type invalidTask struct { taskID, accountID, confirmationID, state, reasonCode string @@ -1619,19 +1646,39 @@ func quarantineInvalid(ctx context.Context, tx *sql.Tx) error { var confirmationVersion sql.NullInt64 if err := rows.Scan(&taskID, &accountID, &confirmationID, &confirmationVersion, &state, &reasonCode); err != nil { rows.Close() - return errors.New("read quarantined task") + return nil, errors.New("read quarantined task") } invalid = append(invalid, invalidTask{taskID, accountID, confirmationID.String, state, reasonCode, confirmationVersion.Int64}) } if err := rows.Close(); err != nil { - return errors.New("close quarantined task rows") + return nil, errors.New("close quarantined task rows") } + transitions := make([]taskstate.Transition, 0, len(invalid)) for _, task := range invalid { if err := appendAudit(ctx, tx, "task_quarantined", task.reasonCode, task.accountID, task.confirmationID, task.confirmationVersion, "", task.taskID, map[string]string{"state": task.state}); err != nil { - return err + return nil, err } + transitions = append(transitions, taskstate.Transition{ + State: task.state, ReasonCode: task.reasonCode, AccountID: task.accountID, TaskID: task.taskID, + }) } - return nil + return transitions, nil +} + +func scanTransitions(rows *sql.Rows, failure string) ([]taskstate.Transition, error) { + defer rows.Close() + var transitions []taskstate.Transition + for rows.Next() { + var transition taskstate.Transition + if err := rows.Scan(&transition.State, &transition.ReasonCode, &transition.AccountID, &transition.TaskID); err != nil { + return nil, errors.New(failure) + } + transitions = append(transitions, transition) + } + if err := rows.Err(); err != nil { + return nil, errors.New(failure) + } + return transitions, nil } func (s *Store) Audit(ctx context.Context) ([]AuditEvent, error) { @@ -1785,6 +1832,14 @@ func commit(tx *sql.Tx) error { return nil } +func (s *Store) commitTransitions(tx *sql.Tx, transitions []taskstate.Transition) error { + if err := commit(tx); err != nil { + return err + } + s.notifyTransitions(transitions) + return nil +} + func rowError(err error) error { if errors.Is(err, sql.ErrNoRows) { return ErrNotFound diff --git a/internal/phasea/store_test.go b/internal/phasea/store_test.go index 018e56b..6c5e130 100644 --- a/internal/phasea/store_test.go +++ b/internal/phasea/store_test.go @@ -12,6 +12,8 @@ import ( "sync/atomic" "testing" "time" + + "git.ipao.vip/rogee/creator-hub/internal/taskstate" ) func TestNewAccountIDIsStoreValidAndUnique(t *testing.T) { @@ -95,6 +97,25 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { t.Fatal(err) } t.Cleanup(func() { _ = store.Close() }) + var notificationMu sync.Mutex + var notifications []taskstate.Transition + store.SetTaskNotifier(func(transition taskstate.Transition) { + notificationMu.Lock() + defer notificationMu.Unlock() + notifications = append(notifications, transition) + }) + assertNotified := func(state, reason, accountID, taskID string) { + t.Helper() + notificationMu.Lock() + defer notificationMu.Unlock() + want := taskstate.Transition{State: state, ReasonCode: reason, AccountID: accountID, TaskID: taskID} + for _, notification := range notifications { + if notification == want { + return + } + } + t.Fatalf("missing task notification %+v in %+v", want, notifications) + } applyHubMigrationsForPhaseATest(t, store) if _, err := store.db.ExecContext(ctx, ` TRUNCATE audit_event, execution_attempt, operation_task, confirmation, content_draft, @@ -299,6 +320,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { t.Fatalf("invalid task was claimed: execution=%+v err=%v", execution, err) } assertTaskGate(t, store, taskID, test.wantState, test.wantReason) + assertNotified(test.wantState, test.wantReason, "gate-"+test.name, taskID) }) } if err := store.VerifyTask(ctx, "task-gate-exit-unhealthy", "not_executed"); err != nil { @@ -345,6 +367,8 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { t.Fatalf("invalid tasks must not execute: execution=%+v err=%v", execution, err) } assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id IN ('task-30', 'task-31') AND state = 'needs_confirmation'`, 2) + assertNotified("needs_confirmation", "confirmation_missing", "account-b", "task-30") + assertNotified("needs_confirmation", "account_version_changed", "account-b", "task-31") assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id IN ('task-30', 'task-31')`, 0) nullConfirmation, err := store.GetTaskDetail(ctx, "task-30") if err != nil || nullConfirmation.Confirmation.ID != "" || nullConfirmation.AllowedAction != "" { @@ -358,6 +382,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { if execution, err := store.ExecuteMock(ctx, "worker-uncertain", "uncertain"); err != nil || execution.State != "needs_confirmation" { t.Fatalf("uncertain outcome was not held: execution=%+v err=%v", execution, err) } + assertNotified("needs_confirmation", "task_result_uncertain", "account-b", "task-32") if execution, err := store.ExecuteMock(ctx, "worker-uncertain", "succeeded"); err != nil || execution.WasClaimed { t.Fatalf("uncertain task was retried: execution=%+v err=%v", execution, err) } @@ -370,6 +395,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { if execution, err := store.ExecuteMock(ctx, "worker-policy", "policy_hold"); err != nil || execution.State != "policy_hold" { t.Fatalf("policy outcome was not held: execution=%+v err=%v", execution, err) } + assertNotified("policy_hold", "task_policy_hold", "account-b", "task-33") if execution, err := store.ExecuteMock(ctx, "worker-policy", "succeeded"); err != nil || execution.WasClaimed { t.Fatalf("policy task was retried: execution=%+v err=%v", execution, err) } @@ -378,6 +404,14 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { if _, _, err := store.Enqueue(ctx, paused); err != nil { t.Fatal(err) } + pausedQueued := approvedTask(340, "account-b", accountB.Version, "draft-b", "confirmation-b") + pausedQueued.ID, pausedQueued.IdempotencyKey = "task-34-queued", "idempotency-34-queued" + if _, _, err := store.Enqueue(ctx, pausedQueued); err != nil { + t.Fatal(err) + } + if claimed, err := store.claim(ctx, "worker-account-pause"); err != nil || claimed.TaskID != paused.ID { + t.Fatalf("claim executing pause fixture: execution=%+v err=%v", claimed, err) + } pauseStarted := time.Now() if err := store.PauseAccount(ctx, "account-b"); err != nil { t.Fatal(err) @@ -388,7 +422,10 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { if execution, err := store.ExecuteMock(ctx, "worker-paused", "succeeded"); err != nil || execution.WasClaimed { t.Fatalf("paused account task was claimed: execution=%+v err=%v", execution, err) } - assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-34' AND state = 'policy_hold'`, 1) + assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-34' AND state = 'needs_confirmation'`, 1) + assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-34-queued' AND state = 'policy_hold'`, 1) + assertNotified("needs_confirmation", "account_paused", "account-b", "task-34") + assertNotified("policy_hold", "account_paused", "account-b", "task-34-queued") pausedAccount, err := store.GetAccount(ctx, "account-b") if err != nil || pausedAccount.Version != accountB.Version+1 { t.Fatalf("idempotent pause must increment once: %#v %v", pausedAccount, err) @@ -398,6 +435,18 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { t.Fatalf("account was not paused within one minute: paused_at=%v err=%v", pausedAt, err) } + revokeQueued := approvedTask(341, "account-b", pausedAccount.Version, "draft-b", "confirmation-b") + revokeQueued.ID, revokeQueued.IdempotencyKey = "task-revoke-queued", "idempotency-revoke-queued" + revokeExecuting := approvedTask(342, "account-b", pausedAccount.Version, "draft-b", "confirmation-b") + revokeExecuting.ID, revokeExecuting.IdempotencyKey = "task-revoke-executing", "idempotency-revoke-executing" + for _, task := range []Task{revokeQueued, revokeExecuting} { + if _, _, err := store.Enqueue(ctx, task); err != nil { + t.Fatal(err) + } + } + if _, err := store.db.ExecContext(ctx, `UPDATE operation_task SET state = 'executing', lease_owner = 'worker-revoke', lease_until = now() + interval '1 minute' WHERE id = $1`, revokeExecuting.ID); err != nil { + t.Fatal(err) + } if err := store.RevokeAccount(ctx, "account-b"); err != nil { t.Fatal(err) } @@ -408,6 +457,8 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { if err != nil || revokedAccount.AuthorizationStatus != "revoked" || revokedAccount.Version != accountB.Version+2 { t.Fatalf("unexpected revoked account: %#v %v", revokedAccount, err) } + assertNotified("policy_hold", "account_revoked", "account-b", revokeQueued.ID) + assertNotified("needs_confirmation", "account_revoked", "account-b", revokeExecuting.ID) if err := store.ResumeAccount(ctx, "account-b"); !errors.Is(err, ErrConflict) { t.Fatalf("revoked account must not resume: %v", err) } @@ -436,6 +487,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { t.Fatalf("expired unknown result was retried: execution=%+v err=%v", execution, err) } assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-36' AND state = 'needs_confirmation'`, 1) + assertNotified("needs_confirmation", "execution_lease_expired", "account-a", "task-36") assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id = 'task-36' AND outcome = 'uncertain'`, 1) if err := store.ResumeTask(ctx, "task-36"); !errors.Is(err, ErrConflict) { t.Fatalf("unknown result resumed without verification: %v", err) @@ -482,6 +534,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { t.Fatalf("stale worker persisted a successful result: execution=%+v err=%v", completed, err) } assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-37' AND state = 'needs_confirmation' AND hold_reason = 'exit_unhealthy'`, 1) + assertNotified("needs_confirmation", "exit_unhealthy", "account-a", "task-37") if _, err := store.db.ExecContext(ctx, `UPDATE network_exit SET health_status = 'healthy' WHERE id = 'exit-shared'`); err != nil { t.Fatal(err) } @@ -532,6 +585,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { } assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-38' AND state = 'needs_confirmation' AND hold_reason = 'runtime_missing'`, 1) + assertNotified("needs_confirmation", "runtime_missing", "account-a", "task-38") concurrentDetail, err := store.GetTaskDetail(ctx, concurrent.ID) if err != nil || concurrentDetail.RuntimeInstanceID != "runtime-instance-a" { t.Fatalf("task detail lost its immutable claim runtime: detail=%+v err=%v", concurrentDetail, err) @@ -562,6 +616,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) { } assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-39' AND state = 'needs_confirmation' AND hold_reason = 'task_result_uncertain'`, 1) + assertNotified("needs_confirmation", "task_result_uncertain", "account-a", "task-39") replacedDetail, err := store.GetTaskDetail(ctx, replaced.ID) if err != nil || replacedDetail.AllowedAction != "verify" || replacedDetail.BrowserEnvAlias != "account-a" || replacedDetail.NetworkExitID != "exit-shared" || replacedDetail.RuntimeInstanceID != "runtime-instance-a2" || replacedDetail.BindingVersion != 1 { diff --git a/internal/taskstate/transition.go b/internal/taskstate/transition.go new file mode 100644 index 0000000..c14548c --- /dev/null +++ b/internal/taskstate/transition.go @@ -0,0 +1,7 @@ +package taskstate + +type Transition struct { + State, ReasonCode, AccountID, TaskID string +} + +type Notifier func(Transition)