858 lines
44 KiB
Go
858 lines
44 KiB
Go
package creator
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"net/url"
|
||
"os"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.ipao.vip/rogee/creator-hub/internal/account"
|
||
hub "git.ipao.vip/rogee/creator-hub/internal/environment"
|
||
_ "github.com/jackc/pgx/v5/stdlib"
|
||
)
|
||
|
||
type integrationCredentialBridge struct{}
|
||
|
||
func (integrationCredentialBridge) Store(context.Context, account.CredentialReference, string, string) error {
|
||
return nil
|
||
}
|
||
|
||
func (integrationCredentialBridge) Delete(context.Context, account.CredentialReference, string) error {
|
||
return nil
|
||
}
|
||
|
||
type integrationAnalyzer struct{}
|
||
|
||
func (integrationAnalyzer) MatchTheme(context.Context, string, string, string) (bool, string, error) {
|
||
return true, "主题匹配", nil
|
||
}
|
||
|
||
func (integrationAnalyzer) MatchLead(context.Context, string, string, string) (bool, string, error) {
|
||
return true, "具备线索意向", nil
|
||
}
|
||
|
||
type integrationExecutor struct{}
|
||
|
||
func (integrationExecutor) Execute(context.Context, ActionRequest) (ActionResult, error) {
|
||
return ActionResult{State: "succeeded", Evidence: map[string]string{"platform_id": "creator-it"}}, nil
|
||
}
|
||
|
||
type integrationCollector struct {
|
||
work WorkInput
|
||
comment CommentInput
|
||
}
|
||
|
||
func (c integrationCollector) VerifyIdentity(context.Context, string) error { return nil }
|
||
|
||
func (c integrationCollector) ListWorks(_ context.Context, _, cursor string) (WorkPage, error) {
|
||
if cursor != "" {
|
||
return WorkPage{Items: nil, HasMore: false}, nil
|
||
}
|
||
return WorkPage{Items: []WorkInput{c.work}, HasMore: true, NextCursor: "1"}, nil
|
||
}
|
||
|
||
func (c integrationCollector) ListTopLevelComments(_ context.Context, _, cursor string) (CommentPage, error) {
|
||
if cursor != "" {
|
||
return CommentPage{Items: nil, HasMore: false}, nil
|
||
}
|
||
return CommentPage{Items: []CommentInput{c.comment}, HasMore: false}, nil
|
||
}
|
||
|
||
func openCreatorIntegrationStore(t *testing.T) (*Store, *account.Store, context.Context) {
|
||
t.Helper()
|
||
databaseURL := os.Getenv("CREATORHUB_POSTGRES_TEST_URL")
|
||
if databaseURL == "" {
|
||
t.Skip("set CREATORHUB_POSTGRES_TEST_URL to run CreatorHub PostgreSQL integration coverage")
|
||
}
|
||
ctx := context.Background()
|
||
admin, err := sql.Open("pgx", databaseURL)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() { _ = admin.Close() })
|
||
if err := admin.PingContext(ctx); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
schema := fmt.Sprintf("creatorhub_it_%d", time.Now().UnixNano())
|
||
if _, err := admin.ExecContext(ctx, "CREATE SCHEMA "+schema); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() {
|
||
if _, err := admin.ExecContext(ctx, "DROP SCHEMA "+schema+" CASCADE"); err != nil {
|
||
t.Errorf("drop test schema: %v", err)
|
||
}
|
||
})
|
||
parsed, err := url.Parse(databaseURL)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
query := parsed.Query()
|
||
query.Set("search_path", schema)
|
||
parsed.RawQuery = query.Encode()
|
||
testURL := parsed.String()
|
||
|
||
phaseAStore, err := account.Open(ctx, testURL)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() { _ = phaseAStore.Close() })
|
||
hubStore, err := hub.Open(ctx, testURL)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() { _ = hubStore.Close() })
|
||
store, err := Open(ctx, testURL)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() { _ = store.Close() })
|
||
return store, phaseAStore, ctx
|
||
}
|
||
|
||
func createIntegrationAccount(t *testing.T, ctx context.Context, phaseAStore *account.Store, suffix string) string {
|
||
t.Helper()
|
||
id := "cit" + suffix
|
||
account := account.Account{
|
||
ID: id,
|
||
Name: "Creator integration " + suffix,
|
||
Platform: PlatformDouyin,
|
||
PlatformAccountKey: "sec_uid_" + id,
|
||
Tags: []string{"integration"},
|
||
CredentialReference: account.CredentialReference{ID: "credential_" + id, Provider: "os_keyring"},
|
||
CredentialKey: "creatorhub/" + id,
|
||
}
|
||
if err := phaseAStore.CreateAccount(ctx, account, integrationCredentialBridge{}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return id
|
||
}
|
||
|
||
func TestCompetitorShareJobLifecycle(t *testing.T) {
|
||
store, _, ctx := openCreatorIntegrationStore(t)
|
||
stamp := fmt.Sprintf("%d", time.Now().UnixNano())
|
||
job, err := store.CreateCompetitorShareJob(ctx, CompetitorShareJobInput{
|
||
Platform: PlatformDouyin,
|
||
ShareURL: "https://v.douyin.com/share-" + stamp + "/",
|
||
Tags: []string{"重点监测"},
|
||
})
|
||
if err != nil || job.Status != CompetitorShareJobQueued || job.Attempts != 0 {
|
||
t.Fatalf("create share job: job=%+v err=%v", job, err)
|
||
}
|
||
due, err := store.ListDueCompetitorShareJobs(ctx, time.Now().UTC())
|
||
if err != nil || len(due) != 1 || due[0].ID != job.ID {
|
||
t.Fatalf("list due share jobs: jobs=%+v err=%v", due, err)
|
||
}
|
||
claimed, leaseToken, ok, err := store.ClaimCompetitorShareJob(ctx, job.ID, time.Now().UTC())
|
||
if err != nil || !ok || leaseToken == "" || claimed.Attempts != 1 || claimed.Status != CompetitorShareJobProcessing {
|
||
t.Fatalf("claim share job: job=%+v token=%q claimed=%v err=%v", claimed, leaseToken, ok, err)
|
||
}
|
||
if err := store.MarkCompetitorShareJob(ctx, job.ID, leaseToken, CompetitorShareJobQueued, "", "temporary failure"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
claimed, leaseToken, ok, err = store.ClaimCompetitorShareJob(ctx, job.ID, time.Now().UTC())
|
||
if err != nil || !ok || leaseToken == "" || claimed.Attempts != 2 {
|
||
t.Fatalf("claim retry share job: job=%+v token=%q claimed=%v err=%v", claimed, leaseToken, ok, err)
|
||
}
|
||
if err := store.MarkCompetitorShareJob(ctx, job.ID, leaseToken, CompetitorShareJobFailed, "", "final failure"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
failed, err := store.GetCompetitorShareJob(ctx, job.ID)
|
||
if err != nil || failed.Status != CompetitorShareJobFailed || failed.Attempts != 2 || failed.FailureReason != "final failure" {
|
||
t.Fatalf("failed share job: job=%+v err=%v", failed, err)
|
||
}
|
||
requeued, err := store.RetryCompetitorShareJob(ctx, job.ID)
|
||
if err != nil || requeued.Status != CompetitorShareJobQueued || requeued.Attempts != 0 || requeued.FailureReason != "final failure" {
|
||
t.Fatalf("requeue share job: job=%+v err=%v", requeued, err)
|
||
}
|
||
}
|
||
|
||
func TestCompetitorUpsertIsIdempotent(t *testing.T) {
|
||
store, _, ctx := openCreatorIntegrationStore(t)
|
||
stamp := fmt.Sprintf("%d", time.Now().UnixNano())
|
||
input := CompetitorInput{
|
||
Platform: PlatformDouyin,
|
||
PlatformAccountKey: "sec_uid_upsert_" + stamp,
|
||
UniqueID: "upsert_" + stamp,
|
||
Nickname: "Competitor",
|
||
HomepageURL: "https://www.douyin.com/user/sec_uid_upsert_" + stamp,
|
||
Tags: []string{"重点监测"},
|
||
}
|
||
first, err := store.UpsertCompetitor(ctx, input)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
second, err := store.UpsertCompetitor(ctx, CompetitorInput{
|
||
Platform: input.Platform,
|
||
PlatformAccountKey: input.PlatformAccountKey,
|
||
Nickname: "Updated Competitor",
|
||
HomepageURL: input.HomepageURL,
|
||
})
|
||
if err != nil || second.ID != first.ID || second.Nickname != "Updated Competitor" || len(second.Tags) != 1 {
|
||
t.Fatalf("upsert competitor: first=%+v second=%+v err=%v", first, second, err)
|
||
}
|
||
}
|
||
|
||
func TestCreatorPostgresContentAndWorkflow(t *testing.T) {
|
||
store, phaseAStore, ctx := openCreatorIntegrationStore(t)
|
||
stamp := fmt.Sprintf("%d", time.Now().UnixNano())
|
||
bigID := createIntegrationAccount(t, ctx, phaseAStore, "big"+stamp)
|
||
smallID := createIntegrationAccount(t, ctx, phaseAStore, "small"+stamp)
|
||
if err := store.EnsureAccountProfile(ctx, bigID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := store.EnsureAccountProfile(ctx, smallID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
profiles, err := store.ListAccountProfiles(ctx)
|
||
if err != nil || len(profiles) != 2 {
|
||
t.Fatalf("list account profiles: profiles=%+v err=%v", profiles, err)
|
||
}
|
||
updatedTags, err := store.UpdateAccountTags(ctx, bigID, []string{"主账号"})
|
||
if err != nil || len(updatedTags) != 1 || updatedTags[0] != "主账号" {
|
||
t.Fatalf("update account tags: tags=%v err=%v", updatedTags, err)
|
||
}
|
||
if _, err := store.UpdateAccountProfile(ctx, bigID, AccountProfileUpdate{RealNameStatus: "unknown", BusinessStatus: "normal", BigAccount: true, ReplyRequirements: "保持准确", CooldownSeconds: 86400}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.UpdateAccountProfile(ctx, smallID, AccountProfileUpdate{RealNameStatus: "unknown", BusinessStatus: "normal", CooldownSeconds: 86400}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.RecordVerifiedLoginResult(ctx, bigID, "sec_uid_"+bigID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.RecordVerifiedLoginResult(ctx, smallID, "sec_uid_"+smallID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := store.SetRelation(ctx, bigID, smallID, true); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
relations, err := store.ListRelations(ctx, bigID)
|
||
if err != nil || len(relations) != 1 {
|
||
t.Fatalf("list relations: relations=%+v err=%v", relations, err)
|
||
}
|
||
if _, err := store.SetBigAccount(ctx, smallID, true); !errors.Is(err, ErrConflict) {
|
||
t.Fatalf("expected small-account promotion to be rejected, got %v", err)
|
||
}
|
||
authorizedID := createIntegrationAccount(t, ctx, phaseAStore, "authorized"+stamp)
|
||
if _, err := store.db.ExecContext(ctx, `UPDATE social_account SET authorization_kind = 'authorized' WHERE id = $1`, authorizedID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := store.SetRelation(ctx, bigID, authorizedID, true); !errors.Is(err, ErrInvalid) {
|
||
t.Fatalf("expected authorized account relation to be rejected, got %v", err)
|
||
}
|
||
|
||
now := time.Now().UTC().Truncate(time.Microsecond)
|
||
ownedDue, err := store.ListDueOwnedAccounts(ctx, now, 1800)
|
||
if err != nil || len(ownedDue) != 2 || ownedDue[0] != bigID || ownedDue[1] != smallID {
|
||
t.Fatalf("list due owned accounts: accounts=%+v err=%v", ownedDue, err)
|
||
}
|
||
published := now.Add(-2 * time.Hour)
|
||
likes, comments, shares := int64(10), int64(2), int64(1)
|
||
plays := int64(777)
|
||
workKey := "creator-it-work-" + stamp
|
||
work, inserted, err := store.UpsertWork(ctx, WorkInput{Platform: PlatformDouyin, WorkKey: workKey, SourceType: SourceOwned, SourceID: bigID, Title: "Topic", Body: "Need consultation", PublishedAt: &published, PublishedAtStatus: "verified", OriginalURL: "https://www.douyin.com/video/" + workKey, Likes: &likes, CommentsCount: &comments, Shares: &shares, PlayCount: &plays}, now)
|
||
if err != nil || !inserted {
|
||
t.Fatalf("insert work: work=%+v inserted=%v err=%v", work, inserted, err)
|
||
}
|
||
if work.PlayCount == nil || *work.PlayCount != 777 {
|
||
t.Fatalf("play_count not persisted on insert: %+v", work)
|
||
}
|
||
// COALESCE 语义：二次 upsert 未带 play_count 时保留旧值。
|
||
if _, _, err := store.UpsertWork(ctx, WorkInput{Platform: PlatformDouyin, WorkKey: workKey, SourceType: SourceOwned, SourceID: bigID, Title: "Topic-renamed", PublishedAt: &published, PublishedAtStatus: "verified"}, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
retained, err := store.GetWorkByKey(ctx, PlatformDouyin, workKey)
|
||
if err != nil || retained.PlayCount == nil || *retained.PlayCount != 777 {
|
||
t.Fatalf("play_count must be retained via COALESCE: work=%+v err=%v", retained, err)
|
||
}
|
||
// 负播放量拒绝。
|
||
if _, _, err := store.UpsertWork(ctx, WorkInput{Platform: PlatformDouyin, WorkKey: workKey + "-neg", SourceType: SourceOwned, SourceID: bigID, Title: "neg", PlayCount: &[]int64{-1}[0]}, now); !errors.Is(err, ErrInvalid) {
|
||
t.Fatalf("negative play_count must be rejected: err=%v", err)
|
||
}
|
||
competitor, err := store.CreateCompetitor(ctx, CompetitorInput{Platform: PlatformDouyin, PlatformAccountKey: "sec_uid_competitor_" + stamp, UniqueID: "competitor_" + stamp, Nickname: "Competitor", HomepageURL: "https://www.douyin.com/user/sec_uid_competitor_" + stamp, Tags: []string{"重点监测"}})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if competitor.UniqueID != "competitor_"+stamp || len(competitor.Tags) != 1 || competitor.Tags[0] != "重点监测" {
|
||
t.Fatalf("create competitor tags: %+v", competitor.Tags)
|
||
}
|
||
updatedCompetitor, err := store.UpdateCompetitorTags(ctx, competitor.ID, []string{"已分类"})
|
||
if err != nil || len(updatedCompetitor.Tags) != 1 || updatedCompetitor.Tags[0] != "已分类" {
|
||
t.Fatalf("update competitor tags: competitor=%+v err=%v", updatedCompetitor, err)
|
||
}
|
||
competitors, err := store.ListCompetitorsWithProfile(ctx, PlatformDouyin)
|
||
if err != nil || len(competitors) != 1 || len(competitors[0].Tags) != 1 || competitors[0].Tags[0] != "已分类" || competitors[0].WorkCount != 0 {
|
||
t.Fatalf("list competitors: competitors=%+v err=%v", competitors, err)
|
||
}
|
||
if _, err := store.SetCompetitorEnabled(ctx, competitor.ID, false); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.SetCompetitorEnabled(ctx, competitor.ID, true); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
dueNow := time.Now().UTC().Add(time.Minute)
|
||
if due, err := store.ListDueCompetitors(ctx, dueNow); err != nil || len(due) != 1 {
|
||
t.Fatalf("list due competitors: due=%+v err=%v", due, err)
|
||
}
|
||
leaseToken, claimed, err := store.ClaimCompetitorSync(ctx, competitor.ID, false, dueNow)
|
||
if err != nil || !claimed || leaseToken == "" {
|
||
t.Fatalf("claim competitor sync: token=%q claimed=%v err=%v", leaseToken, claimed, err)
|
||
}
|
||
if err := store.MarkCompetitorSync(ctx, competitor.ID, leaseToken, "idle", "", "", nil); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// blocked 同步也必须排下一次重试：否则 next_sync_at 置空后定时获取对该账号永久失效。
|
||
blockedToken, blockedClaimed, err := store.ClaimCompetitorSync(ctx, competitor.ID, true, dueNow)
|
||
if err != nil || !blockedClaimed {
|
||
t.Fatalf("claim competitor sync for blocked: err=%v claimed=%v", err, blockedClaimed)
|
||
}
|
||
blockedNext := dueNow.Add(2 * time.Hour)
|
||
if err := store.MarkCompetitorSync(ctx, competitor.ID, blockedToken, "blocked", "", "runtime unavailable", &blockedNext); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if due, err := store.ListDueCompetitors(ctx, blockedNext); err != nil || len(due) != 1 || due[0].ID != competitor.ID {
|
||
t.Fatalf("blocked competitor must reschedule: due=%+v err=%v", due, err)
|
||
}
|
||
if due, err := store.ListDueCompetitors(ctx, dueNow); err != nil || len(due) != 0 {
|
||
t.Fatalf("blocked competitor must not fire before retry time: due=%+v err=%v", due, err)
|
||
}
|
||
// 强制同步语义：next_sync_at 未到期时 force claim 仍立即执行；完成后 MarkCompetitorSync 刷新 next_sync_at。
|
||
if _, claimed, err := store.ClaimCompetitorSync(ctx, competitor.ID, false, dueNow); err != nil || claimed {
|
||
t.Fatalf("non-forced claim must wait for next_sync_at: claimed=%v err=%v", claimed, err)
|
||
}
|
||
forceToken, forceClaimed, err := store.ClaimCompetitorSync(ctx, competitor.ID, true, dueNow)
|
||
if err != nil || !forceClaimed {
|
||
t.Fatalf("forced claim must run before next_sync_at: err=%v claimed=%v", err, forceClaimed)
|
||
}
|
||
forceNext := dueNow.Add(30 * time.Minute)
|
||
if err := store.MarkCompetitorSync(ctx, competitor.ID, forceToken, "idle", "", "", &forceNext); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// Postgres timestamp 微秒精度会截断纳秒，比较用秒级容差。
|
||
if refreshed, err := store.GetCompetitor(ctx, competitor.ID); err != nil || refreshed.NextSyncAt == nil || refreshed.NextSyncAt.Sub(forceNext).Abs() > time.Second {
|
||
t.Fatalf("forced sync must refresh next_sync_at: next=%v err=%v", refreshed.NextSyncAt, err)
|
||
}
|
||
work, inserted, err = store.UpsertWork(ctx, WorkInput{Platform: PlatformDouyin, WorkKey: workKey, SourceType: SourceCompetitor, SourceID: competitor.ID, Title: "", Body: "", PublishedAt: nil, Likes: nil, CommentsCount: nil, Shares: nil}, now)
|
||
if err != nil || inserted || len(work.Sources) != 2 {
|
||
t.Fatalf("work source association was not retained: work=%+v inserted=%v err=%v", work, inserted, err)
|
||
}
|
||
filtered, err := store.ListWorks(ctx, WorkFilter{Platform: PlatformDouyin, SourceType: SourceCompetitor})
|
||
if err != nil || len(filtered) != 1 || filtered[0].ID != work.ID {
|
||
t.Fatalf("competitor source filter failed: works=%+v err=%v", filtered, err)
|
||
}
|
||
|
||
// 画像聚合视图：作品数与最近发布时间来自 creator_work 聚合。
|
||
views, err := store.ListCompetitorsWithProfile(ctx, PlatformDouyin)
|
||
if err != nil || len(views) != 1 {
|
||
t.Fatalf("list competitor views: views=%+v err=%v", views, err)
|
||
}
|
||
if views[0].WorkCount != 1 || views[0].LatestPublishedAt == nil {
|
||
t.Fatalf("competitor view aggregates: work_count=%d latest=%v", views[0].WorkCount, views[0].LatestPublishedAt)
|
||
}
|
||
if _, err := store.UpdateCompetitorProfile(ctx, competitor.ID, "https://example.invalid/avatar.jpg", 1200, 300, 88); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
views, err = store.ListCompetitorsWithProfile(ctx, "")
|
||
if err != nil || len(views) != 1 || views[0].FollowerCount == nil || *views[0].FollowerCount != 1200 ||
|
||
views[0].FollowingCount == nil || *views[0].FollowingCount != 300 || views[0].AwemeCount == nil || *views[0].AwemeCount != 88 ||
|
||
views[0].AvatarURL != "https://example.invalid/avatar.jpg" {
|
||
t.Fatalf("competitor profile fields: %+v err=%v", views, err)
|
||
}
|
||
|
||
comment, inserted, err := store.SaveComment(ctx, CommentInput{Platform: PlatformDouyin, CommentKey: "creator-it-comment-" + stamp, WorkID: work.ID, Content: "咨询价格", CommentType: "top_level"})
|
||
if err != nil || !inserted || comment.AuthorUID != "" {
|
||
t.Fatalf("save comment with missing UID: comment=%+v inserted=%v err=%v", comment, inserted, err)
|
||
}
|
||
if _, duplicate, err := store.SaveComment(ctx, CommentInput{Platform: PlatformDouyin, CommentKey: comment.CommentKey, WorkID: work.ID, Content: "咨询价格", CommentType: "top_level"}); err != nil || duplicate {
|
||
t.Fatalf("comment deduplication failed: duplicate=%v err=%v", duplicate, err)
|
||
}
|
||
collects := int64(4)
|
||
playSnap := int64(5566)
|
||
if _, err := store.RecordMetric(ctx, MetricInput{WorkID: work.ID, CollectedAt: now, Likes: &likes, CommentsCount: &comments, Shares: &shares, CollectCount: &collects, PlayCount: &playSnap}, Settings{LookbackDays: 30, NewWorkIntervalSeconds: 1800, MetricInitialIntervalSeconds: 3600, MetricMultiplier: 2, MetricMaxIntervalSeconds: 86400, MetricAgeSeconds: 30 * 24 * 60 * 60}, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
metrics, err := store.ListMetrics(ctx, work.ID)
|
||
if err != nil || len(metrics) != 1 {
|
||
t.Fatalf("initial metric snapshot failed: metrics=%+v err=%v", metrics, err)
|
||
}
|
||
if metrics[0].CollectCount == nil || *metrics[0].CollectCount != 4 {
|
||
t.Fatalf("collect_count not persisted: %+v", metrics[0])
|
||
}
|
||
if metrics[0].PlayCount == nil || *metrics[0].PlayCount != 5566 {
|
||
t.Fatalf("play_count not persisted in metric point: %+v", metrics[0])
|
||
}
|
||
// 增速计算：独立作品（避免干扰上方计划推进断言），两个采样点。
|
||
growthPublished := now.Add(-2 * time.Hour)
|
||
growthLikes0, growthLikes1 := int64(100), int64(400)
|
||
growthWork, _, err := store.UpsertWork(ctx, WorkInput{Platform: PlatformDouyin, WorkKey: "creator-it-growth-work-" + stamp, SourceType: SourceOwned, SourceID: smallID, Title: "Growth", Body: "body", PublishedAt: &growthPublished, PublishedAtStatus: "verified", Likes: &growthLikes0}, growthPublished.Add(time.Hour))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
growthSettings := Settings{LookbackDays: 30, NewWorkIntervalSeconds: 1800, MetricInitialIntervalSeconds: 3600, MetricMultiplier: 2, MetricMaxIntervalSeconds: 86400, MetricAgeSeconds: 30 * 24 * 60 * 60}
|
||
growthPoint0 := now.Add(-90 * time.Minute) // 首点（晚于发布时间）
|
||
if _, err := store.recordMetricWithPlan(ctx, MetricInput{WorkID: growthWork.ID, CollectedAt: growthPoint0, Likes: &growthLikes0}, growthSettings, growthPoint0); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
growthPoint1 := growthPoint0.Add(2 * time.Hour) // 第二个计划点：next = 首点+2h
|
||
if _, err := store.recordMetricWithPlan(ctx, MetricInput{WorkID: growthWork.ID, CollectedAt: growthPoint1, Likes: &growthLikes1}, growthSettings, growthPoint1); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
growth, err := store.LoadWorksGrowth(ctx, []string{growthWork.ID, work.ID}, WorkGrowthFilter{Hours: 24})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
stat, ok := growth[growthWork.ID]
|
||
if !ok || stat.Likes == nil || *stat.Likes != 300 || stat.Coverage != "partial" {
|
||
t.Fatalf("growth not computed: stat=%+v ok=%v", stat, ok)
|
||
}
|
||
if stat, ok := growth[work.ID]; !ok || stat.Likes == nil || *stat.Likes != 0 || stat.Coverage != "partial" {
|
||
t.Fatalf("single-point work growth: %+v ok=%v", stat, ok)
|
||
}
|
||
if _, err := store.RecordMetric(ctx, MetricInput{WorkID: work.ID, CollectedAt: now.Add(-time.Minute), Likes: &likes, CommentsCount: &comments, Shares: &shares}, Settings{LookbackDays: 30, NewWorkIntervalSeconds: 1800, MetricInitialIntervalSeconds: 3600, MetricMultiplier: 2, MetricMaxIntervalSeconds: 86400, MetricAgeSeconds: 30 * 24 * 60 * 60}, now); !errors.Is(err, ErrConflict) {
|
||
t.Fatalf("expected early metric point to be rejected, got %v", err)
|
||
}
|
||
if _, err := store.RecordMetric(ctx, MetricInput{WorkID: work.ID, CollectedAt: now.Add(time.Minute), Likes: &likes, CommentsCount: &comments, Shares: &shares}, Settings{LookbackDays: 30, NewWorkIntervalSeconds: 1800, MetricInitialIntervalSeconds: 3600, MetricMultiplier: 2, MetricMaxIntervalSeconds: 86400, MetricAgeSeconds: 30 * 24 * 60 * 60}, now.Add(-time.Minute)); !errors.Is(err, ErrInvalid) {
|
||
t.Fatalf("expected future metric point to be rejected, got %v", err)
|
||
}
|
||
|
||
rule, err := store.CreateRule(ctx, LeadRuleInput{Name: "咨询线索", Enabled: true, SourceType: SourceOwned, Topic: "咨询", IncludeKeywords: []string{"咨询"}, ExcludeKeywords: []string{"招聘"}, AIRequirement: "判断购买意向"})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
rules, err := store.ListRules(ctx, false)
|
||
if err != nil || len(rules) != 1 {
|
||
t.Fatalf("list rules: rules=%+v err=%v", rules, err)
|
||
}
|
||
updatedRule, err := store.UpdateRule(ctx, rule.ID, LeadRuleInput{Name: rule.Name, Enabled: true, SourceType: SourceOwned, Topic: rule.Topic, IncludeKeywords: []string{"咨询"}, ExcludeKeywords: []string{"招聘"}, AIRequirement: rule.AIRequirement})
|
||
if err != nil || updatedRule.ID != rule.ID {
|
||
t.Fatalf("update rule: rule=%+v err=%v", updatedRule, err)
|
||
}
|
||
result, err := store.AnalyzeComment(ctx, comment.ID, rule.ID, integrationAnalyzer{})
|
||
if err != nil || result.Status != "lead" {
|
||
t.Fatalf("analyze comment: result=%+v err=%v", result, err)
|
||
}
|
||
ruleResults, err := store.ListRuleResults(ctx, comment.ID, rule.ID)
|
||
if err != nil || len(ruleResults) != 1 {
|
||
t.Fatalf("list rule results: results=%+v err=%v", ruleResults, err)
|
||
}
|
||
leads, err := store.ListLeads(ctx, PlatformDouyin)
|
||
if err != nil || len(leads) != 1 {
|
||
t.Fatalf("list leads: leads=%+v err=%v", leads, err)
|
||
}
|
||
if _, err := store.SetRuleEnabled(ctx, rule.ID, false); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
material, selected, err := store.SelectMaterial(ctx, work.ID)
|
||
if err != nil || !selected || !material.Selected {
|
||
t.Fatalf("select material: material=%+v selected=%v err=%v", material, selected, err)
|
||
}
|
||
for step, status := range map[string]string{"download": "succeeded", "audio": "no_audio", "transcription": "no_speech"} {
|
||
if _, err := store.SetMaterialStep(ctx, work.ID, step, status, "ref-"+step, ""); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
if _, err := store.ConfirmRewrite(ctx, work.ID, "仿写同主题但不复制原文"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.SaveRewrite(ctx, work.ID, "新标题", "新口播"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
collector := integrationCollector{work: WorkInput{Platform: PlatformDouyin, WorkKey: "creator-it-collected-" + stamp, SourceType: SourceOwned, SourceID: bigID, Title: "Collected", Body: "body", PublishedAt: &published, PublishedAtStatus: "verified", Likes: &likes}, comment: CommentInput{Platform: PlatformDouyin, CommentKey: "creator-it-collected-comment-" + stamp, WorkID: "", Content: "hello", CommentType: "top_level"}}
|
||
collector.comment.WorkID = ""
|
||
report, err := store.CollectSource(ctx, PlatformDouyin, SourceOwned, bigID, collector, now)
|
||
if err != nil || !report.PaginationComplete || report.WorksSeen != 1 || report.CommentsSaved != 2 {
|
||
t.Fatalf("collect source workflow failed: report=%+v err=%v", report, err)
|
||
}
|
||
}
|
||
|
||
func TestCreatorPostgresAccountMetricSnapshot(t *testing.T) {
|
||
store, phaseAStore, ctx := openCreatorIntegrationStore(t)
|
||
stamp := fmt.Sprintf("%d", time.Now().UnixNano())
|
||
accountID := createIntegrationAccount(t, ctx, phaseAStore, "metric"+stamp)
|
||
now := time.Now().UTC().Truncate(time.Second)
|
||
follower, following, favorited, aweme := int64(1200), int64(56), int64(56789), int64(34)
|
||
if err := store.RecordAccountMetric(ctx, AccountMetricInput{AccountID: accountID, CollectedAt: now, FollowerCount: &follower, FollowingCount: &following, TotalFavorited: &favorited, AwemeCount: &aweme}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
points, err := store.ListAccountMetrics(ctx, accountID)
|
||
if err != nil || len(points) != 1 {
|
||
t.Fatalf("list account metrics: points=%+v err=%v", points, err)
|
||
}
|
||
if points[0].FollowerCount == nil || *points[0].FollowerCount != 1200 || points[0].TotalFavorited == nil || *points[0].TotalFavorited != 56789 {
|
||
t.Fatalf("account metric snapshot: %+v", points[0])
|
||
}
|
||
// UNIQUE(account_id, collected_at) 覆盖语义：同一时刻重采更新而非新增。
|
||
newFollower := int64(1300)
|
||
if err := store.RecordAccountMetric(ctx, AccountMetricInput{AccountID: accountID, CollectedAt: now, FollowerCount: &newFollower}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
points, err = store.ListAccountMetrics(ctx, accountID)
|
||
if err != nil || len(points) != 1 || points[0].FollowerCount == nil || *points[0].FollowerCount != 1300 {
|
||
t.Fatalf("account metric upsert must overwrite same timestamp: points=%+v err=%v", points, err)
|
||
}
|
||
// 不同时刻追加为独立点。
|
||
if err := store.RecordAccountMetric(ctx, AccountMetricInput{AccountID: accountID, CollectedAt: now.Add(30 * time.Minute), FollowerCount: &newFollower}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
points, err = store.ListAccountMetrics(ctx, accountID)
|
||
if err != nil || len(points) != 2 || !points[0].CollectedAt.Before(points[1].CollectedAt) {
|
||
t.Fatalf("account metric timeline must append: points=%+v err=%v", points, err)
|
||
}
|
||
// 负值拒绝。
|
||
bad := int64(-1)
|
||
if err := store.RecordAccountMetric(ctx, AccountMetricInput{AccountID: accountID, CollectedAt: now, FollowerCount: &bad}); !errors.Is(err, ErrInvalid) {
|
||
t.Fatalf("negative follower_count must be rejected: err=%v", err)
|
||
}
|
||
// 账号不存在时 FK 拒绝。
|
||
if err := store.RecordAccountMetric(ctx, AccountMetricInput{AccountID: "account-missing-" + fmt.Sprint(stamp), CollectedAt: now}); err == nil {
|
||
t.Fatal("missing account must violate FK")
|
||
}
|
||
}
|
||
|
||
func TestCreatorPostgresAccountMonitorViewAndCollectionStatus(t *testing.T) {
|
||
store, phaseAStore, ctx := openCreatorIntegrationStore(t)
|
||
stamp := fmt.Sprintf("%d", time.Now().UnixNano())
|
||
accountID := createIntegrationAccount(t, ctx, phaseAStore, "metric"+stamp)
|
||
now := time.Now().UTC().Truncate(time.Second)
|
||
|
||
// 无 checkpoint：采集状态为 pending。
|
||
status, err := store.GetAccountCollectionStatus(ctx, accountID)
|
||
if err != nil || status.Works.Status != "pending" || status.Comments.Status != "pending" {
|
||
t.Fatalf("pending collection status: %+v err=%v", status, err)
|
||
}
|
||
// 无作品：监控视图作品数为 0。
|
||
views, err := store.ListAccountMonitorViews(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
viewFor := func(id string) *AccountMonitorView {
|
||
for index := range views {
|
||
if views[index].ID == id {
|
||
return &views[index]
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
view := viewFor(accountID)
|
||
if view == nil || view.WorkCount != 0 || view.LatestPublishedAt != nil {
|
||
t.Fatalf("monitor view without works: %+v", view)
|
||
}
|
||
|
||
// 写入 checkpoint（succeeded + 窗口）与 owned 作品后：状态带下次窗口，视图聚合作品数。
|
||
lease, err := store.beginCheckpoint(ctx, SourceOwned, accountID, "works", now.Add(-24*time.Hour), now)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := store.finishCheckpoint(ctx, SourceOwned, accountID, "works", lease, "succeeded", ""); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
published := now.Add(-time.Hour)
|
||
if _, _, err := store.UpsertWork(ctx, WorkInput{Platform: PlatformDouyin, WorkKey: "monitor-view-" + stamp, SourceType: SourceOwned, SourceID: accountID, Title: "MonitorView", PublishedAt: &published, PublishedAtStatus: "verified"}, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
status, err = store.GetAccountCollectionStatus(ctx, accountID)
|
||
if err != nil || status.Works.Status != "succeeded" || status.Works.LastCompletedAt == nil ||
|
||
status.Works.NextWindowStart == nil || status.Works.NextWindowEnd == nil {
|
||
t.Fatalf("succeeded collection status must carry next window: %+v err=%v", status, err)
|
||
}
|
||
if !status.Works.NextWindowStart.Before(*status.Works.NextWindowEnd) {
|
||
t.Fatalf("next window start must precede end: %+v", status.Works)
|
||
}
|
||
view = viewFor(accountID)
|
||
if view == nil || view.WorkCount != 1 || view.LatestPublishedAt == nil {
|
||
// 插入作品后重拉视图：聚合来自 creator_work_source 实时 JOIN。
|
||
views, err = store.ListAccountMonitorViews(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
view = viewFor(accountID)
|
||
if view == nil || view.WorkCount != 1 || view.LatestPublishedAt == nil {
|
||
t.Fatalf("monitor view aggregates: %+v", view)
|
||
}
|
||
}
|
||
// failed checkpoint 带错误信息（普通错误落在 failed，不是 blocked）。
|
||
lease, err = store.beginCheckpoint(ctx, SourceOwned, accountID, "comments", now.Add(-24*time.Hour), now)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := store.failCheckpoint(ctx, SourceOwned, accountID, "comments", lease, errors.New("boom")); err == nil {
|
||
t.Fatal("failCheckpoint must return the primary error")
|
||
}
|
||
status, err = store.GetAccountCollectionStatus(ctx, accountID)
|
||
if err != nil || status.Comments.Status != "failed" || status.Comments.LastError == "" {
|
||
t.Fatalf("failed comments checkpoint surfaces error: %+v err=%v", status.Comments, err)
|
||
}
|
||
// 空账号 ID 拒绝。
|
||
if _, err := store.GetAccountCollectionStatus(ctx, " "); !errors.Is(err, ErrInvalid) {
|
||
t.Fatalf("blank account id must be rejected: err=%v", err)
|
||
}
|
||
}
|
||
|
||
func TestCreatorPostgresCollectionAllowsMutedReadAccount(t *testing.T) {
|
||
store, phaseAStore, ctx := openCreatorIntegrationStore(t)
|
||
stamp := time.Now().UnixNano()
|
||
accountID := createIntegrationAccount(t, ctx, phaseAStore, "muted"+fmt.Sprint(stamp))
|
||
if err := store.EnsureAccountProfile(ctx, accountID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.RecordVerifiedLoginResult(ctx, accountID, "sec_uid_"+accountID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
now := time.Now().UTC()
|
||
if _, err := store.db.ExecContext(ctx, `UPDATE creator_account_profile SET business_status='muted' WHERE account_id=$1`, accountID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
due, err := store.ListDueOwnedAccounts(ctx, now, 1800)
|
||
if err != nil || len(due) != 1 || due[0] != accountID {
|
||
t.Fatalf("muted account was not eligible for read collection: accounts=%v err=%v", due, err)
|
||
}
|
||
if _, err := store.db.ExecContext(ctx, `UPDATE creator_account_profile SET business_status='banned' WHERE account_id=$1`, accountID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
due, err = store.ListDueOwnedAccounts(ctx, now, 1800)
|
||
if err != nil || len(due) != 0 {
|
||
t.Fatalf("banned account was eligible for read collection: accounts=%v err=%v", due, err)
|
||
}
|
||
}
|
||
|
||
func TestCreatorPostgresSettingsResetCollectionCheckpoint(t *testing.T) {
|
||
store, phaseAStore, ctx := openCreatorIntegrationStore(t)
|
||
stamp := time.Now().UnixNano()
|
||
accountID := createIntegrationAccount(t, ctx, phaseAStore, "settings"+fmt.Sprint(stamp))
|
||
before := time.Now().UTC()
|
||
lease, err := store.beginCheckpoint(ctx, SourceOwned, accountID, "works", before.Add(-48*time.Hour), before.Add(-47*time.Hour))
|
||
if err != nil {
|
||
t.Fatalf("begin checkpoint: %v", err)
|
||
}
|
||
if err := store.finishCheckpoint(ctx, SourceOwned, accountID, "works", lease, "succeeded", ""); err != nil {
|
||
t.Fatalf("finish checkpoint: %v", err)
|
||
}
|
||
if _, err := store.db.ExecContext(ctx, `UPDATE creator_collection_checkpoint SET window_start='2020-01-01T00:00:00Z',window_end='2020-01-02T00:00:00Z',cursor='old-cursor' WHERE source_type=$1 AND source_id=$2 AND collection_kind='works'`, SourceOwned, accountID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
settings, err := store.GetSettings(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
settings.LookbackDays = 3
|
||
if _, err := store.UpdateSettings(ctx, SettingsUpdate{LookbackDays: settings.LookbackDays, NewWorkIntervalSeconds: settings.NewWorkIntervalSeconds, MetricInitialIntervalSeconds: settings.MetricInitialIntervalSeconds, MetricMultiplier: settings.MetricMultiplier, MetricMaxIntervalSeconds: settings.MetricMaxIntervalSeconds, MetricAgeSeconds: settings.MetricAgeSeconds, AIProvider: settings.AIProvider, AIModel: settings.AIModel, AIConfigured: settings.AIConfigured, TranscriptionProvider: settings.TranscriptionProvider, TranscriptionModel: settings.TranscriptionModel, TranscriptionConfigured: settings.TranscriptionConfigured}); err != nil {
|
||
t.Fatalf("update settings: %v", err)
|
||
}
|
||
checkpoint, err := store.checkpoint(ctx, SourceOwned, accountID, "works")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if checkpoint.Cursor != "" || checkpoint.Status != "idle" || checkpoint.WindowStart.Before(time.Now().UTC().Add(-4*24*time.Hour)) {
|
||
t.Fatalf("settings did not reset checkpoint: %+v", checkpoint)
|
||
}
|
||
lease, err = store.beginCheckpoint(ctx, SourceOwned, accountID, "works", time.Now().UTC().Add(-time.Hour), time.Now().UTC())
|
||
if err != nil {
|
||
t.Fatalf("begin running checkpoint: %v", err)
|
||
}
|
||
settings.LookbackDays = 4
|
||
if _, err := store.UpdateSettings(ctx, SettingsUpdate{LookbackDays: settings.LookbackDays, NewWorkIntervalSeconds: settings.NewWorkIntervalSeconds, MetricInitialIntervalSeconds: settings.MetricInitialIntervalSeconds, MetricMultiplier: settings.MetricMultiplier, MetricMaxIntervalSeconds: settings.MetricMaxIntervalSeconds, MetricAgeSeconds: settings.MetricAgeSeconds, AIProvider: settings.AIProvider, AIModel: settings.AIModel, AIConfigured: settings.AIConfigured, TranscriptionProvider: settings.TranscriptionProvider, TranscriptionModel: settings.TranscriptionModel, TranscriptionConfigured: settings.TranscriptionConfigured}); !errors.Is(err, ErrConflict) {
|
||
t.Fatalf("settings changed during running checkpoint: err=%v", err)
|
||
}
|
||
if err := store.finishCheckpoint(ctx, SourceOwned, accountID, "works", lease, "succeeded", ""); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
|
||
func TestCreatorPostgresMetricPlanFollowsPublishedAt(t *testing.T) {
|
||
store, phaseAStore, ctx := openCreatorIntegrationStore(t)
|
||
stamp := fmt.Sprintf("%d", time.Now().UnixNano())
|
||
bigID := createIntegrationAccount(t, ctx, phaseAStore, "metric"+stamp)
|
||
if err := store.EnsureAccountProfile(ctx, bigID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
published := time.Now().UTC().Add(-time.Hour).Truncate(time.Microsecond)
|
||
likes, comments, shares := int64(1), int64(1), int64(1)
|
||
work, _, err := store.UpsertWork(ctx, WorkInput{Platform: PlatformDouyin, WorkKey: "creator-it-metric-work-" + stamp, SourceType: SourceOwned, SourceID: bigID, Title: "Metric", Body: "body", PublishedAt: &published, PublishedAtStatus: "verified", Likes: &likes, CommentsCount: &comments, Shares: &shares}, published.Add(time.Hour))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
settings, err := store.GetSettings(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
settings.MetricInitialIntervalSeconds = 3600
|
||
settings.MetricMultiplier = 2
|
||
settings.MetricMaxIntervalSeconds = 55 * 3600
|
||
settings.MetricAgeSeconds = 100 * 3600
|
||
point := func(at time.Time) MetricInput {
|
||
return MetricInput{WorkID: work.ID, CollectedAt: at, Likes: &likes, CommentsCount: &comments, Shares: &shares}
|
||
}
|
||
if _, err := store.recordMetricWithPlan(ctx, point(published.Add(2*time.Hour)), settings, published.Add(2*time.Hour)); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var next time.Time
|
||
var interval int64
|
||
if err := store.db.QueryRowContext(ctx, `SELECT next_plan_at,interval_seconds FROM creator_metric_plan WHERE work_id=$1`, work.ID).Scan(&next, &interval); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if !next.Equal(published.Add(3*time.Hour)) || interval != 4*3600 {
|
||
t.Fatalf("first metric plan drifted: next=%s interval=%d", next, interval)
|
||
}
|
||
if _, err := store.recordMetricWithPlan(ctx, point(published.Add(3*time.Hour)), settings, published.Add(3*time.Hour)); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := store.db.QueryRowContext(ctx, `SELECT next_plan_at,interval_seconds FROM creator_metric_plan WHERE work_id=$1`, work.ID).Scan(&next, &interval); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if !next.Equal(published.Add(7*time.Hour)) || interval != 8*3600 {
|
||
t.Fatalf("second metric plan drifted: next=%s interval=%d", next, interval)
|
||
}
|
||
}
|
||
|
||
func prepareIntegrationActionFixture(t *testing.T, store *Store, phaseAStore *account.Store, ctx context.Context, stamp string) (string, string, Work, Comment, Strategy) {
|
||
t.Helper()
|
||
bigID := createIntegrationAccount(t, ctx, phaseAStore, "big"+stamp)
|
||
smallID := createIntegrationAccount(t, ctx, phaseAStore, "small"+stamp)
|
||
for _, accountID := range []string{bigID, smallID} {
|
||
if err := store.EnsureAccountProfile(ctx, accountID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.UpdateAccountProfile(ctx, accountID, AccountProfileUpdate{RealNameStatus: "unknown", BusinessStatus: "normal", CooldownSeconds: 86400}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.RecordVerifiedLoginResult(ctx, accountID, "sec_uid_"+accountID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
if _, err := store.SetBigAccount(ctx, bigID, true); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := store.SetRelation(ctx, bigID, smallID, true); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
now := time.Now().UTC().Truncate(time.Microsecond)
|
||
published := now.Add(-time.Hour)
|
||
likes, comments, shares := int64(1), int64(1), int64(1)
|
||
work, _, err := store.UpsertWork(ctx, WorkInput{Platform: PlatformDouyin, WorkKey: "creator-it-action-work-" + stamp, SourceType: SourceOwned, SourceID: bigID, Title: "Action", Body: "body", PublishedAt: &published, PublishedAtStatus: "verified", Likes: &likes, CommentsCount: &comments, Shares: &shares}, now)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
comment, _, err := store.SaveComment(ctx, CommentInput{Platform: PlatformDouyin, CommentKey: "creator-it-action-comment-" + stamp, WorkID: work.ID, AuthorUID: "interactor-" + stamp, Content: "hello", CommentType: "top_level"})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
strategy, err := store.CreateStrategy(ctx, bigID, StrategyInput{ExecutionAccountID: smallID, Position: 1, Enabled: true, EventTypes: []string{"comment"}, Action: ActionReplyComment, TargetType: "comment", CandidateTexts: []string{"已收到"}})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
strategies, err := store.ListStrategies(ctx, bigID)
|
||
if err != nil || len(strategies) != 1 {
|
||
t.Fatalf("list strategies: strategies=%+v err=%v", strategies, err)
|
||
}
|
||
return bigID, smallID, work, comment, strategy
|
||
}
|
||
|
||
func TestCreatorPostgresActionsAndMessaging(t *testing.T) {
|
||
store, phaseAStore, ctx := openCreatorIntegrationStore(t)
|
||
stamp := fmt.Sprintf("%d", time.Now().UnixNano())
|
||
bigID, smallID, work, comment, strategy := prepareIntegrationActionFixture(t, store, phaseAStore, ctx, stamp)
|
||
event := InteractionEvent{Platform: PlatformDouyin, ReceivingAccountID: bigID, EventKey: "creator-it-event-" + stamp, EventType: "comment", InteractorUID: comment.AuthorUID, CommentID: comment.ID, WorkID: work.ID}
|
||
automatic, err := store.ProcessAutomaticEvent(ctx, event, integrationExecutor{}, nil)
|
||
if err != nil || automatic.Operation == nil || automatic.Operation.State != "succeeded" || automatic.Event.State != "succeeded" {
|
||
t.Fatalf("automatic action failed: result=%+v err=%v", automatic, err)
|
||
}
|
||
events, err := store.ListEvents(ctx, bigID)
|
||
if err != nil || len(events) != 1 {
|
||
t.Fatalf("list events: events=%+v err=%v", events, err)
|
||
}
|
||
duplicate, err := store.ProcessAutomaticEvent(ctx, event, integrationExecutor{}, nil)
|
||
if err != nil || !duplicate.Duplicate || duplicate.Event.ID != automatic.Event.ID {
|
||
t.Fatalf("automatic event deduplication failed: result=%+v err=%v", duplicate, err)
|
||
}
|
||
cooldownEvent := event
|
||
cooldownEvent.EventKey += "-cooldown"
|
||
blocked, err := store.ProcessAutomaticEvent(ctx, cooldownEvent, integrationExecutor{}, nil)
|
||
if err != nil || blocked.Event.State != "blocked" {
|
||
t.Fatalf("cooldown did not block second event: result=%+v err=%v", blocked, err)
|
||
}
|
||
manualInput := OperationInput{IdempotencyKey: "creator-it-manual-" + stamp, Source: "manual", Action: ActionReplyComment, Platform: PlatformDouyin, AccountID: smallID, TargetUID: comment.AuthorUID, TargetCommentID: comment.ID, Text: "人工回复"}
|
||
manual, inserted, err := store.CreateOperation(ctx, manualInput)
|
||
if err != nil || !inserted {
|
||
t.Fatalf("create manual operation: operation=%+v inserted=%v err=%v", manual, inserted, err)
|
||
}
|
||
manualDuplicate, inserted, err := store.CreateOperation(ctx, manualInput)
|
||
if err != nil || inserted || manualDuplicate.ID != manual.ID {
|
||
t.Fatalf("manual operation idempotency failed: operation=%+v inserted=%v err=%v", manualDuplicate, inserted, err)
|
||
}
|
||
operations, err := store.ListOperations(ctx, smallID)
|
||
if err != nil || len(operations) != 2 {
|
||
t.Fatalf("list operations: operations=%+v err=%v", operations, err)
|
||
}
|
||
manual, err = store.ExecuteManualOperation(ctx, manual.ID, integrationExecutor{})
|
||
if err != nil || manual.State != "succeeded" {
|
||
t.Fatalf("execute manual operation: operation=%+v err=%v", manual, err)
|
||
}
|
||
dmOperation, inserted, err := store.CreateOperation(ctx, OperationInput{IdempotencyKey: "creator-it-dm-" + stamp, Source: "manual", Action: ActionDM, Platform: PlatformDouyin, AccountID: smallID, TargetUID: "peer-" + stamp, Text: "人工私信"})
|
||
if err != nil || !inserted {
|
||
t.Fatalf("create direct-message operation: operation=%+v inserted=%v err=%v", dmOperation, inserted, err)
|
||
}
|
||
dmOperation, err = store.ExecuteManualOperation(ctx, dmOperation.ID, integrationExecutor{})
|
||
if err != nil || dmOperation.State != "succeeded" {
|
||
t.Fatalf("execute direct-message operation: operation=%+v err=%v", dmOperation, err)
|
||
}
|
||
messageAt := time.Now().UTC().Truncate(time.Microsecond)
|
||
message, inserted, err := store.SaveMessage(ctx, MessageInput{Platform: PlatformDouyin, AccountID: smallID, PeerUID: "peer-" + stamp, PeerName: "Peer", PlatformMessageKey: "creator-it-message-" + stamp, Direction: "inbound", MessageType: "text", Text: "hello", MessageAt: &messageAt})
|
||
if err != nil || !inserted {
|
||
t.Fatalf("save message: message=%+v inserted=%v err=%v", message, inserted, err)
|
||
}
|
||
if _, inserted, err := store.SaveMessage(ctx, MessageInput{Platform: PlatformDouyin, AccountID: smallID, PeerUID: "peer-" + stamp, PlatformMessageKey: message.PlatformMessageKey, Direction: "inbound", MessageType: "text", Text: "hello", MessageAt: &messageAt}); err != nil || inserted {
|
||
t.Fatalf("message deduplication failed: inserted=%v err=%v", inserted, err)
|
||
}
|
||
conversations, err := store.ListConversations(ctx, smallID)
|
||
if err != nil || len(conversations) != 1 {
|
||
t.Fatalf("list conversations: conversations=%+v err=%v", conversations, err)
|
||
}
|
||
messages, err := store.ListMessages(ctx, conversations[0].ID)
|
||
if err != nil || len(messages) != 2 {
|
||
t.Fatalf("list messages: messages=%+v err=%v", messages, err)
|
||
}
|
||
if _, err := store.SetEventDisplayed(ctx, automatic.Event.ID, time.Now().UTC()); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
updated, err := store.UpdateStrategy(ctx, strategy.ID, StrategyInput{ExecutionAccountID: smallID, Position: 2, Enabled: true, EventTypes: []string{"comment", "like"}, Action: ActionReplyComment, TargetType: "comment", CandidateTexts: []string{"已更新"}})
|
||
if err != nil || updated.Position != 2 || len(updated.EventTypes) != 2 {
|
||
t.Fatalf("update strategy: strategy=%+v err=%v", updated, err)
|
||
}
|
||
if _, err := store.SetStrategyEnabled(ctx, strategy.ID, false); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := store.DeleteStrategy(ctx, strategy.ID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := store.SetRelation(ctx, bigID, smallID, false); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.SetBigAccount(ctx, bigID, false); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
start, end := time.Now().UTC().Add(-time.Hour), time.Now().UTC()
|
||
lease, err := store.beginCheckpoint(ctx, SourceOwned, bigID, "works", start, end)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.beginCheckpoint(ctx, SourceOwned, bigID, "works", start, end); !errors.Is(err, ErrConflict) {
|
||
t.Fatalf("expected concurrent checkpoint claim conflict, got %v", err)
|
||
}
|
||
if err := store.finishCheckpoint(ctx, SourceOwned, bigID, "works", "stale-lease", "succeeded", ""); !errors.Is(err, ErrConflict) {
|
||
t.Fatalf("expected stale checkpoint completion conflict, got %v", err)
|
||
}
|
||
if err := store.saveCheckpointCursor(ctx, SourceOwned, bigID, "works", lease, "1"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := store.finishCheckpoint(ctx, SourceOwned, bigID, "works", lease, "succeeded", ""); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|