diff --git a/server/internal/tenant/app/schedule_task_service.go b/server/internal/tenant/app/schedule_task_service.go index 6dec6db..36e832e 100644 --- a/server/internal/tenant/app/schedule_task_service.go +++ b/server/internal/tenant/app/schedule_task_service.go @@ -6,12 +6,14 @@ import ( "fmt" "time" + "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "github.com/geo-platform/tenant-api/internal/shared/auditlog" "github.com/geo-platform/tenant-api/internal/shared/auth" "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" ) type ScheduleTaskService struct { @@ -24,30 +26,34 @@ func NewScheduleTaskService(pool *pgxpool.Pool, auditLogs *auditlog.AsyncWriter) } 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"` - TargetPlatform *string `json:"target_platform"` - StartAt *string `json:"start_at"` - EndAt *string `json:"end_at"` + 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"` + TargetPlatform *string `json:"target_platform"` + 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"` - 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"` - TargetPlatform *string `json:"target_platform"` - StartAt *string `json:"start_at"` - EndAt *string `json:"end_at"` - NextRunAt *string `json:"next_run_at"` - Status string `json:"status"` - CreatedAt string `json:"created_at"` - UpdatedAt string `json:"updated_at"` + ID int64 `json:"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"` + TargetPlatform *string `json:"target_platform"` + 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"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` } type ScheduleTaskListParams struct { @@ -95,7 +101,7 @@ func (s *ScheduleTaskService) List(ctx context.Context, params ScheduleTaskListP rows, err := s.pool.Query(ctx, ` SELECT st.id, st.prompt_rule_id, st.brand_id, st.name, - st.cron_expr, st.target_platform, st.start_at, st.end_at, + st.cron_expr, st.target_platform, 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, pr.name AS prompt_rule_name, b.name AS brand_name @@ -122,7 +128,7 @@ func (s *ScheduleTaskService) List(ctx context.Context, params ScheduleTaskListP var ca, ua interface{} var startAt, endAt, nextRunAt interface{} if err := rows.Scan(&r.ID, &r.PromptRuleID, &r.BrandID, &r.Name, - &r.CronExpr, &r.TargetPlatform, &startAt, &endAt, + &r.CronExpr, &r.TargetPlatform, &r.EnableWebSearch, &r.GenerateCount, &startAt, &endAt, &nextRunAt, &r.Status, &ca, &ua, &r.PromptRuleName, &r.BrandName); err != nil { return nil, response.ErrInternal(50010, "scan_failed", err.Error()) @@ -157,7 +163,7 @@ func (s *ScheduleTaskService) Detail(ctx context.Context, id int64) (*ScheduleTa var startAt, endAt, nextRunAt interface{} err := s.pool.QueryRow(ctx, ` SELECT st.id, st.prompt_rule_id, st.brand_id, st.name, - st.cron_expr, st.target_platform, st.start_at, st.end_at, + st.cron_expr, st.target_platform, 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, pr.name AS prompt_rule_name, b.name AS brand_name @@ -166,7 +172,7 @@ func (s *ScheduleTaskService) Detail(ctx context.Context, id int64) (*ScheduleTa LEFT JOIN brands b ON b.id = st.brand_id WHERE st.id = $1 AND st.tenant_id = $2 AND st.deleted_at IS NULL `, id, actor.TenantID).Scan(&r.ID, &r.PromptRuleID, &r.BrandID, &r.Name, - &r.CronExpr, &r.TargetPlatform, &startAt, &endAt, + &r.CronExpr, &r.TargetPlatform, &r.EnableWebSearch, &r.GenerateCount, &startAt, &endAt, &nextRunAt, &r.Status, &ca, &ua, &r.PromptRuleName, &r.BrandName) if err != nil { @@ -192,26 +198,67 @@ func (s *ScheduleTaskService) Detail(ctx context.Context, id int64) (*ScheduleTa func (s *ScheduleTaskService) Create(ctx context.Context, req ScheduleTaskRequest) (*ScheduleTaskResponse, error) { actor := auth.MustActor(ctx) - // Verify prompt rule belongs to same tenant - var exists bool - _ = s.pool.QueryRow(ctx, ` - SELECT EXISTS(SELECT 1 FROM prompt_rules WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL) - `, req.PromptRuleID, actor.TenantID).Scan(&exists) - if !exists { - return nil, response.ErrBadRequest(40001, "invalid_rule", "prompt rule not found or not active") + if err := validateScheduleCronExpr(req.CronExpr); err != nil { + return nil, err } + if err := s.ensurePromptRuleExists(ctx, actor.TenantID, req.PromptRuleID); err != nil { + return nil, err + } + generateCount, err := normalizeScheduleGenerateCount(req.GenerateCount) + if err != nil { + return nil, err + } + enableWebSearch := boolValue(req.EnableWebSearch) + + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, response.ErrInternal(50010, "create_failed", "failed to begin schedule task transaction") + } + defer tx.Rollback(ctx) var id int64 - var ca interface{} - err := s.pool.QueryRow(ctx, ` - INSERT INTO schedule_tasks (tenant_id, prompt_rule_id, brand_id, name, cron_expr, target_platform, start_at, end_at) - VALUES ($1, $2, $3, $4, $5, $6, $7::timestamptz, $8::timestamptz) RETURNING id, created_at - `, actor.TenantID, req.PromptRuleID, req.BrandID, req.Name, req.CronExpr, req.TargetPlatform, req.StartAt, req.EndAt).Scan(&id, &ca) + var createdAt time.Time + var startAt pgtype.Timestamptz + var endAt pgtype.Timestamptz + var status string + err = tx.QueryRow(ctx, ` + INSERT INTO schedule_tasks ( + tenant_id, operator_id, prompt_rule_id, brand_id, name, cron_expr, target_platform, + enable_web_search, generate_count, start_at, end_at + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::timestamptz, $11::timestamptz) + RETURNING id, created_at, start_at, end_at, status + `, actor.TenantID, actor.UserID, req.PromptRuleID, req.BrandID, req.Name, req.CronExpr, req.TargetPlatform, enableWebSearch, generateCount, 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") } - afterJSON, _ := json.Marshal(map[string]interface{}{"id": id, "name": req.Name, "cron_expr": req.CronExpr}) + nextRunAt, err := resolveScheduleNextRunForStatus(status, req.CronExpr, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), time.Now()) + if err != nil { + return nil, response.ErrBadRequest(40015, "invalid_cron_expr", err.Error()) + } + if _, err := tx.Exec(ctx, ` + UPDATE schedule_tasks + SET next_run_at = $1, + updated_at = NOW() + WHERE id = $2 + AND tenant_id = $3 + `, nextRunAt, id, actor.TenantID); err != nil { + return nil, response.ErrInternal(50010, "create_failed", "failed to initialize next run time") + } + + if err := tx.Commit(ctx); err != nil { + return nil, response.ErrInternal(50010, "create_failed", "failed to commit schedule task") + } + + afterJSON, _ := json.Marshal(map[string]interface{}{ + "id": id, + "name": req.Name, + "cron_expr": req.CronExpr, + "enable_web_search": enableWebSearch, + "generate_count": generateCount, + }) result := "success" resourceType := "schedule_task" requestID := middleware.RequestIDFromContext(ctx) @@ -230,20 +277,64 @@ func (s *ScheduleTaskService) Create(ctx context.Context, req ScheduleTaskReques return &ScheduleTaskResponse{ ID: id, PromptRuleID: req.PromptRuleID, BrandID: req.BrandID, Name: req.Name, CronExpr: req.CronExpr, TargetPlatform: req.TargetPlatform, - Status: "enabled", CreatedAt: fmt.Sprintf("%v", ca), + EnableWebSearch: enableWebSearch, GenerateCount: generateCount, + Status: status, CreatedAt: fmt.Sprintf("%v", createdAt), + NextRunAt: timeStringPtr(nextRunAt), }, nil } func (s *ScheduleTaskService) Update(ctx context.Context, id int64, req ScheduleTaskRequest) error { actor := auth.MustActor(ctx) - tag, err := s.pool.Exec(ctx, ` + if err := validateScheduleCronExpr(req.CronExpr); err != nil { + return err + } + if err := s.ensurePromptRuleExists(ctx, actor.TenantID, req.PromptRuleID); err != nil { + return err + } + generateCount, err := normalizeScheduleGenerateCount(req.GenerateCount) + if err != nil { + return err + } + enableWebSearch := boolValue(req.EnableWebSearch) + + tx, err := s.pool.Begin(ctx) + if err != nil { + return response.ErrInternal(50010, "update_failed", "failed to begin schedule task transaction") + } + defer tx.Rollback(ctx) + + var status string + 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, target_platform = $5, start_at = $6::timestamptz, end_at = $7::timestamptz, updated_at = NOW() - WHERE id = $8 AND tenant_id = $9 AND deleted_at IS NULL - `, req.PromptRuleID, req.BrandID, req.Name, req.CronExpr, req.TargetPlatform, req.StartAt, req.EndAt, id, actor.TenantID) - if err != nil || tag.RowsAffected() == 0 { + cron_expr = $4, target_platform = $5, enable_web_search = $6, generate_count = $7, + start_at = $8::timestamptz, end_at = $9::timestamptz, operator_id = $10, updated_at = NOW() + WHERE id = $11 AND tenant_id = $12 AND deleted_at IS NULL + RETURNING status, start_at, end_at + `, req.PromptRuleID, req.BrandID, req.Name, req.CronExpr, req.TargetPlatform, enableWebSearch, generateCount, req.StartAt, req.EndAt, actor.UserID, id, actor.TenantID). + 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()) + if err != nil { + return response.ErrBadRequest(40015, "invalid_cron_expr", err.Error()) + } + if _, err := tx.Exec(ctx, ` + UPDATE schedule_tasks + SET next_run_at = $1, + updated_at = NOW() + WHERE id = $2 + AND tenant_id = $3 + `, nextRunAt, id, actor.TenantID); err != nil { + return response.ErrInternal(50010, "update_failed", "failed to refresh next run time") + } + + if err := tx.Commit(ctx); err != nil { + return response.ErrInternal(50010, "update_failed", "failed to commit schedule task") + } return nil } @@ -277,12 +368,97 @@ func (s *ScheduleTaskService) Delete(ctx context.Context, id int64) error { func (s *ScheduleTaskService) ToggleStatus(ctx context.Context, id int64, req ScheduleTaskStatusRequest) error { actor := auth.MustActor(ctx) - tag, err := s.pool.Exec(ctx, ` - UPDATE schedule_tasks SET status = $1, updated_at = NOW() - WHERE id = $2 AND tenant_id = $3 AND deleted_at IS NULL - `, req.Status, id, actor.TenantID) - if err != nil || tag.RowsAffected() == 0 { + tx, err := s.pool.Begin(ctx) + if err != nil { + return response.ErrInternal(50010, "update_failed", "failed to begin schedule task transaction") + } + defer tx.Rollback(ctx) + + var cronExpr string + var status string + 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 deleted_at IS NULL + RETURNING cron_expr, status, start_at, end_at + `, req.Status, actor.UserID, id, actor.TenantID). + Scan(&cronExpr, &status, &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()) + if err != nil { + return response.ErrBadRequest(40015, "invalid_cron_expr", err.Error()) + } + if _, err := tx.Exec(ctx, ` + UPDATE schedule_tasks + SET next_run_at = $1, + updated_at = NOW() + WHERE id = $2 + AND tenant_id = $3 + `, nextRunAt, id, actor.TenantID); err != nil { + return response.ErrInternal(50010, "update_failed", "failed to refresh next run time") + } + + if err := tx.Commit(ctx); err != nil { + return response.ErrInternal(50010, "update_failed", "failed to commit schedule task") + } return nil } + +func (s *ScheduleTaskService) ensurePromptRuleExists(ctx context.Context, tenantID, promptRuleID int64) error { + var exists bool + _ = s.pool.QueryRow(ctx, ` + SELECT EXISTS(SELECT 1 FROM prompt_rules WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL) + `, promptRuleID, tenantID).Scan(&exists) + if !exists { + return response.ErrBadRequest(40001, "invalid_rule", "prompt rule not found or not active") + } + return nil +} + +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()) + } + return nil +} + +func normalizeScheduleGenerateCount(value *int) (int, error) { + if value == nil { + return 1, nil + } + if *value < 1 || *value > 20 { + return 0, response.ErrBadRequest(40022, "invalid_generate_count", "generate_count must be between 1 and 20") + } + return *value, nil +} + +func boolValue(value *bool) bool { + return value != nil && *value +} + +func timePtrFromTimestamp(value pgtype.Timestamptz) *time.Time { + if !value.Valid { + return nil + } + result := value.Time.Round(0) + return &result +} + +func timeStringPtr(value *time.Time) *string { + if value == nil { + return nil + } + text := value.Format(time.RFC3339) + return &text +} diff --git a/server/internal/tenant/domain/schedule_task.go b/server/internal/tenant/domain/schedule_task.go index 068c55f..022f507 100644 --- a/server/internal/tenant/domain/schedule_task.go +++ b/server/internal/tenant/domain/schedule_task.go @@ -3,17 +3,20 @@ package domain import "time" type ScheduleTask struct { - ID int64 - TenantID int64 - PromptRuleID int64 - BrandID *int64 - Name string - CronExpr string - TargetPlatform *string - StartAt *time.Time - EndAt *time.Time - NextRunAt *time.Time - Status string - CreatedAt time.Time - UpdatedAt time.Time + ID int64 + TenantID int64 + OperatorID *int64 + PromptRuleID int64 + BrandID *int64 + Name string + CronExpr string + TargetPlatform *string + EnableWebSearch bool + GenerateCount int + StartAt *time.Time + EndAt *time.Time + NextRunAt *time.Time + Status string + CreatedAt time.Time + UpdatedAt time.Time } diff --git a/server/migrations/20260413120000_enhance_schedule_tasks.down.sql b/server/migrations/20260413120000_enhance_schedule_tasks.down.sql new file mode 100644 index 0000000..c244717 --- /dev/null +++ b/server/migrations/20260413120000_enhance_schedule_tasks.down.sql @@ -0,0 +1,16 @@ +DROP INDEX IF EXISTS idx_schedule_task_operator; + +ALTER TABLE schedule_tasks + DROP CONSTRAINT IF EXISTS chk_schedule_tasks_generate_count; + +ALTER TABLE schedule_tasks + DROP COLUMN IF EXISTS generate_count, + DROP COLUMN IF EXISTS enable_web_search, + DROP COLUMN IF EXISTS operator_id; + +ALTER TABLE schedule_tasks + ALTER COLUMN target_platform TYPE VARCHAR(50) + USING CASE + WHEN target_platform IS NULL THEN NULL + ELSE LEFT(target_platform, 50) + END; diff --git a/server/migrations/20260413120000_enhance_schedule_tasks.up.sql b/server/migrations/20260413120000_enhance_schedule_tasks.up.sql new file mode 100644 index 0000000..ae3d90d --- /dev/null +++ b/server/migrations/20260413120000_enhance_schedule_tasks.up.sql @@ -0,0 +1,15 @@ +ALTER TABLE schedule_tasks + ALTER COLUMN target_platform TYPE TEXT; + +ALTER TABLE schedule_tasks + ADD COLUMN operator_id BIGINT REFERENCES users(id), + ADD COLUMN enable_web_search BOOLEAN NOT NULL DEFAULT FALSE, + ADD COLUMN generate_count INT NOT NULL DEFAULT 1; + +ALTER TABLE schedule_tasks + ADD CONSTRAINT chk_schedule_tasks_generate_count + CHECK (generate_count >= 1 AND generate_count <= 20); + +CREATE INDEX idx_schedule_task_operator + ON schedule_tasks(tenant_id, operator_id, created_at DESC) + WHERE deleted_at IS NULL;