From 37d75597f97b41f8968d223b8e30dd39b7fb1923 Mon Sep 17 00:00:00 2001 From: liangxu Date: Sun, 21 Jun 2026 09:11:37 +0800 Subject: [PATCH] feat: enhance schedule publish account handling and caching mechanisms --- .../src/components/GenerateTaskDrawer.vue | 32 ++++++++- .../scheduler/schedule_dispatch_worker.go | 11 +++ .../tenant/app/desktop_account_service.go | 9 +++ .../internal/tenant/app/publish_platforms.go | 57 +++++++++++++--- .../tenant/app/schedule_task_service.go | 68 ++++++++----------- .../transport/desktop_account_handler.go | 2 +- 6 files changed, 125 insertions(+), 54 deletions(-) diff --git a/apps/admin-web/src/components/GenerateTaskDrawer.vue b/apps/admin-web/src/components/GenerateTaskDrawer.vue index ced7400..6f63063 100644 --- a/apps/admin-web/src/components/GenerateTaskDrawer.vue +++ b/apps/admin-web/src/components/GenerateTaskDrawer.vue @@ -323,6 +323,12 @@ watch( { immediate: true }, ) +watch( + schedulePublishAccounts, + () => syncSchedulePublishAccountSelection(), + { immediate: true }, +) + watch(currentBrandId, () => { form.promptRuleId = undefined }) @@ -495,6 +501,7 @@ function hydrateForm(): void { normalizeClock(task.random_window_start, randomDefaultScheduleTime())) : randomDefaultScheduleTime() selectDefaultSubscriptionPrompt() + syncSchedulePublishAccountSelection() } function selectDefaultSubscriptionPrompt(): void { @@ -518,7 +525,9 @@ function buildSchedulePayload() { const coverMode = coverEnabled ? form.coverMode : 'specific' const coverAssetUrl = coverEnabled && coverMode === 'specific' ? form.coverAssetUrl.trim() : '' const publishAccountIds = - form.autoPublish && form.publishTargetTab === 'media_accounts' ? form.publishAccountIds : [] + form.autoPublish && form.publishTargetTab === 'media_accounts' + ? currentSchedulePublishAccountIds() + : [] const publishEnterpriseSiteTargets = form.autoPublish && form.publishTargetTab === 'enterprise_sites' ? buildScheduleEnterpriseSiteTargets() @@ -610,6 +619,25 @@ function canToggleSchedulePublishAccount(account: PublishAccountCard): boolean { return account.selectable || isSchedulePublishAccountSelected(account.id) } +function currentSchedulePublishAccountIds(): string[] { + return filterExistingSchedulePublishAccountIds(form.publishAccountIds, schedulePublishAccounts.value) +} + +function syncSchedulePublishAccountSelection(): void { + if (accountsQuery.isPending.value || accountsQuery.isFetching.value) { + return + } + form.publishAccountIds = currentSchedulePublishAccountIds() +} + +function filterExistingSchedulePublishAccountIds( + accountIds: string[], + accounts: PublishAccountCard[], +): string[] { + const existingIds = new Set(accounts.map((account) => account.id)) + return accountIds.filter((id) => existingIds.has(id)) +} + function buildScheduleEnterpriseSiteTargets(): ScheduleEnterpriseSiteTarget[] { return form.publishEnterpriseSiteIds .map((siteId) => ({ @@ -794,7 +822,7 @@ function validateForm(): boolean { } if (isSchedule.value && form.autoPublish) { - if (form.publishTargetTab === 'media_accounts' && form.publishAccountIds.length === 0) { + if (form.publishTargetTab === 'media_accounts' && currentSchedulePublishAccountIds().length === 0) { message.warning(t('custom.messages.missingPublishTargets')) return false } diff --git a/server/internal/scheduler/schedule_dispatch_worker.go b/server/internal/scheduler/schedule_dispatch_worker.go index 3ca72c3..6bf9309 100644 --- a/server/internal/scheduler/schedule_dispatch_worker.go +++ b/server/internal/scheduler/schedule_dispatch_worker.go @@ -469,6 +469,17 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC } task.PublishAccountIDs = decodeDueSchedulePublishAccountIDs(publishAccountIDsJSON) task.PublishEnterpriseSiteTargets = tenantapp.DecodeScheduleEnterpriseSiteTargets(publishEnterpriseSiteTargetsJSON) + if task.AutoPublish && len(task.PublishAccountIDs) > 0 { + accountIDs, _, err := tenantapp.LoadExistingSchedulePublishAccountIDsAndPlatformsByIDs(ctx, tx, task.TenantID, task.WorkspaceID, task.PublishAccountIDs) + if err != nil { + rows.Close() + return nil, err + } + task.PublishAccountIDs = accountIDs + } + if task.AutoPublish && len(task.PublishAccountIDs) == 0 && len(task.PublishEnterpriseSiteTargets) == 0 { + task.AutoPublish = false + } task.CoverRandomFolderID = int64PtrFromInt8(coverRandomFolderID) if task.CoverMode == "" { task.CoverMode = tenantapp.ScheduleCoverModeSpecific diff --git a/server/internal/tenant/app/desktop_account_service.go b/server/internal/tenant/app/desktop_account_service.go index 546da28..8ab67e4 100644 --- a/server/internal/tenant/app/desktop_account_service.go +++ b/server/internal/tenant/app/desktop_account_service.go @@ -16,6 +16,7 @@ import ( goredis "github.com/redis/go-redis/v9" "github.com/geo-platform/tenant-api/internal/shared/auth" + sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache" "github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq" "github.com/geo-platform/tenant-api/internal/shared/response" "github.com/geo-platform/tenant-api/internal/tenant/repository" @@ -27,6 +28,7 @@ type DesktopAccountService struct { pool *pgxpool.Pool repo repository.DesktopAccountRepository clientRepo repository.DesktopClientRepository + cache sharedcache.Cache redis *goredis.Client rabbitMQ *rabbitmq.Client now func() time.Time @@ -49,6 +51,11 @@ func NewDesktopAccountService( } } +func (s *DesktopAccountService) WithCache(c sharedcache.Cache) *DesktopAccountService { + s.cache = c + return s +} + type DesktopAccountView struct { ID string `json:"id"` Platform string `json:"platform"` @@ -389,6 +396,7 @@ func (s *DesktopAccountService) Tombstone(ctx context.Context, client *repositor } view := s.buildDesktopAccountView(account, nil, nil, nil, nil) + invalidateScheduleTaskCaches(ctx, s.cache, client.TenantID, nil) return &view, nil } @@ -451,6 +459,7 @@ func (s *DesktopAccountService) DeleteForActor(ctx context.Context, actor auth.A } view := s.buildDesktopAccountView(account, nil, nil, nil, nil) + invalidateScheduleTaskCaches(ctx, s.cache, actor.TenantID, nil) return &DesktopAccountDeleteResult{ Account: view, CancelledQueuedPublishTasks: cancelled, diff --git a/server/internal/tenant/app/publish_platforms.go b/server/internal/tenant/app/publish_platforms.go index 2d8f867..18c1e08 100644 --- a/server/internal/tenant/app/publish_platforms.go +++ b/server/internal/tenant/app/publish_platforms.go @@ -70,36 +70,71 @@ func loadPublishAccountPlatformsByIDs( tenantID int64, accountIDs []string, ) ([]string, error) { + _, platformIDs, err := loadExistingPublishAccountIDsAndPlatformsByIDs(ctx, db, tenantID, 0, accountIDs) + return platformIDs, err +} + +func LoadExistingSchedulePublishAccountIDsAndPlatformsByIDs( + ctx context.Context, + db scheduleAutoPublishPlatformQuerier, + tenantID int64, + workspaceID int64, + accountIDs []string, +) ([]string, []string, error) { + return loadExistingPublishAccountIDsAndPlatformsByIDs(ctx, db, tenantID, workspaceID, accountIDs) +} + +func loadExistingPublishAccountIDsAndPlatformsByIDs( + ctx context.Context, + db scheduleAutoPublishPlatformQuerier, + tenantID int64, + workspaceID int64, + accountIDs []string, +) ([]string, []string, error) { accountIDs = normalizeSchedulePublishAccountIDs(accountIDs) if db == nil || tenantID <= 0 || len(accountIDs) == 0 { - return []string{}, nil + return []string{}, []string{}, nil } rows, err := db.Query(ctx, ` - SELECT DISTINCT platform_id + SELECT DISTINCT desktop_id::text, platform_id FROM platform_accounts WHERE tenant_id = $1 AND desktop_id::text = ANY($2::text[]) + AND ($3::bigint <= 0 OR workspace_id = $3) AND deleted_at IS NULL - ORDER BY platform_id - `, tenantID, accountIDs) + `, tenantID, accountIDs, workspaceID) if err != nil { - return nil, err + return nil, nil, err } defer rows.Close() - platformIDs := make([]string, 0, len(accountIDs)) + platformsByAccount := make(map[string][]string, len(accountIDs)) + existingByAccount := make(map[string]struct{}, len(accountIDs)) for rows.Next() { + var accountID string var platformID string - if err := rows.Scan(&platformID); err != nil { - return nil, err + if err := rows.Scan(&accountID, &platformID); err != nil { + return nil, nil, err } - platformIDs = append(platformIDs, platformID) + platformsByAccount[accountID] = append(platformsByAccount[accountID], platformID) + existingByAccount[accountID] = struct{}{} } if err := rows.Err(); err != nil { - return nil, err + return nil, nil, err } - return emptyStringSliceIfNil(normalizePlatformIDs(platformIDs)), nil + + existingAccountIDs := make([]string, 0, len(accountIDs)) + platformIDs := make([]string, 0, len(accountIDs)) + for _, accountID := range accountIDs { + if _, ok := existingByAccount[accountID]; !ok { + continue + } + existingAccountIDs = append(existingAccountIDs, accountID) + platformIDs = append(platformIDs, platformsByAccount[accountID]...) + } + + return emptyStringSliceIfNil(existingAccountIDs), emptyStringSliceIfNil(normalizePlatformIDs(platformIDs)), nil } func emptyStringSliceIfNil(values []string) []string { diff --git a/server/internal/tenant/app/schedule_task_service.go b/server/internal/tenant/app/schedule_task_service.go index 20b4fb6..71c3288 100644 --- a/server/internal/tenant/app/schedule_task_service.go +++ b/server/internal/tenant/app/schedule_task_service.go @@ -625,10 +625,15 @@ func (s *ScheduleTaskService) populateScheduleAutoPublishPlatforms(ctx context.C if item == nil || !item.AutoPublish { return nil } - platformIDs, err := loadPublishAccountPlatformsByIDs(ctx, s.pool, tenantID, item.PublishAccountIDs) + workspaceID := int64(0) + if item.WorkspaceID != nil { + workspaceID = *item.WorkspaceID + } + accountIDs, platformIDs, err := loadExistingPublishAccountIDsAndPlatformsByIDs(ctx, s.pool, tenantID, workspaceID, item.PublishAccountIDs) if err != nil { return response.ErrInternal(50010, "query_failed", "failed to resolve schedule auto publish platforms") } + item.PublishAccountIDs = emptyStringSliceIfNil(accountIDs) item.AutoPublishPlatforms = platformIDs return nil } @@ -656,7 +661,7 @@ func (s *ScheduleTaskService) populateScheduleAutoPublishPlatformsBatch(ctx cont accountIDs = append(accountIDs, accountID) } rows, err := s.pool.Query(ctx, ` - SELECT desktop_id::text, platform_id + SELECT DISTINCT desktop_id::text, platform_id FROM platform_accounts WHERE tenant_id = $1 AND desktop_id::text = ANY($2::text[]) @@ -684,14 +689,21 @@ func (s *ScheduleTaskService) populateScheduleAutoPublishPlatformsBatch(ctx cont if !items[idx].AutoPublish { continue } + existingAccountIDs := make([]string, 0, len(items[idx].PublishAccountIDs)) platformIDs := make([]string, 0, len(items[idx].PublishAccountIDs)) for _, accountID := range items[idx].PublishAccountIDs { normalized := normalizeSchedulePublishAccountIDs([]string{accountID}) if len(normalized) == 0 { continue } - platformIDs = append(platformIDs, platformsByAccount[normalized[0]]...) + accountPlatforms := platformsByAccount[normalized[0]] + if len(accountPlatforms) == 0 { + continue + } + existingAccountIDs = append(existingAccountIDs, normalized[0]) + platformIDs = append(platformIDs, accountPlatforms...) } + items[idx].PublishAccountIDs = emptyStringSliceIfNil(existingAccountIDs) items[idx].AutoPublishPlatforms = emptyStringSliceIfNil(normalizePlatformIDs(platformIDs)) } return nil @@ -1135,6 +1147,18 @@ func (s *ScheduleTaskService) normalizeAutoPublishConfig(ctx context.Context, ac if len(accountIDs) == 0 && len(enterpriseTargets) == 0 { return config, response.ErrBadRequest(40024, "publish_targets_required", "开启自动发布后，请至少选择一个发布目标") } + if len(accountIDs) > 0 { + existingAccountIDs, platformIDs, err := loadExistingPublishAccountIDsAndPlatformsByIDs(ctx, s.pool, actor.TenantID, actor.PrimaryWorkspaceID, accountIDs) + if err != nil { + return config, response.ErrInternal(50010, "publish_account_check_failed", "failed to validate publish accounts") + } + accountIDs = existingAccountIDs + config.AccountIDs = accountIDs + config.PlatformIDs = normalizePlatformIDs(platformIDs) + } + if len(accountIDs) == 0 && len(enterpriseTargets) == 0 { + return config, response.ErrBadRequest(40024, "publish_targets_required", "开启自动发布后，请至少选择一个发布目标") + } if err := validateOptionalPositiveInt64(req.CoverImageAssetID, "cover_image_asset_id"); err != nil { return config, err } @@ -1152,12 +1176,7 @@ func (s *ScheduleTaskService) normalizeAutoPublishConfig(ctx context.Context, ac } coverURL := strings.TrimSpace(stringPointerValue(req.CoverAssetURL)) if len(accountIDs) > 0 { - platformIDs, err := s.loadPublishAccountPlatformIDs(ctx, actor.TenantID, actor.PrimaryWorkspaceID, accountIDs) - if err != nil { - return config, err - } - config.PlatformIDs = normalizePlatformIDs(platformIDs) - if schedulePublishPlatformsRequireCover(platformIDs) && coverMode == ScheduleCoverModeSpecific && coverURL == "" { + if schedulePublishPlatformsRequireCover(config.PlatformIDs) && coverMode == ScheduleCoverModeSpecific && coverURL == "" { return config, response.ErrBadRequest(40025, "cover_required", "已选媒体账号的平台要求封面图，请先设置封面") } } @@ -1281,37 +1300,6 @@ func (s *ScheduleTaskService) validateScheduleEnterpriseSiteTargets(ctx context. return nil } -func (s *ScheduleTaskService) loadPublishAccountPlatformIDs(ctx context.Context, tenantID, workspaceID int64, accountIDs []string) ([]string, error) { - rows, err := s.pool.Query(ctx, ` - SELECT platform_id - FROM platform_accounts - WHERE tenant_id = $1 - AND workspace_id = $2 - AND desktop_id::text = ANY($3::text[]) - AND deleted_at IS NULL - `, tenantID, workspaceID, accountIDs) - if err != nil { - return nil, response.ErrInternal(50010, "publish_account_check_failed", "failed to validate publish accounts") - } - defer rows.Close() - - platformIDs := make([]string, 0, len(accountIDs)) - for rows.Next() { - var platformID string - if err := rows.Scan(&platformID); err != nil { - return nil, response.ErrInternal(50010, "publish_account_check_failed", "failed to parse publish accounts") - } - platformIDs = append(platformIDs, strings.TrimSpace(platformID)) - } - if err := rows.Err(); err != nil { - return nil, response.ErrInternal(50010, "publish_account_check_failed", "failed to iterate publish accounts") - } - if len(platformIDs) != len(accountIDs) { - return nil, response.ErrBadRequest(40027, "invalid_publish_accounts", "one or more publish accounts do not exist") - } - return platformIDs, nil -} - func schedulePublishPlatformsRequireCover(platformIDs []string) bool { for _, platformID := range platformIDs { if publishPlatformRequiresCover(platformID) { diff --git a/server/internal/tenant/transport/desktop_account_handler.go b/server/internal/tenant/transport/desktop_account_handler.go index 6d455a0..b07e5cd 100644 --- a/server/internal/tenant/transport/desktop_account_handler.go +++ b/server/internal/tenant/transport/desktop_account_handler.go @@ -25,7 +25,7 @@ func NewDesktopAccountHandler(a *bootstrap.App) *DesktopAccountHandler { repository.NewDesktopClientRepository(a.DB), a.Redis, a.RabbitMQ, - ), + ).WithCache(a.Cache), } }