635 lines
32 KiB
Go
635 lines
32 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)
|
|
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}, now)
|
|
if err != nil || !inserted {
|
|
t.Fatalf("insert work: work=%+v inserted=%v err=%v", work, inserted, 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.ListCompetitors(ctx, PlatformDouyin)
|
|
if err != nil || len(competitors) != 1 || len(competitors[0].Tags) != 1 || competitors[0].Tags[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)
|
|
}
|
|
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)
|
|
}
|
|
|
|
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)
|
|
}
|
|
if _, err := store.RecordMetric(ctx, MetricInput{WorkID: work.ID, CollectedAt: now, Likes: &likes, CommentsCount: &comments, Shares: &shares}, 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 _, 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 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)
|
|
}
|
|
}
|