+
+
+
+
+
{
/>
+
+
+
+
+
{
showSizeChanger: true,
}"
row-key="id"
- :scroll="{ x: 1360 }"
+ :scroll="{ x: 1620 }"
@change="(p: any) => handleTableChange(p.current, p.pageSize)"
>
-
- {{ record.prompt_rule_name || '--' }}
+
+
+ {{ targetTypeLabel(record.target_type) }}
+
+
+
+
+ {{ scheduleTargetName(record) }}
+
+ {{ record.subscription_kol_name || '--' }}
+
+
+
+
+ {{ schedulePlanText(record) }}
+
+
+ {{ formatDateTime(record.next_run_at) }}
@@ -312,8 +479,12 @@ onBeforeUnmount(() => {
-
- {{ formatDateTime(record.execution_time) }}
+
+
+
+ {{ dispatchStatusMeta(record.last_dispatch_status).label }}
+
+
@@ -402,10 +573,7 @@ onBeforeUnmount(() => {
-
+
@@ -413,8 +581,8 @@ onBeforeUnmount(() => {
.task-tab__filters {
display: grid;
grid-template-columns:
- minmax(180px, 240px) minmax(180px, 240px) minmax(240px, 320px) minmax(240px, 1fr)
- auto;
+ repeat(3, minmax(180px, 240px)) minmax(180px, 220px) minmax(240px, 320px)
+ minmax(240px, 1fr) auto;
gap: 16px;
margin-bottom: 18px;
}
@@ -448,6 +616,29 @@ onBeforeUnmount(() => {
gap: 4px;
}
+.task-tab__target {
+ display: flex;
+ min-width: 0;
+ flex-direction: column;
+ gap: 4px;
+}
+
+.task-tab__target strong {
+ overflow: hidden;
+ color: #1f2937;
+ font-size: 13px;
+ font-weight: 700;
+ text-overflow: ellipsis;
+ white-space: nowrap;
+}
+
+.task-tab__target span,
+.task-tab__plan {
+ color: #667085;
+ font-size: 12px;
+ line-height: 1.5;
+}
+
@media (max-width: 1400px) {
.task-tab__filters {
grid-template-columns: repeat(2, minmax(260px, 1fr));
diff --git a/apps/admin-web/src/i18n/messages/en-US.ts b/apps/admin-web/src/i18n/messages/en-US.ts
index 2791403..1e1492c 100644
--- a/apps/admin-web/src/i18n/messages/en-US.ts
+++ b/apps/admin-web/src/i18n/messages/en-US.ts
@@ -1612,6 +1612,20 @@ const enUS = {
coverUpload: 'Upload cover image',
coverUploadHint: 'Local preview only for now',
scheduleCoverUploadHint: 'Choose from local files or the asset library',
+ source: 'Generation source',
+ sourcePromptRule: 'Custom Prompt',
+ sourceKolTemplate: 'Refined Template',
+ kolTemplate: 'Refined template',
+ kolTemplatePlaceholder: 'Select a subscribed refined template',
+ kolTemplateEmptyTitle: 'No refined templates available',
+ kolTemplateEmptyHint:
+ 'This account has no active subscribed refined template package yet. Subscribe in the marketplace first.',
+ scheduleSection: 'Schedule Policy',
+ kolInputSection: 'Refined Variables',
+ kolWebSearch: 'Web search',
+ kolWebSearchHint: 'Enable it when this refined template allows web search.',
+ kolKnowledge: 'Knowledge base',
+ kolKnowledgePlaceholder: 'Optional. Select your knowledge groups for generation',
scheduleTime: 'Schedule article generation',
scheduleEveryDay: 'Every day',
enableWebSearch: 'Enable web search for generation',
@@ -1629,11 +1643,29 @@ const enUS = {
schedule: {
name: 'Task name',
rule: 'Prompt rule',
+ target: 'Target',
+ plan: 'Schedule',
cron: 'Frequency',
+ frequency: 'Frequency',
+ kindDaily: 'Every day',
+ kindWeekly: 'Weekdays',
+ weekdaysLabel: 'Run days',
+ executionTime: 'Run time',
+ executionTimeHint:
+ 'New tasks default to a random time between 01:00 and 05:00 to smooth overnight load. You can edit it before saving.',
+ executionTimeDefault: 'Default staggered time',
+ to: 'to',
autoPublishPlatforms: 'Auto publish platforms',
autoPublish: 'Auto publish',
manualPublish: 'Manual publish',
nextRun: 'Next run',
+ dispatchStatus: 'Last dispatch',
+ dispatchSuccess: 'Success',
+ dispatchFailed: 'Failed',
+ dispatchPartialSuccess: 'Partial success',
+ notDispatched: 'Not dispatched',
+ footerHint:
+ 'New tasks start with a random time between 01:00 and 05:00, then run at the saved time.',
startAt: 'Start time',
endAt: 'End time',
brand: 'Brand',
@@ -1650,6 +1682,15 @@ const enUS = {
weekly: 'Once weekly',
custom: 'Custom Cron',
},
+ weekdays: {
+ mon: 'Mon',
+ tue: 'Tue',
+ wed: 'Wed',
+ thu: 'Thu',
+ fri: 'Fri',
+ sat: 'Sat',
+ sun: 'Sun',
+ },
},
instant: {
title: 'Instant Generate',
@@ -1700,8 +1741,11 @@ const enUS = {
generateSubmitted: 'Generation task submitted.',
missingTaskName: 'Enter a task name first.',
missingPromptRule: 'Choose a prompt first.',
+ missingKolTemplate: 'Choose a subscribed refined template first.',
invalidGenerateCount: 'Article count must be at least 1.',
invalidScheduleTime: 'Choose a valid schedule time.',
+ invalidRandomWindow: 'The schedule time configuration is invalid.',
+ missingScheduleDays: 'Select at least one run day.',
missingPublishAccounts: 'Choose at least one media account when auto publish is enabled.',
missingScheduleCover: 'A selected media account platform requires a cover image.',
missingPromptName: 'Enter a prompt name first.',
diff --git a/apps/admin-web/src/i18n/messages/zh-CN.ts b/apps/admin-web/src/i18n/messages/zh-CN.ts
index 1734012..ca246d2 100644
--- a/apps/admin-web/src/i18n/messages/zh-CN.ts
+++ b/apps/admin-web/src/i18n/messages/zh-CN.ts
@@ -1520,6 +1520,19 @@ const zhCN = {
coverUpload: "上传封面图",
coverUploadHint: "当前仅做本地预览",
scheduleCoverUploadHint: "从本地或素材库选择",
+ source: "生成来源",
+ sourcePromptRule: "自定义 Prompt",
+ sourceKolTemplate: "精调模版",
+ kolTemplate: "精调模版",
+ kolTemplatePlaceholder: "请选择已订阅的精调模版",
+ kolTemplateEmptyTitle: "暂无可用精调模版",
+ kolTemplateEmptyHint: "当前账号还没有订阅可用的精调模版包,请先到精调模版市场完成订阅。",
+ scheduleSection: "调度策略",
+ kolInputSection: "精调变量",
+ kolWebSearch: "联网搜索",
+ kolWebSearchHint: "该精调模版允许联网搜索时可开启。",
+ kolKnowledge: "知识库引用",
+ kolKnowledgePlaceholder: "可选,选择你的知识库分组辅助生成",
scheduleTime: "定时生成文章",
scheduleEveryDay: "每天",
enableWebSearch: "生成文章联网搜索",
@@ -1537,11 +1550,28 @@ const zhCN = {
schedule: {
name: "任务名称",
rule: "关联规则",
+ target: "生成目标",
+ plan: "调度计划",
cron: "执行频率",
+ frequency: "频率",
+ kindDaily: "每天",
+ kindWeekly: "按星期",
+ weekdaysLabel: "执行日",
+ executionTime: "执行时间",
+ executionTimeHint:
+ "新建任务会默认随机落在 01:00-05:00,避免凌晨任务集中;你也可以手动修改为固定执行时间。",
+ executionTimeDefault: "默认错峰生成",
+ to: "至",
autoPublishPlatforms: "自动发布平台",
autoPublish: "自动发布",
manualPublish: "手动发布",
nextRun: "下次执行",
+ dispatchStatus: "最近调度",
+ dispatchSuccess: "成功",
+ dispatchFailed: "失败",
+ dispatchPartialSuccess: "部分成功",
+ notDispatched: "未调度",
+ footerHint: "新建任务默认给出凌晨 1 点到 5 点之间的随机时间,保存后按所选时间执行。",
startAt: "开始时间",
endAt: "结束时间",
brand: "关联品牌",
@@ -1558,6 +1588,15 @@ const zhCN = {
weekly: "每周一次",
custom: "自定义 Cron",
},
+ weekdays: {
+ mon: "周一",
+ tue: "周二",
+ wed: "周三",
+ thu: "周四",
+ fri: "周五",
+ sat: "周六",
+ sun: "周日",
+ },
},
instant: {
title: "即时生成",
@@ -1608,8 +1647,11 @@ const zhCN = {
generateSubmitted: "生成任务已提交",
missingTaskName: "请先填写任务名称",
missingPromptRule: "请先选择 Prompt",
+ missingKolTemplate: "请先选择已订阅的精调模版",
invalidGenerateCount: "生成篇数至少为 1",
invalidScheduleTime: "请选择有效的定时时间",
+ invalidRandomWindow: "执行时间配置不正确",
+ missingScheduleDays: "请至少选择一个执行日",
missingPublishAccounts: "开启自动发布后,请至少选择一个媒体账号",
missingScheduleCover: "已选媒体账号的平台要求封面图,请先设置封面",
missingPromptName: "请先填写 Prompt 名称",
diff --git a/packages/shared-types/src/index.ts b/packages/shared-types/src/index.ts
index b161d7c..4c2ee53 100644
--- a/packages/shared-types/src/index.ts
+++ b/packages/shared-types/src/index.ts
@@ -1675,15 +1675,31 @@ export interface PromptRuleStatusRequest {
// --- Schedule Tasks ---
+export type ScheduleTargetType = 'prompt_rule' | 'kol_subscription_prompt'
+export type ScheduleKind = 'daily' | 'weekly'
+export type ScheduleTimeMode = 'fixed' | 'random_window'
+export type ScheduleDispatchStatus = 'success' | 'failed' | string
+
export interface ScheduleTask {
id: number
workspace_id: number | null
- prompt_rule_id: number
+ target_type: ScheduleTargetType
+ prompt_rule_id: number | null
prompt_rule_name: string | null
+ subscription_prompt_id: number | null
+ subscription_prompt_name: string | null
+ subscription_package_name: string | null
+ subscription_kol_name: string | null
brand_id: number | null
brand_name: string | null
name: string
cron_expr: string
+ schedule_kind: ScheduleKind
+ schedule_days: string[]
+ schedule_time_mode: ScheduleTimeMode
+ random_window_start: string
+ random_window_end: string
+ generation_input: Record
auto_publish: boolean
publish_account_ids: string[]
auto_publish_platforms: string[]
@@ -1695,6 +1711,10 @@ export interface ScheduleTask {
end_at: string | null
next_run_at: string | null
status: 'enabled' | 'disabled'
+ last_run_at: string | null
+ last_dispatch_status: ScheduleDispatchStatus | null
+ last_dispatch_error: string | null
+ consecutive_failures: number
generation_status?: string | null
execution_time?: string | null
generated_articles?: GeneratedArticleLink[] | null
@@ -1703,10 +1723,18 @@ export interface ScheduleTask {
}
export interface ScheduleTaskRequest {
- prompt_rule_id: number
+ target_type?: ScheduleTargetType
+ prompt_rule_id?: number | null
+ subscription_prompt_id?: number | null
brand_id?: number | null
name: string
- cron_expr: string
+ cron_expr?: string
+ schedule_kind?: ScheduleKind
+ schedule_days?: string[]
+ schedule_time_mode?: ScheduleTimeMode
+ random_window_start?: string
+ random_window_end?: string
+ generation_input?: Record
auto_publish?: boolean
publish_account_ids?: string[]
cover_asset_url?: string | null
@@ -1721,6 +1749,8 @@ export interface ScheduleTaskListParams {
page?: number
page_size?: number
prompt_rule_id?: number
+ subscription_prompt_id?: number
+ target_type?: ScheduleTargetType
status?: string
keyword?: string
created_from?: string
diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go
index 106b282..6c9787f 100644
--- a/server/cmd/scheduler/main.go
+++ b/server/cmd/scheduler/main.go
@@ -19,6 +19,7 @@ import (
internalscheduler "github.com/geo-platform/tenant-api/internal/scheduler"
"github.com/geo-platform/tenant-api/internal/shared/response"
tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app"
+ "github.com/geo-platform/tenant-api/internal/tenant/repository"
generateworker "github.com/geo-platform/tenant-api/internal/worker/generate"
"github.com/gin-gonic/gin"
)
@@ -63,6 +64,17 @@ func main() {
generationCfg,
app.Config().LLM.MaxOutputTokens,
).WithCache(app.Cache).WithLogger(app.Logger).WithConfigProvider(generationProvider)
+ kolGenerationSvc := tenantapp.NewKolGenerationService(
+ app.DB,
+ app.KolSubscriptions,
+ app.KolPrompts,
+ repository.NewKolUsageRepository(app.DB),
+ repository.NewArticleRepository(app.DB),
+ repository.NewAuditRepository(app.DB),
+ repository.NewQuotaRepository(app.DB),
+ app.KolPromptAsset,
+ app.RabbitMQ,
+ ).WithCache(app.Cache)
monitoringCallbackService := tenantapp.NewMonitoringCallbackService(
app.DB,
@@ -86,7 +98,7 @@ func main() {
app.Cache,
app.Logger,
app.Config().Scheduler,
- ).WithConfigProvider(app.ConfigStore)
+ ).WithKolGenerationService(kolGenerationSvc).WithConfigProvider(app.ConfigStore)
monitoringDailyTaskWorker := tenantapp.NewMonitoringDailyTaskWorker(monitoringService, monitoringCallbackService, app.Logger)
monitoringResultRecoveryWorker := internalscheduler.NewMonitoringResultRecoveryWorker(app.MonitoringDB, monitoringCallbackService, app.Logger)
monitoringLeaseRecoveryWorker := internalscheduler.NewMonitoringLeaseRecoveryWorker(app.MonitoringDB, app.Logger)
diff --git a/server/internal/scheduler/schedule_dispatch_worker.go b/server/internal/scheduler/schedule_dispatch_worker.go
index 4597889..66d0c29 100644
--- a/server/internal/scheduler/schedule_dispatch_worker.go
+++ b/server/internal/scheduler/schedule_dispatch_worker.go
@@ -3,6 +3,9 @@ package scheduler
import (
"context"
"encoding/json"
+ "errors"
+ "fmt"
+ "strings"
"sync"
"time"
@@ -22,12 +25,14 @@ const (
defaultScheduleDispatchTimeout = 30 * time.Second
defaultScheduleDispatchBatch = 20
scheduleCacheCleanupTimeout = 30 * time.Second
+ maxInvalidScheduleFailures = 3
)
type ScheduleDispatchWorker struct {
pool *pgxpool.Pool
rabbitMQ *rabbitmq.Client
promptRuleService *tenantapp.PromptRuleGenerationService
+ kolService *tenantapp.KolGenerationService
cache sharedcache.Cache
logger *zap.Logger
cfg scheduleDispatchRuntimeConfig
@@ -35,23 +40,30 @@ type ScheduleDispatchWorker struct {
}
type dueScheduleTask struct {
- ID int64
- TenantID int64
- WorkspaceID int64
- OperatorID *int64
- PromptRuleID int64
- BrandID *int64
- Name string
- CronExpr string
- AutoPublish bool
- PublishAccountIDs []string
- CoverAssetURL *string
- CoverImageAssetID *int64
- EnableWebSearch bool
- GenerateCount int
- StartAt *time.Time
- EndAt *time.Time
- ScheduledFor time.Time
+ ID int64
+ TenantID int64
+ WorkspaceID int64
+ OperatorID *int64
+ TargetType string
+ PromptRuleID *int64
+ SubscriptionPromptID *int64
+ BrandID *int64
+ Name string
+ CronExpr string
+ ScheduleDays []string
+ ScheduleTimeMode string
+ RandomWindowStart string
+ RandomWindowEnd string
+ GenerationInput map[string]interface{}
+ AutoPublish bool
+ PublishAccountIDs []string
+ CoverAssetURL *string
+ CoverImageAssetID *int64
+ EnableWebSearch bool
+ GenerateCount int
+ StartAt *time.Time
+ EndAt *time.Time
+ ScheduledFor time.Time
}
type scheduleTaskCacheUpdate struct {
@@ -85,6 +97,11 @@ func NewScheduleDispatchWorker(
}
}
+func (w *ScheduleDispatchWorker) WithKolGenerationService(svc *tenantapp.KolGenerationService) *ScheduleDispatchWorker {
+ w.kolService = svc
+ return w
+}
+
func (w *ScheduleDispatchWorker) WithConfigProvider(provider config.Provider) *ScheduleDispatchWorker {
w.configProvider = provider
return w
@@ -237,7 +254,8 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context, run
defer tx.Rollback(ctx)
rows, err := tx.Query(ctx, `
- SELECT id, tenant_id, cron_expr, start_at, end_at, status
+ SELECT id, tenant_id, cron_expr, schedule_days, schedule_time_mode,
+ random_window_start::text, random_window_end::text, start_at, end_at, status
FROM schedule_tasks
WHERE deleted_at IS NULL
AND status = 'enabled'
@@ -255,29 +273,35 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context, run
updates := make([]struct {
cacheUpdate scheduleTaskCacheUpdate
nextRunAt *time.Time
+ invalidErr error
}, 0, runtimeCfg.BatchSize)
for rows.Next() {
var (
- id int64
- tenantID int64
- cronExpr string
- status string
- startAt pgtype.Timestamptz
- endAt pgtype.Timestamptz
+ id int64
+ tenantID int64
+ cronExpr string
+ scheduleDaysJSON []byte
+ scheduleTimeMode string
+ randomWindowStart string
+ randomWindowEnd string
+ status string
+ startAt pgtype.Timestamptz
+ endAt pgtype.Timestamptz
)
- if err := rows.Scan(&id, &tenantID, &cronExpr, &startAt, &endAt, &status); err != nil {
+ if err := rows.Scan(&id, &tenantID, &cronExpr, &scheduleDaysJSON, &scheduleTimeMode, &randomWindowStart, &randomWindowEnd, &startAt, &endAt, &status); err != nil {
rows.Close()
return 0, err
}
var nextRunAt *time.Time
+ var nextErr error
if status == "enabled" {
- nextRunAt, err = sharedschedule.ComputeNextRun(cronExpr, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), now)
- if err != nil && w.logger != nil {
- w.logger.Warn("schedule dispatch skipped invalid cron while hydrating next_run_at",
+ nextRunAt, nextErr = tenantapp.ComputeScheduleTaskNextRun(status, id, cronExpr, tenantapp.DecodeScheduleDayKeys(scheduleDaysJSON), scheduleTimeMode, randomWindowStart, randomWindowEnd, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), now)
+ if nextErr != nil && w.logger != nil {
+ w.logger.Warn("schedule dispatch skipped invalid schedule while hydrating next_run_at",
zap.Int64("schedule_task_id", id),
zap.String("cron_expr", cronExpr),
- zap.Error(err),
+ zap.Error(nextErr),
)
}
}
@@ -285,12 +309,14 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context, run
updates = append(updates, struct {
cacheUpdate scheduleTaskCacheUpdate
nextRunAt *time.Time
+ invalidErr error
}{
cacheUpdate: scheduleTaskCacheUpdate{
id: id,
tenantID: tenantID,
},
- nextRunAt: nextRunAt,
+ nextRunAt: nextRunAt,
+ invalidErr: nextErr,
})
}
if err := rows.Err(); err != nil {
@@ -300,6 +326,23 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context, run
rows.Close()
for _, update := range updates {
+ if update.invalidErr != nil {
+ message := truncateDispatchError(update.invalidErr)
+ if _, err := tx.Exec(ctx, `
+ UPDATE schedule_tasks
+ SET next_run_at = NULL,
+ last_dispatch_status = 'failed',
+ last_dispatch_error = NULLIF($1, ''),
+ consecutive_failures = consecutive_failures + 1,
+ status = CASE WHEN consecutive_failures + 1 >= $2 THEN 'disabled' ELSE status END,
+ updated_at = NOW()
+ WHERE id = $3
+ `, message, maxInvalidScheduleFailures, update.cacheUpdate.id); err != nil {
+ return 0, err
+ }
+ count++
+ continue
+ }
if _, err := tx.Exec(ctx, `
UPDATE schedule_tasks
SET next_run_at = $1,
@@ -330,7 +373,9 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
defer tx.Rollback(ctx)
rows, err := tx.Query(ctx, `
- SELECT id, tenant_id, workspace_id, operator_id, prompt_rule_id, brand_id, name, cron_expr,
+ SELECT id, tenant_id, workspace_id, operator_id, target_type, prompt_rule_id,
+ subscription_prompt_id, brand_id, name, cron_expr, schedule_days, schedule_time_mode,
+ random_window_start::text, random_window_end::text, generation_input_json,
auto_publish, publish_account_ids, cover_asset_url, cover_image_asset_id,
enable_web_search, generate_count, start_at, end_at, next_run_at
FROM schedule_tasks
@@ -351,12 +396,17 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
updates := make([]struct {
cacheUpdate scheduleTaskCacheUpdate
nextRun *time.Time
+ invalidErr error
}, 0, runtimeCfg.BatchSize)
for rows.Next() {
var (
task dueScheduleTask
workspaceID pgtype.Int8
operatorID pgtype.Int8
+ promptRuleID pgtype.Int8
+ subscriptionPromptID pgtype.Int8
+ generationInputJSON []byte
+ scheduleDaysJSON []byte
publishAccountIDsJSON []byte
startAt pgtype.Timestamptz
endAt pgtype.Timestamptz
@@ -367,10 +417,17 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
&task.TenantID,
&workspaceID,
&operatorID,
- &task.PromptRuleID,
+ &task.TargetType,
+ &promptRuleID,
+ &subscriptionPromptID,
&task.BrandID,
&task.Name,
&task.CronExpr,
+ &scheduleDaysJSON,
+ &task.ScheduleTimeMode,
+ &task.RandomWindowStart,
+ &task.RandomWindowEnd,
+ &generationInputJSON,
&task.AutoPublish,
&publishAccountIDsJSON,
&task.CoverAssetURL,
@@ -391,6 +448,13 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
task.WorkspaceID = workspaceID.Int64
}
task.OperatorID = int64PtrFromInt8(operatorID)
+ task.PromptRuleID = int64PtrFromInt8(promptRuleID)
+ task.SubscriptionPromptID = int64PtrFromInt8(subscriptionPromptID)
+ task.ScheduleDays = tenantapp.DecodeScheduleDayKeys(scheduleDaysJSON)
+ task.GenerationInput = decodeDueScheduleGenerationInput(generationInputJSON)
+ if task.TargetType == "" {
+ task.TargetType = tenantapp.ScheduleTargetPromptRule
+ }
task.PublishAccountIDs = decodeDueSchedulePublishAccountIDs(publishAccountIDsJSON)
if task.GenerateCount <= 0 {
task.GenerateCount = 1
@@ -401,28 +465,29 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
}
task.ScheduledFor = *scheduledFor
- nextRun, nextErr := sharedschedule.ComputeNextRun(task.CronExpr, task.StartAt, task.EndAt, now.Add(time.Nanosecond))
+ nextRun, nextErr := tenantapp.ComputeScheduleTaskNextRun("enabled", task.ID, task.CronExpr, task.ScheduleDays, task.ScheduleTimeMode, task.RandomWindowStart, task.RandomWindowEnd, task.StartAt, task.EndAt, now.Add(time.Nanosecond))
if nextErr != nil && w.logger != nil {
- w.logger.Warn("schedule dispatch encountered invalid cron while claiming due schedule",
+ w.logger.Warn("schedule dispatch encountered invalid schedule while claiming due schedule",
zap.Int64("schedule_task_id", task.ID),
zap.String("cron_expr", task.CronExpr),
zap.Error(nextErr),
)
- nextRun = nil
}
updates = append(updates, struct {
cacheUpdate scheduleTaskCacheUpdate
nextRun *time.Time
+ invalidErr error
}{
cacheUpdate: scheduleTaskCacheUpdate{
id: task.ID,
tenantID: task.TenantID,
},
- nextRun: nextRun,
+ nextRun: nextRun,
+ invalidErr: nextErr,
})
- if sharedschedule.IsRunAllowed(task.ScheduledFor, task.StartAt, task.EndAt) {
+ if nextErr == nil && sharedschedule.IsRunAllowed(task.ScheduledFor, task.StartAt, task.EndAt) {
tasks = append(tasks, task)
}
}
@@ -433,6 +498,22 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
rows.Close()
for _, update := range updates {
+ if update.invalidErr != nil {
+ message := truncateDispatchError(update.invalidErr)
+ if _, err := tx.Exec(ctx, `
+ UPDATE schedule_tasks
+ SET next_run_at = NULL,
+ last_dispatch_status = 'failed',
+ last_dispatch_error = NULLIF($1, ''),
+ consecutive_failures = consecutive_failures + 1,
+ status = CASE WHEN consecutive_failures + 1 >= $2 THEN 'disabled' ELSE status END,
+ updated_at = NOW()
+ WHERE id = $3
+ `, message, maxInvalidScheduleFailures, update.cacheUpdate.id); err != nil {
+ return nil, err
+ }
+ continue
+ }
if _, err := tx.Exec(ctx, `
UPDATE schedule_tasks
SET next_run_at = $1,
@@ -460,31 +541,20 @@ func (w *ScheduleDispatchWorker) dispatch(parent context.Context, task dueSchedu
generateCount = 1
}
+ failures := make([]error, 0)
for idx := 0; idx < generateCount; idx++ {
ctx, cancel := w.stageContext(parent)
- resp, err := w.promptRuleService.EnqueueScheduledGeneration(ctx, tenantapp.SchedulePromptRuleGenerationInput{
- ScheduleTaskID: task.ID,
- OperatorID: task.OperatorID,
- TenantID: task.TenantID,
- PromptRuleID: task.PromptRuleID,
- BrandID: task.BrandID,
- Name: task.Name,
- WorkspaceID: task.WorkspaceID,
- AutoPublish: task.AutoPublish,
- PublishAccountIDs: task.PublishAccountIDs,
- CoverAssetURL: task.CoverAssetURL,
- CoverImageAssetID: task.CoverImageAssetID,
- EnableWebSearch: task.EnableWebSearch,
- GenerateCount: generateCount,
- ScheduledFor: task.ScheduledFor,
- })
+ resp, err := w.enqueueDueGeneration(ctx, task, generateCount)
cancel()
if err != nil {
+ failures = append(failures, err)
if w.logger != nil {
w.logger.Warn("schedule dispatch enqueue generation failed",
zap.Int64("schedule_task_id", task.ID),
zap.Int64("tenant_id", task.TenantID),
- zap.Int64("prompt_rule_id", task.PromptRuleID),
+ zap.String("target_type", task.TargetType),
+ zap.Int64("prompt_rule_id", derefInt64(task.PromptRuleID)),
+ zap.Int64("subscription_prompt_id", derefInt64(task.SubscriptionPromptID)),
zap.Int("generate_index", idx+1),
zap.Int("generate_count", generateCount),
zap.Time("scheduled_for", task.ScheduledFor.UTC()),
@@ -498,7 +568,9 @@ func (w *ScheduleDispatchWorker) dispatch(parent context.Context, task dueSchedu
w.logger.Info("schedule dispatch enqueued generation",
zap.Int64("schedule_task_id", task.ID),
zap.Int64("tenant_id", task.TenantID),
- zap.Int64("prompt_rule_id", task.PromptRuleID),
+ zap.String("target_type", task.TargetType),
+ zap.Int64("prompt_rule_id", derefInt64(task.PromptRuleID)),
+ zap.Int64("subscription_prompt_id", derefInt64(task.SubscriptionPromptID)),
zap.Int64("article_id", resp.ArticleID),
zap.Int64("generation_task_id", resp.TaskID),
zap.Int("generate_index", idx+1),
@@ -507,6 +579,141 @@ func (w *ScheduleDispatchWorker) dispatch(parent context.Context, task dueSchedu
)
}
}
+ w.recordDispatchResult(context.Background(), task, generateCount, failures)
+}
+
+func (w *ScheduleDispatchWorker) enqueueDueGeneration(ctx context.Context, task dueScheduleTask, generateCount int) (*tenantapp.GenerateFromRuleResponse, error) {
+ switch task.TargetType {
+ case "", tenantapp.ScheduleTargetPromptRule:
+ if task.PromptRuleID == nil || *task.PromptRuleID <= 0 {
+ return nil, errors.New("schedule prompt_rule target missing prompt_rule_id")
+ }
+ return w.promptRuleService.EnqueueScheduledGeneration(ctx, tenantapp.SchedulePromptRuleGenerationInput{
+ ScheduleTaskID: task.ID,
+ OperatorID: task.OperatorID,
+ TenantID: task.TenantID,
+ PromptRuleID: *task.PromptRuleID,
+ BrandID: task.BrandID,
+ Name: task.Name,
+ WorkspaceID: task.WorkspaceID,
+ AutoPublish: task.AutoPublish,
+ PublishAccountIDs: task.PublishAccountIDs,
+ CoverAssetURL: task.CoverAssetURL,
+ CoverImageAssetID: task.CoverImageAssetID,
+ EnableWebSearch: task.EnableWebSearch,
+ GenerateCount: generateCount,
+ ScheduledFor: task.ScheduledFor,
+ })
+ case tenantapp.ScheduleTargetKolSubscriptionPrompt:
+ if w.kolService == nil {
+ return nil, errors.New("kol generation service is not configured")
+ }
+ if task.SubscriptionPromptID == nil || *task.SubscriptionPromptID <= 0 {
+ return nil, errors.New("schedule kol target missing subscription_prompt_id")
+ }
+ variables, _ := task.GenerationInput["variables"].(map[string]interface{})
+ kolResp, err := w.kolService.EnqueueScheduledGeneration(ctx, tenantapp.ScheduleKolGenerationInput{
+ ScheduleTaskID: task.ID,
+ OperatorID: task.OperatorID,
+ TenantID: task.TenantID,
+ WorkspaceID: task.WorkspaceID,
+ BrandID: task.BrandID,
+ SubscriptionPromptID: *task.SubscriptionPromptID,
+ Name: task.Name,
+ Variables: variables,
+ EnableWebSearch: scheduleGenerationInputBool(task.GenerationInput, "enable_web_search", task.EnableWebSearch),
+ KnowledgeGroupIDs: scheduleGenerationInputInt64s(task.GenerationInput, "knowledge_group_ids"),
+ AutoPublish: task.AutoPublish,
+ PublishAccountIDs: task.PublishAccountIDs,
+ CoverAssetURL: task.CoverAssetURL,
+ CoverImageAssetID: task.CoverImageAssetID,
+ GenerateCount: generateCount,
+ ScheduledFor: task.ScheduledFor,
+ })
+ if err != nil {
+ return nil, err
+ }
+ return &tenantapp.GenerateFromRuleResponse{
+ ArticleID: kolResp.ArticleID,
+ TaskID: kolResp.GenerationTaskID,
+ }, nil
+ default:
+ return nil, fmt.Errorf("unsupported schedule target type: %s", task.TargetType)
+ }
+}
+
+func (w *ScheduleDispatchWorker) recordDispatchResult(ctx context.Context, task dueScheduleTask, generateCount int, failures []error) {
+ if w == nil || w.pool == nil {
+ return
+ }
+ failedCount := len(failures)
+ result := aggregateDispatchResult(generateCount, failedCount)
+ if result.status == "success" {
+ _, _ = w.pool.Exec(ctx, `
+ UPDATE schedule_tasks
+ SET last_run_at = $1,
+ last_dispatch_status = $2,
+ last_dispatch_error = NULL,
+ consecutive_failures = 0,
+ updated_at = NOW()
+ WHERE id = $3 AND tenant_id = $4
+ `, task.ScheduledFor, result.status, task.ID, task.TenantID)
+ return
+ }
+ message := truncateDispatchErrors(failures)
+ _, _ = w.pool.Exec(ctx, `
+ UPDATE schedule_tasks
+ SET last_run_at = $1,
+ last_dispatch_status = $2,
+ last_dispatch_error = NULLIF($3, ''),
+ consecutive_failures = CASE WHEN $4 THEN consecutive_failures + 1 ELSE 0 END,
+ updated_at = NOW()
+ WHERE id = $5 AND tenant_id = $6
+ `, task.ScheduledFor, result.status, message, result.countAsFailure, task.ID, task.TenantID)
+}
+
+type dispatchAggregateResult struct {
+ status string
+ countAsFailure bool
+}
+
+func aggregateDispatchResult(generateCount, failedCount int) dispatchAggregateResult {
+ if generateCount <= 0 {
+ generateCount = 1
+ }
+ if failedCount <= 0 {
+ return dispatchAggregateResult{status: "success"}
+ }
+ if failedCount >= generateCount {
+ return dispatchAggregateResult{status: "failed", countAsFailure: true}
+ }
+ return dispatchAggregateResult{status: "partial"}
+}
+
+func truncateDispatchError(err error) string {
+ if err == nil {
+ return ""
+ }
+ message := err.Error()
+ if len(message) > 1000 {
+ return message[:1000]
+ }
+ return message
+}
+
+func truncateDispatchErrors(errs []error) string {
+ parts := make([]string, 0, len(errs))
+ for _, err := range errs {
+ if err == nil {
+ continue
+ }
+ parts = append(parts, err.Error())
+ }
+ message := strings.Join(parts, "; ")
+ if len(message) > 1000 {
+ return message[:1000]
+ }
+ return message
}
func (w *ScheduleDispatchWorker) dispatchBatch(parent context.Context, tasks []dueScheduleTask) {
@@ -662,6 +869,24 @@ func int64PtrFromInt8(value pgtype.Int8) *int64 {
return &result
}
+func derefInt64(value *int64) int64 {
+ if value == nil {
+ return 0
+ }
+ return *value
+}
+
+func decodeDueScheduleGenerationInput(raw []byte) map[string]interface{} {
+ if len(raw) == 0 {
+ return map[string]interface{}{}
+ }
+ var payload map[string]interface{}
+ if err := json.Unmarshal(raw, &payload); err != nil || payload == nil {
+ return map[string]interface{}{}
+ }
+ return payload
+}
+
func decodeDueSchedulePublishAccountIDs(raw []byte) []string {
if len(raw) == 0 {
return []string{}
@@ -684,3 +909,65 @@ func decodeDueSchedulePublishAccountIDs(raw []byte) []string {
}
return result
}
+
+func scheduleGenerationInputBool(input map[string]interface{}, key string, fallback bool) bool {
+ if len(input) == 0 {
+ return fallback
+ }
+ raw, ok := input[key]
+ if !ok || raw == nil {
+ return fallback
+ }
+ switch typed := raw.(type) {
+ case bool:
+ return typed
+ case string:
+ return typed == "true" || typed == "1" || typed == "yes"
+ case float64:
+ return typed != 0
+ default:
+ return fallback
+ }
+}
+
+func scheduleGenerationInputInt64s(input map[string]interface{}, key string) []int64 {
+ if len(input) == 0 {
+ return nil
+ }
+ raw, ok := input[key]
+ if !ok || raw == nil {
+ return nil
+ }
+ var ids []int64
+ switch typed := raw.(type) {
+ case []int64:
+ ids = typed
+ case []interface{}:
+ ids = make([]int64, 0, len(typed))
+ for _, item := range typed {
+ switch value := item.(type) {
+ case int64:
+ ids = append(ids, value)
+ case int:
+ ids = append(ids, int64(value))
+ case float64:
+ ids = append(ids, int64(value))
+ }
+ }
+ default:
+ return nil
+ }
+ seen := make(map[int64]struct{}, len(ids))
+ result := make([]int64, 0, len(ids))
+ for _, id := range ids {
+ if id <= 0 {
+ continue
+ }
+ if _, ok := seen[id]; ok {
+ continue
+ }
+ seen[id] = struct{}{}
+ result = append(result, id)
+ }
+ return result
+}
diff --git a/server/internal/scheduler/schedule_dispatch_worker_test.go b/server/internal/scheduler/schedule_dispatch_worker_test.go
new file mode 100644
index 0000000..3ecd7a6
--- /dev/null
+++ b/server/internal/scheduler/schedule_dispatch_worker_test.go
@@ -0,0 +1,30 @@
+package scheduler
+
+import "testing"
+
+func TestAggregateDispatchResult(t *testing.T) {
+ tests := []struct {
+ name string
+ generateCount int
+ failedCount int
+ wantStatus string
+ wantFailureCount bool
+ }{
+ {name: "all success", generateCount: 3, failedCount: 0, wantStatus: "success"},
+ {name: "partial success", generateCount: 3, failedCount: 1, wantStatus: "partial"},
+ {name: "all failed", generateCount: 3, failedCount: 3, wantStatus: "failed", wantFailureCount: true},
+ {name: "zero generate count defaults to one", generateCount: 0, failedCount: 1, wantStatus: "failed", wantFailureCount: true},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got := aggregateDispatchResult(tt.generateCount, tt.failedCount)
+ if got.status != tt.wantStatus {
+ t.Fatalf("status = %q, want %q", got.status, tt.wantStatus)
+ }
+ if got.countAsFailure != tt.wantFailureCount {
+ t.Fatalf("countAsFailure = %v, want %v", got.countAsFailure, tt.wantFailureCount)
+ }
+ })
+ }
+}
diff --git a/server/internal/shared/schedule/policy.go b/server/internal/shared/schedule/policy.go
new file mode 100644
index 0000000..011dbfd
--- /dev/null
+++ b/server/internal/shared/schedule/policy.go
@@ -0,0 +1,177 @@
+package schedule
+
+import (
+ "crypto/sha256"
+ "encoding/binary"
+ "fmt"
+ "strings"
+ "time"
+)
+
+const (
+ TimeModeFixed = "fixed"
+ TimeModeRandomWindow = "random_window"
+
+ maxRandomRunSearchDays = 370
+)
+
+type RandomPolicy struct {
+ Days []time.Weekday
+ TimeMode string
+ WindowStart time.Duration
+ WindowEnd time.Duration
+ Seed int64
+}
+
+func ParseDayKey(value string) (time.Weekday, bool) {
+ switch strings.ToLower(strings.TrimSpace(value)) {
+ case "sun", "sunday", "0", "7":
+ return time.Sunday, true
+ case "mon", "monday", "1":
+ return time.Monday, true
+ case "tue", "tuesday", "2":
+ return time.Tuesday, true
+ case "wed", "wednesday", "3":
+ return time.Wednesday, true
+ case "thu", "thursday", "4":
+ return time.Thursday, true
+ case "fri", "friday", "5":
+ return time.Friday, true
+ case "sat", "saturday", "6":
+ return time.Saturday, true
+ default:
+ return time.Sunday, false
+ }
+}
+
+func DayKey(day time.Weekday) string {
+ switch day {
+ case time.Sunday:
+ return "sun"
+ case time.Monday:
+ return "mon"
+ case time.Tuesday:
+ return "tue"
+ case time.Wednesday:
+ return "wed"
+ case time.Thursday:
+ return "thu"
+ case time.Friday:
+ return "fri"
+ case time.Saturday:
+ return "sat"
+ default:
+ return ""
+ }
+}
+
+func ParseClockDuration(value string) (time.Duration, error) {
+ trimmed := strings.TrimSpace(value)
+ if trimmed == "" {
+ return 0, fmt.Errorf("time value is required")
+ }
+ parsed, err := time.Parse("15:04:05", trimmed)
+ if err != nil {
+ parsed, err = time.Parse("15:04", trimmed)
+ }
+ if err != nil {
+ return 0, fmt.Errorf("parse time %q: %w", trimmed, err)
+ }
+ return time.Duration(parsed.Hour())*time.Hour +
+ time.Duration(parsed.Minute())*time.Minute +
+ time.Duration(parsed.Second())*time.Second, nil
+}
+
+func FormatClockDuration(value time.Duration) string {
+ if value < 0 {
+ value = 0
+ }
+ value = value.Truncate(time.Second)
+ hour := int(value / time.Hour)
+ value -= time.Duration(hour) * time.Hour
+ minute := int(value / time.Minute)
+ value -= time.Duration(minute) * time.Minute
+ second := int(value / time.Second)
+ return fmt.Sprintf("%02d:%02d:%02d", hour, minute, second)
+}
+
+func ComputeNextRandomRun(policy RandomPolicy, startAt, endAt *time.Time, now time.Time) (*time.Time, error) {
+ return computeNextRandomRun(policy, startAt, endAt, now, maxRandomRunSearchDays)
+}
+
+func computeNextRandomRun(policy RandomPolicy, startAt, endAt *time.Time, now time.Time, maxSearchDays int) (*time.Time, error) {
+ days := normalizePolicyDays(policy.Days)
+ if len(days) == 0 {
+ return nil, fmt.Errorf("schedule days are required")
+ }
+ if strings.TrimSpace(policy.TimeMode) == "" {
+ policy.TimeMode = TimeModeRandomWindow
+ }
+ if policy.TimeMode != TimeModeRandomWindow {
+ return nil, fmt.Errorf("unsupported random policy time mode: %s", policy.TimeMode)
+ }
+ if policy.WindowEnd <= policy.WindowStart {
+ return nil, fmt.Errorf("random window end must be after start")
+ }
+
+ notBefore := now.In(time.Local)
+ if startAt != nil {
+ start := startAt.In(time.Local)
+ if start.After(notBefore) {
+ notBefore = start
+ }
+ }
+
+ allowed := make(map[time.Weekday]struct{}, len(days))
+ for _, day := range days {
+ allowed[day] = struct{}{}
+ }
+
+ baseDate := time.Date(notBefore.Year(), notBefore.Month(), notBefore.Day(), 0, 0, 0, 0, notBefore.Location())
+ for offset := 0; offset <= maxSearchDays; offset++ {
+ dayStart := baseDate.AddDate(0, 0, offset)
+ if _, ok := allowed[dayStart.Weekday()]; !ok {
+ continue
+ }
+ runAt := dayStart.Add(policy.WindowStart + stableWindowOffset(policy.Seed, dayStart, policy.WindowEnd-policy.WindowStart))
+ if runAt.Before(notBefore) || runAt.Equal(notBefore) {
+ continue
+ }
+ if endAt != nil && runAt.After(endAt.In(time.Local)) {
+ return nil, nil
+ }
+ return &runAt, nil
+ }
+
+ return nil, nil
+}
+
+func normalizePolicyDays(days []time.Weekday) []time.Weekday {
+ seen := make(map[time.Weekday]struct{}, len(days))
+ result := make([]time.Weekday, 0, len(days))
+ for _, day := range days {
+ if day < time.Sunday || day > time.Saturday {
+ continue
+ }
+ if _, ok := seen[day]; ok {
+ continue
+ }
+ seen[day] = struct{}{}
+ result = append(result, day)
+ }
+ return result
+}
+
+func stableWindowOffset(seed int64, day time.Time, width time.Duration) time.Duration {
+ widthSeconds := int64(width / time.Second)
+ if widthSeconds <= 0 {
+ return 0
+ }
+
+ var buf [16]byte
+ binary.BigEndian.PutUint64(buf[:8], uint64(seed))
+ binary.BigEndian.PutUint64(buf[8:], uint64(day.Year()*10000+int(day.Month())*100+day.Day()))
+ sum := sha256.Sum256(buf[:])
+ value := binary.BigEndian.Uint64(sum[:8])
+ return time.Duration(int64(value%uint64(widthSeconds))) * time.Second
+}
diff --git a/server/internal/shared/schedule/policy_test.go b/server/internal/shared/schedule/policy_test.go
new file mode 100644
index 0000000..a605265
--- /dev/null
+++ b/server/internal/shared/schedule/policy_test.go
@@ -0,0 +1,164 @@
+package schedule
+
+import (
+ "testing"
+ "time"
+)
+
+func TestComputeNextRandomRunStaysInsideWindowAndIsStable(t *testing.T) {
+ loc := time.Local
+ now := time.Date(2026, 5, 18, 0, 30, 0, 0, loc) // Monday
+ policy := RandomPolicy{
+ Days: []time.Weekday{time.Monday},
+ TimeMode: TimeModeRandomWindow,
+ WindowStart: time.Hour,
+ WindowEnd: 5 * time.Hour,
+ Seed: 42,
+ }
+
+ first, err := ComputeNextRandomRun(policy, nil, nil, now)
+ if err != nil {
+ t.Fatalf("ComputeNextRandomRun returned error: %v", err)
+ }
+ second, err := ComputeNextRandomRun(policy, nil, nil, now)
+ if err != nil {
+ t.Fatalf("ComputeNextRandomRun returned error on second call: %v", err)
+ }
+ if first == nil || second == nil {
+ t.Fatal("expected next run")
+ }
+ if !first.Equal(*second) {
+ t.Fatalf("expected stable run for same seed/day, got %v and %v", first, second)
+ }
+ if first.Weekday() != time.Monday {
+ t.Fatalf("expected Monday run, got %s", first.Weekday())
+ }
+ clock := time.Duration(first.Hour())*time.Hour +
+ time.Duration(first.Minute())*time.Minute +
+ time.Duration(first.Second())*time.Second
+ if clock < policy.WindowStart || clock >= policy.WindowEnd {
+ t.Fatalf("run time %s outside window [%s, %s)", clock, policy.WindowStart, policy.WindowEnd)
+ }
+}
+
+func TestComputeNextRandomRunSkipsEqualNotBeforeBoundary(t *testing.T) {
+ loc := time.Local
+ day := time.Date(2026, 5, 18, 0, 0, 0, 0, loc) // Monday
+ offset := stableWindowOffset(7, day, time.Hour)
+ runAt := day.Add(time.Hour + offset)
+ policy := RandomPolicy{
+ Days: []time.Weekday{time.Monday},
+ TimeMode: TimeModeRandomWindow,
+ WindowStart: time.Hour,
+ WindowEnd: 2 * time.Hour,
+ Seed: 7,
+ }
+
+ next, err := ComputeNextRandomRun(policy, nil, nil, runAt)
+ if err != nil {
+ t.Fatalf("ComputeNextRandomRun returned error: %v", err)
+ }
+ if next == nil {
+ t.Fatal("expected next weekly run")
+ }
+ if !next.After(runAt) {
+ t.Fatalf("expected equal boundary to roll forward after %v, got %v", runAt, next)
+ }
+ if next.Weekday() != time.Monday {
+ t.Fatalf("expected next allowed weekday Monday, got %s", next.Weekday())
+ }
+ if next.Year() == runAt.Year() && next.YearDay() == runAt.YearDay() {
+ t.Fatalf("expected equal boundary to skip same day, got %v", next)
+ }
+}
+
+func TestComputeNextRandomRunHonorsStartAtAndEndAt(t *testing.T) {
+ loc := time.Local
+ now := time.Date(2026, 5, 18, 0, 0, 0, 0, loc) // Monday
+ startAt := time.Date(2026, 5, 20, 0, 0, 0, 0, loc) // Wednesday
+ endAt := time.Date(2026, 5, 20, 5, 0, 0, 0, loc)
+ policy := RandomPolicy{
+ Days: []time.Weekday{time.Monday, time.Wednesday},
+ TimeMode: TimeModeRandomWindow,
+ WindowStart: time.Hour,
+ WindowEnd: 2 * time.Hour,
+ Seed: 99,
+ }
+
+ next, err := ComputeNextRandomRun(policy, &startAt, &endAt, now)
+ if err != nil {
+ t.Fatalf("ComputeNextRandomRun returned error: %v", err)
+ }
+ if next == nil {
+ t.Fatal("expected run inside endAt")
+ }
+ if next.Weekday() != time.Wednesday {
+ t.Fatalf("expected Wednesday after startAt, got %s", next.Weekday())
+ }
+ if next.Before(startAt) || next.After(endAt) {
+ t.Fatalf("run %v outside [%v, %v]", next, startAt, endAt)
+ }
+
+ endBeforeNextWindow := time.Date(2026, 5, 20, 0, 30, 0, 0, loc)
+ next, err = ComputeNextRandomRun(policy, &startAt, &endBeforeNextWindow, now)
+ if err != nil {
+ t.Fatalf("ComputeNextRandomRun returned error with endAt: %v", err)
+ }
+ if next != nil {
+ t.Fatalf("expected nil when next run exceeds endAt, got %v", next)
+ }
+}
+
+func TestComputeNextRandomRunRejectsInvalidPolicy(t *testing.T) {
+ _, err := ComputeNextRandomRun(RandomPolicy{}, nil, nil, time.Now())
+ if err == nil {
+ t.Fatal("expected missing days error")
+ }
+
+ _, err = ComputeNextRandomRun(RandomPolicy{
+ Days: []time.Weekday{time.Monday},
+ TimeMode: TimeModeRandomWindow,
+ WindowStart: 2 * time.Hour,
+ WindowEnd: time.Hour,
+ }, nil, nil, time.Now())
+ if err == nil {
+ t.Fatal("expected invalid window error")
+ }
+}
+
+func TestComputeNextRandomRunReturnsNilAfterSearchHorizon(t *testing.T) {
+ loc := time.Local
+ now := time.Date(2026, 5, 18, 23, 0, 0, 0, loc) // Monday after the random window
+ policy := RandomPolicy{
+ Days: []time.Weekday{time.Monday},
+ TimeMode: TimeModeRandomWindow,
+ WindowStart: time.Hour,
+ WindowEnd: 2 * time.Hour,
+ Seed: 1,
+ }
+
+ next, err := computeNextRandomRun(policy, nil, nil, now, 0)
+ if err != nil {
+ t.Fatalf("computeNextRandomRun returned error: %v", err)
+ }
+ if next != nil {
+ t.Fatalf("expected nil after exhausting search horizon, got %v", next)
+ }
+}
+
+func TestStableWindowOffsetIsDeterministicAndBounded(t *testing.T) {
+ day := time.Date(2026, 5, 18, 0, 0, 0, 0, time.Local)
+ width := 4 * time.Hour
+
+ first := stableWindowOffset(123, day, width)
+ second := stableWindowOffset(123, day, width)
+ if first != second {
+ t.Fatalf("expected deterministic offset, got %s and %s", first, second)
+ }
+ if first < 0 || first >= width {
+ t.Fatalf("offset %s outside width %s", first, width)
+ }
+ if got := stableWindowOffset(123, day, 0); got != 0 {
+ t.Fatalf("expected zero width offset to be 0, got %s", got)
+ }
+}
diff --git a/server/internal/tenant/app/instant_task_service.go b/server/internal/tenant/app/instant_task_service.go
index c55d8a5..7eb2bfd 100644
--- a/server/internal/tenant/app/instant_task_service.go
+++ b/server/internal/tenant/app/instant_task_service.go
@@ -3,7 +3,6 @@ package app
import (
"context"
"encoding/json"
- "fmt"
"time"
"github.com/jackc/pgx/v5/pgxpool"
@@ -201,8 +200,8 @@ func (s *InstantTaskService) List(ctx context.Context, params InstantTaskListPar
return nil, response.ErrInternal(50010, "scan_failed", err.Error())
}
- item.ExecutionTime = stringPtrFromDBValue(executionAt)
- item.CreatedAt = fmt.Sprintf("%v", createdAt)
+ item.ExecutionTime = timeStringPtrFromDBValue(executionAt)
+ item.CreatedAt = timeStringFromDBValue(createdAt)
item.Articles = parseInstantTaskArticleLinks(articlesJSON)
autoPublishPlatforms, err := loadScheduleAutoPublishPlatforms(ctx, s.pool, actor.TenantID, inputParamsJSON)
if err != nil {
@@ -228,14 +227,6 @@ func (s *InstantTaskService) List(ctx context.Context, params InstantTaskListPar
}, nil
}
-func stringPtrFromDBValue(value interface{}) *string {
- if value == nil {
- return nil
- }
- text := fmt.Sprintf("%v", value)
- return &text
-}
-
func parseJSONMap(raw []byte) (map[string]interface{}, error) {
if len(raw) == 0 {
return map[string]interface{}{}, nil
diff --git a/server/internal/tenant/app/kol_generation_service.go b/server/internal/tenant/app/kol_generation_service.go
index 485a120..5d7145b 100644
--- a/server/internal/tenant/app/kol_generation_service.go
+++ b/server/internal/tenant/app/kol_generation_service.go
@@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ "strings"
"time"
"github.com/jackc/pgx/v5/pgxpool"
@@ -36,6 +37,25 @@ type KolGenerationSubmitResponse struct {
UsageLogID int64 `json:"usage_log_id"`
}
+type ScheduleKolGenerationInput struct {
+ ScheduleTaskID int64
+ OperatorID *int64
+ TenantID int64
+ WorkspaceID int64
+ BrandID *int64
+ SubscriptionPromptID int64
+ Name string
+ Variables map[string]any
+ EnableWebSearch bool
+ KnowledgeGroupIDs []int64
+ AutoPublish bool
+ PublishAccountIDs []string
+ CoverAssetURL *string
+ CoverImageAssetID *int64
+ GenerateCount int
+ ScheduledFor time.Time
+}
+
type KolSubscriptionPromptSchemaResponse struct {
SubscriptionPromptID int64 `json:"subscription_prompt_id"`
PackageID int64 `json:"package_id"`
@@ -138,13 +158,58 @@ func (s *KolGenerationService) Submit(ctx context.Context, actor auth.Actor, sub
if err != nil {
return nil, err
}
+ return s.enqueue(ctx, kolGenerationEnqueueInput{
+ Actor: actor,
+ BrandID: brandID,
+ SubscriptionPromptID: subPromptID,
+ Variables: req.Variables,
+ EnableWebSearch: req.EnableWebSearch,
+ KnowledgeGroupIDs: req.KnowledgeGroupIDs,
+ })
+}
- row, prompt, err := s.loadAuthorizedPrompt(ctx, actor, subPromptID)
+func (s *KolGenerationService) EnqueueScheduledGeneration(ctx context.Context, input ScheduleKolGenerationInput) (*KolGenerationSubmitResponse, error) {
+ if input.BrandID == nil || *input.BrandID <= 0 {
+ return nil, response.ErrBadRequest(40033, "brand_context_required", "current brand is required")
+ }
+ operatorID := int64(0)
+ if input.OperatorID != nil {
+ operatorID = *input.OperatorID
+ }
+ actor := auth.Actor{
+ TenantID: input.TenantID,
+ UserID: operatorID,
+ PrimaryWorkspaceID: input.WorkspaceID,
+ }
+ return s.enqueue(ctx, kolGenerationEnqueueInput{
+ Actor: actor,
+ BrandID: *input.BrandID,
+ SubscriptionPromptID: input.SubscriptionPromptID,
+ Variables: input.Variables,
+ EnableWebSearch: input.EnableWebSearch,
+ KnowledgeGroupIDs: input.KnowledgeGroupIDs,
+ Schedule: &input,
+ })
+}
+
+type kolGenerationEnqueueInput struct {
+ Actor auth.Actor
+ BrandID int64
+ SubscriptionPromptID int64
+ Variables map[string]any
+ EnableWebSearch bool
+ KnowledgeGroupIDs []int64
+ Schedule *ScheduleKolGenerationInput
+}
+
+func (s *KolGenerationService) enqueue(ctx context.Context, input kolGenerationEnqueueInput) (*KolGenerationSubmitResponse, error) {
+ actor := input.Actor
+ row, prompt, err := s.loadAuthorizedPrompt(ctx, actor, input.SubscriptionPromptID)
if err != nil {
return nil, err
}
- variables := req.Variables
+ variables := input.Variables
if variables == nil {
variables = map[string]any{}
}
@@ -172,16 +237,16 @@ func (s *KolGenerationService) Submit(ctx context.Context, actor auth.Actor, sub
return nil, response.ErrBadRequest(40071, "kol_generation_variables_invalid", err.Error())
}
renderedPromptHash := HashPrompt(renderedPrompt)
- if req.EnableWebSearch && !kolCardConfigAllowsWebSearch(cardConfig) {
+ if input.EnableWebSearch && !kolCardConfigAllowsWebSearch(cardConfig) {
return nil, response.ErrBadRequest(40073, "kol_generation_web_search_disabled", "web search is disabled for this prompt")
}
- knowledgeGroupIDs := normalizeInt64IDs(req.KnowledgeGroupIDs)
+ knowledgeGroupIDs := normalizeInt64IDs(input.KnowledgeGroupIDs)
if len(knowledgeGroupIDs) > 0 && !kolCardConfigAllowsUserKnowledge(cardConfig) {
return nil, response.ErrBadRequest(40074, "kol_generation_knowledge_disabled", "knowledge selection is disabled for this prompt")
}
- enableWebSearch := kolCardConfigAllowsWebSearch(cardConfig) && req.EnableWebSearch
+ enableWebSearch := kolCardConfigAllowsWebSearch(cardConfig) && input.EnableWebSearch
promptRevisionNo := 1
if row.PublishedRevisionNo != nil && *row.PublishedRevisionNo > 0 {
promptRevisionNo = *row.PublishedRevisionNo
@@ -192,26 +257,30 @@ func (s *KolGenerationService) Submit(ctx context.Context, actor auth.Actor, sub
return nil, response.ErrForbidden(40301, "quota_insufficient", "generation quota insufficient, please upgrade your plan")
}
- inputJSON, err := json.Marshal(kolGenerationTaskInput{
- SubscriptionPromptID: subPromptID,
- SubscriptionID: row.SubscriptionID,
- PackageID: row.PackageID,
- PromptID: row.PromptID,
- PromptName: row.PromptName,
- PlatformHint: derefString(row.PlatformHint),
- PromptRevisionNo: promptRevisionNo,
- Variables: variables,
- EnableWebSearch: enableWebSearch,
- KnowledgeGroupIDs: knowledgeGroupIDs,
- SchemaSnapshot: json.RawMessage(prompt.SchemaJSON),
- CardConfigSnapshot: json.RawMessage(prompt.CardConfigJSON),
- RenderedPromptHash: renderedPromptHash,
- RenderedPrompt: renderedPrompt,
- })
+ taskInput := map[string]any{
+ "subscription_prompt_id": input.SubscriptionPromptID,
+ "subscription_id": row.SubscriptionID,
+ "package_id": row.PackageID,
+ "prompt_id": row.PromptID,
+ "prompt_name": row.PromptName,
+ "platform_hint": derefString(row.PlatformHint),
+ "prompt_revision_no": promptRevisionNo,
+ "variables": variables,
+ "enable_web_search": enableWebSearch,
+ "knowledge_group_ids": knowledgeGroupIDs,
+ "schema_snapshot": json.RawMessage(prompt.SchemaJSON),
+ "card_config_snapshot": json.RawMessage(prompt.CardConfigJSON),
+ "rendered_prompt_hash": renderedPromptHash,
+ "rendered_prompt": renderedPrompt,
+ }
+ if input.Schedule != nil {
+ attachKolScheduleTaskInput(taskInput, *input.Schedule)
+ }
+
+ inputJSON, err := json.Marshal(taskInput)
if err != nil {
return nil, response.ErrInternal(50099, "kol_generation_input_encode_failed", err.Error())
}
-
variablesJSON, err := json.Marshal(variables)
if err != nil {
return nil, response.ErrBadRequest(40072, "kol_generation_variables_invalid", "variables must be valid json")
@@ -258,7 +327,7 @@ func (s *KolGenerationService) Submit(ctx context.Context, actor auth.Actor, sub
articleID, err := articleTx.CreateArticle(ctx, repository.CreateArticleInput{
TenantID: actor.TenantID,
- BrandID: brandID,
+ BrandID: input.BrandID,
SourceType: kolGenerationTaskType,
KolPromptID: &row.PromptID,
})
@@ -268,6 +337,22 @@ func (s *KolGenerationService) Submit(ctx context.Context, actor auth.Actor, sub
if err := articleTx.UpdateArticleGenerateStatus(ctx, articleID, actor.TenantID, "generating"); err != nil {
return nil, response.ErrInternal(50112, "kol_generation_article_state_update_failed", err.Error())
}
+ if input.Schedule != nil {
+ coverAssetURL, coverImageAssetID := scheduleCoverState(input.Schedule.CoverAssetURL, input.Schedule.CoverImageAssetID)
+ title := strings.TrimSpace(input.Schedule.Name)
+ if title == "" {
+ title = row.PromptName
+ }
+ wizardStateJSON, _ := mergeArticleWizardState(nil, title, coverAssetURL, coverImageAssetID)
+ if _, err := tx.Exec(ctx, `
+ UPDATE articles
+ SET wizard_state_json = $1,
+ updated_at = NOW()
+ WHERE id = $2 AND tenant_id = $3
+ `, wizardStateJSON, articleID, actor.TenantID); err != nil {
+ return nil, response.ErrInternal(50112, "kol_generation_article_state_update_failed", err.Error())
+ }
+ }
if err := quotaTx.UpdateQuotaReservationResource(ctx, reservationID, actor.TenantID, articleID); err != nil {
return nil, response.ErrInternal(50104, "kol_generation_reservation_update_failed", err.Error())
@@ -357,6 +442,49 @@ func (s *KolGenerationService) Submit(ctx context.Context, actor auth.Actor, sub
}, nil
}
+func attachKolScheduleTaskInput(input map[string]any, schedule ScheduleKolGenerationInput) {
+ input["generation_mode"] = "schedule"
+ input["schedule_task_id"] = schedule.ScheduleTaskID
+ input["target_type"] = ScheduleTargetKolSubscriptionPrompt
+ input["task_name"] = strings.TrimSpace(schedule.Name)
+ if schedule.WorkspaceID > 0 {
+ input["workspace_id"] = schedule.WorkspaceID
+ }
+ if schedule.GenerateCount > 0 {
+ input["generate_count"] = schedule.GenerateCount
+ }
+ if schedule.BrandID != nil && *schedule.BrandID > 0 {
+ input["brand_id"] = *schedule.BrandID
+ }
+ if !schedule.ScheduledFor.IsZero() {
+ input["scheduled_for"] = schedule.ScheduledFor.UTC().Format(time.RFC3339)
+ }
+ if schedule.AutoPublish {
+ input["schedule_auto_publish"] = true
+ input["schedule_publish_account_ids"] = schedule.PublishAccountIDs
+ if schedule.CoverAssetURL != nil {
+ input["schedule_cover_asset_url"] = strings.TrimSpace(*schedule.CoverAssetURL)
+ }
+ if schedule.CoverImageAssetID != nil {
+ input["schedule_cover_image_asset_id"] = *schedule.CoverImageAssetID
+ }
+ }
+}
+
+func scheduleCoverState(coverURL *string, coverImageAssetID *int64) (*string, NullableInt64Input) {
+ var normalizedURL *string
+ var normalizedID NullableInt64Input
+ if coverURL != nil {
+ trimmed := strings.TrimSpace(*coverURL)
+ if trimmed != "" {
+ normalizedURL = &trimmed
+ normalizedID.Set = true
+ normalizedID.Value = coverImageAssetID
+ }
+ }
+ return normalizedURL, normalizedID
+}
+
func (s *KolGenerationService) loadAuthorizedPrompt(ctx context.Context, actor auth.Actor, subPromptID int64) (*repository.KolSubscriptionPrompt, *repository.KolPrompt, error) {
row, err := s.subRepo.GetSubscriptionPromptByID(ctx, actor.TenantID, subPromptID)
if err != nil {
diff --git a/server/internal/tenant/app/schedule_task_policy.go b/server/internal/tenant/app/schedule_task_policy.go
new file mode 100644
index 0000000..b404810
--- /dev/null
+++ b/server/internal/tenant/app/schedule_task_policy.go
@@ -0,0 +1,327 @@
+package app
+
+import (
+ "encoding/json"
+ "fmt"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/geo-platform/tenant-api/internal/shared/response"
+ sharedschedule "github.com/geo-platform/tenant-api/internal/shared/schedule"
+)
+
+const (
+ ScheduleTargetPromptRule = "prompt_rule"
+ ScheduleTargetKolSubscriptionPrompt = "kol_subscription_prompt"
+
+ scheduleKindDaily = "daily"
+ scheduleKindWeekly = "weekly"
+
+ defaultRandomWindowStart = "01:00:00"
+ defaultRandomWindowEnd = "05:00:00"
+)
+
+var scheduleDayOrder = []string{"mon", "tue", "wed", "thu", "fri", "sat", "sun"}
+
+type normalizedSchedulePlan struct {
+ CronExpr string
+ ScheduleKind string
+ ScheduleDays []string
+ ScheduleTimeMode string
+ RandomWindowStart string
+ RandomWindowEnd string
+ windowStart time.Duration
+ windowEnd time.Duration
+}
+
+func normalizeScheduleTarget(req ScheduleTaskRequest) (string, *int64, *int64, error) {
+ targetType := strings.TrimSpace(req.TargetType)
+ if targetType == "" {
+ switch {
+ case req.SubscriptionPromptID != nil && *req.SubscriptionPromptID > 0:
+ targetType = ScheduleTargetKolSubscriptionPrompt
+ default:
+ targetType = ScheduleTargetPromptRule
+ }
+ }
+
+ switch targetType {
+ case ScheduleTargetPromptRule:
+ if req.PromptRuleID == nil || *req.PromptRuleID <= 0 {
+ return "", nil, nil, response.ErrBadRequest(40001, "invalid_rule", "prompt_rule_id is required")
+ }
+ promptRuleID := *req.PromptRuleID
+ return targetType, &promptRuleID, nil, nil
+ case ScheduleTargetKolSubscriptionPrompt:
+ if req.SubscriptionPromptID == nil || *req.SubscriptionPromptID <= 0 {
+ return "", nil, nil, response.ErrBadRequest(40075, "invalid_subscription_prompt", "subscription_prompt_id is required")
+ }
+ subscriptionPromptID := *req.SubscriptionPromptID
+ return targetType, nil, &subscriptionPromptID, nil
+ default:
+ return "", nil, nil, response.ErrBadRequest(40076, "invalid_schedule_target", "target_type must be prompt_rule or kol_subscription_prompt")
+ }
+}
+
+func normalizeSchedulePlan(req ScheduleTaskRequest) (normalizedSchedulePlan, error) {
+ kind := strings.TrimSpace(req.ScheduleKind)
+ if kind == "" {
+ kind = scheduleKindDaily
+ }
+ if kind != scheduleKindDaily && kind != scheduleKindWeekly {
+ return normalizedSchedulePlan{}, response.ErrBadRequest(40077, "invalid_schedule_kind", "schedule_kind must be daily or weekly")
+ }
+
+ dayKeys, err := normalizeScheduleDayKeys(kind, req.ScheduleDays)
+ if err != nil {
+ return normalizedSchedulePlan{}, err
+ }
+
+ timeMode := strings.TrimSpace(req.ScheduleTimeMode)
+ if timeMode == "" {
+ if strings.TrimSpace(req.CronExpr) != "" {
+ timeMode = sharedschedule.TimeModeFixed
+ } else {
+ timeMode = sharedschedule.TimeModeRandomWindow
+ }
+ }
+ if timeMode != sharedschedule.TimeModeFixed && timeMode != sharedschedule.TimeModeRandomWindow {
+ return normalizedSchedulePlan{}, response.ErrBadRequest(40078, "invalid_schedule_time_mode", "schedule_time_mode must be fixed or random_window")
+ }
+
+ windowStart := strings.TrimSpace(req.RandomWindowStart)
+ if windowStart == "" {
+ windowStart = defaultRandomWindowStart
+ }
+ windowEnd := strings.TrimSpace(req.RandomWindowEnd)
+ if windowEnd == "" {
+ windowEnd = defaultRandomWindowEnd
+ }
+ windowStartDuration, err := sharedschedule.ParseClockDuration(windowStart)
+ if err != nil {
+ return normalizedSchedulePlan{}, response.ErrBadRequest(40079, "invalid_random_window_start", err.Error())
+ }
+ windowEndDuration, err := sharedschedule.ParseClockDuration(windowEnd)
+ if err != nil {
+ return normalizedSchedulePlan{}, response.ErrBadRequest(40080, "invalid_random_window_end", err.Error())
+ }
+ if windowEndDuration <= windowStartDuration {
+ return normalizedSchedulePlan{}, response.ErrBadRequest(40081, "invalid_random_window", "random_window_end must be after random_window_start")
+ }
+
+ cronExpr := strings.TrimSpace(req.CronExpr)
+ if timeMode == sharedschedule.TimeModeFixed {
+ if cronExpr == "" {
+ cronExpr = buildScheduleCronFromDays(dayKeys, windowStartDuration)
+ }
+ if err := validateScheduleCronExpr(cronExpr); err != nil {
+ return normalizedSchedulePlan{}, err
+ }
+ } else if cronExpr == "" {
+ cronExpr = buildScheduleCronFromDays(dayKeys, windowStartDuration)
+ }
+
+ return normalizedSchedulePlan{
+ CronExpr: cronExpr,
+ ScheduleKind: kind,
+ ScheduleDays: dayKeys,
+ ScheduleTimeMode: timeMode,
+ RandomWindowStart: sharedschedule.FormatClockDuration(windowStartDuration),
+ RandomWindowEnd: sharedschedule.FormatClockDuration(windowEndDuration),
+ windowStart: windowStartDuration,
+ windowEnd: windowEndDuration,
+ }, nil
+}
+
+func normalizeScheduleDayKeys(kind string, values []string) ([]string, error) {
+ if kind == scheduleKindDaily {
+ return append([]string(nil), scheduleDayOrder...), nil
+ }
+ if len(values) == 0 {
+ return nil, response.ErrBadRequest(40082, "schedule_days_required", "schedule_days is required for weekly schedule")
+ }
+
+ seen := make(map[string]struct{}, len(values))
+ for _, value := range values {
+ day, ok := sharedschedule.ParseDayKey(value)
+ if !ok {
+ return nil, response.ErrBadRequest(40083, "invalid_schedule_day", fmt.Sprintf("invalid schedule day: %s", value))
+ }
+ key := sharedschedule.DayKey(day)
+ if key == "" {
+ return nil, response.ErrBadRequest(40083, "invalid_schedule_day", fmt.Sprintf("invalid schedule day: %s", value))
+ }
+ seen[key] = struct{}{}
+ }
+ if len(seen) == 0 {
+ return nil, response.ErrBadRequest(40082, "schedule_days_required", "schedule_days is required")
+ }
+ if len(seen) > 7 {
+ return nil, response.ErrBadRequest(40084, "too_many_schedule_days", "schedule_days can contain at most 7 days")
+ }
+
+ result := make([]string, 0, len(seen))
+ for _, key := range scheduleDayOrder {
+ if _, ok := seen[key]; ok {
+ result = append(result, key)
+ }
+ }
+ return result, nil
+}
+
+func buildScheduleCronFromDays(dayKeys []string, at time.Duration) string {
+ hour := int(at / time.Hour)
+ minute := int((at - time.Duration(hour)*time.Hour) / time.Minute)
+ if len(dayKeys) == 0 || len(dayKeys) == 7 {
+ return fmt.Sprintf("%d %d * * *", minute, hour)
+ }
+ return fmt.Sprintf("%d %d * * %s", minute, hour, strings.Join(dayKeys, ","))
+}
+
+func scheduleDayKeysToWeekdays(dayKeys []string) []time.Weekday {
+ result := make([]time.Weekday, 0, len(dayKeys))
+ for _, key := range dayKeys {
+ day, ok := sharedschedule.ParseDayKey(key)
+ if ok {
+ result = append(result, day)
+ }
+ }
+ return result
+}
+
+func decodeScheduleDayKeys(raw []byte) []string {
+ if len(raw) == 0 {
+ return append([]string(nil), scheduleDayOrder...)
+ }
+ var values []string
+ if err := json.Unmarshal(raw, &values); err != nil {
+ return append([]string(nil), scheduleDayOrder...)
+ }
+ kind := scheduleKindWeekly
+ if len(values) == 7 {
+ kind = scheduleKindDaily
+ }
+ normalized, err := normalizeScheduleDayKeys(kind, values)
+ if err != nil || len(normalized) == 0 {
+ return append([]string(nil), scheduleDayOrder...)
+ }
+ return normalized
+}
+
+func DecodeScheduleDayKeys(raw []byte) []string {
+ return decodeScheduleDayKeys(raw)
+}
+
+func encodeScheduleDayKeys(dayKeys []string) []byte {
+ if len(dayKeys) == 0 {
+ dayKeys = scheduleDayOrder
+ }
+ raw, err := json.Marshal(dayKeys)
+ if err != nil {
+ return []byte(`["mon","tue","wed","thu","fri","sat","sun"]`)
+ }
+ return raw
+}
+
+func decodeScheduleGenerationInput(raw []byte) map[string]interface{} {
+ if len(raw) == 0 {
+ return map[string]interface{}{}
+ }
+ var value map[string]interface{}
+ if err := json.Unmarshal(raw, &value); err != nil || value == nil {
+ return map[string]interface{}{}
+ }
+ return value
+}
+
+func encodeScheduleGenerationInput(value map[string]interface{}) []byte {
+ if len(value) == 0 {
+ return []byte(`{}`)
+ }
+ raw, err := json.Marshal(value)
+ if err != nil {
+ return []byte(`{}`)
+ }
+ return raw
+}
+
+func scheduleGenerationInputCacheKey(value map[string]interface{}) string {
+ if len(value) == 0 {
+ return "{}"
+ }
+ keys := make([]string, 0, len(value))
+ for key := range value {
+ keys = append(keys, key)
+ }
+ sort.Strings(keys)
+ ordered := make(map[string]interface{}, len(keys))
+ for _, key := range keys {
+ ordered[key] = value[key]
+ }
+ return digestCacheKey(ordered)
+}
+
+func computeScheduleNextRun(
+ status string,
+ taskID int64,
+ cronExpr string,
+ dayKeys []string,
+ timeMode string,
+ windowStart string,
+ windowEnd string,
+ startAt *time.Time,
+ endAt *time.Time,
+ now time.Time,
+) (*time.Time, error) {
+ if status != "enabled" {
+ return nil, nil
+ }
+ if strings.TrimSpace(timeMode) != sharedschedule.TimeModeRandomWindow {
+ return sharedschedule.ComputeNextRun(cronExpr, startAt, endAt, now)
+ }
+
+ windowStartDuration, err := sharedschedule.ParseClockDuration(windowStart)
+ if err != nil {
+ return nil, err
+ }
+ windowEndDuration, err := sharedschedule.ParseClockDuration(windowEnd)
+ if err != nil {
+ return nil, err
+ }
+ return sharedschedule.ComputeNextRandomRun(sharedschedule.RandomPolicy{
+ Days: scheduleDayKeysToWeekdays(dayKeys),
+ TimeMode: sharedschedule.TimeModeRandomWindow,
+ WindowStart: windowStartDuration,
+ WindowEnd: windowEndDuration,
+ Seed: taskID,
+ }, startAt, endAt, now)
+}
+
+func ComputeScheduleTaskNextRun(
+ status string,
+ taskID int64,
+ cronExpr string,
+ dayKeys []string,
+ timeMode string,
+ windowStart string,
+ windowEnd string,
+ startAt *time.Time,
+ endAt *time.Time,
+ now time.Time,
+) (*time.Time, error) {
+ return computeScheduleNextRun(status, taskID, cronExpr, dayKeys, timeMode, windowStart, windowEnd, startAt, endAt, now)
+}
+
+func normalizeKolScheduleGenerationInput(input map[string]interface{}) map[string]interface{} {
+ if input == nil {
+ return map[string]interface{}{}
+ }
+ result := clonePromptRuleExtraParams(input)
+ if variables, ok := result["variables"].(map[string]interface{}); ok && variables != nil {
+ result["variables"] = variables
+ } else {
+ result["variables"] = map[string]interface{}{}
+ }
+ return result
+}
diff --git a/server/internal/tenant/app/schedule_task_policy_test.go b/server/internal/tenant/app/schedule_task_policy_test.go
new file mode 100644
index 0000000..b817780
--- /dev/null
+++ b/server/internal/tenant/app/schedule_task_policy_test.go
@@ -0,0 +1,62 @@
+package app
+
+import (
+ "testing"
+
+ sharedschedule "github.com/geo-platform/tenant-api/internal/shared/schedule"
+)
+
+func TestNormalizeSchedulePlanFixedTimeBuildsCronFromSelectedDays(t *testing.T) {
+ plan, err := normalizeSchedulePlan(ScheduleTaskRequest{
+ ScheduleKind: scheduleKindWeekly,
+ ScheduleDays: []string{"mon", "wed", "fri"},
+ ScheduleTimeMode: sharedschedule.TimeModeFixed,
+ CronExpr: "17 3 * * mon,wed,fri",
+ })
+ if err != nil {
+ t.Fatalf("normalizeSchedulePlan returned error: %v", err)
+ }
+ if plan.ScheduleKind != scheduleKindWeekly {
+ t.Fatalf("expected weekly schedule, got %q", plan.ScheduleKind)
+ }
+ if plan.ScheduleTimeMode != sharedschedule.TimeModeFixed {
+ t.Fatalf("expected fixed mode, got %q", plan.ScheduleTimeMode)
+ }
+ if plan.CronExpr != "17 3 * * mon,wed,fri" {
+ t.Fatalf("unexpected cron expression: %q", plan.CronExpr)
+ }
+ if plan.RandomWindowStart != defaultRandomWindowStart || plan.RandomWindowEnd != defaultRandomWindowEnd {
+ t.Fatalf("unexpected compatibility window: %s-%s", plan.RandomWindowStart, plan.RandomWindowEnd)
+ }
+}
+
+func TestNormalizeSchedulePlanRandomWindowAllowsEditableWindow(t *testing.T) {
+ plan, err := normalizeSchedulePlan(ScheduleTaskRequest{
+ ScheduleKind: scheduleKindWeekly,
+ ScheduleDays: []string{"mon", "wed", "fri"},
+ ScheduleTimeMode: sharedschedule.TimeModeRandomWindow,
+ RandomWindowStart: "02:00:00",
+ RandomWindowEnd: "04:30:00",
+ })
+ if err != nil {
+ t.Fatalf("normalizeSchedulePlan returned error: %v", err)
+ }
+ if plan.ScheduleKind != scheduleKindWeekly {
+ t.Fatalf("expected weekly schedule, got %q", plan.ScheduleKind)
+ }
+ if plan.ScheduleTimeMode != sharedschedule.TimeModeRandomWindow {
+ t.Fatalf("expected random window mode, got %q", plan.ScheduleTimeMode)
+ }
+ if plan.RandomWindowStart != "02:00:00" || plan.RandomWindowEnd != "04:30:00" {
+ t.Fatalf("unexpected random window: %s-%s", plan.RandomWindowStart, plan.RandomWindowEnd)
+ }
+
+ _, err = normalizeSchedulePlan(ScheduleTaskRequest{
+ ScheduleTimeMode: sharedschedule.TimeModeRandomWindow,
+ RandomWindowStart: "05:00:00",
+ RandomWindowEnd: "01:00:00",
+ })
+ if err == nil {
+ t.Fatal("expected inverted random window to be rejected")
+ }
+}
diff --git a/server/internal/tenant/app/schedule_task_service.go b/server/internal/tenant/app/schedule_task_service.go
index af9679b..e830111 100644
--- a/server/internal/tenant/app/schedule_task_service.go
+++ b/server/internal/tenant/app/schedule_task_service.go
@@ -20,6 +20,7 @@ import (
"github.com/geo-platform/tenant-api/internal/shared/middleware"
"github.com/geo-platform/tenant-api/internal/shared/response"
sharedschedule "github.com/geo-platform/tenant-api/internal/shared/schedule"
+ "github.com/geo-platform/tenant-api/internal/tenant/repository"
)
type ScheduleTaskService struct {
@@ -39,55 +40,80 @@ func (s *ScheduleTaskService) WithCache(c sharedcache.Cache) *ScheduleTaskServic
}
type ScheduleTaskRequest struct {
- PromptRuleID int64 `json:"prompt_rule_id" binding:"required"`
- BrandID *int64 `json:"brand_id"`
- Name string `json:"name" binding:"required"`
- CronExpr string `json:"cron_expr" binding:"required"`
- AutoPublish *bool `json:"auto_publish"`
- PublishAccountIDs []string `json:"publish_account_ids"`
- CoverAssetURL *string `json:"cover_asset_url"`
- CoverImageAssetID *int64 `json:"cover_image_asset_id"`
- EnableWebSearch *bool `json:"enable_web_search"`
- GenerateCount *int `json:"generate_count"`
- StartAt *string `json:"start_at"`
- EndAt *string `json:"end_at"`
+ TargetType string `json:"target_type"`
+ PromptRuleID *int64 `json:"prompt_rule_id"`
+ SubscriptionPromptID *int64 `json:"subscription_prompt_id"`
+ BrandID *int64 `json:"brand_id"`
+ Name string `json:"name" binding:"required"`
+ CronExpr string `json:"cron_expr"`
+ ScheduleKind string `json:"schedule_kind"`
+ ScheduleDays []string `json:"schedule_days"`
+ ScheduleTimeMode string `json:"schedule_time_mode"`
+ RandomWindowStart string `json:"random_window_start"`
+ RandomWindowEnd string `json:"random_window_end"`
+ GenerationInput map[string]interface{} `json:"generation_input"`
+ AutoPublish *bool `json:"auto_publish"`
+ PublishAccountIDs []string `json:"publish_account_ids"`
+ CoverAssetURL *string `json:"cover_asset_url"`
+ CoverImageAssetID *int64 `json:"cover_image_asset_id"`
+ EnableWebSearch *bool `json:"enable_web_search"`
+ GenerateCount *int `json:"generate_count"`
+ StartAt *string `json:"start_at"`
+ EndAt *string `json:"end_at"`
}
type ScheduleTaskResponse struct {
- ID int64 `json:"id"`
- WorkspaceID *int64 `json:"workspace_id"`
- PromptRuleID int64 `json:"prompt_rule_id"`
- PromptRuleName *string `json:"prompt_rule_name"`
- BrandID *int64 `json:"brand_id"`
- BrandName *string `json:"brand_name"`
- Name string `json:"name"`
- CronExpr string `json:"cron_expr"`
- AutoPublish bool `json:"auto_publish"`
- PublishAccountIDs []string `json:"publish_account_ids"`
- AutoPublishPlatforms []string `json:"auto_publish_platforms"`
- CoverAssetURL *string `json:"cover_asset_url"`
- CoverImageAssetID *int64 `json:"cover_image_asset_id"`
- EnableWebSearch bool `json:"enable_web_search"`
- GenerateCount int `json:"generate_count"`
- StartAt *string `json:"start_at"`
- EndAt *string `json:"end_at"`
- NextRunAt *string `json:"next_run_at"`
- Status string `json:"status"`
- GenerationStatus *string `json:"generation_status"`
- ExecutionTime *string `json:"execution_time"`
- GeneratedArticles []InstantTaskArticleLink `json:"generated_articles"`
- CreatedAt string `json:"created_at"`
- UpdatedAt string `json:"updated_at"`
+ ID int64 `json:"id"`
+ WorkspaceID *int64 `json:"workspace_id"`
+ TargetType string `json:"target_type"`
+ PromptRuleID *int64 `json:"prompt_rule_id"`
+ PromptRuleName *string `json:"prompt_rule_name"`
+ SubscriptionPromptID *int64 `json:"subscription_prompt_id"`
+ SubscriptionPromptName *string `json:"subscription_prompt_name"`
+ SubscriptionPackageName *string `json:"subscription_package_name"`
+ SubscriptionKolName *string `json:"subscription_kol_name"`
+ BrandID *int64 `json:"brand_id"`
+ BrandName *string `json:"brand_name"`
+ Name string `json:"name"`
+ CronExpr string `json:"cron_expr"`
+ ScheduleKind string `json:"schedule_kind"`
+ ScheduleDays []string `json:"schedule_days"`
+ ScheduleTimeMode string `json:"schedule_time_mode"`
+ RandomWindowStart string `json:"random_window_start"`
+ RandomWindowEnd string `json:"random_window_end"`
+ GenerationInput map[string]interface{} `json:"generation_input"`
+ AutoPublish bool `json:"auto_publish"`
+ PublishAccountIDs []string `json:"publish_account_ids"`
+ AutoPublishPlatforms []string `json:"auto_publish_platforms"`
+ CoverAssetURL *string `json:"cover_asset_url"`
+ CoverImageAssetID *int64 `json:"cover_image_asset_id"`
+ EnableWebSearch bool `json:"enable_web_search"`
+ GenerateCount int `json:"generate_count"`
+ StartAt *string `json:"start_at"`
+ EndAt *string `json:"end_at"`
+ NextRunAt *string `json:"next_run_at"`
+ Status string `json:"status"`
+ LastRunAt *string `json:"last_run_at"`
+ LastDispatchStatus *string `json:"last_dispatch_status"`
+ LastDispatchError *string `json:"last_dispatch_error"`
+ ConsecutiveFailures int `json:"consecutive_failures"`
+ GenerationStatus *string `json:"generation_status"`
+ ExecutionTime *string `json:"execution_time"`
+ GeneratedArticles []InstantTaskArticleLink `json:"generated_articles"`
+ CreatedAt string `json:"created_at"`
+ UpdatedAt string `json:"updated_at"`
}
type ScheduleTaskListParams struct {
- PromptRuleID *int64 `json:"prompt_rule_id"`
- Status *string `json:"status"`
- Keyword *string `json:"keyword"`
- CreatedFrom *time.Time
- CreatedTo *time.Time
- Page int `json:"page"`
- PageSize int `json:"page_size"`
+ PromptRuleID *int64 `json:"prompt_rule_id"`
+ SubscriptionPromptID *int64 `json:"subscription_prompt_id"`
+ TargetType *string `json:"target_type"`
+ Status *string `json:"status"`
+ Keyword *string `json:"keyword"`
+ CreatedFrom *time.Time
+ CreatedTo *time.Time
+ Page int `json:"page"`
+ PageSize int `json:"page_size"`
}
type ScheduleTaskListResponse struct {
@@ -136,12 +162,26 @@ func (s *ScheduleTaskService) Create(ctx context.Context, req ScheduleTaskReques
}
req.BrandID = &brandID
- if err := validateScheduleCronExpr(req.CronExpr); err != nil {
+ targetType, promptRuleID, subscriptionPromptID, err := normalizeScheduleTarget(req)
+ if err != nil {
return nil, err
}
- if err := s.ensurePromptRuleExists(ctx, actor.TenantID, brandID, req.PromptRuleID); err != nil {
+ plan, err := normalizeSchedulePlan(req)
+ if err != nil {
return nil, err
}
+ generationInput := clonePromptRuleExtraParams(req.GenerationInput)
+ if targetType == ScheduleTargetPromptRule {
+ if err := s.ensurePromptRuleExists(ctx, actor.TenantID, brandID, *promptRuleID); err != nil {
+ return nil, err
+ }
+ generationInput = map[string]interface{}{}
+ } else {
+ generationInput, err = s.validateKolScheduleGenerationInput(ctx, actor.TenantID, *subscriptionPromptID, generationInput)
+ if err != nil {
+ return nil, err
+ }
+ }
generateCount, err := normalizeScheduleGenerateCount(req.GenerateCount)
if err != nil {
return nil, err
@@ -165,19 +205,31 @@ func (s *ScheduleTaskService) Create(ctx context.Context, req ScheduleTaskReques
var status string
err = tx.QueryRow(ctx, `
INSERT INTO schedule_tasks (
- tenant_id, workspace_id, operator_id, prompt_rule_id, brand_id, name, cron_expr,
+ tenant_id, workspace_id, operator_id, target_type, prompt_rule_id, subscription_prompt_id,
+ brand_id, name, cron_expr, schedule_kind, schedule_days, schedule_time_mode,
+ random_window_start, random_window_end, generation_input_json,
enable_web_search, generate_count, auto_publish, publish_account_ids, cover_asset_url,
cover_image_asset_id, start_at, end_at
)
- VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::jsonb, $12, $13, $14::timestamptz, $15::timestamptz)
+ VALUES (
+ $1, $2, $3, $4, $5, $6,
+ $7, $8, $9, $10, $11::jsonb, $12,
+ $13::time, $14::time, $15::jsonb,
+ $16, $17, $18, $19::jsonb, $20,
+ $21, $22::timestamptz, $23::timestamptz
+ )
RETURNING id, created_at, start_at, end_at, status
- `, actor.TenantID, actor.PrimaryWorkspaceID, actor.UserID, req.PromptRuleID, req.BrandID, req.Name, req.CronExpr, enableWebSearch, generateCount, publishConfig.AutoPublish, publishConfig.AccountIDsJSON, publishConfig.CoverAssetURL, publishConfig.CoverImageAssetID, req.StartAt, req.EndAt).
+ `, actor.TenantID, actor.PrimaryWorkspaceID, actor.UserID, targetType, promptRuleID, subscriptionPromptID,
+ req.BrandID, req.Name, plan.CronExpr, plan.ScheduleKind, encodeScheduleDayKeys(plan.ScheduleDays), plan.ScheduleTimeMode,
+ plan.RandomWindowStart, plan.RandomWindowEnd, encodeScheduleGenerationInput(generationInput),
+ enableWebSearch, generateCount, publishConfig.AutoPublish, publishConfig.AccountIDsJSON,
+ publishConfig.CoverAssetURL, publishConfig.CoverImageAssetID, req.StartAt, req.EndAt).
Scan(&id, &createdAt, &startAt, &endAt, &status)
if err != nil {
return nil, response.ErrInternal(50010, "create_failed", "failed to create schedule task")
}
- nextRunAt, err := resolveScheduleNextRunForStatus(status, req.CronExpr, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), time.Now())
+ nextRunAt, err := computeScheduleNextRun(status, id, plan.CronExpr, plan.ScheduleDays, plan.ScheduleTimeMode, plan.RandomWindowStart, plan.RandomWindowEnd, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), time.Now())
if err != nil {
return nil, response.ErrBadRequest(40015, "invalid_cron_expr", err.Error())
}
@@ -196,12 +248,15 @@ func (s *ScheduleTaskService) Create(ctx context.Context, req ScheduleTaskReques
}
afterJSON, _ := json.Marshal(map[string]interface{}{
- "id": id,
- "name": req.Name,
- "cron_expr": req.CronExpr,
- "enable_web_search": enableWebSearch,
- "generate_count": generateCount,
- "auto_publish": publishConfig.AutoPublish,
+ "id": id,
+ "name": req.Name,
+ "target_type": targetType,
+ "cron_expr": plan.CronExpr,
+ "schedule_days": plan.ScheduleDays,
+ "schedule_time_mode": plan.ScheduleTimeMode,
+ "enable_web_search": enableWebSearch,
+ "generate_count": generateCount,
+ "auto_publish": publishConfig.AutoPublish,
})
result := "success"
resourceType := "schedule_task"
@@ -221,12 +276,15 @@ func (s *ScheduleTaskService) Create(ctx context.Context, req ScheduleTaskReques
invalidateScheduleTaskCaches(ctx, s.cache, actor.TenantID, &id)
workspaceID := actor.PrimaryWorkspaceID
return &ScheduleTaskResponse{
- ID: id, WorkspaceID: &workspaceID, PromptRuleID: req.PromptRuleID, BrandID: req.BrandID,
- Name: req.Name, CronExpr: req.CronExpr, AutoPublish: publishConfig.AutoPublish,
+ ID: id, WorkspaceID: &workspaceID, TargetType: targetType, PromptRuleID: promptRuleID,
+ SubscriptionPromptID: subscriptionPromptID, BrandID: req.BrandID,
+ Name: req.Name, CronExpr: plan.CronExpr, ScheduleKind: plan.ScheduleKind, ScheduleDays: plan.ScheduleDays,
+ ScheduleTimeMode: plan.ScheduleTimeMode, RandomWindowStart: plan.RandomWindowStart,
+ RandomWindowEnd: plan.RandomWindowEnd, GenerationInput: generationInput, AutoPublish: publishConfig.AutoPublish,
PublishAccountIDs: publishConfig.AccountIDs, AutoPublishPlatforms: publishConfig.PlatformIDs, CoverAssetURL: publishConfig.CoverAssetURL,
CoverImageAssetID: publishConfig.CoverImageAssetID, EnableWebSearch: enableWebSearch,
GenerateCount: generateCount,
- Status: status, CreatedAt: fmt.Sprintf("%v", createdAt),
+ Status: status, CreatedAt: timeStringFromDBValue(createdAt),
NextRunAt: timeStringPtr(nextRunAt),
}, nil
}
@@ -239,12 +297,26 @@ func (s *ScheduleTaskService) Update(ctx context.Context, id int64, req Schedule
}
req.BrandID = &brandID
- if err := validateScheduleCronExpr(req.CronExpr); err != nil {
+ targetType, promptRuleID, subscriptionPromptID, err := normalizeScheduleTarget(req)
+ if err != nil {
return err
}
- if err := s.ensurePromptRuleExists(ctx, actor.TenantID, brandID, req.PromptRuleID); err != nil {
+ plan, err := normalizeSchedulePlan(req)
+ if err != nil {
return err
}
+ generationInput := clonePromptRuleExtraParams(req.GenerationInput)
+ if targetType == ScheduleTargetPromptRule {
+ if err := s.ensurePromptRuleExists(ctx, actor.TenantID, brandID, *promptRuleID); err != nil {
+ return err
+ }
+ generationInput = map[string]interface{}{}
+ } else {
+ generationInput, err = s.validateKolScheduleGenerationInput(ctx, actor.TenantID, *subscriptionPromptID, generationInput)
+ if err != nil {
+ return err
+ }
+ }
generateCount, err := normalizeScheduleGenerateCount(req.GenerateCount)
if err != nil {
return err
@@ -265,20 +337,32 @@ func (s *ScheduleTaskService) Update(ctx context.Context, id int64, req Schedule
var startAt pgtype.Timestamptz
var endAt pgtype.Timestamptz
err = tx.QueryRow(ctx, `
- UPDATE schedule_tasks SET prompt_rule_id = $1, brand_id = $2, name = $3,
- cron_expr = $4, enable_web_search = $5, generate_count = $6,
- workspace_id = $7, auto_publish = $8, publish_account_ids = $9::jsonb,
- cover_asset_url = $10, cover_image_asset_id = $11,
- start_at = $12::timestamptz, end_at = $13::timestamptz, operator_id = $14, updated_at = NOW()
- WHERE id = $15 AND tenant_id = $16 AND brand_id = $17 AND deleted_at IS NULL
+ UPDATE schedule_tasks SET target_type = $1, prompt_rule_id = $2, subscription_prompt_id = $3,
+ brand_id = $4, name = $5, cron_expr = $6,
+ schedule_kind = $7, schedule_days = $8::jsonb, schedule_time_mode = $9,
+ random_window_start = $10::time, random_window_end = $11::time,
+ generation_input_json = $12::jsonb,
+ enable_web_search = $13, generate_count = $14,
+ workspace_id = $15, auto_publish = $16, publish_account_ids = $17::jsonb,
+ cover_asset_url = $18, cover_image_asset_id = $19,
+ start_at = $20::timestamptz, end_at = $21::timestamptz, operator_id = $22,
+ last_dispatch_status = NULL, last_dispatch_error = NULL,
+ consecutive_failures = 0,
+ updated_at = NOW()
+ WHERE id = $23 AND tenant_id = $24 AND brand_id = $25 AND deleted_at IS NULL
RETURNING status, start_at, end_at
- `, req.PromptRuleID, req.BrandID, req.Name, req.CronExpr, enableWebSearch, generateCount, actor.PrimaryWorkspaceID, publishConfig.AutoPublish, publishConfig.AccountIDsJSON, publishConfig.CoverAssetURL, publishConfig.CoverImageAssetID, req.StartAt, req.EndAt, actor.UserID, id, actor.TenantID, brandID).
+ `, targetType, promptRuleID, subscriptionPromptID, req.BrandID, req.Name, plan.CronExpr,
+ plan.ScheduleKind, encodeScheduleDayKeys(plan.ScheduleDays), plan.ScheduleTimeMode,
+ plan.RandomWindowStart, plan.RandomWindowEnd, encodeScheduleGenerationInput(generationInput),
+ enableWebSearch, generateCount, actor.PrimaryWorkspaceID, publishConfig.AutoPublish,
+ publishConfig.AccountIDsJSON, publishConfig.CoverAssetURL, publishConfig.CoverImageAssetID,
+ req.StartAt, req.EndAt, actor.UserID, id, actor.TenantID, brandID).
Scan(&status, &startAt, &endAt)
if err != nil {
return response.ErrNotFound(40432, "schedule_not_found", "schedule task not found")
}
- nextRunAt, err := resolveScheduleNextRunForStatus(status, req.CronExpr, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), time.Now())
+ nextRunAt, err := computeScheduleNextRun(status, id, plan.CronExpr, plan.ScheduleDays, plan.ScheduleTimeMode, plan.RandomWindowStart, plan.RandomWindowEnd, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), time.Now())
if err != nil {
return response.ErrBadRequest(40015, "invalid_cron_expr", err.Error())
}
@@ -346,19 +430,24 @@ func (s *ScheduleTaskService) ToggleStatus(ctx context.Context, id int64, req Sc
var cronExpr string
var status string
+ var scheduleTimeMode string
+ var randomWindowStart string
+ var randomWindowEnd string
+ var scheduleDaysJSON []byte
var startAt pgtype.Timestamptz
var endAt pgtype.Timestamptz
err = tx.QueryRow(ctx, `
UPDATE schedule_tasks SET status = $1, operator_id = $2, updated_at = NOW()
WHERE id = $3 AND tenant_id = $4 AND brand_id = $5 AND deleted_at IS NULL
- RETURNING cron_expr, status, start_at, end_at
+ RETURNING cron_expr, status, schedule_days, schedule_time_mode,
+ random_window_start::text, random_window_end::text, start_at, end_at
`, req.Status, actor.UserID, id, actor.TenantID, brandID).
- Scan(&cronExpr, &status, &startAt, &endAt)
+ Scan(&cronExpr, &status, &scheduleDaysJSON, &scheduleTimeMode, &randomWindowStart, &randomWindowEnd, &startAt, &endAt)
if err != nil {
return response.ErrNotFound(40432, "schedule_not_found", "schedule task not found")
}
- nextRunAt, err := resolveScheduleNextRunForStatus(status, cronExpr, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), time.Now())
+ nextRunAt, err := computeScheduleNextRun(status, id, cronExpr, decodeScheduleDayKeys(scheduleDaysJSON), scheduleTimeMode, randomWindowStart, randomWindowEnd, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), time.Now())
if err != nil {
return response.ErrBadRequest(40015, "invalid_cron_expr", err.Error())
}
@@ -397,19 +486,32 @@ func (s *ScheduleTaskService) loadScheduleTasks(ctx context.Context, tenantID, b
AND ($5::text IS NULL OR name ILIKE '%' || $5 || '%')
AND ($6::timestamptz IS NULL OR created_at >= $6)
AND ($7::timestamptz IS NULL OR created_at <= $7)
- `, tenantID, brandID, params.PromptRuleID, params.Status, params.Keyword, params.CreatedFrom, params.CreatedTo).Scan(&total); err != nil {
+ AND ($8::text IS NULL OR target_type = $8)
+ AND ($9::bigint IS NULL OR subscription_prompt_id = $9)
+ `, tenantID, brandID, params.PromptRuleID, params.Status, params.Keyword, params.CreatedFrom, params.CreatedTo, params.TargetType, params.SubscriptionPromptID).Scan(&total); err != nil {
return nil, response.ErrInternal(50010, "count_failed", "failed to count schedule tasks")
}
rows, err := s.pool.Query(ctx, `
- SELECT st.id, st.workspace_id, st.prompt_rule_id, st.brand_id, st.name,
- st.cron_expr, st.auto_publish, st.publish_account_ids, st.cover_asset_url,
+ SELECT st.id, st.workspace_id, st.target_type, st.prompt_rule_id,
+ st.subscription_prompt_id, st.brand_id, st.name,
+ st.cron_expr, st.schedule_kind, st.schedule_days, st.schedule_time_mode,
+ st.random_window_start::text, st.random_window_end::text, st.generation_input_json,
+ st.auto_publish, st.publish_account_ids, st.cover_asset_url,
st.cover_image_asset_id, st.enable_web_search, st.generate_count, st.start_at, st.end_at,
- st.next_run_at, st.status, st.created_at, st.updated_at,
+ st.next_run_at, st.status, st.last_run_at, st.last_dispatch_status,
+ st.last_dispatch_error, st.consecutive_failures, st.created_at, st.updated_at,
pr.name AS prompt_rule_name,
+ sp_prompt.name AS subscription_prompt_name,
+ sp_pkg.name AS subscription_package_name,
+ sp_profile.display_name AS subscription_kol_name,
b.name AS brand_name
FROM schedule_tasks st
LEFT JOIN prompt_rules pr ON pr.id = st.prompt_rule_id
+ LEFT JOIN kol_subscription_prompts sp ON sp.id = st.subscription_prompt_id
+ LEFT JOIN kol_prompts sp_prompt ON sp_prompt.id = sp.prompt_id
+ LEFT JOIN kol_packages sp_pkg ON sp_pkg.id = sp.package_id
+ LEFT JOIN kol_profiles sp_profile ON sp_profile.id = sp_pkg.kol_profile_id
LEFT JOIN brands b ON b.id = st.brand_id
WHERE st.tenant_id = $1 AND st.brand_id = $2 AND st.deleted_at IS NULL
AND ($3::bigint IS NULL OR st.prompt_rule_id = $3)
@@ -417,9 +519,11 @@ func (s *ScheduleTaskService) loadScheduleTasks(ctx context.Context, tenantID, b
AND ($5::text IS NULL OR st.name ILIKE '%' || $5 || '%')
AND ($6::timestamptz IS NULL OR st.created_at >= $6)
AND ($7::timestamptz IS NULL OR st.created_at <= $7)
+ AND ($10::text IS NULL OR st.target_type = $10)
+ AND ($11::bigint IS NULL OR st.subscription_prompt_id = $11)
ORDER BY st.created_at DESC
LIMIT $8 OFFSET $9
- `, tenantID, brandID, params.PromptRuleID, params.Status, params.Keyword, params.CreatedFrom, params.CreatedTo, params.PageSize, offset)
+ `, tenantID, brandID, params.PromptRuleID, params.Status, params.Keyword, params.CreatedFrom, params.CreatedTo, params.PageSize, offset, params.TargetType, params.SubscriptionPromptID)
if err != nil {
return nil, response.ErrInternal(50010, "query_failed", "failed to list schedule tasks")
}
@@ -431,27 +535,38 @@ func (s *ScheduleTaskService) loadScheduleTasks(ctx context.Context, tenantID, b
if err != nil {
return nil, err
}
- if err := s.populateScheduleAutoPublishPlatforms(ctx, tenantID, &item); err != nil {
- return nil, err
- }
- if err := s.populateScheduleLatestGeneratedArticles(ctx, tenantID, &item); err != nil {
- return nil, err
- }
items = append(items, item)
}
+ if err := rows.Err(); err != nil {
+ return nil, response.ErrInternal(50010, "scan_failed", "failed to iterate schedule tasks")
+ }
+ if err := s.populateScheduleTaskListDecorations(ctx, tenantID, items); err != nil {
+ return nil, err
+ }
return &ScheduleTaskListResponse{Items: items, Total: total}, nil
}
func (s *ScheduleTaskService) loadScheduleTaskDetail(ctx context.Context, tenantID, brandID, taskID int64) (*ScheduleTaskResponse, bool, error) {
row := s.pool.QueryRow(ctx, `
- SELECT st.id, st.workspace_id, st.prompt_rule_id, st.brand_id, st.name,
- st.cron_expr, st.auto_publish, st.publish_account_ids, st.cover_asset_url,
+ SELECT st.id, st.workspace_id, st.target_type, st.prompt_rule_id,
+ st.subscription_prompt_id, st.brand_id, st.name,
+ st.cron_expr, st.schedule_kind, st.schedule_days, st.schedule_time_mode,
+ st.random_window_start::text, st.random_window_end::text, st.generation_input_json,
+ st.auto_publish, st.publish_account_ids, st.cover_asset_url,
st.cover_image_asset_id, st.enable_web_search, st.generate_count, st.start_at, st.end_at,
- st.next_run_at, st.status, st.created_at, st.updated_at,
+ st.next_run_at, st.status, st.last_run_at, st.last_dispatch_status,
+ st.last_dispatch_error, st.consecutive_failures, st.created_at, st.updated_at,
pr.name AS prompt_rule_name,
+ sp_prompt.name AS subscription_prompt_name,
+ sp_pkg.name AS subscription_package_name,
+ sp_profile.display_name AS subscription_kol_name,
b.name AS brand_name
FROM schedule_tasks st
LEFT JOIN prompt_rules pr ON pr.id = st.prompt_rule_id
+ LEFT JOIN kol_subscription_prompts sp ON sp.id = st.subscription_prompt_id
+ LEFT JOIN kol_prompts sp_prompt ON sp_prompt.id = sp.prompt_id
+ LEFT JOIN kol_packages sp_pkg ON sp_pkg.id = sp.package_id
+ LEFT JOIN kol_profiles sp_profile ON sp_profile.id = sp_pkg.kol_profile_id
LEFT JOIN brands b ON b.id = st.brand_id
WHERE st.id = $1 AND st.tenant_id = $2 AND st.brand_id = $3 AND st.deleted_at IS NULL
`, taskID, tenantID, brandID)
@@ -471,6 +586,16 @@ func (s *ScheduleTaskService) loadScheduleTaskDetail(ctx context.Context, tenant
return &item, true, nil
}
+func (s *ScheduleTaskService) populateScheduleTaskListDecorations(ctx context.Context, tenantID int64, items []ScheduleTaskResponse) error {
+ if len(items) == 0 {
+ return nil
+ }
+ if err := s.populateScheduleAutoPublishPlatformsBatch(ctx, tenantID, items); err != nil {
+ return err
+ }
+ return s.populateScheduleLatestGeneratedArticlesBatch(ctx, tenantID, items)
+}
+
func (s *ScheduleTaskService) populateScheduleAutoPublishPlatforms(ctx context.Context, tenantID int64, item *ScheduleTaskResponse) error {
if item == nil || !item.AutoPublish {
return nil
@@ -483,17 +608,85 @@ func (s *ScheduleTaskService) populateScheduleAutoPublishPlatforms(ctx context.C
return nil
}
+func (s *ScheduleTaskService) populateScheduleAutoPublishPlatformsBatch(ctx context.Context, tenantID int64, items []ScheduleTaskResponse) error {
+ accountSet := make(map[string]struct{})
+ for _, item := range items {
+ if !item.AutoPublish {
+ continue
+ }
+ for _, accountID := range item.PublishAccountIDs {
+ normalized := normalizeSchedulePublishAccountIDs([]string{accountID})
+ if len(normalized) == 0 {
+ continue
+ }
+ accountSet[normalized[0]] = struct{}{}
+ }
+ }
+ if len(accountSet) == 0 {
+ return nil
+ }
+
+ accountIDs := make([]string, 0, len(accountSet))
+ for accountID := range accountSet {
+ accountIDs = append(accountIDs, accountID)
+ }
+ rows, err := s.pool.Query(ctx, `
+ SELECT desktop_id::text, platform_id
+ FROM platform_accounts
+ WHERE tenant_id = $1
+ AND desktop_id::text = ANY($2::text[])
+ AND deleted_at IS NULL
+ `, tenantID, accountIDs)
+ if err != nil {
+ return response.ErrInternal(50010, "query_failed", "failed to resolve schedule auto publish platforms")
+ }
+ defer rows.Close()
+
+ platformsByAccount := make(map[string][]string, len(accountIDs))
+ for rows.Next() {
+ var accountID string
+ var platformID string
+ if err := rows.Scan(&accountID, &platformID); err != nil {
+ return response.ErrInternal(50010, "scan_failed", "failed to scan schedule auto publish platforms")
+ }
+ platformsByAccount[accountID] = append(platformsByAccount[accountID], platformID)
+ }
+ if err := rows.Err(); err != nil {
+ return response.ErrInternal(50010, "scan_failed", "failed to iterate schedule auto publish platforms")
+ }
+
+ for idx := range items {
+ if !items[idx].AutoPublish {
+ continue
+ }
+ 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]]...)
+ }
+ items[idx].AutoPublishPlatforms = emptyStringSliceIfNil(normalizePlatformIDs(platformIDs))
+ }
+ return nil
+}
+
func (s *ScheduleTaskService) populateScheduleLatestGeneratedArticles(ctx context.Context, tenantID int64, item *ScheduleTaskResponse) error {
if item == nil || item.ID <= 0 {
return nil
}
+ taskType := "custom_generation"
+ if item.TargetType == ScheduleTargetKolSubscriptionPrompt {
+ taskType = kolGenerationTaskType
+ }
rows, err := s.pool.Query(ctx, `
WITH latest_run AS (
SELECT COALESCE(NULLIF(gt.input_params_json ->> 'scheduled_for', ''), to_char(date_trunc('second', gt.created_at), 'YYYY-MM-DD"T"HH24:MI:SS"Z"')) AS run_key
FROM generation_tasks gt
WHERE gt.tenant_id = $1
- AND gt.task_type = 'custom_generation'
+ AND gt.task_type = $3
AND COALESCE(NULLIF(gt.input_params_json ->> 'generation_mode', ''), '') = 'schedule'
AND NULLIF(gt.input_params_json ->> 'schedule_task_id', '')::bigint = $2
ORDER BY gt.created_at DESC, gt.id DESC
@@ -510,11 +703,11 @@ func (s *ScheduleTaskService) populateScheduleLatestGeneratedArticles(ctx contex
LEFT JOIN articles a ON a.id = gt.article_id AND a.deleted_at IS NULL
LEFT JOIN article_versions av ON av.id = a.current_version_id
WHERE gt.tenant_id = $1
- AND gt.task_type = 'custom_generation'
+ AND gt.task_type = $3
AND COALESCE(NULLIF(gt.input_params_json ->> 'generation_mode', ''), '') = 'schedule'
AND NULLIF(gt.input_params_json ->> 'schedule_task_id', '')::bigint = $2
ORDER BY gt.created_at, gt.id
- `, tenantID, item.ID)
+ `, tenantID, item.ID, taskType)
if err != nil {
return response.ErrInternal(50010, "query_failed", "failed to load schedule generated articles")
}
@@ -534,7 +727,7 @@ func (s *ScheduleTaskService) populateScheduleLatestGeneratedArticles(ctx contex
return response.ErrInternal(50010, "scan_failed", err.Error())
}
if executionTime == nil {
- executionTime = stringPtrFromDBValue(executionAt)
+ executionTime = timeStringPtrFromDBValue(executionAt)
}
statuses = append(statuses, generateStatus)
if articleID != nil && *articleID > 0 {
@@ -542,7 +735,7 @@ func (s *ScheduleTaskService) populateScheduleLatestGeneratedArticles(ctx contex
ArticleID: *articleID,
Title: title,
GenerateStatus: generateStatus,
- CreatedAt: fmt.Sprintf("%v", createdAt),
+ CreatedAt: timeStringFromDBValue(createdAt),
})
} else {
_ = taskID
@@ -561,6 +754,126 @@ func (s *ScheduleTaskService) populateScheduleLatestGeneratedArticles(ctx contex
return nil
}
+func (s *ScheduleTaskService) populateScheduleLatestGeneratedArticlesBatch(ctx context.Context, tenantID int64, items []ScheduleTaskResponse) error {
+ if len(items) == 0 {
+ return nil
+ }
+ promptRuleTaskIDs := make([]int64, 0, len(items))
+ kolTaskIDs := make([]int64, 0, len(items))
+ indexByTaskID := make(map[int64]int, len(items))
+ for idx, item := range items {
+ if item.ID <= 0 {
+ continue
+ }
+ indexByTaskID[item.ID] = idx
+ if item.TargetType == ScheduleTargetKolSubscriptionPrompt {
+ kolTaskIDs = append(kolTaskIDs, item.ID)
+ } else {
+ promptRuleTaskIDs = append(promptRuleTaskIDs, item.ID)
+ }
+ }
+
+ if err := s.populateScheduleLatestGeneratedArticlesBatchByType(ctx, tenantID, items, indexByTaskID, promptRuleTaskIDs, "custom_generation"); err != nil {
+ return err
+ }
+ return s.populateScheduleLatestGeneratedArticlesBatchByType(ctx, tenantID, items, indexByTaskID, kolTaskIDs, kolGenerationTaskType)
+}
+
+func (s *ScheduleTaskService) populateScheduleLatestGeneratedArticlesBatchByType(
+ ctx context.Context,
+ tenantID int64,
+ items []ScheduleTaskResponse,
+ indexByTaskID map[int64]int,
+ taskIDs []int64,
+ taskType string,
+) error {
+ if len(taskIDs) == 0 {
+ return nil
+ }
+
+ rows, err := s.pool.Query(ctx, `
+ WITH latest_run AS (
+ SELECT DISTINCT ON (NULLIF(gt.input_params_json ->> 'schedule_task_id', '')::bigint)
+ NULLIF(gt.input_params_json ->> 'schedule_task_id', '')::bigint AS schedule_task_id,
+ COALESCE(NULLIF(gt.input_params_json ->> 'scheduled_for', ''), to_char(date_trunc('second', gt.created_at), 'YYYY-MM-DD"T"HH24:MI:SS"Z"')) AS run_key
+ FROM generation_tasks gt
+ WHERE gt.tenant_id = $1
+ AND gt.task_type = $2
+ AND COALESCE(NULLIF(gt.input_params_json ->> 'generation_mode', ''), '') = 'schedule'
+ AND NULLIF(gt.input_params_json ->> 'schedule_task_id', '')::bigint = ANY($3::bigint[])
+ ORDER BY NULLIF(gt.input_params_json ->> 'schedule_task_id', '')::bigint, gt.created_at DESC, gt.id DESC
+ )
+ SELECT lr.schedule_task_id,
+ gt.id,
+ COALESCE(a.id, gt.article_id) AS article_id,
+ COALESCE(av.title, NULLIF(a.wizard_state_json ->> 'title', ''), NULLIF(gt.input_params_json ->> 'title', ''), NULLIF(gt.input_params_json ->> 'task_name', '')) AS article_title,
+ COALESCE(a.generate_status, gt.status) AS article_generate_status,
+ COALESCE(a.created_at, gt.created_at) AS article_created_at,
+ COALESCE(gt.started_at, gt.completed_at, gt.created_at) AS execution_time
+ FROM latest_run lr
+ JOIN generation_tasks gt
+ ON gt.tenant_id = $1
+ AND gt.task_type = $2
+ AND COALESCE(NULLIF(gt.input_params_json ->> 'generation_mode', ''), '') = 'schedule'
+ AND NULLIF(gt.input_params_json ->> 'schedule_task_id', '')::bigint = lr.schedule_task_id
+ AND COALESCE(NULLIF(gt.input_params_json ->> 'scheduled_for', ''), to_char(date_trunc('second', gt.created_at), 'YYYY-MM-DD"T"HH24:MI:SS"Z"')) = lr.run_key
+ LEFT JOIN articles a ON a.id = gt.article_id AND a.deleted_at IS NULL
+ LEFT JOIN article_versions av ON av.id = a.current_version_id
+ ORDER BY lr.schedule_task_id, gt.created_at, gt.id
+ `, tenantID, taskType, taskIDs)
+ if err != nil {
+ return response.ErrInternal(50010, "query_failed", "failed to load schedule generated articles")
+ }
+ defer rows.Close()
+
+ statusesByTask := make(map[int64][]string, len(taskIDs))
+ for rows.Next() {
+ var scheduleTaskID int64
+ var taskID int64
+ var articleID *int64
+ var title *string
+ var generateStatus string
+ var createdAt interface{}
+ var executionAt interface{}
+ if err := rows.Scan(&scheduleTaskID, &taskID, &articleID, &title, &generateStatus, &createdAt, &executionAt); err != nil {
+ return response.ErrInternal(50010, "scan_failed", err.Error())
+ }
+ idx, ok := indexByTaskID[scheduleTaskID]
+ if !ok {
+ continue
+ }
+ if items[idx].ExecutionTime == nil {
+ items[idx].ExecutionTime = timeStringPtrFromDBValue(executionAt)
+ }
+ statusesByTask[scheduleTaskID] = append(statusesByTask[scheduleTaskID], generateStatus)
+ if articleID != nil && *articleID > 0 {
+ items[idx].GeneratedArticles = append(items[idx].GeneratedArticles, InstantTaskArticleLink{
+ ArticleID: *articleID,
+ Title: title,
+ GenerateStatus: generateStatus,
+ CreatedAt: timeStringFromDBValue(createdAt),
+ })
+ } else {
+ _ = taskID
+ }
+ }
+ if err := rows.Err(); err != nil {
+ return response.ErrInternal(50010, "scan_failed", "failed to iterate schedule generated articles")
+ }
+
+ for scheduleTaskID, statuses := range statusesByTask {
+ status := aggregateGeneratedArticleStatus(statuses)
+ if status == "" {
+ continue
+ }
+ idx, ok := indexByTaskID[scheduleTaskID]
+ if ok {
+ items[idx].GenerationStatus = &status
+ }
+ }
+ return nil
+}
+
func aggregateGeneratedArticleStatus(statuses []string) string {
if len(statuses) == 0 {
return ""
@@ -608,32 +921,49 @@ func scanScheduleTaskResponse(scanner interface {
var startAt interface{}
var endAt interface{}
var nextRunAt interface{}
+ var lastRunAt interface{}
var publishAccountIDsJSON []byte
- if err := scanner.Scan(&item.ID, &item.WorkspaceID, &item.PromptRuleID, &item.BrandID, &item.Name,
- &item.CronExpr, &item.AutoPublish, &publishAccountIDsJSON,
+ var scheduleDaysJSON []byte
+ var generationInputJSON []byte
+ if err := scanner.Scan(&item.ID, &item.WorkspaceID, &item.TargetType, &item.PromptRuleID,
+ &item.SubscriptionPromptID, &item.BrandID, &item.Name,
+ &item.CronExpr, &item.ScheduleKind, &scheduleDaysJSON, &item.ScheduleTimeMode,
+ &item.RandomWindowStart, &item.RandomWindowEnd, &generationInputJSON,
+ &item.AutoPublish, &publishAccountIDsJSON,
&item.CoverAssetURL, &item.CoverImageAssetID, &item.EnableWebSearch, &item.GenerateCount, &startAt, &endAt,
- &nextRunAt, &item.Status, &createdAt, &updatedAt,
- &item.PromptRuleName, &item.BrandName); err != nil {
+ &nextRunAt, &item.Status, &lastRunAt, &item.LastDispatchStatus,
+ &item.LastDispatchError, &item.ConsecutiveFailures, &createdAt, &updatedAt,
+ &item.PromptRuleName, &item.SubscriptionPromptName, &item.SubscriptionPackageName,
+ &item.SubscriptionKolName, &item.BrandName); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ScheduleTaskResponse{}, pgx.ErrNoRows
}
return ScheduleTaskResponse{}, response.ErrInternal(50010, "scan_failed", err.Error())
}
+ if item.TargetType == "" {
+ item.TargetType = ScheduleTargetPromptRule
+ }
+ if item.ScheduleKind == "" {
+ item.ScheduleKind = scheduleKindDaily
+ }
+ if item.ScheduleTimeMode == "" {
+ item.ScheduleTimeMode = sharedschedule.TimeModeFixed
+ }
+ if item.RandomWindowStart == "" {
+ item.RandomWindowStart = defaultRandomWindowStart
+ }
+ if item.RandomWindowEnd == "" {
+ item.RandomWindowEnd = defaultRandomWindowEnd
+ }
+ item.ScheduleDays = decodeScheduleDayKeys(scheduleDaysJSON)
+ item.GenerationInput = decodeScheduleGenerationInput(generationInputJSON)
item.PublishAccountIDs = decodeSchedulePublishAccountIDs(publishAccountIDsJSON)
- item.CreatedAt = fmt.Sprintf("%v", createdAt)
- item.UpdatedAt = fmt.Sprintf("%v", updatedAt)
- if startAt != nil {
- value := fmt.Sprintf("%v", startAt)
- item.StartAt = &value
- }
- if endAt != nil {
- value := fmt.Sprintf("%v", endAt)
- item.EndAt = &value
- }
- if nextRunAt != nil {
- value := fmt.Sprintf("%v", nextRunAt)
- item.NextRunAt = &value
- }
+ item.CreatedAt = timeStringFromDBValue(createdAt)
+ item.UpdatedAt = timeStringFromDBValue(updatedAt)
+ item.StartAt = timeStringPtrFromDBValue(startAt)
+ item.EndAt = timeStringPtrFromDBValue(endAt)
+ item.NextRunAt = timeStringPtrFromDBValue(nextRunAt)
+ item.LastRunAt = timeStringPtrFromDBValue(lastRunAt)
return item, nil
}
@@ -648,6 +978,80 @@ func (s *ScheduleTaskService) ensurePromptRuleExists(ctx context.Context, tenant
return nil
}
+func (s *ScheduleTaskService) validateKolScheduleGenerationInput(ctx context.Context, tenantID, subscriptionPromptID int64, input map[string]interface{}) (map[string]interface{}, error) {
+ input = normalizeKolScheduleGenerationInput(input)
+ row, err := repository.NewKolSubscriptionRepository(s.pool).GetSubscriptionPromptByID(ctx, tenantID, subscriptionPromptID)
+ if err != nil {
+ return input, response.ErrInternal(50109, "kol_subscription_prompt_query_failed", err.Error())
+ }
+ if row == nil {
+ return input, ErrKolGenerationForbidden
+ }
+
+ now := time.Now()
+ switch {
+ case row.Status != "active":
+ return input, ErrKolGenerationForbidden
+ case row.SubscriptionStatus != "active":
+ return input, ErrKolGenerationForbidden
+ case row.SubscriptionEndAt != nil && row.SubscriptionEndAt.Before(now):
+ return input, ErrKolGenerationForbidden
+ case row.PackageStatus != "published":
+ return input, ErrKolGenerationForbidden
+ case row.PromptStatus != "active":
+ return input, ErrKolGenerationForbidden
+ }
+
+ prompt, err := repository.NewKolPromptRepository(s.pool).GetByID(ctx, row.CreatorTenantID, row.PromptID)
+ if err != nil {
+ return input, response.ErrInternal(50110, "kol_prompt_query_failed", err.Error())
+ }
+ if prompt == nil || prompt.PromptAssetKey == nil || *prompt.PromptAssetKey == "" {
+ return input, response.ErrInternal(50111, "kol_prompt_content_missing", "published prompt content not found")
+ }
+ schema, err := decodeKolSchema(prompt.SchemaJSON)
+ if err != nil {
+ return input, response.ErrInternal(50096, "kol_generation_schema_decode_failed", err.Error())
+ }
+ cardConfig, err := decodeKolCardConfig(prompt.CardConfigJSON)
+ if err != nil {
+ return input, response.ErrInternal(50097, "kol_generation_card_config_decode_failed", err.Error())
+ }
+
+ variables, _ := input["variables"].(map[string]interface{})
+ if variables == nil {
+ variables = map[string]interface{}{}
+ input["variables"] = variables
+ }
+ missing := make([]string, 0)
+ for _, variable := range schema.Variables {
+ _, hasValue, err := resolveKolVariableValue(variables, variable, variable.Key)
+ if err != nil {
+ return input, response.ErrBadRequest(40071, "kol_generation_variables_invalid", err.Error())
+ }
+ if variable.Required && !hasValue {
+ missing = append(missing, variable.Label)
+ }
+ }
+ if len(missing) > 0 {
+ return input, response.ErrBadRequest(40071, "kol_generation_variables_invalid", "missing variables: "+strings.Join(missing, ", "))
+ }
+
+ if extractBool(input, "enable_web_search") && !kolCardConfigAllowsWebSearch(cardConfig) {
+ return input, response.ErrBadRequest(40073, "kol_generation_web_search_disabled", "web search is disabled for this prompt")
+ }
+ knowledgeGroupIDs := extractKnowledgeGroupIDs(input["knowledge_group_ids"])
+ if len(knowledgeGroupIDs) > 0 && !kolCardConfigAllowsUserKnowledge(cardConfig) {
+ return input, response.ErrBadRequest(40074, "kol_generation_knowledge_disabled", "knowledge selection is disabled for this prompt")
+ }
+ if len(knowledgeGroupIDs) > 0 {
+ input["knowledge_group_ids"] = knowledgeGroupIDs
+ } else {
+ delete(input, "knowledge_group_ids")
+ }
+ return input, nil
+}
+
type scheduleAutoPublishConfig struct {
AutoPublish bool
AccountIDs []string
@@ -744,13 +1148,6 @@ func schedulePublishPlatformsRequireCover(platformIDs []string) bool {
return false
}
-func resolveScheduleNextRunForStatus(status, cronExpr string, startAt, endAt *time.Time, now time.Time) (*time.Time, error) {
- if status != "enabled" {
- return nil, nil
- }
- return sharedschedule.ComputeNextRun(cronExpr, startAt, endAt, now)
-}
-
func validateScheduleCronExpr(cronExpr string) error {
if _, err := sharedschedule.Parse(cronExpr); err != nil {
return response.ErrBadRequest(40015, "invalid_cron_expr", err.Error())
@@ -827,3 +1224,21 @@ func timeStringPtr(value *time.Time) *string {
text := value.Format(time.RFC3339)
return &text
}
+
+func timeStringFromDBValue(value interface{}) string {
+ if value == nil {
+ return ""
+ }
+ if timestamp, ok := value.(time.Time); ok {
+ return timestamp.Round(0).Format(time.RFC3339)
+ }
+ return fmt.Sprintf("%v", value)
+}
+
+func timeStringPtrFromDBValue(value interface{}) *string {
+ if value == nil {
+ return nil
+ }
+ text := timeStringFromDBValue(value)
+ return &text
+}
diff --git a/server/internal/tenant/domain/schedule_task.go b/server/internal/tenant/domain/schedule_task.go
index 901e143..c0a5a9d 100644
--- a/server/internal/tenant/domain/schedule_task.go
+++ b/server/internal/tenant/domain/schedule_task.go
@@ -3,24 +3,36 @@ package domain
import "time"
type ScheduleTask struct {
- ID int64
- TenantID int64
- WorkspaceID *int64
- OperatorID *int64
- PromptRuleID int64
- BrandID *int64
- Name string
- CronExpr string
- AutoPublish bool
- PublishAccountIDs []string
- CoverAssetURL *string
- CoverImageAssetID *int64
- EnableWebSearch bool
- GenerateCount int
- StartAt *time.Time
- EndAt *time.Time
- NextRunAt *time.Time
- Status string
- CreatedAt time.Time
- UpdatedAt time.Time
+ ID int64
+ TenantID int64
+ WorkspaceID *int64
+ OperatorID *int64
+ TargetType string
+ PromptRuleID *int64
+ SubscriptionPromptID *int64
+ BrandID *int64
+ Name string
+ CronExpr string
+ ScheduleKind string
+ ScheduleDays []string
+ ScheduleTimeMode string
+ RandomWindowStart string
+ RandomWindowEnd string
+ GenerationInput map[string]interface{}
+ AutoPublish bool
+ PublishAccountIDs []string
+ CoverAssetURL *string
+ CoverImageAssetID *int64
+ EnableWebSearch bool
+ GenerateCount int
+ StartAt *time.Time
+ EndAt *time.Time
+ NextRunAt *time.Time
+ LastRunAt *time.Time
+ LastDispatchStatus *string
+ LastDispatchError *string
+ ConsecutiveFailures int
+ Status string
+ CreatedAt time.Time
+ UpdatedAt time.Time
}
diff --git a/server/internal/tenant/transport/schedule_task_handler.go b/server/internal/tenant/transport/schedule_task_handler.go
index 14ad66d..628b718 100644
--- a/server/internal/tenant/transport/schedule_task_handler.go
+++ b/server/internal/tenant/transport/schedule_task_handler.go
@@ -39,6 +39,14 @@ func (h *ScheduleTaskHandler) List(c *gin.Context) {
params.PromptRuleID = &rid
}
}
+ if v := c.Query("subscription_prompt_id"); v != "" {
+ if rid, err := strconv.ParseInt(v, 10, 64); err == nil {
+ params.SubscriptionPromptID = &rid
+ }
+ }
+ if v := c.Query("target_type"); v != "" {
+ params.TargetType = &v
+ }
if v := c.Query("status"); v != "" {
params.Status = &v
}
diff --git a/server/migrations/20260522100000_extend_schedule_tasks_for_random_and_kol.down.sql b/server/migrations/20260522100000_extend_schedule_tasks_for_random_and_kol.down.sql
new file mode 100644
index 0000000..825d052
--- /dev/null
+++ b/server/migrations/20260522100000_extend_schedule_tasks_for_random_and_kol.down.sql
@@ -0,0 +1,26 @@
+DROP INDEX IF EXISTS idx_schedule_task_subscription_prompt;
+DROP INDEX IF EXISTS idx_schedule_task_target;
+
+ALTER TABLE schedule_tasks
+ DROP CONSTRAINT IF EXISTS chk_schedule_tasks_consecutive_failures,
+ DROP CONSTRAINT IF EXISTS chk_schedule_tasks_generation_input_json,
+ DROP CONSTRAINT IF EXISTS chk_schedule_tasks_random_window,
+ DROP CONSTRAINT IF EXISTS chk_schedule_tasks_schedule_time_mode,
+ DROP CONSTRAINT IF EXISTS chk_schedule_tasks_schedule_days_json,
+ DROP CONSTRAINT IF EXISTS chk_schedule_tasks_schedule_kind,
+ DROP CONSTRAINT IF EXISTS chk_schedule_tasks_target_ref,
+ DROP CONSTRAINT IF EXISTS chk_schedule_tasks_target_type;
+
+ALTER TABLE schedule_tasks
+ DROP COLUMN IF EXISTS consecutive_failures,
+ DROP COLUMN IF EXISTS last_dispatch_error,
+ DROP COLUMN IF EXISTS last_dispatch_status,
+ DROP COLUMN IF EXISTS last_run_at,
+ DROP COLUMN IF EXISTS generation_input_json,
+ DROP COLUMN IF EXISTS random_window_end,
+ DROP COLUMN IF EXISTS random_window_start,
+ DROP COLUMN IF EXISTS schedule_time_mode,
+ DROP COLUMN IF EXISTS schedule_days,
+ DROP COLUMN IF EXISTS schedule_kind,
+ DROP COLUMN IF EXISTS subscription_prompt_id,
+ DROP COLUMN IF EXISTS target_type;
diff --git a/server/migrations/20260522100000_extend_schedule_tasks_for_random_and_kol.up.sql b/server/migrations/20260522100000_extend_schedule_tasks_for_random_and_kol.up.sql
new file mode 100644
index 0000000..d2fc2e8
--- /dev/null
+++ b/server/migrations/20260522100000_extend_schedule_tasks_for_random_and_kol.up.sql
@@ -0,0 +1,109 @@
+ALTER TABLE schedule_tasks
+ ALTER COLUMN prompt_rule_id DROP NOT NULL;
+
+ALTER TABLE schedule_tasks
+ ADD COLUMN IF NOT EXISTS target_type VARCHAR(32) NOT NULL DEFAULT 'prompt_rule',
+ ADD COLUMN IF NOT EXISTS subscription_prompt_id BIGINT REFERENCES kol_subscription_prompts(id),
+ ADD COLUMN IF NOT EXISTS schedule_kind VARCHAR(20) NOT NULL DEFAULT 'daily',
+ ADD COLUMN IF NOT EXISTS schedule_days JSONB NOT NULL DEFAULT '["mon","tue","wed","thu","fri","sat","sun"]'::jsonb,
+ ADD COLUMN IF NOT EXISTS schedule_time_mode VARCHAR(20) NOT NULL DEFAULT 'fixed',
+ ADD COLUMN IF NOT EXISTS random_window_start TIME NOT NULL DEFAULT '01:00:00',
+ ADD COLUMN IF NOT EXISTS random_window_end TIME NOT NULL DEFAULT '05:00:00',
+ ADD COLUMN IF NOT EXISTS generation_input_json JSONB NOT NULL DEFAULT '{}'::jsonb,
+ ADD COLUMN IF NOT EXISTS last_run_at TIMESTAMPTZ,
+ ADD COLUMN IF NOT EXISTS last_dispatch_status VARCHAR(20),
+ ADD COLUMN IF NOT EXISTS last_dispatch_error TEXT,
+ ADD COLUMN IF NOT EXISTS consecutive_failures INT NOT NULL DEFAULT 0;
+
+UPDATE schedule_tasks
+SET target_type = 'prompt_rule'
+WHERE target_type IS NULL OR btrim(target_type) = '';
+
+UPDATE schedule_tasks
+SET schedule_kind = 'daily',
+ schedule_days = '["mon","tue","wed","thu","fri","sat","sun"]'::jsonb
+WHERE schedule_days IS NULL
+ OR jsonb_typeof(schedule_days) <> 'array'
+ OR jsonb_array_length(schedule_days) = 0;
+
+UPDATE schedule_tasks
+SET schedule_time_mode = 'fixed'
+WHERE schedule_time_mode IS NULL OR btrim(schedule_time_mode) = '';
+
+DO $$
+BEGIN
+ IF NOT EXISTS (
+ SELECT 1 FROM pg_constraint WHERE conname = 'chk_schedule_tasks_target_type'
+ ) THEN
+ ALTER TABLE schedule_tasks
+ ADD CONSTRAINT chk_schedule_tasks_target_type
+ CHECK (target_type IN ('prompt_rule', 'kol_subscription_prompt'));
+ END IF;
+
+ IF NOT EXISTS (
+ SELECT 1 FROM pg_constraint WHERE conname = 'chk_schedule_tasks_target_ref'
+ ) THEN
+ ALTER TABLE schedule_tasks
+ ADD CONSTRAINT chk_schedule_tasks_target_ref
+ CHECK (
+ (target_type = 'prompt_rule' AND prompt_rule_id IS NOT NULL)
+ OR (target_type = 'kol_subscription_prompt' AND subscription_prompt_id IS NOT NULL)
+ );
+ END IF;
+
+ IF NOT EXISTS (
+ SELECT 1 FROM pg_constraint WHERE conname = 'chk_schedule_tasks_schedule_kind'
+ ) THEN
+ ALTER TABLE schedule_tasks
+ ADD CONSTRAINT chk_schedule_tasks_schedule_kind
+ CHECK (schedule_kind IN ('daily', 'weekly'));
+ END IF;
+
+ IF NOT EXISTS (
+ SELECT 1 FROM pg_constraint WHERE conname = 'chk_schedule_tasks_schedule_days_json'
+ ) THEN
+ ALTER TABLE schedule_tasks
+ ADD CONSTRAINT chk_schedule_tasks_schedule_days_json
+ CHECK (jsonb_typeof(schedule_days) = 'array' AND jsonb_array_length(schedule_days) BETWEEN 1 AND 7);
+ END IF;
+
+ IF NOT EXISTS (
+ SELECT 1 FROM pg_constraint WHERE conname = 'chk_schedule_tasks_schedule_time_mode'
+ ) THEN
+ ALTER TABLE schedule_tasks
+ ADD CONSTRAINT chk_schedule_tasks_schedule_time_mode
+ CHECK (schedule_time_mode IN ('fixed', 'random_window'));
+ END IF;
+
+ IF NOT EXISTS (
+ SELECT 1 FROM pg_constraint WHERE conname = 'chk_schedule_tasks_random_window'
+ ) THEN
+ ALTER TABLE schedule_tasks
+ ADD CONSTRAINT chk_schedule_tasks_random_window
+ CHECK (random_window_end > random_window_start);
+ END IF;
+
+ IF NOT EXISTS (
+ SELECT 1 FROM pg_constraint WHERE conname = 'chk_schedule_tasks_generation_input_json'
+ ) THEN
+ ALTER TABLE schedule_tasks
+ ADD CONSTRAINT chk_schedule_tasks_generation_input_json
+ CHECK (jsonb_typeof(generation_input_json) = 'object');
+ END IF;
+
+ IF NOT EXISTS (
+ SELECT 1 FROM pg_constraint WHERE conname = 'chk_schedule_tasks_consecutive_failures'
+ ) THEN
+ ALTER TABLE schedule_tasks
+ ADD CONSTRAINT chk_schedule_tasks_consecutive_failures
+ CHECK (consecutive_failures >= 0);
+ END IF;
+END $$;
+
+CREATE INDEX IF NOT EXISTS idx_schedule_task_target
+ ON schedule_tasks(tenant_id, target_type, created_at DESC)
+ WHERE deleted_at IS NULL;
+
+CREATE INDEX IF NOT EXISTS idx_schedule_task_subscription_prompt
+ ON schedule_tasks(tenant_id, subscription_prompt_id, created_at DESC)
+ WHERE deleted_at IS NULL AND subscription_prompt_id IS NOT NULL;