rogee/creator-hub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: finish Douyin production readiness flow

Browse Source
This commit is contained in:
rogee
2026-09-15 09:20:51 +08:00
parent 5867aac20f
commit 977e541fef
28 changed files with 1257 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/control-plane/creator.go
+148
View File
@@ -743,6 +743,66 @@ func registerCreatorWithServices(app *fiber.App, store *creator.Store, phaseASto
}
return c.JSON(items)
})
app.Post("/api/creator/conversations/:id/sync", func(c fiber.Ctx) error {
conversation, err := store.GetConversation(c.Context(), c.Params("id"))
if err != nil {
return creatorError(c, err)
}
if conversation.Platform != creator.PlatformDouyin {
return creatorError(c, creator.ErrUnavailable)
}
limit := 100
if value := c.Query("limit"); value != "" {
limit, err = strconv.Atoi(value)
if err != nil || limit < 1 || limit > 200 {
return creatorError(c, creator.ErrInvalid)
}
}
profile, err := store.GetAccountProfile(c.Context(), conversation.AccountID)
if err != nil {
return creatorError(c, err)
}
account, err := phaseAStore.GetAccount(c.Context(), conversation.AccountID)
if err != nil {
return creatorError(c, err)
}
if account.Platform != creator.PlatformDouyin || account.AuthorizationStatus != "authorized" ||
profile.Platform != creator.PlatformDouyin || profile.BusinessStatus != "normal" || profile.LoginStatus != "logged_in" ||
profile.PlatformAccountKey == "" || account.PlatformAccountKey != profile.PlatformAccountKey {
return creatorError(c, creator.ErrConflict)
}
environment, err := hubStore.GetEnvironmentContextForAccount(c.Context(), conversation.AccountID)
if err != nil {
return creatorError(c, err)
}
gateway, err := hubStore.GetGateway(c.Context(), environment.Gateway)
if err != nil {
return creatorError(c, err)
}
browser := creatorGatewayBrowser{gateway: gateway, environment: environment}
accountUID, err := browser.Identity(c.Context(), profile.PlatformAccountKey)
if err != nil {
return creatorError(c, fmt.Errorf("%w: account identity verification failed: %v", creator.ErrConflict, err))
}
history, err := browser.MessageHistory(c.Context(), accountUID, conversation.PeerUID, limit)
if err != nil {
return creatorError(c, err)
}
if history.AccountUID != accountUID {
return creatorError(c, creator.ErrConflict)
}
inserted, err := persistDouyinMessageHistory(c.Context(), store, conversation, accountUID, history.Messages)
if err != nil {
return creatorError(c, err)
}
creatorUpdates.publish()
return c.JSON(map[string]any{
"conversation_id": conversation.ID,
"messages": inserted,
"history_source": history.HistorySource,
"history_has_more": history.HistoryHasMore,
})
})
app.Post("/api/creator/messages/send", func(c fiber.Ctx) error {
var input creator.MessageInput
if err := decodeCreator(c, &input); err != nil {
@@ -795,6 +855,69 @@ func registerCreatorWithServices(app *fiber.App, store *creator.Store, phaseASto
})
}
func persistDouyinMessageHistory(ctx context.Context, store *creator.Store, conversation creator.Conversation, accountUID string, messages []douyinHistoryMessage) (int, error) {
if store == nil || conversation.Platform != creator.PlatformDouyin || strings.TrimSpace(accountUID) == "" || strings.TrimSpace(conversation.PeerUID) == "" {
return 0, creator.ErrInvalid
}
inserted := 0
for _, item := range messages {
if strings.TrimSpace(item.ServerID) == "" || strings.TrimSpace(item.SenderUID) == "" {
return inserted, creator.ErrInvalid
}
messageType := creator.MessageTypeUnknown
text := ""
if len(item.Content) > 0 && string(item.Content) != "null" {
var payload struct {
Text string `json:"text"`
}
var encoded string
if err := json.Unmarshal(item.Content, &encoded); err == nil {
if err := json.Unmarshal([]byte(encoded), &payload); err != nil {
return inserted, creator.ErrInvalid
}
} else if err := json.Unmarshal(item.Content, &payload); err != nil {
return inserted, creator.ErrInvalid
}
text = payload.Text
if text != "" {
messageType = creator.MessageTypeText
}
}
var messageAt *time.Time
if item.CreatedAt != "" {
milliseconds, err := strconv.ParseInt(item.CreatedAt, 10, 64)
if err != nil || milliseconds <= 0 {
return inserted, creator.ErrInvalid
}
value := time.UnixMilli(milliseconds).UTC()
messageAt = &value
}
direction, state := "inbound", "received"
if item.SenderUID == accountUID {
direction, state = "outbound", "succeeded"
}
_, wasInserted, err := store.SaveMessage(ctx, creator.MessageInput{
Platform: creator.PlatformDouyin,
AccountID: conversation.AccountID,
PeerUID: conversation.PeerUID,
PeerName: conversation.PeerName,
PlatformMessageKey: item.ServerID,
Direction: direction,
MessageType: messageType,
Text: text,
SentState: state,
MessageAt: messageAt,
})
if err != nil {
return inserted, err
}
if wasInserted {
inserted++
}
}
return inserted, nil
}
func setStrategyEnabled(c fiber.Ctx, store *creator.Store, enabled bool) error {
item, err := store.SetStrategyEnabled(c.Context(), c.Params("id"), enabled)
if err != nil {
@@ -916,6 +1039,9 @@ func (executor creatorGatewayActionExecutor) Execute(ctx context.Context, reques
// the verified platform keys and the comment's owning work key.
if request.TargetCommentID != "" {
comment, targetErr := executor.store.GetComment(ctx, request.TargetCommentID)
if errors.Is(targetErr, creator.ErrNotFound) {
comment, targetErr = executor.store.GetCommentByKey(ctx, request.Platform, request.TargetCommentID)
}
if targetErr != nil {
return creator.ActionResult{}, targetErr
}
@@ -926,6 +1052,9 @@ func (executor creatorGatewayActionExecutor) Execute(ctx context.Context, reques
}
if request.TargetWorkID != "" {
work, targetErr := executor.store.GetWork(ctx, request.TargetWorkID)
if errors.Is(targetErr, creator.ErrNotFound) {
work, targetErr = executor.store.GetWorkByKey(ctx, request.Platform, request.TargetWorkID)
}
if targetErr != nil {
return creator.ActionResult{}, targetErr
}
@@ -1012,6 +1141,25 @@ func (browser creatorGatewayBrowser) Identity(ctx context.Context, expectedKey s
return identity.UID, nil
}
func (browser creatorGatewayBrowser) MessageHistory(ctx context.Context, expectedUID, targetUID string, limit int) (douyinMessageHistory, error) {
if strings.TrimSpace(expectedUID) == "" || strings.TrimSpace(targetUID) == "" || limit < 1 || limit > 200 {
return douyinMessageHistory{}, errors.New("invalid message history request")
}
payload := gatewayGenerationPayload(browser.environment)
payload["expected_uid"] = expectedUID
payload["target_uid"] = targetUID
payload["limit"] = limit
status, body, err := gatewayCall(ctx, browser.gateway, http.MethodPost, "/v1/browsers/"+url.PathEscape(browser.environment.Alias)+"/douyin/messages", payload, 30*time.Second)
if err != nil || status != http.StatusOK {
return douyinMessageHistory{}, errors.New("restricted message history operation failed")
}
var response douyinMessageHistory
if json.Unmarshal(body, &response) != nil || response.Status != "succeeded" || response.HistorySource == "" {
return douyinMessageHistory{}, errors.New("restricted message history response is invalid")
}
return response, nil
}
func verifyCreatorAccount(ctx context.Context, store *creator.Store, phaseAStore *phasea.Store, hubStore *hub.Store, accountID string) (creator.LoginResult, error) {
if store == nil || phaseAStore == nil || hubStore == nil || strings.TrimSpace(accountID) == "" {
return creator.LoginResult{}, creator.ErrUnavailable
cmd/control-plane/creator_events.go
+25 -24
View File
@@ -19,13 +19,14 @@ import (
const creatorEventReconcileInterval = 10 * time.Second
type creatorGatewayEvent struct {
DeliveryID string `json:"delivery_id,omitempty"`
Kind string `json:"kind"`
Reason string `json:"reason,omitempty"`
Continuity string `json:"continuity,omitempty"`
BoundaryAt string `json:"boundary_at,omitempty"`
Baseline bool `json:"baseline,omitempty"`
Notice *creatorGatewayEventNotice `json:"notice,omitempty"`
DeliveryID string `json:"delivery_id,omitempty"`
Kind string `json:"kind"`
Reason string `json:"reason,omitempty"`
Continuity string `json:"continuity,omitempty"`
BoundaryAt string `json:"boundary_at,omitempty"`
BoundarySource string `json:"boundary_source,omitempty"`
Baseline bool `json:"baseline,omitempty"`
Notice *creatorGatewayEventNotice `json:"notice,omitempty"`
}
type creatorGatewayEventNotice struct {
@@ -290,23 +291,9 @@ func runCreatorEventListener(ctx context.Context, store *creator.Store, binding
}
for _, event := range events {
if event.Kind == "baseline" {
if event.BoundaryAt != "" {
parsedBoundary, boundaryErr := time.Parse(time.RFC3339Nano, event.BoundaryAt)
if boundaryErr != nil {
logrus.WithError(boundaryErr).WithField("account_id", binding.accountID).Warn("creator event baseline timestamp is invalid")
ready = false
boundaryAt = time.Time{}
} else {
boundaryAt = parsedBoundary.UTC()
ready = true
}
} else {
// A marker without a platform-verified boundary cannot prove
// continuity. Keep all timestamped notices non-actionable.
ready = false
boundaryAt = time.Time{}
}
status, reason := "gap", "平台边界无效"
parsedBoundary, boundaryReady, reason := creatorGatewayBoundary(event)
boundaryAt, ready = parsedBoundary, boundaryReady
status := "gap"
if ready {
status, reason = "ready", ""
}
@@ -474,6 +461,20 @@ func handleCreatorGatewayEvent(ctx context.Context, store *creator.Store, bindin
}()
}
func creatorGatewayBoundary(event creatorGatewayEvent) (time.Time, bool, string) {
if event.BoundarySource != "douyin_identity_extra_now" {
return time.Time{}, false, "平台边界来源未验证"
}
if event.BoundaryAt == "" {
return time.Time{}, false, "平台边界无效"
}
parsedBoundary, err := time.Parse(time.RFC3339Nano, event.BoundaryAt)
if err != nil {
return time.Time{}, false, "平台边界无效"
}
return parsedBoundary.UTC(), true, ""
}
func creatorEventBeforeBoundary(event creatorGatewayEvent, boundaryAt time.Time) bool {
if boundaryAt.IsZero() || event.Notice == nil {
return false
cmd/control-plane/creator_events_test.go
+13
View File
@@ -52,6 +52,19 @@ func TestCreatorGatewayEventNeedsBaseline(t *testing.T) {
}
}
func TestCreatorGatewayBoundary(t *testing.T) {
boundary, ready, reason := creatorGatewayBoundary(creatorGatewayEvent{
BoundaryAt: "2024-01-01T00:00:00Z",
BoundarySource: "douyin_identity_extra_now",
})
if !ready || !boundary.Equal(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)) || reason != "" {
t.Fatalf("unexpected valid boundary: %v %v %q", boundary, ready, reason)
}
if _, ready, reason := creatorGatewayBoundary(creatorGatewayEvent{BoundaryAt: "2024-01-01T00:00:00Z"}); ready || reason != "平台边界来源未验证" {
t.Fatalf("unverified boundary must stay blocked: %v %q", ready, reason)
}
}
func TestCreatorEventBeforeBoundary(t *testing.T) {
boundary := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
before := creatorGatewayEvent{Notice: &creatorGatewayEventNotice{PlatformEventAt: "2023-12-31T23:59:59Z"}}
cmd/control-plane/douyin.go
+18
View File
@@ -49,6 +49,24 @@ func (browser douyinGatewayBrowser) Get(ctx context.Context, target string) (dou
return douyin.Response{Status: response.Status, Body: []byte(response.Body), Challenge: response.Challenge}, nil
}
type douyinHistoryMessage struct {
ServerID string `json:"server_id"`
SenderUID string `json:"sender_uid"`
MessageType string `json:"message_type"`
Content json.RawMessage `json:"content"`
CreatedAt string `json:"created_at"`
ServerState *int `json:"server_status"`
}
type douyinMessageHistory struct {
Status string `json:"status"`
HistorySource string `json:"history_source"`
HistoryHasMore bool `json:"history_has_more"`
AccountUID string `json:"account_uid"`
Conversation map[string]any `json:"conversation"`
Messages []douyinHistoryMessage `json:"messages"`
}
func (browser douyinGatewayBrowser) request() (douyinGatewayRequest, error) {
environment := browser.environment
if browser.gateway.Endpoint == "" || browser.gateway.Token == "" || !accountRunnable(environment) ||
cmd/control-plane/main.go
+39
View File
@@ -286,6 +286,30 @@ func newHandlerWithCreatorAndAI(webDirectory, username, password string, phaseAS
c.Status(fiber.StatusNoContent)
return nil
})
app.Get("/readyz", func(c fiber.Ctx) error {
if phaseAStore == nil || hubStore == nil || creatorStore == nil {
return c.Status(fiber.StatusServiceUnavailable).JSON(fiber.Map{"error": "service is not ready"})
}
ctx, cancel := context.WithTimeout(c.Context(), 2*time.Second)
defer cancel()
checks := []struct {
name string
fn func(context.Context) error
}{
{"phase_a", phaseAStore.Ping},
{"hub", hubStore.Ping},
{"creator", creatorStore.Ping},
{"creator_schema", creatorStore.EnsureSchema},
}
for _, check := range checks {
if err := check.fn(ctx); err != nil {
logrus.WithError(err).WithField("check", check.name).Warn("control plane readiness check failed")
return c.Status(fiber.StatusServiceUnavailable).JSON(fiber.Map{"error": "service is not ready"})
}
}
c.Status(fiber.StatusNoContent)
return nil
})
app.Use(authenticate(username, password))
if hubStore != nil {
registerHub(app, hubStore)
@@ -304,10 +328,25 @@ func newHandlerWithCreatorAndAI(webDirectory, username, password string, phaseAS
registerCreatorWithServices(app, creatorStore, phaseAStore, hubStore, executor, generator, analyzer)
}
}
app.Use(func(c fiber.Ctx) error {
if isControlPlaneAPIPath(c.Path()) {
return c.Status(fiber.StatusNotFound).JSON(fiber.Map{"error": "not found"})
}
return c.Next()
})
app.Get("/*", spaHandler(webDirectory))
return app
}
func isControlPlaneAPIPath(path string) bool {
for _, prefix := range []string{"/api", "/phase-a", "/gateways", "/browser-images", "/browsers", "/network-exits"} {
if path == prefix || strings.HasPrefix(path, prefix+"/") {
return true
}
}
return false
}
func authenticate(username, password string) fiber.Handler {
wantUser, wantPassword := sha256.Sum256([]byte(username)), sha256.Sum256([]byte(password))
return func(c fiber.Ctx) error {
cmd/control-plane/main_test.go
+19 -2
View File
@@ -134,6 +134,23 @@ func TestControlPlaneAuthentication(t *testing.T) {
}
health.Body.Close()
ready, err := app.Test(httptest.NewRequest(http.MethodGet, "/readyz", nil))
if err != nil || ready.StatusCode != http.StatusServiceUnavailable {
t.Fatalf("readiness must fail without stores: status=%d err=%v", ready.StatusCode, err)
}
ready.Body.Close()
unknown := httptest.NewRequest(http.MethodGet, "/api/not-registered", nil)
unknown.SetBasicAuth("operator", "unit-test-password")
unknownResponse, err := app.Test(unknown)
if err != nil || unknownResponse.StatusCode != http.StatusNotFound {
t.Fatalf("unknown API must be JSON 404: status=%d err=%v", unknownResponse.StatusCode, err)
}
if got := unknownResponse.Header.Get("Content-Type"); !strings.HasPrefix(got, "application/json") {
t.Fatalf("unknown API content type=%q", got)
}
unknownResponse.Body.Close()
for _, path := range []string{
"/", "/api/phase-a/accounts", "/api/browsers", "/api/network-exits", "/api/phase-a/tasks", "/api/phase-a/audit",
} {
@@ -262,8 +279,8 @@ func TestControlPlaneRegisteredRouteMatrix(t *testing.T) {
method, path string
want int
}{
{http.MethodPost, "/api/not-registered", http.StatusMethodNotAllowed},
{http.MethodDelete, "/api/not-registered", http.StatusMethodNotAllowed},
{http.MethodPost, "/api/not-registered", http.StatusNotFound},
{http.MethodDelete, "/api/not-registered", http.StatusNotFound},
} {
t.Run("unregistered "+route.method, func(t *testing.T) {
if response := do(app, route.method, route.path, ""); response.Code != http.StatusUnauthorized {
cmd/control-plane/xiaohongshu.go
+3 -2
View File
@@ -211,10 +211,11 @@ func newCreatorCollector(ctx context.Context, platform string, gateway hub.Gatew
return nil, "", err
}
collector := douyinCollector(browser, targetAccountKey, sourceType, sourceID)
if _, err := collector.CanonicalSecUID(ctx, viewerAccountKey); err != nil {
canonicalKey, err := collector.CanonicalTargetSecUID(ctx, targetAccountKey)
if err != nil {
return nil, "", err
}
return &collector, targetAccountKey, nil
return &collector, canonicalKey, nil
case creator.PlatformXiaohongshu:
if homepageURL != "" {
if err := xiaohongshu.ValidateSourceURL(homepageURL, targetAccountKey); err != nil {
cmd/control-plane/xiaohongshu_test.go
+28
View File
@@ -51,6 +51,34 @@ func TestXiaohongshuGatewayBrowserResolvesShareLinks(t *testing.T) {
}
}
func TestNewDouyinCollectorResolvesTargetIdentitySeparately(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
var body map[string]any
if json.NewDecoder(request.Body).Decode(&body) != nil {
t.Fatal("invalid gateway request")
}
switch request.URL.Path {
case "/v1/browsers/account-a/douyin/identity":
if body["expected_account_key"] != "viewer-1" {
t.Fatalf("viewer identity key=%#v", body["expected_account_key"])
}
_ = json.NewEncoder(response).Encode(map[string]any{"uid": "viewer-1", "sec_uid": "viewer-sec", "unique_id": "viewer"})
case "/v1/browsers/account-a/douyin/get":
if body["url"] != "https://www.douyin.com/aweme/v1/web/user/profile/other/?aid=6383&device_platform=webapp&user_id=2328120603967913" {
t.Fatalf("target profile URL=%#v", body["url"])
}
_ = json.NewEncoder(response).Encode(map[string]any{"status": 200, "body": `{"status_code":0,"user":{"uid":"2328120603967913","sec_uid":"target-sec","unique_id":"target"}}`, "challenge": ""})
default:
t.Fatalf("unexpected path: %s", request.URL.Path)
}
}))
defer server.Close()
collector, target, err := newCreatorCollector(context.Background(), creator.PlatformDouyin, hub.Gateway{Endpoint: server.URL, Token: "gateway-token-1"}, readyDouyinEnvironment(), "viewer-1", "2328120603967913", "", creator.SourceCompetitor, "source-1")
if err != nil || collector == nil || target != "target-sec" {
t.Fatalf("new Douyin collector: collector=%#v target=%q err=%v", collector, target, err)
}
}
func TestNewXiaohongshuCollectorKeepsViewerAndTargetSeparate(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
if request.URL.Path != "/v1/browsers/account-a/xiaohongshu/identity" {