HH-847 / HH-799: add draft review and idempotent enqueue (#24)

cmd/control-plane/hub_test.go

@@ -87,6 +87,28 @@ func TestResumeBlockReasonIsStable(t *testing.T) {
 	}
 }
 
+func TestPhaseAReadinessErrorsAreStructured(t *testing.T) {
+	for _, test := range []struct {
+		name, reason string
+		unavailable  bool
+		status       int
+	}{
+		{name: "version conflict", reason: "draft_version_changed", status: http.StatusConflict},
+		{name: "resource unavailable", reason: "network_exit_unhealthy", unavailable: true, status: http.StatusServiceUnavailable},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			app := fiber.New()
+			app.Get("/", func(c fiber.Ctx) error {
+				return phaseAError(c, &phasea.ReadinessError{Reason: test.reason, Unavailable: test.unavailable})
+			})
+			response := do(app, http.MethodGet, "/", "")
+			if response.Code != test.status || !strings.Contains(response.Body.String(), `"reason_code":"`+test.reason+`"`) {
+				t.Fatalf("unexpected response: %d %s", response.Code, response.Body.String())
+			}
+		})
+	}
+}
+
 func (s *memoryStore) CreateGateway(_ context.Context, _, _, _ string) (hub.Gateway, error) {
 	return hub.Gateway{}, nil
 }

cmd/control-plane/phasea.go

@@ -23,6 +23,21 @@ type accountRequest struct {
 	} `json:"credential_reference"`
 }
 
+type draftRequest struct {
+	AccountID string `json:"account_id"`
+	Content   string `json:"content"`
+}
+
+type confirmationRequest struct {
+	DraftID        string `json:"draft_id"`
+	AccountVersion int64  `json:"account_version"`
+	DraftVersion   int64  `json:"draft_version"`
+}
+
+type taskRequest struct {
+	ConfirmationID string `json:"confirmation_id"`
+}
+
 func registerPhaseA(app *fiber.App, store *phasea.Store, runtimeStore runtimeStopStore) {
 	app.Post("/api/phase-a/accounts", func(c fiber.Ctx) error {
 		var input accountRequest
@@ -102,33 +117,71 @@ func registerPhaseA(app *fiber.App, store *phasea.Store, runtimeStore runtimeSto
 	})
 
 	app.Post("/api/phase-a/drafts", func(c fiber.Ctx) error {
-		var input phasea.Draft
+		var input draftRequest
 		if err := decodePhaseA(c, &input); err != nil {
 			return phaseAError(c, err)
 		}
-		if err := store.CreateDraft(c.Context(), input); err != nil {
+		draft, err := store.CreateDraftVersion(c.Context(), input.AccountID, input.Content)
+		if err != nil {
 			return phaseAError(c, err)
 		}
-		return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID})
+		return c.Status(fiber.StatusCreated).JSON(draft)
+	})
+
+	app.Get("/api/phase-a/drafts", func(c fiber.Ctx) error {
+		drafts, err := store.ListDrafts(c.Context(), c.Query("account_id"))
+		if err != nil {
+			return phaseAError(c, err)
+		}
+		return c.JSON(drafts)
+	})
+
+	app.Get("/api/phase-a/drafts/:id", func(c fiber.Ctx) error {
+		draft, err := store.GetDraftDetail(c.Context(), c.Params("id"))
+		if err != nil {
+			return phaseAError(c, err)
+		}
+		return c.JSON(draft)
 	})
 
 	app.Post("/api/phase-a/confirmations", func(c fiber.Ctx) error {
-		var input phasea.Confirmation
+		var input confirmationRequest
 		if err := decodePhaseA(c, &input); err != nil {
 			return phaseAError(c, err)
 		}
-		if err := store.Confirm(c.Context(), input); err != nil {
+		confirmation, inserted, err := store.ConfirmDraft(c.Context(), input.DraftID, input.AccountVersion, input.DraftVersion)
+		if err != nil {
 			return phaseAError(c, err)
 		}
-		return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID})
+		status := fiber.StatusOK
+		if inserted {
+			status = fiber.StatusCreated
+		}
+		return c.Status(status).JSON(confirmation)
+	})
+
+	app.Get("/api/phase-a/confirmations", func(c fiber.Ctx) error {
+		confirmations, err := store.ListConfirmations(c.Context(), c.Query("draft_id"))
+		if err != nil {
+			return phaseAError(c, err)
+		}
+		return c.JSON(confirmations)
+	})
+
+	app.Get("/api/phase-a/confirmations/:id", func(c fiber.Ctx) error {
+		confirmation, err := store.GetConfirmation(c.Context(), c.Params("id"))
+		if err != nil {
+			return phaseAError(c, err)
+		}
+		return c.JSON(confirmation)
 	})
 
 	app.Post("/api/phase-a/tasks", func(c fiber.Ctx) error {
-		var input phasea.Task
+		var input taskRequest
 		if err := decodePhaseA(c, &input); err != nil {
 			return phaseAError(c, err)
 		}
-		task, inserted, err := store.Enqueue(c.Context(), input)
+		task, inserted, err := store.EnqueueConfirmation(c.Context(), input.ConfirmationID)
 		if err != nil {
 			return phaseAError(c, err)
 		}
@@ -139,6 +192,22 @@ func registerPhaseA(app *fiber.App, store *phasea.Store, runtimeStore runtimeSto
 		return c.Status(status).JSON(task)
 	})
 
+	app.Get("/api/phase-a/tasks", func(c fiber.Ctx) error {
+		tasks, err := store.ListTasks(c.Context(), c.Query("account_id"), c.Query("draft_id"))
+		if err != nil {
+			return phaseAError(c, err)
+		}
+		return c.JSON(tasks)
+	})
+
+	app.Get("/api/phase-a/tasks/:id", func(c fiber.Ctx) error {
+		task, err := store.GetTask(c.Context(), c.Params("id"))
+		if err != nil {
+			return phaseAError(c, err)
+		}
+		return c.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)
@@ -223,7 +292,14 @@ func decodePhaseA(c fiber.Ctx, destination any) error {
 func phaseAError(c fiber.Ctx, err error) error {
 	status := fiber.StatusInternalServerError
 	message := "phase A operation failed"
+	var readiness *phasea.ReadinessError
 	switch {
+	case errors.As(err, &readiness):
+		status, message = fiber.StatusConflict, "phase A version or account state changed"
+		if readiness.Unavailable {
+			status, message = fiber.StatusServiceUnavailable, "phase A resources are not ready"
+		}
+		return c.Status(status).JSON(map[string]string{"error": message, "reason_code": readiness.Reason})
 	case errors.Is(err, phasea.ErrInvalid):
 		status, message = fiber.StatusBadRequest, phasea.ErrInvalid.Error()
 	case errors.Is(err, phasea.ErrConflict):