creator-hub/cmd/control-plane/phasea.go

package main

import (
	"bytes"
	"encoding/json"
	"errors"
	"io"
	"strconv"
	"strings"
	"time"

	"git.ipao.vip/rogee/creator-hub/internal/hub"
	"git.ipao.vip/rogee/creator-hub/internal/phasea"
	"github.com/gofiber/fiber/v3"
)

type accountRequest struct {
	Name               string   `json:"name"`
	Platform           string   `json:"platform"`
	PlatformAccountKey string   `json:"platform_account_key"`
	Tags               []string `json:"tags"`
	Cookies            string   `json:"cookies"`
}

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"`
}

type taskVerificationRequest struct {
	Result string `json:"result"`
}

func registerPhaseA(app *fiber.App, store *phasea.Store, runtimeStore runtimeStopStore, credentials phasea.CredentialBridge) {
	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)
		}
		tags := input.Tags
		if tags == nil {
			tags = []string{}
		}
		for index := range tags {
			tags[index] = strings.TrimSpace(tags[index])
		}
		accountID := phasea.NewAccountID()
		account := phasea.Account{
			ID: accountID, Name: strings.TrimSpace(input.Name), Platform: strings.TrimSpace(input.Platform),
			PlatformAccountKey: strings.TrimSpace(input.PlatformAccountKey), Tags: tags, Cookies: strings.TrimSpace(input.Cookies),
			CredentialReference: phasea.CredentialReference{ID: accountID + "-cookies", Provider: "os_keyring"},
			CredentialKey:       "creatorhub/" + accountID + "/cookies",
			AuthorizationStatus: "authorized", RuntimeStatus: "paused", Version: 1,
		}
		if err := store.CreateAccount(c.Context(), account, credentials); err != nil {
			if errors.Is(err, phasea.ErrAccountCreationUnknown) {
				return c.Status(fiber.StatusServiceUnavailable).JSON(map[string]string{
					"error": "account creation result is unknown", "reason_code": "account_creation_result_unknown", "account_id": accountID,
				})
			}
			return phaseAError(c, err)
		}
		return c.Status(fiber.StatusCreated).JSON(account)
	})

	app.Get("/api/phase-a/accounts", func(c fiber.Ctx) error {
		accounts, err := store.ListAccounts(c.Context())
		if err != nil {
			return phaseAError(c, err)
		}
		return c.JSON(accounts)
	})

	app.Get("/api/phase-a/accounts/:id", func(c fiber.Ctx) error {
		account, err := store.GetAccount(c.Context(), c.Params("id"))
		if err != nil {
			return phaseAError(c, err)
		}
		return c.JSON(account)
	})

	app.Post("/api/phase-a/accounts/:id/pause", func(c fiber.Ctx) error {
		unlock, err := lockAccountResources(c.Context(), runtimeStore, c.Params("id"))
		if err != nil {
			return hubError(c, err)
		}
		defer unlock()
		if err := store.PauseAccount(c.Context(), c.Params("id")); err != nil {
			return phaseAError(c, err)
		}
		if runtimeStore != nil {
			if err := stopAccountRuntime(c.Context(), runtimeStore, c.Params("id")); err != nil {
				return hubError(c, err)
			}
		}
		return c.SendStatus(fiber.StatusNoContent)
	})

	app.Post("/api/phase-a/accounts/:id/resume", func(c fiber.Ctx) error {
		unlock, err := lockAccountResources(c.Context(), runtimeStore, c.Params("id"))
		if err != nil {
			return hubError(c, err)
		}
		defer unlock()
		if err := store.ResumeAccount(c.Context(), c.Params("id")); err != nil {
			if errors.Is(err, phasea.ErrConflict) {
				return accountResumeConflict(c, store, runtimeStore, c.Params("id"))
			}
			return phaseAError(c, err)
		}
		return c.SendStatus(fiber.StatusNoContent)
	})

	app.Post("/api/phase-a/accounts/:id/revoke", func(c fiber.Ctx) error {
		unlock, err := lockAccountResources(c.Context(), runtimeStore, c.Params("id"))
		if err != nil {
			return hubError(c, err)
		}
		defer unlock()
		if err := store.RevokeAccount(c.Context(), c.Params("id")); err != nil {
			return phaseAError(c, err)
		}
		if runtimeStore != nil {
			if err := stopAccountRuntime(c.Context(), runtimeStore, c.Params("id")); err != nil {
				return hubError(c, err)
			}
		}
		return c.SendStatus(fiber.StatusNoContent)
	})

	app.Post("/api/phase-a/drafts", func(c fiber.Ctx) error {
		var input draftRequest
		if err := decodePhaseA(c, &input); err != nil {
			return phaseAError(c, err)
		}
		draft, err := store.CreateDraftVersion(c.Context(), input.AccountID, input.Content)
		if err != nil {
			return phaseAError(c, err)
		}
		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 confirmationRequest
		if err := decodePhaseA(c, &input); err != nil {
			return phaseAError(c, err)
		}
		confirmation, inserted, err := store.ConfirmDraft(c.Context(), input.DraftID, input.AccountVersion, input.DraftVersion)
		if err != nil {
			return phaseAError(c, err)
		}
		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 taskRequest
		if err := decodePhaseA(c, &input); err != nil {
			return phaseAError(c, err)
		}
		task, inserted, err := store.EnqueueConfirmation(c.Context(), input.ConfirmationID)
		if err != nil {
			return phaseAError(c, err)
		}
		status := fiber.StatusOK
		if inserted {
			status = fiber.StatusCreated
		}
		return c.Status(status).JSON(task)
	})

	app.Get("/api/phase-a/tasks", func(c fiber.Ctx) error {
		tasks, err := store.ListTasksFiltered(c.Context(), c.Query("account_id"), c.Query("draft_id"), c.Query("state"))
		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.GetTaskDetail(c.Context(), c.Params("id"))
		if err != nil {
			return phaseAError(c, err)
		}
		return c.JSON(task)
	})

	app.Get("/api/phase-a/attempts/:id", func(c fiber.Ctx) error {
		attempt, err := store.GetTaskAttemptDetail(c.Context(), c.Params("id"))
		if err != nil {
			return phaseAError(c, err)
		}
		return c.JSON(attempt)
	})

	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/tasks/:id/verify", func(c fiber.Ctx) error {
		var input taskVerificationRequest
		if err := decodePhaseA(c, &input); err != nil {
			return phaseAError(c, err)
		}
		if err := store.VerifyTask(c.Context(), c.Params("id"), input.Result); err != nil {
			return phaseAError(c, err)
		}
		return c.SendStatus(fiber.StatusNoContent)
	})

	app.Post("/api/phase-a/tasks/:id/resume", func(c fiber.Ctx) error {
		if err := store.ResumeTask(c.Context(), c.Params("id")); err != nil {
			return phaseAError(c, err)
		}
		return c.SendStatus(fiber.StatusNoContent)
	})

	app.Post("/api/phase-a/tasks/:id/finish", func(c fiber.Ctx) error {
		if err := store.FinishTask(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 {
		filter, err := auditFilter(c)
		if err != nil {
			return phaseAError(c, err)
		}
		events, err := store.ListAudit(c.Context(), filter)
		if err != nil {
			return phaseAError(c, err)
		}
		return c.JSON(events)
	})
}

func auditFilter(c fiber.Ctx) (phasea.AuditFilter, error) {
	filter := phasea.AuditFilter{
		AccountID: c.Query("account_id"), TaskID: c.Query("task_id"), AttemptID: c.Query("attempt_id"),
		BrowserEnvAlias: c.Query("browser_env_alias"), NetworkExitID: c.Query("network_exit_id"),
		EventType: c.Query("event_type"), Page: 1, PageSize: 25,
	}
	for _, field := range []struct {
		value  string
		target *int
	}{{c.Query("page"), &filter.Page}, {c.Query("page_size"), &filter.PageSize}} {
		value, target := field.value, field.target
		if value == "" {
			continue
		}
		parsed, err := strconv.Atoi(value)
		if err != nil {
			return phasea.AuditFilter{}, phasea.ErrInvalid
		}
		*target = parsed
	}
	for _, field := range []struct {
		value  string
		target **time.Time
	}{{c.Query("from"), &filter.From}, {c.Query("to"), &filter.To}} {
		value, target := field.value, field.target
		if value == "" {
			continue
		}
		parsed, err := time.Parse(time.RFC3339, value)
		if err != nil {
			return phasea.AuditFilter{}, phasea.ErrInvalid
		}
		*target = &parsed
	}
	return filter, nil
}

func resumeBlockReason(account phasea.Account, environment hub.EnvironmentContext, bindingFound bool) string {
	switch {
	case account.AuthorizationStatus != "authorized":
		return "account_revoked"
	case !bindingFound:
		return "binding_missing"
	case environment.Exit.ID != "" && environment.Exit.HealthStatus != "healthy":
		return "network_exit_unhealthy"
	case environment.RuntimeCleanupPending:
		return "runtime_stop_pending"
	case environment.RuntimeInstanceID != "":
		return "runtime_active"
	default:
		return "account_conflict"
	}
}

func accountResumeConflict(c fiber.Ctx, store *phasea.Store, runtimeStore runtimeStopStore, accountID string) error {
	account, err := store.GetAccount(c.Context(), accountID)
	if err != nil {
		return phaseAError(c, err)
	}
	var environment hub.EnvironmentContext
	bindingFound := false
	if runtimeStore != nil {
		environment, err = runtimeStore.GetEnvironmentContextForAccount(c.Context(), accountID)
		bindingFound = err == nil
		if err != nil && !errors.Is(err, hub.ErrNotFound) {
			return phaseAError(c, err)
		}
	}
	return c.Status(fiber.StatusConflict).JSON(map[string]string{
		"error": phasea.ErrConflict.Error(), "reason_code": resumeBlockReason(account, environment, bindingFound), "readiness": "blocked",
	})
}

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"
	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):
		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})
}