diff --git a/Dockerfile b/Dockerfile index ae7678c..65c204f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -9,6 +9,7 @@ FROM golang:1.26-alpine@sha256:28d89ee9cc0ff9fec75c82ca201e6bf7fdf9a679d4b7b24df WORKDIR /src COPY go.mod go.sum ./ COPY cmd/ ./cmd/ +COPY internal/ ./internal/ RUN CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /out/control-plane ./cmd/control-plane \ && CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /out/docker-gateway ./cmd/docker-gateway diff --git a/README.md b/README.md index 515030d..83304b9 100644 --- a/README.md +++ b/README.md @@ -4,14 +4,14 @@ ## 当前阶段 -项目处于阶段 A 技术验证。首个可运行切片提供浏览器容器的创建、列表、启停和回收，功能参考公开上游项目独立实现，不直接复制其代码。 +项目处于阶段 A 技术验证。当前可运行切片提供浏览器容器控制，以及单用户、Mock 连接器的离线业务闭环，功能参考公开上游项目独立实现，不直接复制其代码。 详细的五领域优先级、系统边界、合规知识库与分阶段验证方案见 [《合规产品规划与上游调研》](docs/product/compliance-product-plan.md)。 ## 本地运行 -需要 Docker Compose。控制面仅绑定本机；浏览器网络在阶段 A 默认禁止外网访问： +需要 Docker Compose。控制面仅绑定本机；PostgreSQL 仅在内部 control 网络可见，浏览器网络在阶段 A 默认禁止外网访问： ```bash docker pull git.ipao.vip/rogee/fingerprint-chromium@sha256:b9f23b8e3ac640174db0dfa49e9095fe7eb06f5db55a4e7550d979b35ff3a1b7 diff --git a/cmd/control-plane/main.go b/cmd/control-plane/main.go index bdd8f91..42df587 100644 --- a/cmd/control-plane/main.go +++ b/cmd/control-plane/main.go @@ -15,6 +15,7 @@ import ( "syscall" "time" + "git.ipao.vip/rogee/creator-hub/internal/phasea" "github.com/gofiber/fiber/v3" "github.com/gofiber/fiber/v3/middleware/adaptor" "github.com/sirupsen/logrus" @@ -23,22 +24,27 @@ import ( ) type config struct { - listenAddr string - gateway *url.URL - webDir string - logLevel logrus.Level + listenAddr string + gateway *url.URL + webDir string + databaseURL string + logLevel logrus.Level } func main() { logrus.SetFormatter(&logrus.JSONFormatter{}) - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - defer stop() - if err := newCommand().ExecuteContext(ctx); err != nil { + if err := execute(newCommand()); err != nil { logrus.WithField("service", "control-plane").WithError(err).Error("service stopped") os.Exit(1) } } +func execute(command *cobra.Command) error { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + return command.ExecuteContext(ctx) +} + func newCommand() *cobra.Command { command := &cobra.Command{ Use: "control-plane", @@ -52,12 +58,13 @@ func newCommand() *cobra.Command { return err } logrus.SetLevel(cfg.logLevel) - logrus.WithFields(logrus.Fields{ - "service": "control-plane", - "listen_addr": cfg.listenAddr, - "docker_gateway": cfg.gateway.Scheme + "://" + cfg.gateway.Host, - }).Info("service starting") - return newHandler(cfg.gateway, cfg.webDir).Listen(cfg.listenAddr, fiber.ListenConfig{ + store, err := phasea.Open(command.Context(), cfg.databaseURL) + if err != nil { + return err + } + defer store.Close() + logStartup(cfg) + return newHandlerWithPhaseA(cfg.gateway, cfg.webDir, store).Listen(cfg.listenAddr, fiber.ListenConfig{ GracefulContext: command.Context(), DisableStartupMessage: true, }) @@ -66,15 +73,25 @@ func newCommand() *cobra.Command { return command } +func logStartup(cfg config) { + logrus.WithFields(logrus.Fields{ + "service": "control-plane", + "listen_addr": cfg.listenAddr, + "docker_gateway": cfg.gateway.Scheme + "://" + cfg.gateway.Host, + }).Info("service starting") +} + func loadConfig() (config, error) { v := viper.New() v.SetDefault("listen_addr", ":8080") v.SetDefault("docker_gateway_url", "http://127.0.0.1:8081") v.SetDefault("web_dir", "web/dist") + v.SetDefault("database_url", "postgres://creatorhub@127.0.0.1:5432/creatorhub?sslmode=disable") v.SetDefault("log_level", "info") _ = v.BindEnv("listen_addr", "LISTEN_ADDR") _ = v.BindEnv("docker_gateway_url", "DOCKER_GATEWAY_URL") _ = v.BindEnv("web_dir", "WEB_DIR") + _ = v.BindEnv("database_url", "DATABASE_URL") _ = v.BindEnv("log_level", "LOG_LEVEL") gateway, err := url.Parse(v.GetString("docker_gateway_url")) @@ -86,10 +103,11 @@ func loadConfig() (config, error) { return config{}, errors.New("LOG_LEVEL must be panic, fatal, error, warn, info, debug, or trace") } cfg := config{ - listenAddr: strings.TrimSpace(v.GetString("listen_addr")), - gateway: gateway, - webDir: strings.TrimSpace(v.GetString("web_dir")), - logLevel: level, + listenAddr: strings.TrimSpace(v.GetString("listen_addr")), + gateway: gateway, + webDir: strings.TrimSpace(v.GetString("web_dir")), + databaseURL: strings.TrimSpace(v.GetString("database_url")), + logLevel: level, } if cfg.listenAddr == "" { return config{}, errors.New("LISTEN_ADDR must not be empty") @@ -100,6 +118,11 @@ func loadConfig() (config, error) { if cfg.webDir == "" { return config{}, errors.New("WEB_DIR must not be empty") } + databaseURL, err := url.Parse(cfg.databaseURL) + if err != nil || databaseURL.Host == "" || databaseURL.Path == "" || databaseURL.Path == "/" || + (databaseURL.Scheme != "postgres" && databaseURL.Scheme != "postgresql") { + return config{}, errors.New("DATABASE_URL must be a postgres URL with a host") + } return cfg, nil } @@ -116,6 +139,10 @@ func validateListenAddr(addr string) error { } func newHandler(gateway *url.URL, webDirectory string) *fiber.App { + return newHandlerWithPhaseA(gateway, webDirectory, nil) +} + +func newHandlerWithPhaseA(gateway *url.URL, webDirectory string, store *phasea.Store) *fiber.App { proxy := httputil.NewSingleHostReverseProxy(gateway) originalDirector := proxy.Director proxy.Director = func(request *http.Request) { @@ -127,6 +154,7 @@ func newHandler(gateway *url.URL, webDirectory string) *fiber.App { app := fiber.New(fiber.Config{ AppName: "CreatorHub control plane", + BodyLimit: 1 << 20, ReadTimeout: 5 * time.Second, IdleTimeout: 60 * time.Second, }) @@ -136,6 +164,9 @@ func newHandler(gateway *url.URL, webDirectory string) *fiber.App { }) app.All("/api/browsers", proxyHandler) app.All("/api/browsers/*", proxyHandler) + if store != nil { + registerPhaseA(app, store) + } app.Get("/*", spaHandler(webDirectory)) return app } diff --git a/cmd/control-plane/main_test.go b/cmd/control-plane/main_test.go index caa7bfe..3f2c57f 100644 --- a/cmd/control-plane/main_test.go +++ b/cmd/control-plane/main_test.go @@ -1,13 +1,26 @@ package main import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" "io" "net/http" "net/http/httptest" "net/url" "os" + "os/exec" "path/filepath" + "strings" + "syscall" "testing" + "time" + + "github.com/gofiber/fiber/v3" + "github.com/sirupsen/logrus" + "github.com/spf13/cobra" ) func TestAPIProxyPreservesMethodPathAndStatus(t *testing.T) { @@ -45,6 +58,122 @@ func TestLoadConfigRejectsInvalidGateway(t *testing.T) { } } +func TestLoadConfigRejectsInvalidDatabase(t *testing.T) { + t.Setenv("DATABASE_URL", "file:///tmp/creatorhub.db") + if _, err := loadConfig(); err == nil { + t.Fatal("expected non-Postgres database URL to be rejected") + } +} + +func TestExecuteContextStopsOnSIGTERM(t *testing.T) { + if os.Getenv("CREATORHUB_SIGTERM_HELPER") == "1" { + command := &cobra.Command{Use: "shutdown-test", RunE: func(command *cobra.Command, _ []string) error { + _, _ = os.Stdout.WriteString("ready\n") + <-command.Context().Done() + _, _ = os.Stdout.WriteString("stopped\n") + return nil + }} + if err := execute(command); err != nil { + t.Fatal(err) + } + return + } + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + process := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestExecuteContextStopsOnSIGTERM$") + process.Env = append(os.Environ(), "CREATORHUB_SIGTERM_HELPER=1") + stdout, err := process.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := process.Start(); err != nil { + t.Fatal(err) + } + scanner := bufio.NewScanner(stdout) + if !scanner.Scan() || scanner.Text() != "ready" { + t.Fatalf("helper did not become ready: %q err=%v", scanner.Text(), scanner.Err()) + } + if err := process.Process.Signal(syscall.SIGTERM); err != nil { + t.Fatal(err) + } + if !scanner.Scan() || scanner.Text() != "stopped" { + t.Fatalf("helper did not finish cleanup: %q err=%v", scanner.Text(), scanner.Err()) + } + if err := process.Wait(); err != nil || ctx.Err() != nil { + t.Fatalf("process did not exit gracefully after SIGTERM: wait=%v context=%v", err, ctx.Err()) + } +} + +func TestStartupLogRedactsGatewayURL(t *testing.T) { + gateway, err := url.Parse("https://internal-user:internal-pass@docker-gateway:8443/v1?access_token=internal-token") + if err != nil { + t.Fatal(err) + } + logger := logrus.StandardLogger() + previousOutput, previousFormatter, previousLevel := logger.Out, logger.Formatter, logger.Level + t.Cleanup(func() { + logrus.SetOutput(previousOutput) + logrus.SetFormatter(previousFormatter) + logrus.SetLevel(previousLevel) + }) + var output bytes.Buffer + logrus.SetOutput(&output) + logrus.SetFormatter(&logrus.JSONFormatter{}) + logrus.SetLevel(logrus.InfoLevel) + + logStartup(config{listenAddr: ":8080", gateway: gateway}) + var entry map[string]any + if err := json.Unmarshal(output.Bytes(), &entry); err != nil { + t.Fatal(err) + } + if entry["docker_gateway"] != "https://docker-gateway:8443" { + t.Fatalf("unexpected redacted gateway: %#v", entry["docker_gateway"]) + } + for _, sensitive := range []string{"internal-user", "internal-pass", "access_token", "internal-token"} { + if strings.Contains(output.String(), sensitive) { + t.Fatalf("startup log leaked %q: %s", sensitive, output.String()) + } + } +} + +func TestPhaseARequestRejectsUnknownFields(t *testing.T) { + app := fiber.New() + app.Post("/", func(c fiber.Ctx) error { + var input struct { + ID string `json:"id"` + } + if err := decodePhaseA(c, &input); err != nil { + return phaseAError(c, err) + } + return c.SendStatus(http.StatusNoContent) + }) + response, err := app.Test(httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id":"account-a","password":"must-not-be-accepted"}`))) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusBadRequest { + t.Fatalf("expected unknown sensitive field to be rejected, got %d", response.StatusCode) + } +} + +func TestPhaseAErrorRedactsInternalDetails(t *testing.T) { + app := fiber.New() + app.Get("/", func(c fiber.Ctx) error { + return phaseAError(c, errors.New("database exposed a secret value")) + }) + response, err := app.Test(httptest.NewRequest(http.MethodGet, "/", nil)) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, _ := io.ReadAll(response.Body) + if response.StatusCode != http.StatusInternalServerError || strings.Contains(string(body), "secret") || !strings.Contains(string(body), "phase A operation failed") { + t.Fatalf("internal error was not redacted: status=%d body=%s", response.StatusCode, body) + } +} + func TestSPAServesFileAndIndexFallback(t *testing.T) { directory := t.TempDir() files := map[string]string{ diff --git a/cmd/control-plane/phasea.go b/cmd/control-plane/phasea.go new file mode 100644 index 0000000..8817238 --- /dev/null +++ b/cmd/control-plane/phasea.go @@ -0,0 +1,157 @@ +package main + +import ( + "bytes" + "encoding/json" + "errors" + "io" + + "git.ipao.vip/rogee/creator-hub/internal/phasea" + "github.com/gofiber/fiber/v3" +) + +type accountRequest struct { + ID string `json:"id"` + ProfileID string `json:"profile_id"` + CredentialReference struct { + ID string `json:"id"` + Provider string `json:"provider"` + Key string `json:"key"` + } `json:"credential_reference"` +} + +func registerPhaseA(app *fiber.App, store *phasea.Store) { + app.Post("/api/phase-a/accounts", func(c fiber.Ctx) error { + var input accountRequest + if err := decodePhaseA(c, &input); err != nil { + return phaseAError(c, err) + } + err := store.CreateAccount(c.Context(), phasea.Account{ + ID: input.ID, ProfileID: input.ProfileID, CredentialReferenceID: input.CredentialReference.ID, + CredentialProvider: input.CredentialReference.Provider, CredentialKey: input.CredentialReference.Key, + }) + if err != nil { + return phaseAError(c, err) + } + return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID, "profile_id": input.ProfileID}) + }) + + app.Post("/api/phase-a/accounts/:id/pause", func(c fiber.Ctx) error { + if err := store.PauseAccount(c.Context(), c.Params("id")); err != nil { + return phaseAError(c, err) + } + return c.SendStatus(fiber.StatusNoContent) + }) + + app.Post("/api/phase-a/runtimes", func(c fiber.Ctx) error { + var input phasea.RuntimeBinding + if err := decodePhaseA(c, &input); err != nil { + return phaseAError(c, err) + } + if err := store.BindRuntime(c.Context(), input); err != nil { + return phaseAError(c, err) + } + return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID}) + }) + + app.Delete("/api/phase-a/runtimes/:id", func(c fiber.Ctx) error { + if err := store.ReleaseRuntime(c.Context(), c.Params("id")); err != nil { + return phaseAError(c, err) + } + return c.SendStatus(fiber.StatusNoContent) + }) + + app.Post("/api/phase-a/drafts", func(c fiber.Ctx) error { + var input phasea.Draft + if err := decodePhaseA(c, &input); err != nil { + return phaseAError(c, err) + } + if err := store.CreateDraft(c.Context(), input); err != nil { + return phaseAError(c, err) + } + return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID}) + }) + + app.Post("/api/phase-a/confirmations", func(c fiber.Ctx) error { + var input phasea.Confirmation + if err := decodePhaseA(c, &input); err != nil { + return phaseAError(c, err) + } + if err := store.Confirm(c.Context(), input); err != nil { + return phaseAError(c, err) + } + return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID}) + }) + + app.Post("/api/phase-a/tasks", func(c fiber.Ctx) error { + var input phasea.Task + if err := decodePhaseA(c, &input); err != nil { + return phaseAError(c, err) + } + task, inserted, err := store.Enqueue(c.Context(), input) + if err != nil { + return phaseAError(c, err) + } + status := fiber.StatusOK + if inserted { + status = fiber.StatusCreated + } + return c.Status(status).JSON(task) + }) + + app.Post("/api/phase-a/tasks/:id/cancel", func(c fiber.Ctx) error { + if err := store.CancelTask(c.Context(), c.Params("id")); err != nil { + return phaseAError(c, err) + } + return c.SendStatus(fiber.StatusNoContent) + }) + + app.Post("/api/phase-a/mock/execute", func(c fiber.Ctx) error { + var input struct { + WorkerID string `json:"worker_id"` + Outcome string `json:"outcome"` + } + if err := decodePhaseA(c, &input); err != nil { + return phaseAError(c, err) + } + execution, err := store.ExecuteMock(c.Context(), input.WorkerID, input.Outcome) + if err != nil { + return phaseAError(c, err) + } + return c.JSON(execution) + }) + + app.Get("/api/phase-a/audit", func(c fiber.Ctx) error { + events, err := store.Audit(c.Context()) + if err != nil { + return phaseAError(c, err) + } + return c.JSON(events) + }) +} + +func decodePhaseA(c fiber.Ctx, destination any) error { + decoder := json.NewDecoder(bytes.NewReader(c.Body())) + decoder.DisallowUnknownFields() + if err := decoder.Decode(destination); err != nil { + return phasea.ErrInvalid + } + if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) { + return phasea.ErrInvalid + } + return nil +} + +func phaseAError(c fiber.Ctx, err error) error { + status := fiber.StatusInternalServerError + message := "phase A operation failed" + switch { + case errors.Is(err, phasea.ErrInvalid): + status, message = fiber.StatusBadRequest, phasea.ErrInvalid.Error() + case errors.Is(err, phasea.ErrConflict): + status, message = fiber.StatusConflict, phasea.ErrConflict.Error() + case errors.Is(err, phasea.ErrNotFound): + status, message = fiber.StatusNotFound, phasea.ErrNotFound.Error() + } + return c.Status(status).JSON(map[string]string{"error": message}) +} diff --git a/compose.yaml b/compose.yaml index d17fbdc..962b6b0 100644 --- a/compose.yaml +++ b/compose.yaml @@ -3,6 +3,7 @@ services: build: . environment: DOCKER_GATEWAY_URL: http://docker-gateway:8081 + DATABASE_URL: postgres://creatorhub@postgres/creatorhub?sslmode=disable ports: - "127.0.0.1:${CREATORHUB_PORT:-8080}:8080" read_only: true @@ -10,7 +11,34 @@ services: - /tmp:size=16m,noexec,nosuid,nodev cap_drop: [ALL] security_opt: [no-new-privileges:true] - depends_on: [docker-gateway] + depends_on: + docker-gateway: + condition: service_started + postgres: + condition: service_healthy + networks: [control] + restart: unless-stopped + + postgres: + image: postgres:17-alpine@sha256:18cfe3ef5e6815560c98237d6216d1e5119702fb0f3894c8785dd58b8bbe5d73 + environment: + POSTGRES_DB: creatorhub + POSTGRES_USER: creatorhub + POSTGRES_HOST_AUTH_METHOD: trust + healthcheck: + test: [CMD-SHELL, pg_isready -h 127.0.0.1 -U creatorhub -d creatorhub] + interval: 2s + timeout: 2s + retries: 15 + volumes: + - creatorhub_postgres:/var/lib/postgresql/data + read_only: true + tmpfs: + - /run/postgresql:size=1m,nosuid,nodev + - /tmp:size=16m,nosuid,nodev + cap_drop: [ALL] + cap_add: [CHOWN, FOWNER, DAC_OVERRIDE, SETUID, SETGID] + security_opt: [no-new-privileges:true] networks: [control] restart: unless-stopped @@ -35,3 +63,6 @@ networks: control: name: creatorhub_control internal: true + +volumes: + creatorhub_postgres: diff --git a/docs/architecture/container-control.md b/docs/architecture/container-control.md index 9480a1b..854a8dd 100644 --- a/docs/architecture/container-control.md +++ b/docs/architecture/container-control.md @@ -4,15 +4,16 @@ - 前端：React + Vite。阶段 A 只有一个运行环境页面，不引入路由、状态库或组件库。 - 后端：Go 模块化单体，Fiber v3 提供 HTTP 路由，Viper 读取并校验启动配置，Logrus 输出 JSON 结构化日志，Cobra 保持当前两个服务入口。控制面提供同源 API 和静态文件，受限网关单独封装 Docker Engine API。 -- 数据：本任务不引入业务数据库；Docker 容器标签是运行态事实，Profile 使用命名卷持久化。账号、任务和审计实体仍按产品规划在后续阶段落入 Postgres。 +- 数据：Docker 容器标签仍是运行态事实，Profile 使用命名卷持久化；阶段 A 账号、凭据引用、确认、任务、尝试和审计实体由控制面使用原生 SQL 持久化到 Postgres。 - 部署：Docker Compose 启动控制面和受限网关；浏览器容器由网关动态创建。 ## 调用链与契约 ```text -React ── /api/browsers ──> control-plane ── /v1/browsers ──> docker-gateway ──> docker.sock - │ - └─> browser container +React ──> control-plane ── /api/browsers ──> docker-gateway ──> docker.sock + │ │ + │ └─> browser container + └─ /api/phase-a ──> PostgreSQL ``` `POST /api/browsers` 接受 `{name, seed}` 并创建后启动；`GET /api/browsers` 列出受管容器；`POST /api/browsers/{name}/start|stop` 改变状态；`DELETE /api/browsers/{name}` 仅回收容器，保留 Profile 数据卷。名称唯一约束由 Docker 容器名保证，重复创建返回冲突；启停接受 Docker 的幂等响应，不自动重试未知结果。 @@ -43,3 +44,11 @@ DOCKER_GID=$(stat -c %g /var/run/docker.sock) docker compose up --build 打开 。浏览器镜像固定为 `git.ipao.vip/rogee/fingerprint-chromium@sha256:b9f23b8e3ac640174db0dfa49e9095fe7eb06f5db55a4e7550d979b35ff3a1b7`。阶段 A 网络默认失败关闭且不能访问外网；后续授权沙箱必须先设计受管出口，再调整 `browser` 网络。 创建成功但启动失败时，网关会立即删除失败容器并保留命名 Profile 卷，允许同名请求安全重试。跨容器隔离回归可用 `CREATORHUB_DOCKER_INTEGRATION=1 go test -run TestDockerIntegrationBlocksCrossContainerTraffic ./cmd/docker-gateway` 运行。 + +## 阶段 A 离线闭环 + +`POST /api/phase-a/accounts` 只接受 OS Keyring/Secret Manager 的引用标识，不接受秘密值；账号的 `profile_id` 全局唯一。`POST /api/phase-a/runtimes` 通过部分唯一索引保证一个账号和一个运行时都只有一条活动绑定。 + +草稿经 `POST /api/phase-a/confirmations` 显式确认后才可投递到 `/api/phase-a/tasks`。任务由幂等键去重；`POST /api/phase-a/mock/execute` 使用 `FOR UPDATE SKIP LOCKED` 领取一分钟租约，执行前统一核对账号、草稿和确认版本。缺少确认或版本不一致会进入 `needs_confirmation`，暂停账号或 Mock 策略结果会进入 `policy_hold`，不确定结果与过期租约进入 `needs_confirmation`；这些状态都不会自动重试。`GET /api/phase-a/audit` 只导出账号、确认版本、尝试和结果等非秘密证据。 + +启动时控制面在事务和 advisory lock 下应用前向迁移 `internal/phasea/migrations/001_phase_a.sql`。本迁移只新建表、索引、约束和追加式审计触发器，不删除或改写现有数据；回滚需停服务后人工删除阶段 A 新表，本阶段不提供自动破坏性回滚。 diff --git a/go.mod b/go.mod index 4b75c8a..f008678 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.26 require ( github.com/gofiber/fiber/v3 v3.5.0 + github.com/jackc/pgx/v5 v5.10.0 github.com/sirupsen/logrus v1.10.0 github.com/spf13/cobra v1.10.2 github.com/spf13/viper v1.21.0 @@ -17,6 +18,9 @@ require ( github.com/gofiber/utils/v2 v2.4.1 // indirect github.com/google/uuid v1.6.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/klauspost/compress v1.19.2 // indirect github.com/mattn/go-colorable v0.1.15 // indirect github.com/mattn/go-isatty v0.0.24 // indirect @@ -34,6 +38,7 @@ require ( go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/crypto v0.54.0 // indirect golang.org/x/net v0.57.0 // indirect + golang.org/x/sync v0.22.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.40.0 // indirect ) diff --git a/go.sum b/go.sum index c088c85..f13eae1 100644 --- a/go.sum +++ b/go.sum @@ -1,6 +1,7 @@ github.com/andybalholm/brotli v1.2.2 h1:HzTuoo2ErYQqf5qvcJInB8uvqSVxRttzkFexPWtnceM= github.com/andybalholm/brotli v1.2.2/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= @@ -23,6 +24,14 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -61,6 +70,9 @@ github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/spf13/viper v1.21.0 h1:x5S+0EU27Lbphp4UKm1C+1oQO+rKx36vfCoaVebLFSU= github.com/spf13/viper v1.21.0/go.mod h1:P0lhsswPGWD/1lZJ9ny3fYnVqxiegrlNrEmgLjbTCAY= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= @@ -81,12 +93,15 @@ golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= -gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/phasea/migrations/001_phase_a.sql b/internal/phasea/migrations/001_phase_a.sql new file mode 100644 index 0000000..42121d4 --- /dev/null +++ b/internal/phasea/migrations/001_phase_a.sql @@ -0,0 +1,107 @@ +CREATE TABLE credential_reference ( + id text PRIMARY KEY, + provider text NOT NULL CHECK (provider IN ('os_keyring', 'secret_manager')), + reference_key text NOT NULL UNIQUE, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE social_account ( + id text PRIMARY KEY CHECK (id ~ '^[a-z0-9][a-z0-9-]{0,31}$'), + credential_reference_id text NOT NULL UNIQUE REFERENCES credential_reference(id), + profile_id text NOT NULL UNIQUE, + version bigint NOT NULL DEFAULT 1 CHECK (version > 0), + status text NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'paused')), + paused_at timestamptz, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE runtime_instance ( + id text PRIMARY KEY, + account_id text NOT NULL REFERENCES social_account(id), + runtime_id text NOT NULL, + acquired_at timestamptz NOT NULL DEFAULT now(), + lease_until timestamptz NOT NULL, + released_at timestamptz +); + +CREATE UNIQUE INDEX one_active_runtime_per_account + ON runtime_instance (account_id) WHERE released_at IS NULL; +CREATE UNIQUE INDEX one_account_per_active_runtime + ON runtime_instance (runtime_id) WHERE released_at IS NULL; + +CREATE TABLE content_draft ( + id text PRIMARY KEY, + account_id text NOT NULL REFERENCES social_account(id), + version bigint NOT NULL CHECK (version > 0), + content text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE confirmation ( + id text PRIMARY KEY, + account_id text NOT NULL REFERENCES social_account(id), + account_version bigint NOT NULL CHECK (account_version > 0), + draft_id text NOT NULL REFERENCES content_draft(id), + draft_version bigint NOT NULL CHECK (draft_version > 0), + version bigint NOT NULL CHECK (version > 0), + confirmed_by text NOT NULL DEFAULT 'local-user' CHECK (confirmed_by = 'local-user'), + confirmed_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE operation_task ( + id text PRIMARY KEY, + idempotency_key text NOT NULL UNIQUE, + account_id text NOT NULL REFERENCES social_account(id), + account_version bigint NOT NULL CHECK (account_version > 0), + draft_id text NOT NULL REFERENCES content_draft(id), + draft_version bigint NOT NULL CHECK (draft_version > 0), + confirmation_id text REFERENCES confirmation(id), + confirmation_version bigint CHECK (confirmation_version > 0), + state text NOT NULL DEFAULT 'queued' CHECK (state IN ( + 'queued', 'executing', 'succeeded', 'failed', + 'needs_confirmation', 'policy_hold', 'cancelled' + )), + lease_owner text, + lease_until timestamptz, + current_attempt_id text, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + CHECK ((confirmation_id IS NULL) = (confirmation_version IS NULL)) +); + +CREATE TABLE execution_attempt ( + id text PRIMARY KEY, + task_id text NOT NULL UNIQUE REFERENCES operation_task(id), + started_at timestamptz NOT NULL DEFAULT now(), + finished_at timestamptz, + outcome text CHECK (outcome IN ('succeeded', 'failed', 'uncertain', 'policy_hold')), + result jsonb NOT NULL DEFAULT '{}'::jsonb +); + +ALTER TABLE operation_task + ADD CONSTRAINT operation_task_current_attempt_fk + FOREIGN KEY (current_attempt_id) REFERENCES execution_attempt(id); + +CREATE TABLE audit_event ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + event_type text NOT NULL, + account_id text REFERENCES social_account(id), + confirmation_id text REFERENCES confirmation(id), + confirmation_version bigint, + attempt_id text REFERENCES execution_attempt(id), + task_id text REFERENCES operation_task(id), + details jsonb NOT NULL DEFAULT '{}'::jsonb, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE FUNCTION reject_audit_event_mutation() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + RAISE EXCEPTION 'audit_event is append-only'; +END; +$$; + +CREATE TRIGGER audit_event_append_only + BEFORE UPDATE OR DELETE ON audit_event + FOR EACH ROW EXECUTE FUNCTION reject_audit_event_mutation(); diff --git a/internal/phasea/store.go b/internal/phasea/store.go new file mode 100644 index 0000000..79fff72 --- /dev/null +++ b/internal/phasea/store.go @@ -0,0 +1,658 @@ +package phasea + +import ( + "context" + "crypto/rand" + "database/sql" + _ "embed" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "regexp" + "strings" + "time" + + "github.com/jackc/pgx/v5/pgconn" + _ "github.com/jackc/pgx/v5/stdlib" +) + +//go:embed migrations/001_phase_a.sql +var migration001 string + +var ( + ErrConflict = errors.New("resource conflicts with existing state") + ErrInvalid = errors.New("invalid phase A input") + ErrNotFound = errors.New("resource not found") + idPattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`) + refPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$`) + credentialKeyPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}/[A-Za-z0-9][A-Za-z0-9._/-]{0,126}$`) +) + +type Store struct{ db *sql.DB } + +type Account struct { + ID string `json:"id"` + CredentialReferenceID string `json:"credential_reference_id"` + CredentialProvider string `json:"credential_provider"` + CredentialKey string `json:"-"` + ProfileID string `json:"profile_id"` +} + +type RuntimeBinding struct { + ID string `json:"id"` + AccountID string `json:"account_id"` + RuntimeID string `json:"runtime_id"` + LeaseUntil time.Time `json:"lease_until"` +} + +type Draft struct { + ID string `json:"id"` + AccountID string `json:"account_id"` + Version int64 `json:"version"` + Content string `json:"content"` +} + +type Confirmation struct { + ID string `json:"id"` + AccountID string `json:"account_id"` + AccountVersion int64 `json:"account_version"` + DraftID string `json:"draft_id"` + DraftVersion int64 `json:"draft_version"` + Version int64 `json:"version"` +} + +type Task struct { + ID string `json:"id"` + IdempotencyKey string `json:"idempotency_key"` + AccountID string `json:"account_id"` + AccountVersion int64 `json:"account_version"` + DraftID string `json:"draft_id"` + DraftVersion int64 `json:"draft_version"` + ConfirmationID string `json:"confirmation_id"` + ConfirmationVersion int64 `json:"confirmation_version"` + State string `json:"state"` +} + +type Execution struct { + TaskID string `json:"task_id"` + AttemptID string `json:"attempt_id"` + AccountID string `json:"account_id"` + ConfirmationID string `json:"confirmation_id"` + ConfirmationVersion int64 `json:"confirmation_version"` + State string `json:"state"` + WasClaimed bool `json:"was_claimed"` +} + +type AuditEvent struct { + ID int64 `json:"id"` + EventType string `json:"event_type"` + AccountID string `json:"account_id,omitempty"` + ConfirmationID string `json:"confirmation_id,omitempty"` + ConfirmationVersion int64 `json:"confirmation_version,omitempty"` + AttemptID string `json:"attempt_id,omitempty"` + TaskID string `json:"task_id,omitempty"` + Details json.RawMessage `json:"details"` + CreatedAt time.Time `json:"created_at"` +} + +func Open(ctx context.Context, databaseURL string) (*Store, error) { + db, err := sql.Open("pgx", databaseURL) + if err != nil { + return nil, errors.New("open phase A database") + } + db.SetMaxOpenConns(10) + db.SetMaxIdleConns(2) + db.SetConnMaxIdleTime(5 * time.Minute) + if err := db.PingContext(ctx); err != nil { + db.Close() + return nil, errors.New("connect to phase A database") + } + store := &Store{db: db} + if err := store.migrate(ctx); err != nil { + db.Close() + return nil, err + } + return store, nil +} + +func (s *Store) Close() error { return s.db.Close() } + +func (s *Store) migrate(ctx context.Context) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return errors.New("begin schema migration") + } + defer tx.Rollback() + if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(1542738011)`); err != nil { + return errors.New("lock schema migration") + } + if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migration (version integer PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`); err != nil { + return errors.New("create schema migration table") + } + var applied bool + if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = 1)`).Scan(&applied); err != nil { + return errors.New("read schema migration state") + } + if !applied { + if _, err := tx.ExecContext(ctx, migration001); err != nil { + return fmt.Errorf("apply schema migration 1: %w", err) + } + if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migration (version) VALUES (1)`); err != nil { + return errors.New("record schema migration 1") + } + } + if err := tx.Commit(); err != nil { + return errors.New("commit schema migration") + } + return nil +} + +func (s *Store) CreateAccount(ctx context.Context, account Account) error { + if !idPattern.MatchString(account.ID) || !refPattern.MatchString(account.CredentialReferenceID) || + !credentialKeyPattern.MatchString(account.CredentialKey) || !refPattern.MatchString(account.ProfileID) || + (account.CredentialProvider != "os_keyring" && account.CredentialProvider != "secret_manager") { + return ErrInvalid + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return errors.New("begin account transaction") + } + defer tx.Rollback() + if _, err := tx.ExecContext(ctx, `INSERT INTO credential_reference (id, provider, reference_key) VALUES ($1, $2, $3)`, + account.CredentialReferenceID, account.CredentialProvider, account.CredentialKey); err != nil { + return publicDatabaseError(err) + } + if _, err := tx.ExecContext(ctx, `INSERT INTO social_account (id, credential_reference_id, profile_id) VALUES ($1, $2, $3)`, + account.ID, account.CredentialReferenceID, account.ProfileID); err != nil { + return publicDatabaseError(err) + } + if err := appendAudit(ctx, tx, "account_created", account.ID, "", 0, "", "", map[string]string{"profile_id": account.ProfileID}); err != nil { + return err + } + return commit(tx) +} + +func (s *Store) BindRuntime(ctx context.Context, binding RuntimeBinding) error { + if !refPattern.MatchString(binding.ID) || !idPattern.MatchString(binding.AccountID) || + !refPattern.MatchString(binding.RuntimeID) || !binding.LeaseUntil.After(time.Now()) { + return ErrInvalid + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return errors.New("begin runtime transaction") + } + defer tx.Rollback() + if _, err := tx.ExecContext(ctx, `UPDATE runtime_instance SET released_at = now() WHERE released_at IS NULL AND lease_until < now()`); err != nil { + return errors.New("expire runtime bindings") + } + var accountID string + if err := tx.QueryRowContext(ctx, ` + INSERT INTO runtime_instance (id, account_id, runtime_id, lease_until) + SELECT $1, id, $3, $4 FROM social_account WHERE id = $2 AND status = 'active' + RETURNING account_id`, binding.ID, binding.AccountID, binding.RuntimeID, binding.LeaseUntil).Scan(&accountID); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return ErrConflict + } + return rowError(err) + } + if err := appendAudit(ctx, tx, "runtime_bound", accountID, "", 0, "", "", map[string]string{"runtime_id": binding.RuntimeID}); err != nil { + return err + } + return commit(tx) +} + +func (s *Store) ReleaseRuntime(ctx context.Context, id string) error { + if !refPattern.MatchString(id) { + return ErrInvalid + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return errors.New("begin runtime release") + } + defer tx.Rollback() + var accountID string + if err := tx.QueryRowContext(ctx, `UPDATE runtime_instance SET released_at = now() WHERE id = $1 AND released_at IS NULL RETURNING account_id`, id).Scan(&accountID); err != nil { + return rowError(err) + } + if err := appendAudit(ctx, tx, "runtime_released", accountID, "", 0, "", "", map[string]string{"binding_id": id}); err != nil { + return err + } + return commit(tx) +} + +func (s *Store) CreateDraft(ctx context.Context, draft Draft) error { + if !refPattern.MatchString(draft.ID) || !idPattern.MatchString(draft.AccountID) || draft.Version < 1 || strings.TrimSpace(draft.Content) == "" { + return ErrInvalid + } + _, err := s.db.ExecContext(ctx, `INSERT INTO content_draft (id, account_id, version, content) VALUES ($1, $2, $3, $4)`, + draft.ID, draft.AccountID, draft.Version, draft.Content) + return publicDatabaseError(err) +} + +func (s *Store) Confirm(ctx context.Context, confirmation Confirmation) error { + if !refPattern.MatchString(confirmation.ID) || !idPattern.MatchString(confirmation.AccountID) || + !refPattern.MatchString(confirmation.DraftID) || confirmation.AccountVersion < 1 || confirmation.DraftVersion < 1 || confirmation.Version < 1 { + return ErrInvalid + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return errors.New("begin confirmation transaction") + } + defer tx.Rollback() + var valid bool + if err := tx.QueryRowContext(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM social_account a JOIN content_draft d ON d.account_id = a.id + WHERE a.id = $1 AND a.version = $2 AND d.id = $3 AND d.version = $4 + )`, confirmation.AccountID, confirmation.AccountVersion, confirmation.DraftID, confirmation.DraftVersion).Scan(&valid); err != nil { + return errors.New("validate confirmation versions") + } + if !valid { + return ErrConflict + } + if _, err := tx.ExecContext(ctx, ` + INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version) + VALUES ($1, $2, $3, $4, $5, $6)`, confirmation.ID, confirmation.AccountID, confirmation.AccountVersion, + confirmation.DraftID, confirmation.DraftVersion, confirmation.Version); err != nil { + return publicDatabaseError(err) + } + if err := appendAudit(ctx, tx, "draft_confirmed", confirmation.AccountID, confirmation.ID, confirmation.Version, "", "", map[string]any{ + "account_version": confirmation.AccountVersion, "draft_id": confirmation.DraftID, "draft_version": confirmation.DraftVersion, + }); err != nil { + return err + } + return commit(tx) +} + +func (s *Store) Enqueue(ctx context.Context, task Task) (Task, bool, error) { + if !refPattern.MatchString(task.ID) || !refPattern.MatchString(task.IdempotencyKey) || !idPattern.MatchString(task.AccountID) || + !refPattern.MatchString(task.DraftID) || task.AccountVersion < 1 || task.DraftVersion < 1 || + ((task.ConfirmationID == "") != (task.ConfirmationVersion == 0)) || + (task.ConfirmationID != "" && (!refPattern.MatchString(task.ConfirmationID) || task.ConfirmationVersion < 1)) { + return Task{}, false, ErrInvalid + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return Task{}, false, errors.New("begin task transaction") + } + defer tx.Rollback() + var insertedID string + err = tx.QueryRowContext(ctx, ` + INSERT INTO operation_task (id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version) + VALUES ($1, $2, $3, $4, $5, $6, NULLIF($7, ''), NULLIF($8, 0)) + ON CONFLICT (idempotency_key) DO NOTHING RETURNING id`, task.ID, task.IdempotencyKey, task.AccountID, task.AccountVersion, + task.DraftID, task.DraftVersion, task.ConfirmationID, task.ConfirmationVersion).Scan(&insertedID) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return Task{}, false, publicDatabaseError(err) + } + if insertedID != "" { + task.State = "queued" + if err := appendAudit(ctx, tx, "task_queued", task.AccountID, task.ConfirmationID, task.ConfirmationVersion, "", task.ID, nil); err != nil { + return Task{}, false, err + } + if err := commit(tx); err != nil { + return Task{}, false, err + } + return task, true, nil + } + + var existing Task + var confirmationID sql.NullString + var confirmationVersion sql.NullInt64 + err = tx.QueryRowContext(ctx, ` + SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version, state + FROM operation_task WHERE idempotency_key = $1`, task.IdempotencyKey).Scan(&existing.ID, &existing.IdempotencyKey, + &existing.AccountID, &existing.AccountVersion, &existing.DraftID, &existing.DraftVersion, &confirmationID, &confirmationVersion, &existing.State) + if err != nil { + return Task{}, false, rowError(err) + } + existing.ConfirmationID = confirmationID.String + existing.ConfirmationVersion = confirmationVersion.Int64 + if existing.AccountID != task.AccountID || existing.AccountVersion != task.AccountVersion || existing.DraftID != task.DraftID || + existing.DraftVersion != task.DraftVersion || existing.ConfirmationID != task.ConfirmationID || existing.ConfirmationVersion != task.ConfirmationVersion { + return Task{}, false, ErrConflict + } + if err := commit(tx); err != nil { + return Task{}, false, err + } + return existing, false, nil +} + +func (s *Store) PauseAccount(ctx context.Context, accountID string) error { + if !idPattern.MatchString(accountID) { + return ErrInvalid + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return errors.New("begin pause transaction") + } + defer tx.Rollback() + var version int64 + var status string + if err := tx.QueryRowContext(ctx, `SELECT version, status FROM social_account WHERE id = $1 FOR UPDATE`, accountID).Scan(&version, &status); err != nil { + return rowError(err) + } + if status == "paused" { + return commit(tx) + } + if err := tx.QueryRowContext(ctx, ` + UPDATE social_account SET status = 'paused', paused_at = now(), version = version + 1, updated_at = now() + WHERE id = $1 RETURNING version`, accountID).Scan(&version); err != nil { + return errors.New("pause account") + } + result, err := tx.ExecContext(ctx, ` + UPDATE operation_task SET state = 'policy_hold', updated_at = now() + WHERE account_id = $1 AND state = 'queued'`, accountID) + if err != nil { + return errors.New("hold queued account tasks") + } + held, _ := result.RowsAffected() + if err := appendAudit(ctx, tx, "account_paused", accountID, "", 0, "", "", map[string]any{"account_version": version, "tasks_held": held}); err != nil { + return err + } + return commit(tx) +} + +func (s *Store) CancelTask(ctx context.Context, taskID string) error { + if !refPattern.MatchString(taskID) { + return ErrInvalid + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return errors.New("begin cancel transaction") + } + defer tx.Rollback() + var accountID, state string + var attemptID sql.NullString + var confirmationID sql.NullString + var confirmationVersion sql.NullInt64 + err = tx.QueryRowContext(ctx, ` + UPDATE operation_task SET state = CASE WHEN state = 'executing' THEN 'needs_confirmation' ELSE 'cancelled' END, + 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) + if err != nil { + return rowError(err) + } + if attemptID.Valid { + if _, err := tx.ExecContext(ctx, ` + UPDATE execution_attempt SET finished_at = now(), outcome = 'uncertain' + WHERE id = $1 AND finished_at IS NULL`, attemptID.String); err != nil { + return errors.New("mark cancelled attempt uncertain") + } + } + if err := appendAudit(ctx, tx, "task_cancel_requested", accountID, confirmationID.String, confirmationVersion.Int64, attemptID.String, taskID, map[string]string{"state": state}); err != nil { + return err + } + return commit(tx) +} + +func (s *Store) ExecuteMock(ctx context.Context, workerID, outcome string) (Execution, error) { + if !refPattern.MatchString(workerID) || (outcome != "succeeded" && outcome != "failed" && outcome != "uncertain" && outcome != "policy_hold") { + return Execution{}, ErrInvalid + } + execution, err := s.claim(ctx, workerID) + if err != nil || !execution.WasClaimed { + return execution, err + } + return s.complete(ctx, workerID, execution, outcome) +} + +func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return Execution{}, errors.New("begin claim transaction") + } + defer tx.Rollback() + if err := quarantineExpired(ctx, tx); err != nil { + return Execution{}, err + } + if err := quarantineInvalid(ctx, tx); err != nil { + return Execution{}, err + } + var execution Execution + err = tx.QueryRowContext(ctx, ` + WITH candidate AS ( + SELECT t.id + FROM operation_task t + JOIN social_account a ON a.id = t.account_id + JOIN content_draft d ON d.id = t.draft_id + JOIN confirmation c ON c.id = t.confirmation_id + WHERE t.state = 'queued' AND a.status = 'active' + AND a.version = t.account_version + AND d.account_id = t.account_id AND d.version = t.draft_version + AND c.account_id = t.account_id AND c.account_version = t.account_version + AND c.draft_id = t.draft_id AND c.draft_version = t.draft_version + AND c.version = t.confirmation_version + ORDER BY t.created_at, t.id + FOR UPDATE OF t SKIP LOCKED LIMIT 1 + ) + UPDATE operation_task t SET state = 'executing', lease_owner = $1, + lease_until = now() + interval '1 minute', updated_at = now() + FROM candidate WHERE t.id = candidate.id + 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 { + return Execution{}, err + } + return Execution{}, nil + } + if err != nil { + return Execution{}, errors.New("claim task") + } + execution.AttemptID = newID() + execution.State = "executing" + execution.WasClaimed = true + if _, err := tx.ExecContext(ctx, `INSERT INTO execution_attempt (id, task_id) VALUES ($1, $2)`, execution.AttemptID, execution.TaskID); err != nil { + return Execution{}, publicDatabaseError(err) + } + if _, err := tx.ExecContext(ctx, `UPDATE operation_task SET current_attempt_id = $1 WHERE id = $2`, execution.AttemptID, execution.TaskID); err != nil { + return Execution{}, errors.New("attach execution attempt") + } + if err := appendAudit(ctx, tx, "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 { + return Execution{}, err + } + return execution, nil +} + +func (s *Store) complete(ctx context.Context, workerID string, execution Execution, outcome string) (Execution, error) { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return Execution{}, errors.New("begin completion transaction") + } + defer tx.Rollback() + var leaseUntil time.Time + err = tx.QueryRowContext(ctx, ` + SELECT lease_until FROM operation_task + WHERE id = $1 AND state = 'executing' AND lease_owner = $2 AND current_attempt_id = $3 + FOR UPDATE`, execution.TaskID, workerID, execution.AttemptID).Scan(&leaseUntil) + if err != nil { + return Execution{}, rowError(err) + } + state := map[string]string{"succeeded": "succeeded", "failed": "failed", "uncertain": "needs_confirmation", "policy_hold": "policy_hold"}[outcome] + if leaseUntil.Before(time.Now()) { + outcome, state = "uncertain", "needs_confirmation" + } + result, _ := json.Marshal(map[string]string{"mock_outcome": outcome}) + if _, err := tx.ExecContext(ctx, `UPDATE execution_attempt SET finished_at = now(), outcome = $1, result = $2 WHERE id = $3`, outcome, result, execution.AttemptID); err != nil { + return Execution{}, errors.New("finish execution attempt") + } + if _, err := tx.ExecContext(ctx, ` + UPDATE operation_task SET state = $1, lease_owner = NULL, lease_until = NULL, updated_at = now() WHERE id = $2`, state, execution.TaskID); err != nil { + return Execution{}, errors.New("finish task") + } + if err := appendAudit(ctx, tx, "task_finished", 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 { + return Execution{}, err + } + execution.State = state + return execution, nil +} + +func quarantineExpired(ctx context.Context, tx *sql.Tx) error { + rows, err := tx.QueryContext(ctx, ` + UPDATE operation_task SET state = 'needs_confirmation', lease_owner = NULL, lease_until = NULL, updated_at = now() + 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") + } + type expiredTask struct { + taskID, accountID, attemptID, confirmationID string + confirmationVersion int64 + } + var expired []expiredTask + for rows.Next() { + var taskID, accountID string + var attemptID sql.NullString + var confirmationID sql.NullString + var confirmationVersion sql.NullInt64 + if err := rows.Scan(&taskID, &accountID, &attemptID, &confirmationID, &confirmationVersion); err != nil { + rows.Close() + return 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") + } + 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") + } + } + if err := appendAudit(ctx, tx, "task_lease_expired", task.accountID, task.confirmationID, task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil { + return err + } + } + return nil +} + +func quarantineInvalid(ctx context.Context, tx *sql.Tx) error { + rows, err := tx.QueryContext(ctx, ` + UPDATE operation_task t SET + state = CASE WHEN EXISTS ( + SELECT 1 FROM social_account paused WHERE paused.id = t.account_id AND paused.status = 'paused' + ) THEN 'policy_hold' ELSE 'needs_confirmation' END, + updated_at = now() + WHERE t.state = 'queued' AND NOT EXISTS ( + SELECT 1 + FROM social_account a + JOIN content_draft d ON d.id = t.draft_id + JOIN confirmation c ON c.id = t.confirmation_id + WHERE a.id = t.account_id AND a.status = 'active' AND a.version = t.account_version + AND d.account_id = t.account_id AND d.version = t.draft_version + AND c.account_id = t.account_id AND c.account_version = t.account_version + AND c.draft_id = t.draft_id AND c.draft_version = t.draft_version + AND c.version = t.confirmation_version + ) + RETURNING t.id, t.account_id, t.confirmation_id, t.confirmation_version, t.state`) + if err != nil { + return errors.New("quarantine invalid tasks") + } + type invalidTask struct { + taskID, accountID, confirmationID, state string + confirmationVersion int64 + } + var invalid []invalidTask + for rows.Next() { + var taskID, accountID, state string + var confirmationID sql.NullString + var confirmationVersion sql.NullInt64 + if err := rows.Scan(&taskID, &accountID, &confirmationID, &confirmationVersion, &state); err != nil { + rows.Close() + return errors.New("read quarantined task") + } + invalid = append(invalid, invalidTask{taskID, accountID, confirmationID.String, state, confirmationVersion.Int64}) + } + if err := rows.Close(); err != nil { + return errors.New("close quarantined task rows") + } + for _, task := range invalid { + if err := appendAudit(ctx, tx, "task_quarantined", task.accountID, task.confirmationID, task.confirmationVersion, "", task.taskID, map[string]string{"state": task.state}); err != nil { + return err + } + } + return nil +} + +func (s *Store) Audit(ctx context.Context) ([]AuditEvent, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT id, event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id, details, created_at + FROM audit_event ORDER BY id`) + if err != nil { + return nil, errors.New("read audit events") + } + defer rows.Close() + var events []AuditEvent + for rows.Next() { + var event AuditEvent + var accountID, confirmationID, attemptID, taskID sql.NullString + var confirmationVersion sql.NullInt64 + if err := rows.Scan(&event.ID, &event.EventType, &accountID, &confirmationID, &confirmationVersion, &attemptID, &taskID, &event.Details, &event.CreatedAt); err != nil { + return nil, errors.New("decode audit event") + } + event.AccountID, event.ConfirmationID, event.ConfirmationVersion = accountID.String, confirmationID.String, confirmationVersion.Int64 + event.AttemptID, event.TaskID = attemptID.String, taskID.String + events = append(events, event) + } + return events, rows.Err() +} + +func appendAudit(ctx context.Context, tx *sql.Tx, eventType, accountID, confirmationID string, confirmationVersion int64, attemptID, taskID string, details any) error { + if details == nil { + details = map[string]any{} + } + encoded, err := json.Marshal(details) + if err != nil { + return errors.New("encode audit details") + } + _, err = tx.ExecContext(ctx, ` + INSERT INTO audit_event (event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id, details) + VALUES ($1, NULLIF($2, ''), NULLIF($3, ''), NULLIF($4, 0), NULLIF($5, ''), NULLIF($6, ''), $7)`, + eventType, accountID, confirmationID, confirmationVersion, attemptID, taskID, encoded) + if err != nil { + return errors.New("append audit event") + } + return nil +} + +func newID() string { + var value [16]byte + _, _ = rand.Read(value[:]) + return hex.EncodeToString(value[:]) +} + +func commit(tx *sql.Tx) error { + if err := tx.Commit(); err != nil { + return errors.New("commit transaction") + } + return nil +} + +func rowError(err error) error { + if errors.Is(err, sql.ErrNoRows) { + return ErrNotFound + } + return publicDatabaseError(err) +} + +func publicDatabaseError(err error) error { + if err == nil { + return nil + } + var postgresError *pgconn.PgError + if errors.As(err, &postgresError) && (postgresError.Code == "23505" || postgresError.Code == "23503" || postgresError.Code == "23514") { + return ErrConflict + } + return errors.New("phase A persistence operation failed") +} diff --git a/internal/phasea/store_test.go b/internal/phasea/store_test.go new file mode 100644 index 0000000..bc721e7 --- /dev/null +++ b/internal/phasea/store_test.go @@ -0,0 +1,263 @@ +package phasea + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestValidationRejectsInvalidInputsBeforePersistence(t *testing.T) { + store := &Store{} + if err := store.CreateAccount(context.Background(), Account{ID: "INVALID"}); !errors.Is(err, ErrInvalid) { + t.Fatalf("expected invalid account, got %v", err) + } + if err := store.CreateAccount(context.Background(), Account{ + ID: "account-a", CredentialReferenceID: "credential-a", CredentialProvider: "os_keyring", + CredentialKey: "raw-value", ProfileID: "profile-a", + }); !errors.Is(err, ErrInvalid) { + t.Fatalf("expected non-reference credential value to be rejected, got %v", err) + } + if _, _, err := store.Enqueue(context.Background(), Task{ID: "task-a"}); !errors.Is(err, ErrInvalid) { + t.Fatalf("expected invalid task, got %v", err) + } + if _, err := store.ExecuteMock(context.Background(), "worker-a", "retry"); !errors.Is(err, ErrInvalid) { + t.Fatalf("expected unsupported outcome to be rejected, got %v", err) + } +} + +func TestPhaseAOfflineWorkflow(t *testing.T) { + databaseURL := os.Getenv("CREATORHUB_POSTGRES_TEST_URL") + if databaseURL == "" { + t.Skip("set CREATORHUB_POSTGRES_TEST_URL to run PostgreSQL integration coverage") + } + ctx := context.Background() + store, err := Open(ctx, databaseURL) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + if _, err := store.db.ExecContext(ctx, ` + TRUNCATE audit_event, execution_attempt, operation_task, confirmation, content_draft, + runtime_instance, social_account, credential_reference RESTART IDENTITY CASCADE`); err != nil { + t.Fatal(err) + } + + accounts := []Account{ + {ID: "account-a", CredentialReferenceID: "credential-a", CredentialProvider: "os_keyring", CredentialKey: "creatorhub/account-a", ProfileID: "profile-a"}, + {ID: "account-b", CredentialReferenceID: "credential-b", CredentialProvider: "os_keyring", CredentialKey: "creatorhub/account-b", ProfileID: "profile-b"}, + } + for _, account := range accounts { + if err := store.CreateAccount(ctx, account); err != nil { + t.Fatal(err) + } + } + if err := store.CreateAccount(ctx, Account{ + ID: "account-c", CredentialReferenceID: "credential-c", CredentialProvider: "os_keyring", + CredentialKey: "creatorhub/account-c", ProfileID: "profile-a", + }); !errors.Is(err, ErrConflict) { + t.Fatalf("expected profiles to be unique, got %v", err) + } + + lease := time.Now().Add(time.Minute) + if err := store.BindRuntime(ctx, RuntimeBinding{ID: "binding-a", AccountID: "account-a", RuntimeID: "runtime-a", LeaseUntil: lease}); err != nil { + t.Fatal(err) + } + if err := store.BindRuntime(ctx, RuntimeBinding{ID: "binding-a-2", AccountID: "account-a", RuntimeID: "runtime-b", LeaseUntil: lease}); !errors.Is(err, ErrConflict) { + t.Fatalf("expected one active runtime per account, got %v", err) + } + if err := store.BindRuntime(ctx, RuntimeBinding{ID: "binding-b", AccountID: "account-b", RuntimeID: "runtime-a", LeaseUntil: lease}); !errors.Is(err, ErrConflict) { + t.Fatalf("expected one account per runtime, got %v", err) + } + if err := store.ReleaseRuntime(ctx, "binding-a"); err != nil { + t.Fatal(err) + } + + createApprovedDraft(t, store, "account-a", "draft-a", "confirmation-a") + for index := range 20 { + task := approvedTask(index, "account-a", "draft-a", "confirmation-a") + if _, inserted, err := store.Enqueue(ctx, task); err != nil || !inserted { + t.Fatalf("enqueue task %d: inserted=%v err=%v", index, inserted, err) + } + task.ID = fmt.Sprintf("duplicate-%02d", index) + if existing, inserted, err := store.Enqueue(ctx, task); err != nil || inserted || existing.ID == task.ID { + t.Fatalf("redeliver task %d: existing=%q inserted=%v err=%v", index, existing.ID, inserted, err) + } + } + + var executed atomic.Int64 + var workers sync.WaitGroup + errorsFromWorkers := make(chan error, 4) + for worker := range 4 { + workers.Add(1) + go func() { + defer workers.Done() + for { + execution, err := store.ExecuteMock(ctx, fmt.Sprintf("worker-%d", worker), "succeeded") + if err != nil { + errorsFromWorkers <- err + return + } + if !execution.WasClaimed { + return + } + executed.Add(1) + } + }() + } + workers.Wait() + close(errorsFromWorkers) + for err := range errorsFromWorkers { + t.Fatal(err) + } + if executed.Load() != 20 { + t.Fatalf("expected exactly 20 executions, got %d", executed.Load()) + } + assertCount(t, store, `SELECT count(*) FROM operation_task WHERE state = 'succeeded'`, 20) + assertCount(t, store, `SELECT count(*) FROM execution_attempt`, 20) + + createApprovedDraft(t, store, "account-b", "draft-b", "confirmation-b") + unconfirmed := approvedTask(30, "account-b", "draft-b", "") + unconfirmed.ConfirmationVersion = 0 + if _, _, err := store.Enqueue(ctx, unconfirmed); err != nil { + t.Fatal(err) + } + mismatch := approvedTask(31, "account-b", "draft-b", "confirmation-b") + mismatch.AccountVersion = 2 + if _, _, err := store.Enqueue(ctx, mismatch); err != nil { + t.Fatal(err) + } + if execution, err := store.ExecuteMock(ctx, "worker-validation", "succeeded"); err != nil || execution.WasClaimed { + 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) + assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id IN ('task-30', 'task-31')`, 0) + + uncertain := approvedTask(32, "account-b", "draft-b", "confirmation-b") + if _, _, err := store.Enqueue(ctx, uncertain); err != nil { + t.Fatal(err) + } + 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) + } + 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) + } + assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id = 'task-32'`, 1) + + policy := approvedTask(33, "account-b", "draft-b", "confirmation-b") + if _, _, err := store.Enqueue(ctx, policy); err != nil { + t.Fatal(err) + } + 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) + } + 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) + } + + paused := approvedTask(34, "account-b", "draft-b", "confirmation-b") + if _, _, err := store.Enqueue(ctx, paused); err != nil { + t.Fatal(err) + } + pauseStarted := time.Now() + if err := store.PauseAccount(ctx, "account-b"); err != nil { + t.Fatal(err) + } + if err := store.PauseAccount(ctx, "account-b"); err != nil { + t.Fatalf("pause should be idempotent: %v", err) + } + 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) + var pausedAt time.Time + if err := store.db.QueryRowContext(ctx, `SELECT paused_at FROM social_account WHERE id = 'account-b'`).Scan(&pausedAt); err != nil || pausedAt.Sub(pauseStarted) >= time.Minute { + t.Fatalf("account was not paused within one minute: paused_at=%v err=%v", pausedAt, err) + } + + cancelled := approvedTask(35, "account-a", "draft-a", "confirmation-a") + if _, _, err := store.Enqueue(ctx, cancelled); err != nil { + t.Fatal(err) + } + if err := store.CancelTask(ctx, cancelled.ID); err != nil { + t.Fatal(err) + } + assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-35' AND state = 'cancelled'`, 1) + + expired := approvedTask(36, "account-a", "draft-a", "confirmation-a") + if _, _, err := store.Enqueue(ctx, expired); err != nil { + t.Fatal(err) + } + claimed, err := store.claim(ctx, "worker-crashed") + if err != nil || !claimed.WasClaimed { + t.Fatalf("claim task for lease test: execution=%+v err=%v", claimed, err) + } + if _, err := store.db.ExecContext(ctx, `UPDATE operation_task SET lease_until = now() - interval '1 second' WHERE id = $1`, claimed.TaskID); err != nil { + t.Fatal(err) + } + if execution, err := store.ExecuteMock(ctx, "worker-after-crash", "succeeded"); err != nil || execution.WasClaimed { + 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) + assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id = 'task-36' AND outcome = 'uncertain'`, 1) + + events, err := store.Audit(ctx) + if err != nil { + t.Fatal(err) + } + traceable := false + for _, event := range events { + if event.EventType == "task_finished" && event.ConfirmationID != "" && event.ConfirmationVersion == 1 && event.AttemptID != "" && event.TaskID != "" { + traceable = true + break + } + } + if !traceable { + t.Fatal("audit does not trace confirmation version, task, and attempt") + } + exported, _ := json.Marshal(events) + for _, forbidden := range []string{"password", "cookie", "token", "credential-a", "creatorhub/account-a"} { + if strings.Contains(strings.ToLower(string(exported)), forbidden) { + t.Fatalf("audit export contains sensitive field or credential reference %q", forbidden) + } + } + if _, err := store.db.ExecContext(ctx, `UPDATE audit_event SET event_type = 'rewritten' WHERE id = 1`); err == nil { + t.Fatal("audit events must be append-only") + } +} + +func createApprovedDraft(t *testing.T, store *Store, accountID, draftID, confirmationID string) { + t.Helper() + ctx := context.Background() + if err := store.CreateDraft(ctx, Draft{ID: draftID, AccountID: accountID, Version: 1, Content: "offline mock content"}); err != nil { + t.Fatal(err) + } + if err := store.Confirm(ctx, Confirmation{ + ID: confirmationID, AccountID: accountID, AccountVersion: 1, DraftID: draftID, DraftVersion: 1, Version: 1, + }); err != nil { + t.Fatal(err) + } +} + +func approvedTask(index int, accountID, draftID, confirmationID string) Task { + return Task{ + ID: fmt.Sprintf("task-%02d", index), IdempotencyKey: fmt.Sprintf("idempotency-%02d", index), + AccountID: accountID, AccountVersion: 1, DraftID: draftID, DraftVersion: 1, + ConfirmationID: confirmationID, ConfirmationVersion: 1, + } +} + +func assertCount(t *testing.T, store *Store, query string, expected int) { + t.Helper() + var actual int + if err := store.db.QueryRowContext(context.Background(), query).Scan(&actual); err != nil || actual != expected { + t.Fatalf("count mismatch: expected=%d actual=%d err=%v query=%s", expected, actual, err, query) + } +}