root
8958cb44c0
- Removed ListQuestionVersions method and associated types from BrandService and Queries. - Updated PromptRuleGenerationService to change task naming and handling. - Enhanced ScheduleTaskService to support filtering by created date range and keyword. - Introduced InstantTaskService for managing instant tasks with appropriate filtering and response structures. - Added InstantTaskHandler for handling API requests related to instant tasks. - Created InstantTaskTab component for the admin web interface to display and manage instant tasks. - Updated database migrations to rename source types for articles and generation tasks. - Adjusted models and repository queries to reflect the removal of question version handling.
289 lines
10 KiB
Go
289 lines
10 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
|
|
"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"
|
|
)
|
|
|
|
type ScheduleTaskService struct {
|
|
pool *pgxpool.Pool
|
|
auditLogs *auditlog.AsyncWriter
|
|
}
|
|
|
|
func NewScheduleTaskService(pool *pgxpool.Pool, auditLogs *auditlog.AsyncWriter) *ScheduleTaskService {
|
|
return &ScheduleTaskService{pool: pool, auditLogs: auditLogs}
|
|
}
|
|
|
|
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"`
|
|
}
|
|
|
|
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"`
|
|
}
|
|
|
|
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"`
|
|
}
|
|
|
|
type ScheduleTaskListResponse struct {
|
|
Items []ScheduleTaskResponse `json:"items"`
|
|
Total int64 `json:"total"`
|
|
}
|
|
|
|
type ScheduleTaskStatusRequest struct {
|
|
Status string `json:"status" binding:"required,oneof=enabled disabled"`
|
|
}
|
|
|
|
func (s *ScheduleTaskService) List(ctx context.Context, params ScheduleTaskListParams) (*ScheduleTaskListResponse, error) {
|
|
actor := auth.MustActor(ctx)
|
|
if params.Page < 1 {
|
|
params.Page = 1
|
|
}
|
|
if params.PageSize < 1 || params.PageSize > 100 {
|
|
params.PageSize = 20
|
|
}
|
|
offset := (params.Page - 1) * params.PageSize
|
|
|
|
var total int64
|
|
err := s.pool.QueryRow(ctx, `
|
|
SELECT COUNT(*) FROM schedule_tasks
|
|
WHERE tenant_id = $1 AND deleted_at IS NULL
|
|
AND ($2::bigint IS NULL OR prompt_rule_id = $2)
|
|
AND ($3::text IS NULL OR status = $3)
|
|
AND ($4::text IS NULL OR name ILIKE '%' || $4 || '%')
|
|
AND ($5::timestamptz IS NULL OR created_at >= $5)
|
|
AND ($6::timestamptz IS NULL OR created_at <= $6)
|
|
`, actor.TenantID, params.PromptRuleID, params.Status, params.Keyword, params.CreatedFrom, params.CreatedTo).Scan(&total)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50010, "count_failed", "failed to count schedule tasks")
|
|
}
|
|
|
|
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.next_run_at, st.status, st.created_at, st.updated_at,
|
|
pr.name AS prompt_rule_name,
|
|
b.name AS brand_name
|
|
FROM schedule_tasks st
|
|
LEFT JOIN prompt_rules pr ON pr.id = st.prompt_rule_id
|
|
LEFT JOIN brands b ON b.id = st.brand_id
|
|
WHERE st.tenant_id = $1 AND st.deleted_at IS NULL
|
|
AND ($2::bigint IS NULL OR st.prompt_rule_id = $2)
|
|
AND ($3::text IS NULL OR st.status = $3)
|
|
AND ($4::text IS NULL OR st.name ILIKE '%' || $4 || '%')
|
|
AND ($5::timestamptz IS NULL OR st.created_at >= $5)
|
|
AND ($6::timestamptz IS NULL OR st.created_at <= $6)
|
|
ORDER BY st.created_at DESC
|
|
LIMIT $7 OFFSET $8
|
|
`, actor.TenantID, params.PromptRuleID, params.Status, params.Keyword, params.CreatedFrom, params.CreatedTo, params.PageSize, offset)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50010, "query_failed", "failed to list schedule tasks")
|
|
}
|
|
defer rows.Close()
|
|
|
|
var items []ScheduleTaskResponse
|
|
for rows.Next() {
|
|
var r ScheduleTaskResponse
|
|
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,
|
|
&nextRunAt, &r.Status, &ca, &ua,
|
|
&r.PromptRuleName, &r.BrandName); err != nil {
|
|
return nil, response.ErrInternal(50010, "scan_failed", err.Error())
|
|
}
|
|
r.CreatedAt = fmt.Sprintf("%v", ca)
|
|
r.UpdatedAt = fmt.Sprintf("%v", ua)
|
|
if startAt != nil {
|
|
s := fmt.Sprintf("%v", startAt)
|
|
r.StartAt = &s
|
|
}
|
|
if endAt != nil {
|
|
s := fmt.Sprintf("%v", endAt)
|
|
r.EndAt = &s
|
|
}
|
|
if nextRunAt != nil {
|
|
s := fmt.Sprintf("%v", nextRunAt)
|
|
r.NextRunAt = &s
|
|
}
|
|
items = append(items, r)
|
|
}
|
|
if items == nil {
|
|
items = []ScheduleTaskResponse{}
|
|
}
|
|
|
|
return &ScheduleTaskListResponse{Items: items, Total: total}, nil
|
|
}
|
|
|
|
func (s *ScheduleTaskService) Detail(ctx context.Context, id int64) (*ScheduleTaskResponse, error) {
|
|
actor := auth.MustActor(ctx)
|
|
var r ScheduleTaskResponse
|
|
var ca, ua interface{}
|
|
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.next_run_at, st.status, st.created_at, st.updated_at,
|
|
pr.name AS prompt_rule_name,
|
|
b.name AS brand_name
|
|
FROM schedule_tasks st
|
|
LEFT JOIN prompt_rules pr ON pr.id = st.prompt_rule_id
|
|
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,
|
|
&nextRunAt, &r.Status, &ca, &ua,
|
|
&r.PromptRuleName, &r.BrandName)
|
|
if err != nil {
|
|
return nil, response.ErrNotFound(40432, "schedule_not_found", "schedule task not found")
|
|
}
|
|
r.CreatedAt = fmt.Sprintf("%v", ca)
|
|
r.UpdatedAt = fmt.Sprintf("%v", ua)
|
|
if startAt != nil {
|
|
sv := fmt.Sprintf("%v", startAt)
|
|
r.StartAt = &sv
|
|
}
|
|
if endAt != nil {
|
|
sv := fmt.Sprintf("%v", endAt)
|
|
r.EndAt = &sv
|
|
}
|
|
if nextRunAt != nil {
|
|
sv := fmt.Sprintf("%v", nextRunAt)
|
|
r.NextRunAt = &sv
|
|
}
|
|
return &r, nil
|
|
}
|
|
|
|
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")
|
|
}
|
|
|
|
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)
|
|
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})
|
|
result := "success"
|
|
resourceType := "schedule_task"
|
|
requestID := middleware.RequestIDFromContext(ctx)
|
|
s.auditLogs.Log(auditlog.Entry{
|
|
OperatorID: actor.UserID,
|
|
TenantID: &actor.TenantID,
|
|
Module: "schedule_task",
|
|
Action: "create",
|
|
ResourceType: &resourceType,
|
|
ResourceID: &id,
|
|
RequestID: nilIfEmptyString(requestID),
|
|
AfterJSON: afterJSON,
|
|
Result: &result,
|
|
})
|
|
|
|
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),
|
|
}, nil
|
|
}
|
|
|
|
func (s *ScheduleTaskService) Update(ctx context.Context, id int64, req ScheduleTaskRequest) error {
|
|
actor := auth.MustActor(ctx)
|
|
tag, err := s.pool.Exec(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 {
|
|
return response.ErrNotFound(40432, "schedule_not_found", "schedule task not found")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *ScheduleTaskService) Delete(ctx context.Context, id int64) error {
|
|
actor := auth.MustActor(ctx)
|
|
tag, err := s.pool.Exec(ctx, `
|
|
UPDATE schedule_tasks SET deleted_at = NOW(), updated_at = NOW()
|
|
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
|
|
`, id, actor.TenantID)
|
|
if err != nil || tag.RowsAffected() == 0 {
|
|
return response.ErrNotFound(40432, "schedule_not_found", "schedule task not found")
|
|
}
|
|
|
|
afterJSON, _ := json.Marshal(map[string]interface{}{"id": id})
|
|
result := "success"
|
|
resourceType := "schedule_task"
|
|
requestID := middleware.RequestIDFromContext(ctx)
|
|
s.auditLogs.Log(auditlog.Entry{
|
|
OperatorID: actor.UserID,
|
|
TenantID: &actor.TenantID,
|
|
Module: "schedule_task",
|
|
Action: "delete",
|
|
ResourceType: &resourceType,
|
|
ResourceID: &id,
|
|
RequestID: nilIfEmptyString(requestID),
|
|
AfterJSON: afterJSON,
|
|
Result: &result,
|
|
})
|
|
return nil
|
|
}
|
|
|
|
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 {
|
|
return response.ErrNotFound(40432, "schedule_not_found", "schedule task not found")
|
|
}
|
|
return nil
|
|
}
|