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

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"time"

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

func registerCreator(app *fiber.App, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, credentials phasea.CredentialBridge) {
	registerCreatorWithServices(app, store, phaseAStore, hubStore, credentials, nil, nil, nil)
}

func registerCreatorWithServices(app *fiber.App, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, credentials phasea.CredentialBridge, executor creator.ActionExecutor, generator creator.TextGenerator, analyzer creator.ThemeAnalyzer) {
	app.Get("/api/creator/settings", func(c fiber.Ctx) error {
		settings, err := store.GetSettings(c.Context())
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(settings)
	})
	app.Put("/api/creator/settings", func(c fiber.Ctx) error {
		var input creator.SettingsUpdate
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		settings, err := store.UpdateSettings(c.Context(), input)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(settings)
	})

	app.Get("/api/creator/accounts", func(c fiber.Ctx) error {
		profiles, err := store.ListAccountProfiles(c.Context())
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(profiles)
	})
	app.Get("/api/creator/accounts/:id/profile", func(c fiber.Ctx) error {
		profile, err := store.GetAccountProfile(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(profile)
	})
	app.Put("/api/creator/accounts/:id/profile", func(c fiber.Ctx) error {
		var input creator.AccountProfileUpdate
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		profile, err := store.UpdateAccountProfile(c.Context(), c.Params("id"), input)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(profile)
	})
	app.Post("/api/creator/accounts/:id/login-result", func(c fiber.Ctx) error {
		var input struct {
			Status    string `json:"status"`
			Reason    string `json:"reason"`
			ActualKey string `json:"actual_platform_account_key"`
		}
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		result, err := store.RecordLoginResult(c.Context(), c.Params("id"), input.Status, input.Reason, input.ActualKey)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(result)
	})
	app.Post("/api/creator/accounts/:id/big-account", func(c fiber.Ctx) error {
		var input struct {
			Enabled bool `json:"enabled"`
		}
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		profile, err := store.SetBigAccount(c.Context(), c.Params("id"), input.Enabled)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(profile)
	})
	app.Get("/api/creator/relations", func(c fiber.Ctx) error {
		relations, err := store.ListRelations(c.Context(), c.Query("big_account_id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(relations)
	})
	app.Post("/api/creator/relations", func(c fiber.Ctx) error {
		var input struct {
			creator.Relation
			Enabled bool `json:"enabled"`
		}
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		if err := store.SetRelation(c.Context(), input.BigAccountID, input.SmallAccountID, input.Enabled); err != nil {
			return creatorError(c, err)
		}
		return c.SendStatus(fiber.StatusNoContent)
	})
	app.Get("/api/creator/accounts/:id/strategies", func(c fiber.Ctx) error {
		strategies, err := store.ListStrategies(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(strategies)
	})
	app.Post("/api/creator/accounts/:id/strategies", func(c fiber.Ctx) error {
		var input creator.StrategyInput
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		strategy, err := store.CreateStrategy(c.Context(), c.Params("id"), input)
		if err != nil {
			return creatorError(c, err)
		}
		return c.Status(fiber.StatusCreated).JSON(strategy)
	})
	app.Put("/api/creator/strategies/:id", func(c fiber.Ctx) error {
		var input creator.StrategyInput
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		strategy, err := store.UpdateStrategy(c.Context(), c.Params("id"), input)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(strategy)
	})
	app.Post("/api/creator/strategies/:id/enable", func(c fiber.Ctx) error { return setStrategyEnabled(c, store, true) })
	app.Post("/api/creator/strategies/:id/disable", func(c fiber.Ctx) error { return setStrategyEnabled(c, store, false) })
	app.Delete("/api/creator/strategies/:id", func(c fiber.Ctx) error {
		if err := store.DeleteStrategy(c.Context(), c.Params("id")); err != nil {
			return creatorError(c, err)
		}
		return c.SendStatus(fiber.StatusNoContent)
	})

	app.Get("/api/creator/competitors", func(c fiber.Ctx) error {
		items, err := store.ListCompetitors(c.Context(), c.Query("platform"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Post("/api/creator/competitors", func(c fiber.Ctx) error {
		var input creator.CompetitorInput
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		item, err := store.CreateCompetitor(c.Context(), input)
		if err != nil {
			return creatorError(c, err)
		}
		return c.Status(fiber.StatusCreated).JSON(item)
	})
	app.Get("/api/creator/competitors/:id", func(c fiber.Ctx) error {
		item, err := store.GetCompetitor(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Post("/api/creator/competitors/:id/pause", func(c fiber.Ctx) error {
		item, err := store.SetCompetitorEnabled(c.Context(), c.Params("id"), false)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Post("/api/creator/competitors/:id/resume", func(c fiber.Ctx) error {
		item, err := store.SetCompetitorEnabled(c.Context(), c.Params("id"), true)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Post("/api/creator/competitors/:id/sync", func(c fiber.Ctx) error {
		var input struct {
			AccountID string `json:"account_id"`
		}
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		report, err := syncCreatorCompetitor(c.Context(), store, phaseAStore, hubStore, credentials, c.Params("id"), input.AccountID)
		if err != nil {
			return creatorError(c, err)
		}
		return c.Status(fiber.StatusAccepted).JSON(report)
	})

	app.Get("/api/creator/works", func(c fiber.Ctx) error {
		filter, err := workFilter(c)
		if err != nil {
			return creatorError(c, err)
		}
		items, err := store.ListWorks(c.Context(), filter)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Post("/api/creator/works", func(c fiber.Ctx) error {
		var input creator.WorkInput
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		item, inserted, err := store.UpsertWork(c.Context(), input, time.Now().UTC())
		if err != nil {
			return creatorError(c, err)
		}
		status := fiber.StatusOK
		if inserted {
			status = fiber.StatusCreated
		}
		return c.Status(status).JSON(item)
	})
	app.Get("/api/creator/works/:id", func(c fiber.Ctx) error {
		item, err := store.GetWork(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Get("/api/creator/works/:id/metrics", func(c fiber.Ctx) error {
		items, err := store.ListMetrics(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Post("/api/creator/works/:id/metrics", func(c fiber.Ctx) error {
		var input struct {
			CollectedAt   time.Time `json:"collected_at"`
			Likes         *int64    `json:"likes"`
			CommentsCount *int64    `json:"comments_count"`
			Shares        *int64    `json:"shares"`
		}
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		if input.CollectedAt.IsZero() {
			input.CollectedAt = time.Now().UTC()
		}
		settings, err := store.GetSettings(c.Context())
		if err != nil {
			return creatorError(c, err)
		}
		point, err := store.RecordMetric(c.Context(), creator.MetricInput{WorkID: c.Params("id"), CollectedAt: input.CollectedAt, Likes: input.Likes, CommentsCount: input.CommentsCount, Shares: input.Shares}, settings, time.Now().UTC())
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(point)
	})
	app.Get("/api/creator/works/:id/material", func(c fiber.Ctx) error {
		item, err := store.GetMaterial(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Post("/api/creator/works/:id/material/select", func(c fiber.Ctx) error {
		item, inserted, err := store.SelectMaterial(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		status := fiber.StatusOK
		if inserted {
			status = fiber.StatusCreated
		}
		return c.Status(status).JSON(item)
	})
	app.Post("/api/creator/works/:id/material/step", func(c fiber.Ctx) error {
		var input struct {
			Step      string `json:"step"`
			Status    string `json:"status"`
			Reference string `json:"reference"`
			Reason    string `json:"reason"`
		}
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		item, err := store.SetMaterialStep(c.Context(), c.Params("id"), input.Step, input.Status, input.Reference, input.Reason)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Post("/api/creator/works/:id/material/rewrite/confirm", func(c fiber.Ctx) error {
		var input struct {
			Requirement string `json:"requirement"`
		}
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		item, err := store.ConfirmRewrite(c.Context(), c.Params("id"), input.Requirement)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Put("/api/creator/works/:id/material/rewrite", func(c fiber.Ctx) error {
		var input struct {
			Title  string `json:"title"`
			Script string `json:"script"`
		}
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		item, err := store.SaveRewrite(c.Context(), c.Params("id"), input.Title, input.Script)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})

	app.Get("/api/creator/comments", func(c fiber.Ctx) error {
		items, err := store.ListComments(c.Context(), c.Query("platform"), c.Query("work_id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Post("/api/creator/comments", func(c fiber.Ctx) error {
		var input creator.CommentInput
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		item, inserted, err := store.SaveComment(c.Context(), input)
		if err != nil {
			return creatorError(c, err)
		}
		status := fiber.StatusOK
		if inserted {
			status = fiber.StatusCreated
		}
		return c.Status(status).JSON(item)
	})
	app.Get("/api/creator/comments/:id", func(c fiber.Ctx) error {
		item, err := store.GetComment(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Get("/api/creator/rules", func(c fiber.Ctx) error {
		items, err := store.ListRules(c.Context(), c.Query("enabled_only") == "true")
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Post("/api/creator/rules", func(c fiber.Ctx) error {
		var input creator.LeadRuleInput
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		item, err := store.CreateRule(c.Context(), input)
		if err != nil {
			return creatorError(c, err)
		}
		return c.Status(fiber.StatusCreated).JSON(item)
	})
	app.Get("/api/creator/rules/:id", func(c fiber.Ctx) error {
		item, err := store.GetRule(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Put("/api/creator/rules/:id", func(c fiber.Ctx) error {
		var input creator.LeadRuleInput
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		item, err := store.UpdateRule(c.Context(), c.Params("id"), input)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Post("/api/creator/rules/:id/enable", func(c fiber.Ctx) error {
		item, err := store.SetRuleEnabled(c.Context(), c.Params("id"), true)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Post("/api/creator/rules/:id/disable", func(c fiber.Ctx) error {
		item, err := store.SetRuleEnabled(c.Context(), c.Params("id"), false)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Get("/api/creator/rule-results", func(c fiber.Ctx) error {
		items, err := store.ListRuleResults(c.Context(), c.Query("comment_id"), c.Query("rule_id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Get("/api/creator/leads", func(c fiber.Ctx) error {
		items, err := store.ListLeads(c.Context(), c.Query("platform"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Post("/api/creator/comments/:id/analyze", func(c fiber.Ctx) error {
		var input struct {
			RuleID string `json:"rule_id"`
		}
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		result, err := store.AnalyzeComment(c.Context(), c.Params("id"), input.RuleID, analyzer)
		if err != nil && !errors.Is(err, creator.ErrUnavailable) {
			return creatorError(c, err)
		}
		status := fiber.StatusOK
		if errors.Is(err, creator.ErrUnavailable) {
			status = fiber.StatusServiceUnavailable
		}
		return c.Status(status).JSON(result)
	})

	app.Get("/api/creator/events", func(c fiber.Ctx) error {
		items, err := store.ListEvents(c.Context(), c.Query("account_id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Post("/api/creator/events", func(c fiber.Ctx) error {
		var input creator.InteractionEvent
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		result, err := store.RecordEvent(c.Context(), input)
		if err != nil {
			return creatorError(c, err)
		}
		status := fiber.StatusOK
		if !result.Duplicate {
			status = fiber.StatusCreated
		}
		return c.Status(status).JSON(result)
	})
	app.Post("/api/creator/events/process", func(c fiber.Ctx) error {
		var input creator.InteractionEvent
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		result, err := store.ProcessAutomaticEvent(c.Context(), input, executor, generator)
		if err != nil && !errors.Is(err, creator.ErrUnavailable) {
			return creatorError(c, err)
		}
		status := fiber.StatusOK
		if errors.Is(err, creator.ErrUnavailable) {
			status = fiber.StatusServiceUnavailable
		}
		return c.Status(status).JSON(result)
	})
	app.Post("/api/creator/events/:id/display", func(c fiber.Ctx) error {
		event, err := store.SetEventDisplayed(c.Context(), c.Params("id"), time.Now().UTC())
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(event)
	})
	app.Get("/api/creator/operations", func(c fiber.Ctx) error {
		items, err := store.ListOperations(c.Context(), c.Query("account_id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Post("/api/creator/operations", func(c fiber.Ctx) error {
		var input creator.OperationInput
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		item, inserted, err := store.CreateOperation(c.Context(), input)
		if err != nil {
			return creatorError(c, err)
		}
		status := fiber.StatusOK
		if inserted {
			status = fiber.StatusCreated
		}
		return c.Status(status).JSON(item)
	})
	app.Get("/api/creator/operations/:id", func(c fiber.Ctx) error {
		item, err := store.GetOperation(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})
	app.Post("/api/creator/operations/:id/execute", func(c fiber.Ctx) error {
		item, err := store.ExecuteManualOperation(c.Context(), c.Params("id"), executor)
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(item)
	})

	app.Get("/api/creator/conversations", func(c fiber.Ctx) error {
		items, err := store.ListConversations(c.Context(), c.Query("account_id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Get("/api/creator/conversations/:id/messages", func(c fiber.Ctx) error {
		items, err := store.ListMessages(c.Context(), c.Params("id"))
		if err != nil {
			return creatorError(c, err)
		}
		return c.JSON(items)
	})
	app.Post("/api/creator/messages", func(c fiber.Ctx) error {
		var input creator.MessageInput
		if err := decodeCreator(c, &input); err != nil {
			return creatorError(c, err)
		}
		item, inserted, err := store.SaveMessage(c.Context(), input)
		if err != nil {
			return creatorError(c, err)
		}
		status := fiber.StatusOK
		if inserted {
			status = fiber.StatusCreated
		}
		return c.Status(status).JSON(item)
	})
}

func setStrategyEnabled(c fiber.Ctx, store *creator.Store, enabled bool) error {
	item, err := store.SetStrategyEnabled(c.Context(), c.Params("id"), enabled)
	if err != nil {
		return creatorError(c, err)
	}
	return c.JSON(item)
}

func workFilter(c fiber.Ctx) (creator.WorkFilter, error) {
	filter := creator.WorkFilter{Platform: c.Query("platform"), SourceID: c.Query("source_id"), SourceType: c.Query("source_type")}
	for _, field := range []struct {
		name   string
		target **int64
	}{{"min_likes", &filter.MinLikes}, {"min_comments", &filter.MinComments}, {"min_shares", &filter.MinShares}} {
		value := c.Query(field.name)
		if value == "" {
			continue
		}
		parsed, err := strconv.ParseInt(value, 10, 64)
		if err != nil {
			return creator.WorkFilter{}, creator.ErrInvalid
		}
		*field.target = &parsed
	}
	for _, field := range []struct {
		name   string
		target **time.Time
	}{{"published_after", &filter.PublishedAfter}, {"published_before", &filter.PublishedBefore}} {
		value := c.Query(field.name)
		if value == "" {
			continue
		}
		parsed, err := time.Parse(time.RFC3339, value)
		if err != nil {
			return creator.WorkFilter{}, creator.ErrInvalid
		}
		*field.target = &parsed
	}
	return filter, nil
}

func decodeCreator(c fiber.Ctx, destination any) error {
	decoder := json.NewDecoder(bytes.NewReader(c.Body()))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(destination); err != nil {
		return creator.ErrInvalid
	}
	if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
		return creator.ErrInvalid
	}
	return nil
}

func creatorError(c fiber.Ctx, err error) error {
	logrus.WithError(err).WithField("service", "control-plane").Error("creator operation failed")
	status, message := fiber.StatusInternalServerError, "creator operation failed"
	switch {
	case errors.Is(err, creator.ErrInvalid):
		status, message = fiber.StatusBadRequest, creator.ErrInvalid.Error()
	case errors.Is(err, creator.ErrConflict):
		status, message = fiber.StatusConflict, creator.ErrConflict.Error()
	case errors.Is(err, creator.ErrNotFound):
		status, message = fiber.StatusNotFound, creator.ErrNotFound.Error()
	case errors.Is(err, creator.ErrUnavailable):
		status, message = fiber.StatusServiceUnavailable, creator.ErrUnavailable.Error()
	case errors.Is(err, creator.ErrUncertain):
		status, message = fiber.StatusConflict, creator.ErrUncertain.Error()
	}
	return c.Status(status).JSON(map[string]string{"error": message})
}

type creatorGatewayBrowser struct {
	gateway     hub.Gateway
	environment hub.EnvironmentContext
}

func (browser creatorGatewayBrowser) SetCookies(ctx context.Context, cookies []douyin.Cookie) error {
	payload := gatewayGenerationPayload(browser.environment)
	payload["cookies"] = cookies
	status, body, err := gatewayCall(ctx, browser.gateway, http.MethodPost, "/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/douyin/cookies", payload, 30*time.Second)
	if err != nil {
		return err
	}
	if status != http.StatusNoContent && status != http.StatusNotModified {
		return fmt.Errorf("douyin cookie injection rejected with HTTP %d: %s", status, string(body))
	}
	return nil
}

func (browser creatorGatewayBrowser) Get(ctx context.Context, target string) (douyin.Response, error) {
	payload := gatewayGenerationPayload(browser.environment)
	payload["url"] = target
	status, body, err := gatewayCall(ctx, browser.gateway, http.MethodPost, "/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/douyin/get", payload, 30*time.Second)
	if err != nil {
		return douyin.Response{}, err
	}
	if status != http.StatusOK {
		return douyin.Response{}, fmt.Errorf("douyin browser request rejected with HTTP %d: %s", status, string(body))
	}
	var response struct {
		Status    int              `json:"status"`
		Body      string           `json:"body"`
		Challenge douyin.Challenge `json:"challenge"`
	}
	if err := json.Unmarshal(body, &response); err != nil {
		return douyin.Response{}, fmt.Errorf("decode douyin browser response: %w", err)
	}
	return douyin.Response{Status: response.Status, Body: []byte(response.Body), Challenge: response.Challenge}, nil
}

func syncCreatorCompetitor(ctx context.Context, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, credentials phasea.CredentialBridge, competitorID, accountID string) (creator.CollectionReport, error) {
	return syncCreatorCompetitorWithClaim(ctx, store, phaseAStore, hubStore, credentials, competitorID, accountID, true)
}

func syncCreatorCompetitorDue(ctx context.Context, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, credentials phasea.CredentialBridge, competitorID, accountID string) (creator.CollectionReport, error) {
	return syncCreatorCompetitorWithClaim(ctx, store, phaseAStore, hubStore, credentials, competitorID, accountID, false)
}

func syncCreatorCompetitorWithClaim(ctx context.Context, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, credentials phasea.CredentialBridge, competitorID, accountID string, force bool) (creator.CollectionReport, error) {
	if store == nil || phaseAStore == nil || hubStore == nil || credentials == nil || accountID == "" {
		return creator.CollectionReport{}, creator.ErrUnavailable
	}
	competitor, err := store.GetCompetitor(ctx, competitorID)
	if err != nil {
		return creator.CollectionReport{}, err
	}
	settings, err := store.GetSettings(ctx)
	if err != nil {
		return creator.CollectionReport{}, err
	}
	now := time.Now().UTC()
	claimed, err := store.ClaimCompetitorSync(ctx, competitorID, force, now)
	if err != nil {
		return creator.CollectionReport{}, err
	}
	if !claimed {
		return creator.CollectionReport{}, creator.ErrConflict
	}
	blocked := func(blockErr error) (creator.CollectionReport, error) {
		markErr := store.MarkCompetitorSync(ctx, competitorID, "blocked", "", blockErr.Error(), nil)
		return creator.CollectionReport{}, errors.Join(blockErr, markErr)
	}
	if competitor.Platform != creator.PlatformDouyin {
		return blocked(fmt.Errorf("%w: 小红书采集器尚未完成平台能力验证", creator.ErrUnavailable))
	}
	account, err := phaseAStore.GetAccount(ctx, accountID)
	if err != nil {
		return blocked(err)
	}
	if account.Platform != creator.PlatformDouyin || account.AuthorizationStatus != "authorized" {
		return blocked(creator.ErrConflict)
	}
	if _, err := store.GetAccountProfile(ctx, accountID); err != nil {
		return blocked(err)
	}
	resolver, ok := credentials.(phasea.CredentialResolver)
	if !ok {
		return blocked(creator.ErrUnavailable)
	}
	rawCredential, err := phaseAStore.ResolveAccountCredential(ctx, accountID, resolver)
	if err != nil {
		return blocked(fmt.Errorf("%w: resolve account credential: %v", creator.ErrUnavailable, err))
	}
	cookies, err := douyin.ParseCookieHeader(rawCredential)
	if err != nil {
		cookies, err = douyin.ParseCookieBundle(rawCredential)
	}
	if err != nil {
		return blocked(fmt.Errorf("%w: invalid account cookie bundle", creator.ErrConflict))
	}
	environment, err := hubStore.GetEnvironmentContextForAccount(ctx, accountID)
	if err != nil {
		return blocked(fmt.Errorf("%w: account environment unavailable: %v", creator.ErrUnavailable, err))
	}
	if environment.RuntimeID == "" || environment.RuntimeNetworkID == "" || environment.BindingVersion <= 0 {
		return blocked(fmt.Errorf("%w: account runtime is not running", creator.ErrUnavailable))
	}
	gateway, err := hubStore.GetGateway(ctx, environment.Gateway)
	if err != nil {
		return blocked(fmt.Errorf("%w: gateway unavailable: %v", creator.ErrUnavailable, err))
	}
	browser := creatorGatewayBrowser{gateway: gateway, environment: environment}
	if err := browser.SetCookies(ctx, cookies); err != nil {
		return blocked(fmt.Errorf("%w: set account cookies: %v", creator.ErrUnavailable, err))
	}
	collector := douyin.CreatorCollector{Browser: browser, AccountKey: competitor.PlatformAccountKey, SourceType: creator.SourceCompetitor, SourceID: competitor.ID}
	if err := collector.VerifyIdentity(ctx, account.PlatformAccountKey); err != nil {
		return blocked(fmt.Errorf("%w: account identity verification failed: %v", creator.ErrConflict, err))
	}
	report, collectErr := store.CollectSource(ctx, competitor.Platform, creator.SourceCompetitor, competitor.ID, collector, now)
	if collectErr != nil {
		status, next := "failed", now.Add(time.Duration(settings.NewWorkIntervalSeconds)*time.Second)
		if errors.Is(collectErr, creator.ErrUnavailable) || errors.Is(collectErr, creator.ErrConflict) {
			status, next = "blocked", time.Time{}
		}
		var nextAt *time.Time
		if !next.IsZero() {
			nextAt = &next
		}
		markErr := store.MarkCompetitorSync(ctx, competitorID, status, "", collectErr.Error(), nextAt)
		return report, errors.Join(collectErr, markErr)
	}
	next := time.Now().UTC().Add(time.Duration(settings.NewWorkIntervalSeconds) * time.Second)
	if err := store.MarkCompetitorSync(ctx, competitorID, "idle", "", "", &next); err != nil {
		return report, err
	}
	return report, nil
}

func runCreatorScheduleOnce(ctx context.Context, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, credentials phasea.CredentialBridge) error {
	if store == nil {
		return creator.ErrUnavailable
	}
	settings, err := store.GetSettings(ctx)
	if err != nil {
		return err
	}
	now := time.Now().UTC()
	competitors, err := store.ListDueCompetitors(ctx, now)
	if err != nil {
		return err
	}
	for _, competitor := range competitors {
		accountID, err := creatorCollectionAccount(ctx, store, phaseAStore, hubStore, competitor.Platform)
		if err != nil {
			_ = store.MarkCompetitorSync(ctx, competitor.ID, "blocked", "", err.Error(), nil)
			logrus.WithError(err).WithField("competitor_id", competitor.ID).Warn("creator competitor sync blocked")
			continue
		}
		if _, err := syncCreatorCompetitorDue(ctx, store, phaseAStore, hubStore, credentials, competitor.ID, accountID); err != nil {
			logrus.WithError(err).WithField("competitor_id", competitor.ID).Warn("creator competitor scheduled sync failed")
		}
	}
	ownedAccounts, err := store.ListDueOwnedAccounts(ctx, now, settings.NewWorkIntervalSeconds)
	if err != nil {
		return err
	}
	for _, accountID := range ownedAccounts {
		if err := syncCreatorOwned(ctx, store, phaseAStore, hubStore, credentials, accountID, now); err != nil {
			logrus.WithError(err).WithField("account_id", accountID).Warn("creator owned scheduled sync failed")
		}
	}
	return nil
}

func syncCreatorOwned(ctx context.Context, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, credentials phasea.CredentialBridge, accountID string, now time.Time) error {
	if store == nil || phaseAStore == nil || hubStore == nil || credentials == nil || accountID == "" {
		return creator.ErrUnavailable
	}
	account, err := phaseAStore.GetAccount(ctx, accountID)
	if err != nil {
		return err
	}
	if account.Platform != creator.PlatformDouyin || account.AuthorizationStatus != "authorized" {
		return creator.ErrConflict
	}
	if _, err := store.GetAccountProfile(ctx, accountID); err != nil {
		return err
	}
	resolver, ok := credentials.(phasea.CredentialResolver)
	if !ok {
		return creator.ErrUnavailable
	}
	rawCredential, err := phaseAStore.ResolveAccountCredential(ctx, accountID, resolver)
	if err != nil {
		return fmt.Errorf("%w: resolve account credential: %v", creator.ErrUnavailable, err)
	}
	cookies, err := douyin.ParseCookieHeader(rawCredential)
	if err != nil {
		cookies, err = douyin.ParseCookieBundle(rawCredential)
	}
	if err != nil {
		return fmt.Errorf("%w: invalid account cookie bundle", creator.ErrConflict)
	}
	environment, err := hubStore.GetEnvironmentContextForAccount(ctx, accountID)
	if err != nil {
		return fmt.Errorf("%w: account environment unavailable: %v", creator.ErrUnavailable, err)
	}
	if environment.RuntimeID == "" || environment.RuntimeNetworkID == "" || environment.BindingVersion <= 0 {
		return fmt.Errorf("%w: account runtime is not running", creator.ErrUnavailable)
	}
	gateway, err := hubStore.GetGateway(ctx, environment.Gateway)
	if err != nil {
		return fmt.Errorf("%w: gateway unavailable: %v", creator.ErrUnavailable, err)
	}
	browser := creatorGatewayBrowser{gateway: gateway, environment: environment}
	if err := browser.SetCookies(ctx, cookies); err != nil {
		return fmt.Errorf("%w: set account cookies: %v", creator.ErrUnavailable, err)
	}
	collector := douyin.CreatorCollector{Browser: browser, AccountKey: account.PlatformAccountKey, SourceType: creator.SourceOwned, SourceID: account.ID}
	if err := collector.VerifyIdentity(ctx, account.PlatformAccountKey); err != nil {
		return fmt.Errorf("%w: account identity verification failed: %v", creator.ErrConflict, err)
	}
	_, err = store.CollectSource(ctx, account.Platform, creator.SourceOwned, account.ID, collector, now)
	return err
}

func creatorCollectionAccount(ctx context.Context, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, platform string) (string, error) {
	if store == nil || phaseAStore == nil || hubStore == nil || platform == "" {
		return "", creator.ErrUnavailable
	}
	accounts, err := phaseAStore.ListAccounts(ctx)
	if err != nil {
		return "", err
	}
	for _, account := range accounts {
		if account.Platform != platform || account.AuthorizationStatus != "authorized" {
			continue
		}
		if _, err := store.GetAccountProfile(ctx, account.ID); err != nil {
			continue
		}
		if _, err := hubStore.GetEnvironmentContextForAccount(ctx, account.ID); err != nil {
			continue
		}
		return account.ID, nil
	}
	return "", fmt.Errorf("%w: no authorized creator collection account", creator.ErrUnavailable)
}

func runCreatorScheduler(ctx context.Context, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, credentials phasea.CredentialBridge) {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()
	for {
		if err := runCreatorScheduleOnce(ctx, store, phaseAStore, hubStore, credentials); err != nil {
			logrus.WithError(err).Error("creator scheduler failed")
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}