package usecase import ( "context" "strings" "time" "haixun-backend/internal/library/clock" app "haixun-backend/internal/library/errors" "haixun-backend/internal/library/errors/code" libthreads "haixun-backend/internal/library/threadsapi" "haixun-backend/internal/library/threadspost" copydraftdomain "haixun-backend/internal/model/copy_draft/domain/usecase" padomain "haixun-backend/internal/model/publish_analytics/domain/usecase" guarddomain "haixun-backend/internal/model/publish_guard/domain/usecase" "haixun-backend/internal/model/publish_queue/domain/entity" domrepo "haixun-backend/internal/model/publish_queue/domain/repository" domusecase "haixun-backend/internal/model/publish_queue/domain/usecase" evententity "haixun-backend/internal/model/publish_queue_event/domain/entity" eventdomain "haixun-backend/internal/model/publish_queue_event/domain/usecase" threadsaccountdomain "haixun-backend/internal/model/threads_account/domain/usecase" "github.com/google/uuid" ) type publishQueueUseCase struct { repo domrepo.Repository threadsAccount threadsaccountdomain.UseCase publishAnalytics padomain.UseCase guard guarddomain.UseCase events eventdomain.UseCase copyDraft copydraftdomain.UseCase } func NewUseCase( repo domrepo.Repository, threadsAccount threadsaccountdomain.UseCase, publishAnalytics padomain.UseCase, guard guarddomain.UseCase, events eventdomain.UseCase, copyDraft copydraftdomain.UseCase, ) domusecase.UseCase { return &publishQueueUseCase{ repo: repo, threadsAccount: threadsAccount, publishAnalytics: publishAnalytics, guard: guard, events: events, copyDraft: copyDraft, } } func (u *publishQueueUseCase) Create(ctx context.Context, req domusecase.CreateRequest) (*domusecase.QueueItemSummary, error) { tenantID := strings.TrimSpace(req.TenantID) ownerUID := strings.TrimSpace(req.OwnerUID) accountID := strings.TrimSpace(req.AccountID) text := strings.TrimSpace(req.Text) if tenantID == "" || ownerUID == "" { return nil, app.For(code.Auth).AuthUnauthorized("missing actor") } if accountID == "" { return nil, app.For(code.ThreadsAccount).InputMissingRequired("account_id is required") } if text == "" { return nil, app.For(code.ThreadsAccount).InputMissingRequired("text is required") } if err := threadspost.ValidatePublish(text); err != nil { return nil, app.For(code.ThreadsAccount).InputInvalidFormat(err.Error()) } if _, err := u.threadsAccount.Get(ctx, tenantID, ownerUID, accountID); err != nil { return nil, err } now := clock.NowUnixNano() scheduledAt := req.ScheduledAt if scheduledAt <= 0 { scheduledAt = now } item := &entity.QueueItem{ ID: uuid.NewString(), TenantID: tenantID, OwnerUID: ownerUID, AccountID: accountID, PersonaID: strings.TrimSpace(req.PersonaID), CopyMissionID: strings.TrimSpace(req.CopyMissionID), CopyDraftID: strings.TrimSpace(req.CopyDraftID), Text: text, ScheduledAt: scheduledAt, Status: entity.StatusScheduled, CreateAt: now, UpdateAt: now, } if err := u.repo.Create(ctx, item); err != nil { return nil, err } u.recordEvent(ctx, item, evententity.EventCreated, "", item.Status, "加入發文庫存") if scheduledAt <= now { _, _ = u.DispatchDue(ctx, 1) fresh, err := u.repo.Get(ctx, tenantID, ownerUID, accountID, item.ID) if err == nil && fresh != nil { out := toSummary(*fresh) return &out, nil } } out := toSummary(*item) return &out, nil } func (u *publishQueueUseCase) RecordPublished(ctx context.Context, req domusecase.RecordPublishedRequest) (*domusecase.QueueItemSummary, error) { tenantID := strings.TrimSpace(req.TenantID) ownerUID := strings.TrimSpace(req.OwnerUID) accountID := strings.TrimSpace(req.AccountID) mediaID := strings.TrimSpace(req.MediaID) if err := requireActor(tenantID, ownerUID); err != nil { return nil, err } if accountID == "" || mediaID == "" { return nil, app.For(code.ThreadsAccount).InputMissingRequired("account_id and media_id are required") } if _, err := u.threadsAccount.Get(ctx, tenantID, ownerUID, accountID); err != nil { return nil, err } now := clock.NowUnixNano() publishedAt := req.PublishedAt if publishedAt <= 0 { publishedAt = now } item := &entity.QueueItem{ ID: uuid.NewString(), TenantID: tenantID, OwnerUID: ownerUID, AccountID: accountID, PersonaID: strings.TrimSpace(req.PersonaID), CopyMissionID: strings.TrimSpace(req.CopyMissionID), CopyDraftID: strings.TrimSpace(req.CopyDraftID), Text: strings.TrimSpace(req.Text), ScheduledAt: publishedAt, Status: entity.StatusPublished, MediaID: mediaID, Permalink: strings.TrimSpace(req.Permalink), PublishedAt: publishedAt, CreateAt: now, UpdateAt: now, } if err := u.repo.Create(ctx, item); err != nil { return nil, err } u.recordEvent(ctx, item, evententity.EventPublished, "", item.Status, "已記錄發布結果") _ = u.publishAnalytics.ScheduleCheckpoints(ctx, padomain.ScheduleCheckpointsRequest{ TenantID: tenantID, OwnerUID: ownerUID, AccountID: accountID, MediaID: mediaID, Permalink: item.Permalink, PublishedAt: publishedAt, }) out := toSummary(*item) return &out, nil } func (u *publishQueueUseCase) Get(ctx context.Context, tenantID, ownerUID, accountID, queueID string) (*domusecase.QueueItemSummary, error) { if err := requireActor(tenantID, ownerUID); err != nil { return nil, err } item, err := u.repo.Get(ctx, tenantID, ownerUID, accountID, queueID) if err != nil { return nil, err } out := toSummary(*item) return &out, nil } func (u *publishQueueUseCase) ListByIDs(ctx context.Context, tenantID, ownerUID string, ids []string) ([]domusecase.QueueItemSummary, error) { if err := requireActor(tenantID, ownerUID); err != nil { return nil, err } items, err := u.repo.ListByIDs(ctx, tenantID, ownerUID, ids) if err != nil { return nil, err } out := make([]domusecase.QueueItemSummary, 0, len(items)) for _, item := range items { out = append(out, toSummary(item)) } return out, nil } func (u *publishQueueUseCase) List(ctx context.Context, req domusecase.ListRequest) (*domusecase.ListResult, error) { if err := requireActor(req.TenantID, req.OwnerUID); err != nil { return nil, err } page, pageSize := req.Page, req.PageSize if page <= 0 { page = 1 } if pageSize <= 0 { pageSize = 10 } if pageSize > 50 { pageSize = 50 } result, err := u.repo.List(ctx, domrepo.ListFilter{ TenantID: req.TenantID, OwnerUID: req.OwnerUID, AccountID: req.AccountID, Status: req.Status, }, page, pageSize) if err != nil { return nil, err } items := make([]domusecase.QueueItemSummary, 0, len(result.Items)) for _, item := range result.Items { items = append(items, toSummary(item)) } totalPages := 0 if pageSize > 0 { totalPages = int((result.Total + int64(pageSize) - 1) / int64(pageSize)) } return &domusecase.ListResult{ Items: items, Total: result.Total, Page: page, PageSize: pageSize, TotalPages: totalPages, }, nil } func (u *publishQueueUseCase) ListRange(ctx context.Context, req domusecase.RangeRequest) (*domusecase.ListResult, error) { if err := requireActor(req.TenantID, req.OwnerUID); err != nil { return nil, err } page, pageSize := req.Page, req.PageSize if page <= 0 { page = 1 } if pageSize <= 0 { pageSize = 50 } if pageSize > 200 { pageSize = 200 } result, err := u.repo.List(ctx, domrepo.ListFilter{ TenantID: req.TenantID, OwnerUID: req.OwnerUID, AccountID: req.AccountID, Status: req.Status, StartAt: req.StartAt, EndAt: req.EndAt, }, page, pageSize) if err != nil { return nil, err } items := make([]domusecase.QueueItemSummary, 0, len(result.Items)) for _, item := range result.Items { items = append(items, toSummary(item)) } totalPages := 0 if pageSize > 0 { totalPages = int((result.Total + int64(pageSize) - 1) / int64(pageSize)) } return &domusecase.ListResult{Items: items, Total: result.Total, Page: page, PageSize: pageSize, TotalPages: totalPages}, nil } func (u *publishQueueUseCase) Update(ctx context.Context, req domusecase.UpdateRequest) (*domusecase.QueueItemSummary, error) { if err := requireActor(req.TenantID, req.OwnerUID); err != nil { return nil, err } item, err := u.repo.Get(ctx, req.TenantID, req.OwnerUID, req.AccountID, req.QueueID) if err != nil { return nil, err } if item.Status != entity.StatusScheduled { return nil, app.For(code.ThreadsAccount).ResInvalidState("僅排程中的項目可修改") } if req.Text != nil { text := strings.TrimSpace(*req.Text) if text == "" { return nil, app.For(code.ThreadsAccount).InputMissingRequired("text is required") } if err := threadspost.ValidatePublish(text); err != nil { return nil, app.For(code.ThreadsAccount).InputInvalidFormat(err.Error()) } item.Text = text } if req.ScheduledAt != nil && *req.ScheduledAt > 0 { item.ScheduledAt = *req.ScheduledAt } item.UpdateAt = clock.NowUnixNano() updated, err := u.repo.UpdateIfStatus(ctx, item, []string{entity.StatusScheduled}) if err != nil { return nil, err } out := toSummary(*updated) return &out, nil } func (u *publishQueueUseCase) Cancel(ctx context.Context, tenantID, ownerUID, accountID, queueID string) (*domusecase.QueueItemSummary, error) { if err := requireActor(tenantID, ownerUID); err != nil { return nil, err } item, err := u.repo.Get(ctx, tenantID, ownerUID, accountID, queueID) if err != nil { return nil, err } if item.Status != entity.StatusScheduled { return nil, app.For(code.ThreadsAccount).ResInvalidState("僅排程中的項目可取消") } item.Status = entity.StatusCancelled item.UpdateAt = clock.NowUnixNano() updated, err := u.repo.UpdateIfStatus(ctx, item, []string{entity.StatusScheduled}) if err != nil { return nil, err } u.recordEvent(ctx, updated, evententity.EventCancelled, entity.StatusScheduled, updated.Status, "已取消排程") out := toSummary(*updated) return &out, nil } func (u *publishQueueUseCase) Retry(ctx context.Context, tenantID, ownerUID, accountID, queueID string) (*domusecase.QueueItemSummary, error) { if err := requireActor(tenantID, ownerUID); err != nil { return nil, err } item, err := u.repo.Get(ctx, tenantID, ownerUID, accountID, queueID) if err != nil { return nil, err } if item.Status != entity.StatusFailed && item.Status != entity.StatusCancelled && item.Status != entity.StatusMissed { return nil, app.For(code.ThreadsAccount).ResInvalidState("僅 failed/cancelled/missed 項目可重試") } from := item.Status item.Status = entity.StatusScheduled item.ErrorMessage = "" item.PausedReason = "" item.MissedAt = 0 item.RetryCount++ item.NextAttemptAt = 0 now := clock.NowUnixNano() item.UpdateAt = now if item.ScheduledAt < now { item.ScheduledAt = now } updated, err := u.repo.UpdateIfStatus(ctx, item, []string{entity.StatusFailed, entity.StatusCancelled, entity.StatusMissed}) if err != nil { return nil, err } u.recordEvent(ctx, updated, evententity.EventRetry, from, updated.Status, "已重新排程") out := toSummary(*updated) return &out, nil } func (u *publishQueueUseCase) Delete(ctx context.Context, tenantID, ownerUID, accountID, queueID string) error { if err := requireActor(tenantID, ownerUID); err != nil { return err } item, err := u.repo.Get(ctx, tenantID, ownerUID, accountID, queueID) if err != nil { return err } switch item.Status { case entity.StatusPublishing: return app.For(code.ThreadsAccount).ResInvalidState("發佈進行中,無法刪除") case entity.StatusPublished: return app.For(code.ThreadsAccount).ResInvalidState("已發佈項目不可刪除") case entity.StatusScheduled: _, err = u.Cancel(ctx, tenantID, ownerUID, accountID, queueID) return err default: return u.repo.Delete(ctx, tenantID, ownerUID, accountID, queueID) } } func (u *publishQueueUseCase) DispatchDue(ctx context.Context, limit int) (int, error) { if u.repo == nil { return 0, nil } now := clock.NowUnixNano() due, err := u.repo.ListDue(ctx, now, limit) if err != nil { return 0, err } dispatched := 0 for _, item := range due { if err := u.dispatchOne(ctx, &item); err != nil { continue } dispatched++ } return dispatched, nil } func (u *publishQueueUseCase) dispatchOne(ctx context.Context, item *entity.QueueItem) error { if result, err := u.checkGuard(ctx, item); err == nil && result != nil && !result.Allowed { if result.ShouldPause && u.guard != nil { _, _ = u.guard.Pause(ctx, item.TenantID, item.OwnerUID, item.AccountID, result.Reason) } item.PausedReason = result.Reason if result.DelayUntil > 0 { item.ScheduledAt = result.DelayUntil } else { item.Status = entity.StatusFailed item.ErrorMessage = result.Reason } item.UpdateAt = clock.NowUnixNano() updated, updateErr := u.repo.UpdateIfStatus(ctx, item, []string{entity.StatusScheduled}) if updateErr != nil { return updateErr } u.recordEvent(ctx, updated, evententity.EventGuardDelayed, entity.StatusScheduled, updated.Status, result.Reason) return nil } item.Status = entity.StatusPublishing item.LastAttemptAt = clock.NowUnixNano() item.UpdateAt = clock.NowUnixNano() claimed, err := u.repo.UpdateIfStatus(ctx, item, []string{entity.StatusScheduled}) if err != nil { return err } u.recordEvent(ctx, claimed, evententity.EventPublishing, entity.StatusScheduled, entity.StatusPublishing, "開始發布") account, err := u.threadsAccount.Get(ctx, claimed.TenantID, claimed.OwnerUID, claimed.AccountID) if err != nil { return u.markFailed(ctx, claimed, "取得帳號失敗:"+err.Error()) } token, err := u.threadsAccount.ResolveAccountAPIAccessToken(ctx, claimed.TenantID, claimed.OwnerUID, claimed.AccountID) if err != nil { return u.markFailed(ctx, claimed, err.Error()) } userID := strings.TrimSpace(account.ThreadsUserID) if userID == "" || token == "" { return u.markFailed(ctx, claimed, "Threads API 未連線") } result, err := libthreads.PublishText(ctx, libthreads.PublishTextInput{ ThreadsUserID: userID, AccessToken: token, Text: claimed.Text, }) if err != nil { return u.markFailed(ctx, claimed, "發文失敗:"+err.Error()) } now := clock.NowUnixNano() claimed.Status = entity.StatusPublished claimed.MediaID = result.MediaID claimed.Permalink = result.Permalink claimed.PublishedAt = now claimed.UpdateAt = now claimed.ErrorMessage = "" if _, err := u.repo.UpdateIfStatus(ctx, claimed, []string{entity.StatusPublishing}); err != nil { return err } u.recordEvent(ctx, claimed, evententity.EventPublished, entity.StatusPublishing, entity.StatusPublished, "Threads API 發布完成") if claimed.CopyDraftID != "" && u.copyDraft != nil { _, _ = u.copyDraft.MarkPublished(ctx, copydraftdomain.MarkPublishedRequest{ TenantID: claimed.TenantID, OwnerUID: claimed.OwnerUID, PersonaID: claimed.PersonaID, DraftID: claimed.CopyDraftID, MediaID: result.MediaID, Permalink: result.Permalink, }) } _ = u.publishAnalytics.ScheduleCheckpoints(ctx, padomain.ScheduleCheckpointsRequest{ TenantID: claimed.TenantID, OwnerUID: claimed.OwnerUID, AccountID: claimed.AccountID, MediaID: result.MediaID, Permalink: result.Permalink, PublishedAt: now, }) return nil } func (u *publishQueueUseCase) markFailed(ctx context.Context, item *entity.QueueItem, message string) error { item.Status = entity.StatusFailed item.ErrorMessage = message item.LastAttemptAt = clock.NowUnixNano() item.UpdateAt = clock.NowUnixNano() updated, err := u.repo.UpdateIfStatus(ctx, item, []string{entity.StatusPublishing, entity.StatusScheduled}) if err == nil { u.recordEvent(ctx, updated, evententity.EventFailed, entity.StatusPublishing, entity.StatusFailed, message) failureStreak := u.failureStreak(ctx, item.TenantID, item.OwnerUID, item.AccountID) if u.guard != nil && failureStreak >= 3 { _, _ = u.guard.Pause(ctx, item.TenantID, item.OwnerUID, item.AccountID, "連續發文失敗,已暫停帳號") } } return err } func (u *publishQueueUseCase) ListPublishHealth(ctx context.Context, tenantID, ownerUID, accountID string, page, pageSize int) (*domusecase.PublishHealthResult, error) { if err := requireActor(tenantID, ownerUID); err != nil { return nil, err } if page <= 0 { page = 1 } if pageSize <= 0 { pageSize = 10 } if pageSize > 30 { pageSize = 30 } pending, _ := u.repo.CountByStatus(ctx, tenantID, ownerUID, accountID, entity.StatusScheduled) failed, _ := u.repo.CountByStatus(ctx, tenantID, ownerUID, accountID, entity.StatusFailed) publishedList, err := u.repo.List(ctx, domrepo.ListFilter{ TenantID: tenantID, OwnerUID: ownerUID, AccountID: accountID, Status: entity.StatusPublished, }, page, pageSize) if err != nil { return nil, err } cutoff7d := clock.NowUnixNano() - int64(7*24*time.Hour) var published7d int64 var totalLikes7d, totalReplies7d int items := make([]domusecase.PublishHealthItem, 0, len(publishedList.Items)) for _, row := range publishedList.Items { if row.PublishedAt >= cutoff7d { published7d++ } checkpoints, err := u.publishAnalytics.ListByMedia(ctx, tenantID, ownerUID, row.MediaID) if err != nil { checkpoints = nil } healthCPs := make([]domusecase.PublishHealthCheckpoint, 0, len(checkpoints)) var latestLike, latestReply, latestRepost, latestQuote, latestViews, latestShares int var latestChecked int64 for _, cp := range checkpoints { healthCPs = append(healthCPs, domusecase.PublishHealthCheckpoint{ Checkpoint: cp.Checkpoint, LikeCount: cp.LikeCount, ReplyCount: cp.ReplyCount, RepostCount: cp.RepostCount, QuoteCount: cp.QuoteCount, Views: cp.ViewCount, CheckedAt: cp.CheckedAt, }) if cp.CheckedAt >= latestChecked { latestChecked = cp.CheckedAt latestLike = cp.LikeCount latestReply = cp.ReplyCount latestRepost = cp.RepostCount latestQuote = cp.QuoteCount latestViews = cp.ViewCount latestShares = cp.ShareCount } } if row.PublishedAt >= cutoff7d { totalLikes7d += latestLike totalReplies7d += latestReply } items = append(items, domusecase.PublishHealthItem{ MediaID: row.MediaID, Permalink: row.Permalink, Text: row.Text, PublishedAt: row.PublishedAt, LikeCount: latestLike, ReplyCount: latestReply, RepostCount: latestRepost, QuoteCount: latestQuote, Views: latestViews, Shares: latestShares, Checkpoints: healthCPs, }) } totalPages := 0 if pageSize > 0 { totalPages = int((publishedList.Total + int64(pageSize) - 1) / int64(pageSize)) } return &domusecase.PublishHealthResult{ Summary: domusecase.PublishHealthSummary{ PendingScheduled: pending, FailedCount: failed, Published7d: published7d, TotalLikes7d: totalLikes7d, TotalReplies7d: totalReplies7d, }, Items: items, Total: publishedList.Total, Page: page, PageSize: pageSize, TotalPages: totalPages, }, nil } func (u *publishQueueUseCase) MarkMissed(ctx context.Context, graceNs int64, limit int) (int, error) { if graceNs <= 0 { graceNs = int64(10 * time.Minute) } before := clock.NowUnixNano() - graceNs items, err := u.repo.ListMissedCandidates(ctx, before, limit) if err != nil { return 0, err } count := 0 for _, item := range items { item.Status = entity.StatusMissed item.MissedAt = clock.NowUnixNano() item.UpdateAt = item.MissedAt updated, err := u.repo.UpdateIfStatus(ctx, &item, []string{entity.StatusScheduled}) if err != nil { continue } u.recordEvent(ctx, updated, evententity.EventMissed, entity.StatusScheduled, entity.StatusMissed, "超過 grace period 未發布") count++ } return count, nil } func (u *publishQueueUseCase) ListAlerts(ctx context.Context, tenantID, ownerUID, accountID string) ([]domusecase.Alert, error) { if err := requireActor(tenantID, ownerUID); err != nil { return nil, err } result, err := u.repo.List(ctx, domrepo.ListFilter{TenantID: tenantID, OwnerUID: ownerUID, AccountID: accountID}, 1, 50) if err != nil { return nil, err } alerts := make([]domusecase.Alert, 0) for _, item := range result.Items { switch item.Status { case entity.StatusFailed: alerts = append(alerts, domusecase.Alert{Type: "failed", Severity: "danger", Message: firstNonEmpty(item.ErrorMessage, "發文失敗"), QueueID: item.ID, AccountID: item.AccountID, CreateAt: item.UpdateAt}) case entity.StatusMissed: alerts = append(alerts, domusecase.Alert{Type: "missed", Severity: "warning", Message: "排程已漏發", QueueID: item.ID, AccountID: item.AccountID, CreateAt: item.MissedAt}) } if item.PausedReason != "" { alerts = append(alerts, domusecase.Alert{Type: "guard", Severity: "warning", Message: item.PausedReason, QueueID: item.ID, AccountID: item.AccountID, CreateAt: item.UpdateAt}) } } return alerts, nil } func (u *publishQueueUseCase) ListSlotInsights(ctx context.Context, tenantID, ownerUID, accountID, timezone string) ([]domusecase.SlotInsight, error) { if err := requireActor(tenantID, ownerUID); err != nil { return nil, err } health, err := u.ListPublishHealth(ctx, tenantID, ownerUID, accountID, 1, 30) if err != nil { return nil, err } type agg struct { likes7d int replies7d int samples int } slots := map[string]*agg{} for _, item := range health.Items { t := time.Unix(0, item.PublishedAt).UTC() key := t.Format("15:04") if _, ok := slots[key]; !ok { slots[key] = &agg{} } slots[key].likes7d += item.LikeCount slots[key].replies7d += item.ReplyCount slots[key].samples++ } out := make([]domusecase.SlotInsight, 0, len(slots)) for key, item := range slots { avgLikes := 0 avgReplies := 0 if item.samples > 0 { avgLikes = item.likes7d / item.samples avgReplies = item.replies7d / item.samples } recommendation := "neutral" if avgLikes+avgReplies >= 5 { recommendation = "recommended" } else if item.samples >= 2 { recommendation = "low" } out = append(out, domusecase.SlotInsight{ Weekday: -1, Time: key, AvgLikes1h: avgLikes, AvgReplies1h: avgReplies, AvgLikes24h: avgLikes, AvgReplies24h: avgReplies, AvgLikes7d: avgLikes, AvgReplies7d: avgReplies, SampleSize: item.samples, Recommendation: recommendation, }) } if len(out) == 0 { out = []domusecase.SlotInsight{ {Weekday: 1, Time: "09:30", Recommendation: "recommended"}, {Weekday: 3, Time: "12:30", Recommendation: "neutral"}, {Weekday: 5, Time: "18:30", Recommendation: "recommended"}, } } return out, nil } func (u *publishQueueUseCase) checkGuard(ctx context.Context, item *entity.QueueItem) (*guarddomain.CheckResult, error) { if u.guard == nil || item == nil { return &guarddomain.CheckResult{Allowed: true}, nil } now := clock.NowUnixNano() publishedToday, _ := u.repo.CountPublishedSince(ctx, item.TenantID, item.OwnerUID, item.AccountID, startOfUTCDay(now)) lastPublishedAt, _ := u.repo.LastPublishedAt(ctx, item.TenantID, item.OwnerUID, item.AccountID) return u.guard.Check(ctx, guarddomain.CheckRequest{ TenantID: item.TenantID, OwnerUID: item.OwnerUID, AccountID: item.AccountID, Now: now, PublishedToday: publishedToday, LastPublishedAt: lastPublishedAt, FailureStreak: u.failureStreak(ctx, item.TenantID, item.OwnerUID, item.AccountID), }) } func (u *publishQueueUseCase) failureStreak(ctx context.Context, tenantID, ownerUID, accountID string) int { result, err := u.repo.List(ctx, domrepo.ListFilter{TenantID: tenantID, OwnerUID: ownerUID, AccountID: accountID}, 1, 10) if err != nil { return 0 } streak := 0 for _, item := range result.Items { if item.Status == entity.StatusFailed { streak++ continue } if item.Status == entity.StatusPublished { break } } return streak } func (u *publishQueueUseCase) recordEvent(ctx context.Context, item *entity.QueueItem, eventType, fromStatus, toStatus, message string) { if u.events == nil || item == nil { return } _ = u.events.Record(ctx, eventdomain.RecordRequest{ TenantID: item.TenantID, OwnerUID: item.OwnerUID, AccountID: item.AccountID, QueueID: item.ID, EventType: eventType, FromStatus: fromStatus, ToStatus: toStatus, Message: message, }) } func startOfUTCDay(ns int64) int64 { t := time.Unix(0, ns).UTC() return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.UTC).UnixNano() } func firstNonEmpty(values ...string) string { for _, value := range values { if strings.TrimSpace(value) != "" { return strings.TrimSpace(value) } } return "" } func requireActor(tenantID, ownerUID string) error { if strings.TrimSpace(tenantID) == "" || strings.TrimSpace(ownerUID) == "" { return app.For(code.Auth).AuthUnauthorized("missing actor") } return nil } func toSummary(item entity.QueueItem) domusecase.QueueItemSummary { return domusecase.QueueItemSummary{ ID: item.ID, AccountID: item.AccountID, PersonaID: item.PersonaID, CopyMissionID: item.CopyMissionID, CopyDraftID: item.CopyDraftID, Text: item.Text, ScheduledAt: item.ScheduledAt, Status: item.Status, MediaID: item.MediaID, Permalink: item.Permalink, PublishedAt: item.PublishedAt, ErrorMessage: item.ErrorMessage, RetryCount: item.RetryCount, LastAttemptAt: item.LastAttemptAt, NextAttemptAt: item.NextAttemptAt, MissedAt: item.MissedAt, PausedReason: item.PausedReason, CreateAt: item.CreateAt, UpdateAt: item.UpdateAt, } }