geo/server/internal/tenant/app/instant_task_service.go

package app

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/geo-platform/tenant-api/internal/shared/auth"
	"github.com/geo-platform/tenant-api/internal/shared/response"
)

type InstantTaskService struct {
	pool *pgxpool.Pool
}

func NewInstantTaskService(pool *pgxpool.Pool) *InstantTaskService {
	return &InstantTaskService{pool: pool}
}

type InstantTaskListParams struct {
	PromptRuleID *int64
	Status       *string
	Keyword      *string
	CreatedFrom  *time.Time
	CreatedTo    *time.Time
	Page         int
	PageSize     int
}

type InstantTaskResponse struct {
	ID             int64    `json:"id"`
	ArticleID      *int64   `json:"article_id"`
	PromptRuleID   *int64   `json:"prompt_rule_id"`
	PromptRuleName *string  `json:"prompt_rule_name"`
	Name           string   `json:"name"`
	Status         string   `json:"status"`
	ExecutionTime  *string  `json:"execution_time"`
	Platforms      []string `json:"platforms"`
	GenerateCount  int      `json:"generate_count"`
	CreatedAt      string   `json:"created_at"`
}

type InstantTaskListResponse struct {
	Items []InstantTaskResponse `json:"items"`
	Total int64                 `json:"total"`
}

func (s *InstantTaskService) List(ctx context.Context, params InstantTaskListParams) (*InstantTaskListResponse, 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 generation_tasks gt
		LEFT JOIN articles a ON a.id = gt.article_id
		LEFT JOIN prompt_rules pr ON pr.id = COALESCE(a.prompt_rule_id, NULLIF(gt.input_params_json ->> 'prompt_rule_id', '')::bigint)
		WHERE gt.tenant_id = $1
		  AND gt.task_type = 'custom_generation'
		  AND ($2::bigint IS NULL OR pr.id = $2)
		  AND ($3::text IS NULL OR gt.status = $3)
		  AND ($4::text IS NULL OR COALESCE(NULLIF(gt.input_params_json ->> 'task_name', ''), pr.name, '') ILIKE '%' || $4 || '%')
		  AND ($5::timestamptz IS NULL OR gt.created_at >= $5)
		  AND ($6::timestamptz IS NULL OR gt.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 instant tasks")
	}

	rows, err := s.pool.Query(ctx, `
		SELECT gt.id, gt.article_id, pr.id, pr.name,
		       COALESCE(NULLIF(gt.input_params_json ->> 'task_name', ''), pr.name, '即时任务') AS task_name,
		       gt.status,
		       COALESCE(gt.started_at, gt.completed_at, gt.created_at) AS execution_time,
		       gt.input_params_json,
		       gt.created_at
		FROM generation_tasks gt
		LEFT JOIN articles a ON a.id = gt.article_id
		LEFT JOIN prompt_rules pr ON pr.id = COALESCE(a.prompt_rule_id, NULLIF(gt.input_params_json ->> 'prompt_rule_id', '')::bigint)
		WHERE gt.tenant_id = $1
		  AND gt.task_type = 'custom_generation'
		  AND ($2::bigint IS NULL OR pr.id = $2)
		  AND ($3::text IS NULL OR gt.status = $3)
		  AND ($4::text IS NULL OR COALESCE(NULLIF(gt.input_params_json ->> 'task_name', ''), pr.name, '') ILIKE '%' || $4 || '%')
		  AND ($5::timestamptz IS NULL OR gt.created_at >= $5)
		  AND ($6::timestamptz IS NULL OR gt.created_at <= $6)
		ORDER BY gt.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 instant tasks")
	}
	defer rows.Close()

	items := make([]InstantTaskResponse, 0)
	for rows.Next() {
		var item InstantTaskResponse
		var executionAt interface{}
		var createdAt interface{}
		var inputParamsJSON []byte
		if err := rows.Scan(
			&item.ID,
			&item.ArticleID,
			&item.PromptRuleID,
			&item.PromptRuleName,
			&item.Name,
			&item.Status,
			&executionAt,
			&inputParamsJSON,
			&createdAt,
		); err != nil {
			return nil, response.ErrInternal(50010, "scan_failed", err.Error())
		}

		item.ExecutionTime = stringPtrFromDBValue(executionAt)
		item.CreatedAt = fmt.Sprintf("%v", createdAt)
		item.Platforms = resolveArticlePlatforms(nil, inputParamsJSON)
		item.GenerateCount = 1

		if len(inputParamsJSON) > 0 {
			payload, err := parseJSONMap(inputParamsJSON)
			if err == nil {
				if count := extractInt(payload, "generate_count"); count > 0 {
					item.GenerateCount = count
				}
			}
		}

		items = append(items, item)
	}

	return &InstantTaskListResponse{
		Items: items,
		Total: total,
	}, 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
	}

	payload := make(map[string]interface{})
	if err := json.Unmarshal(raw, &payload); err != nil {
		return nil, err
	}
	return payload, nil
}