root
7b4d7ccf68
Provide a unified ops console for inspecting, retrying and cancelling jobs across generation, template/kol assist, knowledge parse, desktop publish/task, compliance review and monitoring collect sources. Wires RabbitMQ for retry republish and consolidates the desktop_publish_jobs columns into the base migration. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1746 lines
50 KiB
Go
1746 lines
50 KiB
Go
package repository
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"github.com/geo-platform/tenant-api/internal/ops/domain"
|
|
)
|
|
|
|
var ErrJobNotFound = errors.New("job not found")
|
|
|
|
type JobRepository struct {
|
|
pool *pgxpool.Pool
|
|
monitoringPool *pgxpool.Pool
|
|
}
|
|
|
|
func NewJobRepository(pool, monitoringPool *pgxpool.Pool) *JobRepository {
|
|
return &JobRepository{pool: pool, monitoringPool: monitoringPool}
|
|
}
|
|
|
|
func (r *JobRepository) List(ctx context.Context, f domain.JobFilter) ([]domain.JobSummary, int64, domain.JobCounts, error) {
|
|
limit := f.Limit
|
|
if limit <= 0 || limit > 500 {
|
|
limit = 50
|
|
}
|
|
offset := f.Offset
|
|
if offset < 0 {
|
|
offset = 0
|
|
}
|
|
|
|
businessItems, businessErr := r.listBusinessJobs(ctx, f)
|
|
if businessErr != nil {
|
|
return nil, 0, domain.JobCounts{}, businessErr
|
|
}
|
|
monitoringItems, monitoringErr := r.listMonitoringJobs(ctx, f)
|
|
if monitoringErr != nil {
|
|
return nil, 0, domain.JobCounts{}, monitoringErr
|
|
}
|
|
|
|
all := append(businessItems, monitoringItems...)
|
|
sortJobs(all)
|
|
counts := buildJobCounts(all)
|
|
|
|
total := int64(len(all))
|
|
if offset >= len(all) {
|
|
return []domain.JobSummary{}, total, counts, nil
|
|
}
|
|
end := offset + limit
|
|
if end > len(all) {
|
|
end = len(all)
|
|
}
|
|
return all[offset:end], total, counts, nil
|
|
}
|
|
|
|
func (r *JobRepository) Get(ctx context.Context, source, id string) (*domain.JobDetail, error) {
|
|
source = normalizeJobSource(source)
|
|
id = strings.TrimSpace(id)
|
|
switch source {
|
|
case "generation":
|
|
return r.getGenerationJob(ctx, id)
|
|
case "template_assist":
|
|
return r.getTemplateAssistJob(ctx, id)
|
|
case "kol_assist":
|
|
return r.getKolAssistJob(ctx, id)
|
|
case "knowledge_parse":
|
|
return r.getKnowledgeParseJob(ctx, id)
|
|
case "desktop_publish":
|
|
return r.getDesktopPublishJob(ctx, id)
|
|
case "desktop_task":
|
|
return r.getDesktopTaskJob(ctx, id)
|
|
case "compliance_review":
|
|
return r.getComplianceReviewJob(ctx, id)
|
|
case "monitoring_collect":
|
|
return r.getMonitoringCollectJob(ctx, id)
|
|
default:
|
|
return nil, ErrJobNotFound
|
|
}
|
|
}
|
|
|
|
func (r *JobRepository) Retry(ctx context.Context, source, id string) (*domain.JobDetail, error) {
|
|
source = normalizeJobSource(source)
|
|
switch source {
|
|
case "generation":
|
|
return r.retryGenerationJob(ctx, id)
|
|
case "template_assist":
|
|
return r.retryTemplateAssistJob(ctx, id)
|
|
case "kol_assist":
|
|
return r.retryKolAssistJob(ctx, id)
|
|
case "knowledge_parse":
|
|
return r.retryKnowledgeParseJob(ctx, id)
|
|
case "desktop_task":
|
|
return r.retryDesktopTaskJob(ctx, id)
|
|
case "compliance_review":
|
|
return r.retryComplianceReviewJob(ctx, id)
|
|
case "monitoring_collect":
|
|
return r.retryMonitoringCollectJob(ctx, id)
|
|
default:
|
|
return nil, ErrJobNotFound
|
|
}
|
|
}
|
|
|
|
func (r *JobRepository) Cancel(ctx context.Context, source, id, reason string) (*domain.JobDetail, error) {
|
|
source = normalizeJobSource(source)
|
|
reason = strings.TrimSpace(reason)
|
|
if reason == "" {
|
|
reason = "cancelled from ops job console"
|
|
}
|
|
switch source {
|
|
case "generation":
|
|
return r.cancelGenerationJob(ctx, id, reason)
|
|
case "template_assist":
|
|
return r.cancelTemplateAssistJob(ctx, id, reason)
|
|
case "kol_assist":
|
|
return r.cancelKolAssistJob(ctx, id, reason)
|
|
case "knowledge_parse":
|
|
return r.cancelKnowledgeParseJob(ctx, id, reason)
|
|
case "desktop_publish":
|
|
return r.cancelDesktopPublishJob(ctx, id, reason)
|
|
case "desktop_task":
|
|
return r.cancelDesktopTaskJob(ctx, id, reason)
|
|
case "compliance_review":
|
|
return r.cancelComplianceReviewJob(ctx, id, reason)
|
|
case "monitoring_collect":
|
|
return r.cancelMonitoringCollectJob(ctx, id, reason)
|
|
default:
|
|
return nil, ErrJobNotFound
|
|
}
|
|
}
|
|
|
|
func (r *JobRepository) listBusinessJobs(ctx context.Context, f domain.JobFilter) ([]domain.JobSummary, error) {
|
|
if r == nil || r.pool == nil {
|
|
return nil, nil
|
|
}
|
|
args := make([]any, 0)
|
|
sourceFilter := normalizeJobSource(f.Source)
|
|
sourcePredicate := "TRUE"
|
|
if sourceFilter != "" {
|
|
args = append(args, sourceFilter)
|
|
sourcePredicate = fmt.Sprintf("source = $%d", len(args))
|
|
}
|
|
sourceArg := len(args) + 1
|
|
args = append(args, nullableTextValue(sourceFilter))
|
|
phaseArg := len(args) + 1
|
|
args = append(args, nullableTextValue(f.Phase))
|
|
statusArg := len(args) + 1
|
|
args = append(args, nullableTextValue(f.Status))
|
|
kindArg := len(args) + 1
|
|
args = append(args, nullableTextValue(f.Kind))
|
|
keywordArg := len(args) + 1
|
|
args = append(args, nullableKeyword(f.Keyword))
|
|
tenantArg := len(args) + 1
|
|
args = append(args, f.TenantID)
|
|
workspaceArg := len(args) + 1
|
|
args = append(args, f.WorkspaceID)
|
|
userArg := len(args) + 1
|
|
args = append(args, f.UserID)
|
|
startArg := len(args) + 1
|
|
args = append(args, f.StartAt)
|
|
endArg := len(args) + 1
|
|
args = append(args, f.EndAt)
|
|
|
|
rows, err := r.pool.Query(ctx, fmt.Sprintf(`
|
|
WITH jobs AS (
|
|
%s
|
|
)
|
|
SELECT
|
|
source, id, numeric_id, uuid, kind, status, phase,
|
|
tenant_id, tenant_name, workspace_id, workspace_name, user_id, user_name,
|
|
title, target, queue, worker, attempts, priority, error_message,
|
|
created_at, updated_at, started_at, completed_at, lease_expires_at,
|
|
available_at, last_heartbeat_at
|
|
FROM jobs
|
|
WHERE %s
|
|
AND ($%d::text IS NULL OR source = $%d)
|
|
AND ($%d::text IS NULL OR phase = $%d)
|
|
AND ($%d::text IS NULL OR status = $%d)
|
|
AND ($%d::text IS NULL OR kind ILIKE '%%' || $%d || '%%')
|
|
AND ($%d::text IS NULL OR
|
|
id ILIKE '%%' || $%d || '%%' OR
|
|
COALESCE(uuid, '') ILIKE '%%' || $%d || '%%' OR
|
|
COALESCE(title, '') ILIKE '%%' || $%d || '%%' OR
|
|
COALESCE(target, '') ILIKE '%%' || $%d || '%%' OR
|
|
COALESCE(error_message, '') ILIKE '%%' || $%d || '%%' OR
|
|
COALESCE(tenant_name, '') ILIKE '%%' || $%d || '%%' OR
|
|
COALESCE(workspace_name, '') ILIKE '%%' || $%d || '%%' OR
|
|
COALESCE(user_name, '') ILIKE '%%' || $%d || '%%')
|
|
AND ($%d::bigint IS NULL OR tenant_id = $%d)
|
|
AND ($%d::bigint IS NULL OR workspace_id = $%d)
|
|
AND ($%d::bigint IS NULL OR user_id = $%d)
|
|
AND ($%d::timestamptz IS NULL OR created_at >= $%d)
|
|
AND ($%d::timestamptz IS NULL OR created_at < $%d)
|
|
ORDER BY created_at DESC, source ASC, id DESC
|
|
LIMIT 2000`,
|
|
businessJobUnionSQL(),
|
|
sourcePredicate,
|
|
sourceArg, sourceArg,
|
|
phaseArg, phaseArg,
|
|
statusArg, statusArg,
|
|
kindArg, kindArg,
|
|
keywordArg, keywordArg, keywordArg, keywordArg, keywordArg, keywordArg, keywordArg, keywordArg, keywordArg,
|
|
tenantArg, tenantArg,
|
|
workspaceArg, workspaceArg,
|
|
userArg, userArg,
|
|
startArg, startArg,
|
|
endArg, endArg,
|
|
), args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
items := make([]domain.JobSummary, 0)
|
|
for rows.Next() {
|
|
item, err := scanJobSummary(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, *item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (r *JobRepository) listMonitoringJobs(ctx context.Context, f domain.JobFilter) ([]domain.JobSummary, error) {
|
|
if r == nil || r.monitoringPool == nil {
|
|
return nil, nil
|
|
}
|
|
sourceFilter := normalizeJobSource(f.Source)
|
|
if sourceFilter != "" && sourceFilter != "monitoring_collect" {
|
|
return nil, nil
|
|
}
|
|
|
|
args := make([]any, 0)
|
|
phaseArg := len(args) + 1
|
|
args = append(args, nullableTextValue(f.Phase))
|
|
statusArg := len(args) + 1
|
|
args = append(args, nullableTextValue(f.Status))
|
|
kindArg := len(args) + 1
|
|
args = append(args, nullableTextValue(f.Kind))
|
|
keywordArg := len(args) + 1
|
|
args = append(args, nullableKeyword(f.Keyword))
|
|
tenantArg := len(args) + 1
|
|
args = append(args, f.TenantID)
|
|
workspaceArg := len(args) + 1
|
|
args = append(args, f.WorkspaceID)
|
|
userArg := len(args) + 1
|
|
args = append(args, f.UserID)
|
|
startArg := len(args) + 1
|
|
args = append(args, f.StartAt)
|
|
endArg := len(args) + 1
|
|
args = append(args, f.EndAt)
|
|
|
|
rows, err := r.monitoringPool.Query(ctx, fmt.Sprintf(`
|
|
WITH jobs AS (
|
|
SELECT
|
|
'monitoring_collect'::text AS source,
|
|
t.id::text AS id,
|
|
t.id::bigint AS numeric_id,
|
|
NULL::text AS uuid,
|
|
('monitor:' || t.ai_platform_id || ':' || t.run_mode)::text AS kind,
|
|
t.status::text AS status,
|
|
CASE
|
|
WHEN t.status IN ('pending', 'accepted', 'dispatching') THEN 'queued'
|
|
WHEN t.status IN ('leased', 'running', 'received') THEN 'running'
|
|
WHEN t.status IN ('completed', 'succeeded') THEN 'succeeded'
|
|
WHEN t.status IN ('failed', 'expired') THEN 'failed'
|
|
WHEN t.status IN ('skipped', 'superseded') THEN 'canceled'
|
|
ELSE 'unknown'
|
|
END AS phase,
|
|
t.tenant_id::bigint AS tenant_id,
|
|
NULL::text AS tenant_name,
|
|
r.workspace_id::bigint AS workspace_id,
|
|
NULL::text AS workspace_name,
|
|
r.requested_by_user_id::bigint AS user_id,
|
|
NULL::text AS user_name,
|
|
('Brand #' || t.brand_id || ' / Question #' || t.question_id)::text AS title,
|
|
(t.ai_platform_id || ' / ' || t.business_date::text)::text AS target,
|
|
('monitoring:' || COALESCE(t.dispatch_lane, 'normal'))::text AS queue,
|
|
COALESCE(t.leased_to_executor, t.target_client_id::text)::text AS worker,
|
|
COALESCE(t.dispatch_attempts, 0)::int AS attempts,
|
|
COALESCE(t.dispatch_priority, 100)::int AS priority,
|
|
NULLIF(COALESCE(t.error_message, t.skip_reason), '')::text AS error_message,
|
|
t.created_at,
|
|
t.updated_at,
|
|
t.leased_at AS started_at,
|
|
COALESCE(t.completed_at, t.callback_received_at) AS completed_at,
|
|
t.lease_expires_at,
|
|
COALESCE(t.dispatch_after, t.planned_at) AS available_at,
|
|
NULL::timestamptz AS last_heartbeat_at
|
|
FROM monitoring_collect_tasks t
|
|
LEFT JOIN monitoring_collect_requests r ON r.request_id = t.superseded_by_request_id
|
|
)
|
|
SELECT
|
|
source, id, numeric_id, uuid, kind, status, phase,
|
|
tenant_id, tenant_name, workspace_id, workspace_name, user_id, user_name,
|
|
title, target, queue, worker, attempts, priority, error_message,
|
|
created_at, updated_at, started_at, completed_at, lease_expires_at,
|
|
available_at, last_heartbeat_at
|
|
FROM jobs
|
|
WHERE ($%d::text IS NULL OR phase = $%d)
|
|
AND ($%d::text IS NULL OR status = $%d)
|
|
AND ($%d::text IS NULL OR kind ILIKE '%%' || $%d || '%%')
|
|
AND ($%d::text IS NULL OR
|
|
id ILIKE '%%' || $%d || '%%' OR
|
|
COALESCE(title, '') ILIKE '%%' || $%d || '%%' OR
|
|
COALESCE(target, '') ILIKE '%%' || $%d || '%%' OR
|
|
COALESCE(error_message, '') ILIKE '%%' || $%d || '%%')
|
|
AND ($%d::bigint IS NULL OR tenant_id = $%d)
|
|
AND ($%d::bigint IS NULL OR workspace_id = $%d)
|
|
AND ($%d::bigint IS NULL OR user_id = $%d)
|
|
AND ($%d::timestamptz IS NULL OR created_at >= $%d)
|
|
AND ($%d::timestamptz IS NULL OR created_at < $%d)
|
|
ORDER BY created_at DESC, id DESC
|
|
LIMIT 2000`,
|
|
phaseArg, phaseArg,
|
|
statusArg, statusArg,
|
|
kindArg, kindArg,
|
|
keywordArg, keywordArg, keywordArg, keywordArg, keywordArg,
|
|
tenantArg, tenantArg,
|
|
workspaceArg, workspaceArg,
|
|
userArg, userArg,
|
|
startArg, startArg,
|
|
endArg, endArg,
|
|
), args...)
|
|
if err != nil {
|
|
if isUndefinedMonitoringRelation(err) {
|
|
return nil, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
items := make([]domain.JobSummary, 0)
|
|
for rows.Next() {
|
|
item, err := scanJobSummary(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, *item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func businessJobUnionSQL() string {
|
|
return strings.Join([]string{
|
|
generationJobSelectSQL(),
|
|
templateAssistJobSelectSQL(),
|
|
kolAssistJobSelectSQL(),
|
|
knowledgeParseJobSelectSQL(),
|
|
desktopPublishJobSelectSQL(),
|
|
desktopTaskJobSelectSQL(),
|
|
complianceReviewJobSelectSQL(),
|
|
}, "\nUNION ALL\n")
|
|
}
|
|
|
|
func generationJobSelectSQL() string {
|
|
return `
|
|
SELECT
|
|
'generation'::text AS source,
|
|
gt.id::text AS id,
|
|
gt.id::bigint AS numeric_id,
|
|
NULL::text AS uuid,
|
|
gt.task_type::text AS kind,
|
|
gt.status::text AS status,
|
|
CASE
|
|
WHEN gt.status = 'queued' THEN 'queued'
|
|
WHEN gt.status = 'running' THEN 'running'
|
|
WHEN gt.status = 'completed' THEN 'succeeded'
|
|
WHEN gt.status = 'failed' THEN 'failed'
|
|
WHEN gt.status IN ('cancelled', 'canceled', 'aborted') THEN 'canceled'
|
|
ELSE 'unknown'
|
|
END AS phase,
|
|
gt.tenant_id::bigint AS tenant_id,
|
|
t.name::text AS tenant_name,
|
|
w.id::bigint AS workspace_id,
|
|
w.name::text AS workspace_name,
|
|
gt.operator_id::bigint AS user_id,
|
|
u.name::text AS user_name,
|
|
COALESCE(av.title, 'Article #' || gt.article_id::text, 'Generation task #' || gt.id::text)::text AS title,
|
|
CASE WHEN gt.article_id IS NOT NULL THEN 'article #' || gt.article_id::text ELSE NULL END::text AS target,
|
|
'generation.task.run'::text AS queue,
|
|
gt.lease_owner::text AS worker,
|
|
COALESCE(gt.attempt_count, 0)::int AS attempts,
|
|
NULL::int AS priority,
|
|
NULLIF(gt.error_message, '')::text AS error_message,
|
|
gt.created_at,
|
|
gt.updated_at,
|
|
gt.started_at,
|
|
gt.completed_at,
|
|
gt.lease_expires_at,
|
|
NULL::timestamptz AS available_at,
|
|
gt.last_heartbeat_at
|
|
FROM generation_tasks gt
|
|
LEFT JOIN tenants t ON t.id = gt.tenant_id
|
|
LEFT JOIN users u ON u.id = gt.operator_id
|
|
LEFT JOIN articles a ON a.id = gt.article_id
|
|
LEFT JOIN article_versions av ON av.id = a.current_version_id
|
|
LEFT JOIN LATERAL (
|
|
SELECT wm.workspace_id AS id, ws.name
|
|
FROM workspace_memberships wm
|
|
JOIN workspaces ws ON ws.id = wm.workspace_id
|
|
WHERE wm.tenant_id = gt.tenant_id
|
|
AND wm.user_id = gt.operator_id
|
|
AND wm.is_primary = TRUE
|
|
ORDER BY wm.created_at ASC, wm.id ASC
|
|
LIMIT 1
|
|
) w ON TRUE`
|
|
}
|
|
|
|
func templateAssistJobSelectSQL() string {
|
|
return `
|
|
SELECT
|
|
'template_assist'::text AS source,
|
|
tat.id::text AS id,
|
|
NULL::bigint AS numeric_id,
|
|
tat.id::text AS uuid,
|
|
tat.task_type::text AS kind,
|
|
tat.status::text AS status,
|
|
CASE
|
|
WHEN tat.status = 'queued' THEN 'queued'
|
|
WHEN tat.status = 'running' THEN 'running'
|
|
WHEN tat.status = 'completed' THEN 'succeeded'
|
|
WHEN tat.status = 'failed' THEN 'failed'
|
|
WHEN tat.status IN ('cancelled', 'canceled', 'aborted') THEN 'canceled'
|
|
ELSE 'unknown'
|
|
END AS phase,
|
|
tat.tenant_id::bigint AS tenant_id,
|
|
t.name::text AS tenant_name,
|
|
w.id::bigint AS workspace_id,
|
|
w.name::text AS workspace_name,
|
|
tat.operator_id::bigint AS user_id,
|
|
u.name::text AS user_name,
|
|
COALESCE(at.template_name, 'Template #' || tat.template_id::text)::text AS title,
|
|
('template #' || tat.template_id::text)::text AS target,
|
|
'template.assist.run'::text AS queue,
|
|
NULL::text AS worker,
|
|
NULL::int AS attempts,
|
|
NULL::int AS priority,
|
|
NULLIF(tat.error_message, '')::text AS error_message,
|
|
tat.created_at,
|
|
tat.updated_at,
|
|
tat.started_at,
|
|
tat.completed_at,
|
|
NULL::timestamptz AS lease_expires_at,
|
|
NULL::timestamptz AS available_at,
|
|
NULL::timestamptz AS last_heartbeat_at
|
|
FROM template_assist_tasks tat
|
|
LEFT JOIN tenants t ON t.id = tat.tenant_id
|
|
LEFT JOIN users u ON u.id = tat.operator_id
|
|
LEFT JOIN article_templates at ON at.id = tat.template_id
|
|
LEFT JOIN LATERAL (
|
|
SELECT wm.workspace_id AS id, ws.name
|
|
FROM workspace_memberships wm
|
|
JOIN workspaces ws ON ws.id = wm.workspace_id
|
|
WHERE wm.tenant_id = tat.tenant_id
|
|
AND wm.user_id = tat.operator_id
|
|
AND wm.is_primary = TRUE
|
|
ORDER BY wm.created_at ASC, wm.id ASC
|
|
LIMIT 1
|
|
) w ON TRUE`
|
|
}
|
|
|
|
func kolAssistJobSelectSQL() string {
|
|
return `
|
|
SELECT
|
|
'kol_assist'::text AS source,
|
|
kat.id::text AS id,
|
|
NULL::bigint AS numeric_id,
|
|
kat.id::text AS uuid,
|
|
kat.task_type::text AS kind,
|
|
kat.status::text AS status,
|
|
CASE
|
|
WHEN kat.status = 'queued' THEN 'queued'
|
|
WHEN kat.status = 'running' THEN 'running'
|
|
WHEN kat.status = 'completed' THEN 'succeeded'
|
|
WHEN kat.status = 'failed' THEN 'failed'
|
|
WHEN kat.status IN ('cancelled', 'canceled', 'aborted') THEN 'canceled'
|
|
ELSE 'unknown'
|
|
END AS phase,
|
|
kat.tenant_id::bigint AS tenant_id,
|
|
t.name::text AS tenant_name,
|
|
w.id::bigint AS workspace_id,
|
|
w.name::text AS workspace_name,
|
|
kat.operator_id::bigint AS user_id,
|
|
u.name::text AS user_name,
|
|
COALESCE(kpr.name, kp.display_name, 'KOL assist #' || kat.id)::text AS title,
|
|
COALESCE(kp.display_name, 'profile #' || kat.kol_profile_id::text)::text AS target,
|
|
'kol.assist.run'::text AS queue,
|
|
NULL::text AS worker,
|
|
NULL::int AS attempts,
|
|
NULL::int AS priority,
|
|
NULLIF(kat.error_message, '')::text AS error_message,
|
|
kat.created_at,
|
|
kat.updated_at,
|
|
kat.started_at,
|
|
kat.completed_at,
|
|
NULL::timestamptz AS lease_expires_at,
|
|
NULL::timestamptz AS available_at,
|
|
NULL::timestamptz AS last_heartbeat_at
|
|
FROM kol_assist_tasks kat
|
|
LEFT JOIN tenants t ON t.id = kat.tenant_id
|
|
LEFT JOIN users u ON u.id = kat.operator_id
|
|
LEFT JOIN kol_profiles kp ON kp.id = kat.kol_profile_id
|
|
LEFT JOIN kol_prompts kpr ON kpr.id = kat.prompt_id
|
|
LEFT JOIN LATERAL (
|
|
SELECT wm.workspace_id AS id, ws.name
|
|
FROM workspace_memberships wm
|
|
JOIN workspaces ws ON ws.id = wm.workspace_id
|
|
WHERE wm.tenant_id = kat.tenant_id
|
|
AND wm.user_id = kat.operator_id
|
|
AND wm.is_primary = TRUE
|
|
ORDER BY wm.created_at ASC, wm.id ASC
|
|
LIMIT 1
|
|
) w ON TRUE`
|
|
}
|
|
|
|
func knowledgeParseJobSelectSQL() string {
|
|
return `
|
|
SELECT
|
|
'knowledge_parse'::text AS source,
|
|
kpt.id::text AS id,
|
|
kpt.id::bigint AS numeric_id,
|
|
NULL::text AS uuid,
|
|
('knowledge:' || kpt.source_type)::text AS kind,
|
|
kpt.status::text AS status,
|
|
CASE
|
|
WHEN kpt.status = 'pending' THEN 'queued'
|
|
WHEN kpt.status = 'running' THEN 'running'
|
|
WHEN kpt.status = 'completed' THEN 'succeeded'
|
|
WHEN kpt.status = 'failed' THEN 'failed'
|
|
WHEN kpt.status IN ('deleted', 'cancelled', 'canceled', 'aborted') THEN 'canceled'
|
|
ELSE 'unknown'
|
|
END AS phase,
|
|
kpt.tenant_id::bigint AS tenant_id,
|
|
t.name::text AS tenant_name,
|
|
NULL::bigint AS workspace_id,
|
|
NULL::text AS workspace_name,
|
|
NULL::bigint AS user_id,
|
|
NULL::text AS user_name,
|
|
COALESCE(ki.name, 'Knowledge item #' || kpt.knowledge_item_id::text)::text AS title,
|
|
('knowledge item #' || kpt.knowledge_item_id::text)::text AS target,
|
|
'in_process.knowledge_parse'::text AS queue,
|
|
NULL::text AS worker,
|
|
NULL::int AS attempts,
|
|
NULL::int AS priority,
|
|
NULLIF(kpt.error_message, '')::text AS error_message,
|
|
kpt.created_at,
|
|
kpt.updated_at,
|
|
kpt.started_at,
|
|
kpt.completed_at,
|
|
NULL::timestamptz AS lease_expires_at,
|
|
NULL::timestamptz AS available_at,
|
|
NULL::timestamptz AS last_heartbeat_at
|
|
FROM knowledge_parse_tasks kpt
|
|
LEFT JOIN tenants t ON t.id = kpt.tenant_id
|
|
LEFT JOIN knowledge_items ki ON ki.id = kpt.knowledge_item_id`
|
|
}
|
|
|
|
func desktopPublishJobSelectSQL() string {
|
|
return `
|
|
SELECT
|
|
'desktop_publish'::text AS source,
|
|
j.desktop_id::text AS id,
|
|
j.id::bigint AS numeric_id,
|
|
j.desktop_id::text AS uuid,
|
|
'publish_job'::text AS kind,
|
|
COALESCE(j.status, 'queued')::text AS status,
|
|
CASE
|
|
WHEN COALESCE(j.status, 'queued') = 'queued' THEN 'queued'
|
|
WHEN j.status = 'blocked_by_compliance' THEN 'blocked'
|
|
WHEN j.status IN ('completed', 'succeeded') THEN 'succeeded'
|
|
WHEN j.status = 'failed' THEN 'failed'
|
|
WHEN j.status IN ('cancelled', 'canceled', 'aborted') THEN 'canceled'
|
|
ELSE 'unknown'
|
|
END AS phase,
|
|
j.tenant_id::bigint AS tenant_id,
|
|
t.name::text AS tenant_name,
|
|
j.workspace_id::bigint AS workspace_id,
|
|
w.name::text AS workspace_name,
|
|
j.created_by_user_id::bigint AS user_id,
|
|
u.name::text AS user_name,
|
|
j.title::text AS title,
|
|
CASE WHEN j.article_id IS NOT NULL THEN 'article #' || j.article_id::text ELSE NULL END::text AS target,
|
|
'desktop.publish'::text AS queue,
|
|
NULL::text AS worker,
|
|
NULL::int AS attempts,
|
|
NULL::int AS priority,
|
|
NULLIF(j.compliance_blocked_reason, '')::text AS error_message,
|
|
j.created_at,
|
|
j.updated_at,
|
|
NULL::timestamptz AS started_at,
|
|
NULL::timestamptz AS completed_at,
|
|
NULL::timestamptz AS lease_expires_at,
|
|
j.scheduled_at AS available_at,
|
|
NULL::timestamptz AS last_heartbeat_at
|
|
FROM desktop_publish_jobs j
|
|
LEFT JOIN tenants t ON t.id = j.tenant_id
|
|
LEFT JOIN workspaces w ON w.id = j.workspace_id
|
|
LEFT JOIN users u ON u.id = j.created_by_user_id`
|
|
}
|
|
|
|
func desktopTaskJobSelectSQL() string {
|
|
return `
|
|
SELECT
|
|
'desktop_task'::text AS source,
|
|
dt.desktop_id::text AS id,
|
|
dt.id::bigint AS numeric_id,
|
|
dt.desktop_id::text AS uuid,
|
|
dt.kind::text AS kind,
|
|
dt.status::text AS status,
|
|
CASE
|
|
WHEN dt.status = 'queued' THEN 'queued'
|
|
WHEN dt.status = 'in_progress' THEN 'running'
|
|
WHEN dt.status = 'succeeded' THEN 'succeeded'
|
|
WHEN dt.status = 'failed' THEN 'failed'
|
|
WHEN dt.status = 'aborted' THEN 'canceled'
|
|
WHEN dt.status = 'unknown' THEN 'unknown'
|
|
ELSE 'unknown'
|
|
END AS phase,
|
|
dt.tenant_id::bigint AS tenant_id,
|
|
t.name::text AS tenant_name,
|
|
dt.workspace_id::bigint AS workspace_id,
|
|
w.name::text AS workspace_name,
|
|
dc.user_id::bigint AS user_id,
|
|
u.name::text AS user_name,
|
|
COALESCE(j.title, dt.kind || ' task #' || dt.id::text)::text AS title,
|
|
COALESCE(mp.name, dt.platform_id, dt.target_client_id::text)::text AS target,
|
|
('desktop.' || COALESCE(dt.lane, 'normal'))::text AS queue,
|
|
dt.target_client_id::text AS worker,
|
|
COALESCE(dt.attempts, 0)::int AS attempts,
|
|
COALESCE(dt.priority, 0)::int AS priority,
|
|
NULLIF(dt.error::text, '')::text AS error_message,
|
|
dt.created_at,
|
|
dt.updated_at,
|
|
dt.started_at,
|
|
CASE WHEN dt.status IN ('succeeded','failed','aborted') THEN dt.updated_at ELSE NULL END AS completed_at,
|
|
dt.lease_expires_at,
|
|
dt.enqueued_at AS available_at,
|
|
NULL::timestamptz AS last_heartbeat_at
|
|
FROM desktop_tasks dt
|
|
LEFT JOIN desktop_publish_jobs j ON j.desktop_id = dt.job_id
|
|
LEFT JOIN tenants t ON t.id = dt.tenant_id
|
|
LEFT JOIN workspaces w ON w.id = dt.workspace_id
|
|
LEFT JOIN desktop_clients dc ON dc.id = dt.target_client_id
|
|
LEFT JOIN users u ON u.id = dc.user_id
|
|
LEFT JOIN media_platforms mp ON mp.platform_id = dt.platform_id`
|
|
}
|
|
|
|
func complianceReviewJobSelectSQL() string {
|
|
return `
|
|
SELECT
|
|
'compliance_review'::text AS source,
|
|
crj.id::text AS id,
|
|
crj.id::bigint AS numeric_id,
|
|
crj.message_id::text AS uuid,
|
|
'compliance_review'::text AS kind,
|
|
crj.status::text AS status,
|
|
CASE
|
|
WHEN crj.status = 'pending' THEN 'queued'
|
|
WHEN crj.status = 'running' THEN 'running'
|
|
WHEN crj.status = 'done' THEN 'succeeded'
|
|
WHEN crj.status = 'failed' THEN 'failed'
|
|
ELSE 'unknown'
|
|
END AS phase,
|
|
crj.tenant_id::bigint AS tenant_id,
|
|
t.name::text AS tenant_name,
|
|
w.id::bigint AS workspace_id,
|
|
w.name::text AS workspace_name,
|
|
ccr.checked_by::bigint AS user_id,
|
|
u.name::text AS user_name,
|
|
COALESCE(av.title, 'Compliance review #' || crj.id::text)::text AS title,
|
|
('check record #' || crj.check_record_id::text)::text AS target,
|
|
'compliance.review.run'::text AS queue,
|
|
crj.locked_by::text AS worker,
|
|
COALESCE(crj.attempts, 0)::int AS attempts,
|
|
NULL::int AS priority,
|
|
NULLIF(crj.last_error, '')::text AS error_message,
|
|
crj.created_at,
|
|
crj.updated_at,
|
|
crj.locked_at AS started_at,
|
|
CASE WHEN crj.status IN ('done','failed') THEN crj.updated_at ELSE NULL END AS completed_at,
|
|
NULL::timestamptz AS lease_expires_at,
|
|
crj.available_at,
|
|
NULL::timestamptz AS last_heartbeat_at
|
|
FROM compliance_review_jobs crj
|
|
LEFT JOIN tenants t ON t.id = crj.tenant_id
|
|
LEFT JOIN compliance_check_records ccr ON ccr.id = crj.check_record_id
|
|
LEFT JOIN users u ON u.id = ccr.checked_by
|
|
LEFT JOIN article_versions av ON av.id = crj.article_version_id
|
|
LEFT JOIN LATERAL (
|
|
SELECT wm.workspace_id AS id, ws.name
|
|
FROM workspace_memberships wm
|
|
JOIN workspaces ws ON ws.id = wm.workspace_id
|
|
WHERE wm.tenant_id = crj.tenant_id
|
|
AND wm.user_id = ccr.checked_by
|
|
AND wm.is_primary = TRUE
|
|
ORDER BY wm.created_at ASC, wm.id ASC
|
|
LIMIT 1
|
|
) w ON TRUE`
|
|
}
|
|
|
|
func (r *JobRepository) getGenerationJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getBusinessJob(ctx, "generation", jobID, `
|
|
SELECT
|
|
gt.input_params_json,
|
|
NULL::jsonb,
|
|
CASE WHEN gt.error_message IS NULL OR gt.error_message = '' THEN NULL ELSE to_jsonb(gt.error_message) END,
|
|
jsonb_strip_nulls(jsonb_build_object(
|
|
'article_id', gt.article_id,
|
|
'quota_reservation_id', gt.quota_reservation_id,
|
|
'request_hash', gt.request_hash,
|
|
'task_batch_id', gt.task_batch_id
|
|
))
|
|
FROM generation_tasks gt
|
|
WHERE gt.id = $1`, jobID)
|
|
}
|
|
|
|
func (r *JobRepository) getTemplateAssistJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
return r.getBusinessJobByTextID(ctx, "template_assist", id, `
|
|
SELECT
|
|
tat.request_json,
|
|
tat.result_json,
|
|
CASE WHEN tat.error_message IS NULL OR tat.error_message = '' THEN NULL ELSE to_jsonb(tat.error_message) END,
|
|
jsonb_strip_nulls(jsonb_build_object(
|
|
'template_id', tat.template_id,
|
|
'template_key', at.template_key,
|
|
'template_name', at.template_name,
|
|
'analyze_prompt', COALESCE(
|
|
at.card_config_json #>> '{wizard,structure,analyze_prompt_template}',
|
|
at.card_config_json #>> '{wizard,analyze_prompt_template}'
|
|
),
|
|
'title_prompt', COALESCE(
|
|
at.card_config_json #>> '{wizard,structure,title_prompt_template}',
|
|
at.card_config_json #>> '{wizard,title_prompt_template}'
|
|
),
|
|
'outline_prompt', COALESCE(
|
|
at.card_config_json #>> '{wizard,structure,outline_prompt_template}',
|
|
at.card_config_json #>> '{wizard,outline_prompt_template}'
|
|
)
|
|
))
|
|
FROM template_assist_tasks tat
|
|
LEFT JOIN article_templates at ON at.id = tat.template_id
|
|
WHERE tat.id = $1`)
|
|
}
|
|
|
|
func (r *JobRepository) getKolAssistJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
return r.getBusinessJobByTextID(ctx, "kol_assist", id, `
|
|
SELECT
|
|
kat.request_json,
|
|
kat.result_json,
|
|
CASE WHEN kat.error_message IS NULL OR kat.error_message = '' THEN NULL ELSE to_jsonb(kat.error_message) END,
|
|
jsonb_strip_nulls(jsonb_build_object(
|
|
'kol_profile_id', kat.kol_profile_id,
|
|
'prompt_id', kat.prompt_id
|
|
))
|
|
FROM kol_assist_tasks kat
|
|
WHERE kat.id = $1`)
|
|
}
|
|
|
|
func (r *JobRepository) getKnowledgeParseJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getBusinessJob(ctx, "knowledge_parse", jobID, `
|
|
SELECT
|
|
jsonb_strip_nulls(jsonb_build_object(
|
|
'knowledge_item_id', kpt.knowledge_item_id,
|
|
'source_type', kpt.source_type,
|
|
'source_uri', ki.source_uri,
|
|
'storage_key', ki.storage_key
|
|
)),
|
|
NULL::jsonb,
|
|
CASE WHEN kpt.error_message IS NULL OR kpt.error_message = '' THEN NULL ELSE to_jsonb(kpt.error_message) END,
|
|
jsonb_strip_nulls(jsonb_build_object(
|
|
'group_id', ki.group_id,
|
|
'item_status', ki.status,
|
|
'size_bytes', ki.size_bytes,
|
|
'item_version', ki.item_version
|
|
))
|
|
FROM knowledge_parse_tasks kpt
|
|
LEFT JOIN knowledge_items ki ON ki.id = kpt.knowledge_item_id
|
|
WHERE kpt.id = $1`, jobID)
|
|
}
|
|
|
|
func (r *JobRepository) getDesktopPublishJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
return r.getBusinessJobByTextID(ctx, "desktop_publish", id, `
|
|
SELECT
|
|
j.content_ref,
|
|
NULL::jsonb,
|
|
CASE WHEN j.compliance_blocked_reason IS NULL OR j.compliance_blocked_reason = '' THEN NULL ELSE to_jsonb(j.compliance_blocked_reason) END,
|
|
jsonb_strip_nulls(jsonb_build_object(
|
|
'article_id', j.article_id,
|
|
'article_version_id', j.article_version_id,
|
|
'compliance_blocked_record_id', j.compliance_blocked_record_id,
|
|
'compliance_blocked_at', j.compliance_blocked_at,
|
|
'scheduled_at', j.scheduled_at
|
|
))
|
|
FROM desktop_publish_jobs j
|
|
WHERE j.desktop_id::text = $1`)
|
|
}
|
|
|
|
func (r *JobRepository) getDesktopTaskJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
return r.getBusinessJobByTextID(ctx, "desktop_task", id, `
|
|
SELECT
|
|
dt.payload,
|
|
dt.result,
|
|
dt.error,
|
|
jsonb_strip_nulls(jsonb_build_object(
|
|
'job_id', dt.job_id,
|
|
'target_account_id', dt.target_account_id,
|
|
'target_client_id', dt.target_client_id,
|
|
'platform_id', dt.platform_id,
|
|
'dedup_key', dt.dedup_key,
|
|
'active_attempt_id', dt.active_attempt_id,
|
|
'lane', dt.lane,
|
|
'lane_weight', dt.lane_weight,
|
|
'source', dt.source,
|
|
'scheduler_group_key', dt.scheduler_group_key,
|
|
'monitor_task_id', dt.monitor_task_id,
|
|
'supersedes_task_id', dt.supersedes_task_id,
|
|
'control_flags', dt.control_flags,
|
|
'interrupt_generation', dt.interrupt_generation,
|
|
'interrupted_at', dt.interrupted_at,
|
|
'interrupt_reason', dt.interrupt_reason
|
|
))
|
|
FROM desktop_tasks dt
|
|
WHERE dt.desktop_id::text = $1`)
|
|
}
|
|
|
|
func (r *JobRepository) getComplianceReviewJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getBusinessJob(ctx, "compliance_review", jobID, `
|
|
SELECT
|
|
jsonb_strip_nulls(jsonb_build_object(
|
|
'message_id', crj.message_id,
|
|
'article_id', crj.article_id,
|
|
'article_version_id', crj.article_version_id,
|
|
'check_record_id', crj.check_record_id
|
|
)),
|
|
NULL::jsonb,
|
|
CASE WHEN crj.last_error IS NULL OR crj.last_error = '' THEN NULL ELSE to_jsonb(crj.last_error) END,
|
|
jsonb_strip_nulls(jsonb_build_object(
|
|
'available_at', crj.available_at,
|
|
'locked_at', crj.locked_at,
|
|
'locked_by', crj.locked_by
|
|
))
|
|
FROM compliance_review_jobs crj
|
|
WHERE crj.id = $1`, jobID)
|
|
}
|
|
|
|
func (r *JobRepository) getMonitoringCollectJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
if r == nil || r.monitoringPool == nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
summary, err := r.getMonitoringSummary(ctx, jobID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var payloadRaw, metaRaw []byte
|
|
err = r.monitoringPool.QueryRow(ctx, `
|
|
SELECT
|
|
request_payload_json,
|
|
jsonb_strip_nulls(jsonb_build_object(
|
|
'brand_id', brand_id,
|
|
'question_id', question_id,
|
|
'question_hash', encode(question_hash, 'hex'),
|
|
'ai_platform_id', ai_platform_id,
|
|
'collector_type', collector_type,
|
|
'trigger_source', trigger_source,
|
|
'run_mode', run_mode,
|
|
'business_date', business_date,
|
|
'planned_at', planned_at,
|
|
'lease_token_hash', lease_token_hash,
|
|
'leased_to_executor', leased_to_executor,
|
|
'callback_received_at', callback_received_at,
|
|
'skip_reason', skip_reason,
|
|
'dispatch_priority', dispatch_priority,
|
|
'dispatch_lane', dispatch_lane,
|
|
'target_client_id', target_client_id,
|
|
'dispatch_after', dispatch_after,
|
|
'interrupt_generation', interrupt_generation,
|
|
'superseded_by_request_id', superseded_by_request_id,
|
|
'last_dispatched_at', last_dispatched_at,
|
|
'dispatch_attempts', dispatch_attempts,
|
|
'ingest_shard_key', ingest_shard_key,
|
|
'execution_owner', execution_owner
|
|
))
|
|
FROM monitoring_collect_tasks
|
|
WHERE id = $1
|
|
`, jobID).Scan(&payloadRaw, &metaRaw)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
if isUndefinedMonitoringRelation(err) {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
return &domain.JobDetail{
|
|
JobSummary: *summary,
|
|
Payload: decodeJSONValue(payloadRaw),
|
|
Metadata: decodeJSONMap(metaRaw),
|
|
}, nil
|
|
}
|
|
|
|
func (r *JobRepository) getBusinessJob(ctx context.Context, source string, numericID int64, detailSQL string, args ...any) (*domain.JobDetail, error) {
|
|
summary, err := r.getBusinessSummary(ctx, source, strconv.FormatInt(numericID, 10))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return r.attachBusinessDetail(ctx, summary, detailSQL, args...)
|
|
}
|
|
|
|
func (r *JobRepository) getBusinessJobByTextID(ctx context.Context, source, id, detailSQL string) (*domain.JobDetail, error) {
|
|
summary, err := r.getBusinessSummary(ctx, source, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return r.attachBusinessDetail(ctx, summary, detailSQL, id)
|
|
}
|
|
|
|
func (r *JobRepository) getBusinessSummary(ctx context.Context, source, id string) (*domain.JobSummary, error) {
|
|
items, err := r.listBusinessJobs(ctx, domain.JobFilter{
|
|
Source: source,
|
|
Keyword: id,
|
|
Limit: 2000,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for i := range items {
|
|
if items[i].Source == source && items[i].ID == id {
|
|
return &items[i], nil
|
|
}
|
|
}
|
|
return nil, ErrJobNotFound
|
|
}
|
|
|
|
func (r *JobRepository) getMonitoringSummary(ctx context.Context, id int64) (*domain.JobSummary, error) {
|
|
items, err := r.listMonitoringJobs(ctx, domain.JobFilter{
|
|
Source: "monitoring_collect",
|
|
Keyword: strconv.FormatInt(id, 10),
|
|
Limit: 2000,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for i := range items {
|
|
if items[i].ID == strconv.FormatInt(id, 10) {
|
|
return &items[i], nil
|
|
}
|
|
}
|
|
return nil, ErrJobNotFound
|
|
}
|
|
|
|
func (r *JobRepository) attachBusinessDetail(ctx context.Context, summary *domain.JobSummary, query string, args ...any) (*domain.JobDetail, error) {
|
|
if summary == nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
var payloadRaw, resultRaw, errorRaw, metaRaw []byte
|
|
err := r.pool.QueryRow(ctx, query, args...).Scan(&payloadRaw, &resultRaw, &errorRaw, &metaRaw)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
return &domain.JobDetail{
|
|
JobSummary: *summary,
|
|
Payload: decodeJSONValue(payloadRaw),
|
|
Result: decodeJSONValue(resultRaw),
|
|
Error: decodeJSONValue(errorRaw),
|
|
Metadata: decodeJSONMap(metaRaw),
|
|
}, nil
|
|
}
|
|
|
|
func (r *JobRepository) retryGenerationJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
UPDATE generation_tasks
|
|
SET status = 'queued',
|
|
error_message = NULL,
|
|
started_at = NULL,
|
|
completed_at = NULL,
|
|
lease_token = NULL,
|
|
lease_owner = NULL,
|
|
lease_expires_at = NULL,
|
|
updated_at = NOW()
|
|
WHERE id = $1
|
|
AND status = 'failed'
|
|
`, jobID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getGenerationJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) retryTemplateAssistJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
UPDATE template_assist_tasks
|
|
SET status = 'queued',
|
|
result_json = NULL,
|
|
error_message = NULL,
|
|
started_at = NULL,
|
|
completed_at = NULL,
|
|
updated_at = NOW()
|
|
WHERE id = $1
|
|
AND status = 'failed'
|
|
`, strings.TrimSpace(id))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getTemplateAssistJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) retryKolAssistJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
UPDATE kol_assist_tasks
|
|
SET status = 'queued',
|
|
result_json = NULL,
|
|
error_message = NULL,
|
|
started_at = NULL,
|
|
completed_at = NULL,
|
|
updated_at = NOW()
|
|
WHERE id = $1
|
|
AND status = 'failed'
|
|
`, strings.TrimSpace(id))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getKolAssistJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) retryKnowledgeParseJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
WITH target AS (
|
|
SELECT kpt.id, kpt.tenant_id, kpt.knowledge_item_id
|
|
FROM knowledge_parse_tasks kpt
|
|
WHERE kpt.id = $1
|
|
AND kpt.status = 'failed'
|
|
),
|
|
task_update AS (
|
|
UPDATE knowledge_parse_tasks kpt
|
|
SET status = 'pending',
|
|
error_message = NULL,
|
|
started_at = NULL,
|
|
completed_at = NULL,
|
|
updated_at = NOW()
|
|
FROM target
|
|
WHERE kpt.id = target.id
|
|
RETURNING target.tenant_id, target.knowledge_item_id
|
|
)
|
|
UPDATE knowledge_items ki
|
|
SET status = 'pending',
|
|
error_message = NULL,
|
|
updated_at = NOW()
|
|
FROM task_update
|
|
WHERE ki.id = task_update.knowledge_item_id
|
|
AND ki.tenant_id = task_update.tenant_id
|
|
`, jobID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getKnowledgeParseJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) retryDesktopTaskJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
errorJSON, _ := json.Marshal(map[string]any{
|
|
"message": "retried from ops job console",
|
|
"retried_at": time.Now().UTC().Format(time.RFC3339),
|
|
})
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
UPDATE desktop_tasks
|
|
SET status = 'queued',
|
|
error = $2,
|
|
result = NULL,
|
|
active_attempt_id = NULL,
|
|
lease_token_hash = NULL,
|
|
lease_expires_at = NULL,
|
|
attempts = attempts + 1,
|
|
started_at = NULL,
|
|
interrupted_at = NULL,
|
|
interrupt_reason = NULL,
|
|
enqueued_at = NOW(),
|
|
updated_at = NOW()
|
|
WHERE desktop_id::text = $1
|
|
AND status IN ('failed', 'unknown', 'aborted')
|
|
`, strings.TrimSpace(id), errorJSON)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getDesktopTaskJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) retryComplianceReviewJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
UPDATE compliance_review_jobs
|
|
SET status = 'pending',
|
|
available_at = NOW(),
|
|
locked_at = NULL,
|
|
locked_by = NULL,
|
|
last_error = NULL,
|
|
updated_at = NOW()
|
|
WHERE id = $1
|
|
AND status = 'failed'
|
|
`, jobID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getComplianceReviewJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) retryMonitoringCollectJob(ctx context.Context, id string) (*domain.JobDetail, error) {
|
|
if r == nil || r.monitoringPool == nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
cmd, err := r.monitoringPool.Exec(ctx, `
|
|
UPDATE monitoring_collect_tasks
|
|
SET status = 'pending',
|
|
lease_token_hash = NULL,
|
|
leased_to_executor = NULL,
|
|
leased_at = NULL,
|
|
lease_expires_at = NULL,
|
|
callback_received_at = NULL,
|
|
completed_at = NULL,
|
|
skip_reason = NULL,
|
|
error_message = NULL,
|
|
dispatch_after = NOW(),
|
|
updated_at = NOW()
|
|
WHERE id = $1
|
|
AND status IN ('failed', 'expired', 'skipped')
|
|
`, jobID)
|
|
if err != nil {
|
|
if isUndefinedMonitoringRelation(err) {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getMonitoringCollectJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) cancelGenerationJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) {
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
UPDATE generation_tasks
|
|
SET status = 'failed',
|
|
error_message = $2,
|
|
completed_at = NOW(),
|
|
lease_token = NULL,
|
|
lease_owner = NULL,
|
|
lease_expires_at = NULL,
|
|
updated_at = NOW()
|
|
WHERE id = $1
|
|
AND status IN ('queued', 'running')
|
|
`, jobID, reason)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getGenerationJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) cancelTemplateAssistJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) {
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
UPDATE template_assist_tasks
|
|
SET status = 'failed',
|
|
error_message = $2,
|
|
completed_at = NOW(),
|
|
updated_at = NOW()
|
|
WHERE id = $1
|
|
AND status IN ('queued', 'running')
|
|
`, strings.TrimSpace(id), reason)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getTemplateAssistJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) cancelKolAssistJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) {
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
UPDATE kol_assist_tasks
|
|
SET status = 'failed',
|
|
error_message = $2,
|
|
completed_at = NOW(),
|
|
updated_at = NOW()
|
|
WHERE id = $1
|
|
AND status IN ('queued', 'running')
|
|
`, strings.TrimSpace(id), reason)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getKolAssistJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) cancelKnowledgeParseJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) {
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
WITH target AS (
|
|
SELECT kpt.id, kpt.tenant_id, kpt.knowledge_item_id
|
|
FROM knowledge_parse_tasks kpt
|
|
WHERE kpt.id = $1
|
|
AND kpt.status IN ('pending', 'running')
|
|
),
|
|
task_update AS (
|
|
UPDATE knowledge_parse_tasks kpt
|
|
SET status = 'deleted',
|
|
error_message = $2,
|
|
completed_at = NOW(),
|
|
updated_at = NOW()
|
|
FROM target
|
|
WHERE kpt.id = target.id
|
|
RETURNING target.tenant_id, target.knowledge_item_id
|
|
)
|
|
UPDATE knowledge_items ki
|
|
SET status = 'failed',
|
|
error_message = $2,
|
|
updated_at = NOW()
|
|
FROM task_update
|
|
WHERE ki.id = task_update.knowledge_item_id
|
|
AND ki.tenant_id = task_update.tenant_id
|
|
`, jobID, reason)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getKnowledgeParseJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) cancelDesktopPublishJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
cmd, err := tx.Exec(ctx, `
|
|
UPDATE desktop_publish_jobs
|
|
SET status = 'aborted',
|
|
compliance_blocked_reason = $2,
|
|
updated_at = NOW()
|
|
WHERE desktop_id::text = $1
|
|
AND status IN ('queued', 'blocked_by_compliance')
|
|
`, strings.TrimSpace(id), reason)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
|
|
errorJSON, _ := json.Marshal(map[string]any{"message": reason})
|
|
if _, err := tx.Exec(ctx, `
|
|
UPDATE desktop_tasks
|
|
SET status = 'aborted',
|
|
error = $2,
|
|
active_attempt_id = NULL,
|
|
lease_token_hash = NULL,
|
|
lease_expires_at = NULL,
|
|
interrupted_at = COALESCE(interrupted_at, NOW()),
|
|
interrupt_reason = 'ops_cancel',
|
|
updated_at = NOW()
|
|
WHERE job_id::text = $1
|
|
AND status IN ('queued', 'in_progress', 'unknown')
|
|
`, strings.TrimSpace(id), errorJSON); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
return r.getDesktopPublishJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) cancelDesktopTaskJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) {
|
|
errorJSON, _ := json.Marshal(map[string]any{"message": reason})
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
UPDATE desktop_tasks
|
|
SET status = 'aborted',
|
|
error = $2,
|
|
active_attempt_id = NULL,
|
|
lease_token_hash = NULL,
|
|
lease_expires_at = NULL,
|
|
interrupted_at = COALESCE(interrupted_at, NOW()),
|
|
interrupt_reason = 'ops_cancel',
|
|
updated_at = NOW()
|
|
WHERE desktop_id::text = $1
|
|
AND status IN ('queued', 'in_progress', 'unknown')
|
|
`, strings.TrimSpace(id), errorJSON)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getDesktopTaskJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) cancelComplianceReviewJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) {
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
cmd, err := r.pool.Exec(ctx, `
|
|
UPDATE compliance_review_jobs
|
|
SET status = 'failed',
|
|
locked_at = NULL,
|
|
locked_by = NULL,
|
|
last_error = $2,
|
|
updated_at = NOW()
|
|
WHERE id = $1
|
|
AND status IN ('pending', 'running')
|
|
`, jobID, reason)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getComplianceReviewJob(ctx, id)
|
|
}
|
|
|
|
func (r *JobRepository) cancelMonitoringCollectJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) {
|
|
if r == nil || r.monitoringPool == nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
jobID, err := parseInt64ID(id)
|
|
if err != nil {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
cmd, err := r.monitoringPool.Exec(ctx, `
|
|
UPDATE monitoring_collect_tasks
|
|
SET status = 'skipped',
|
|
skip_reason = 'ops_cancel',
|
|
error_message = $2,
|
|
lease_token_hash = NULL,
|
|
leased_to_executor = NULL,
|
|
leased_at = NULL,
|
|
lease_expires_at = NULL,
|
|
completed_at = NOW(),
|
|
updated_at = NOW()
|
|
WHERE id = $1
|
|
AND status IN ('pending', 'leased', 'received', 'expired')
|
|
`, jobID, reason)
|
|
if err != nil {
|
|
if isUndefinedMonitoringRelation(err) {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
if cmd.RowsAffected() == 0 {
|
|
return nil, ErrJobNotFound
|
|
}
|
|
return r.getMonitoringCollectJob(ctx, id)
|
|
}
|
|
|
|
type jobScanner interface {
|
|
Scan(dest ...any) error
|
|
}
|
|
|
|
func scanJobSummary(row jobScanner) (*domain.JobSummary, error) {
|
|
var (
|
|
item domain.JobSummary
|
|
numericID pgtype.Int8
|
|
uuidValue pgtype.Text
|
|
tenantID pgtype.Int8
|
|
tenantName pgtype.Text
|
|
workspaceID pgtype.Int8
|
|
workspaceName pgtype.Text
|
|
userID pgtype.Int8
|
|
userName pgtype.Text
|
|
title pgtype.Text
|
|
target pgtype.Text
|
|
queue pgtype.Text
|
|
worker pgtype.Text
|
|
attempts pgtype.Int4
|
|
priority pgtype.Int4
|
|
errorMessage pgtype.Text
|
|
startedAt pgtype.Timestamptz
|
|
completedAt pgtype.Timestamptz
|
|
leaseExpires pgtype.Timestamptz
|
|
availableAt pgtype.Timestamptz
|
|
heartbeatAt pgtype.Timestamptz
|
|
)
|
|
|
|
if err := row.Scan(
|
|
&item.Source,
|
|
&item.ID,
|
|
&numericID,
|
|
&uuidValue,
|
|
&item.Kind,
|
|
&item.Status,
|
|
&item.Phase,
|
|
&tenantID,
|
|
&tenantName,
|
|
&workspaceID,
|
|
&workspaceName,
|
|
&userID,
|
|
&userName,
|
|
&title,
|
|
&target,
|
|
&queue,
|
|
&worker,
|
|
&attempts,
|
|
&priority,
|
|
&errorMessage,
|
|
&item.CreatedAt,
|
|
&item.UpdatedAt,
|
|
&startedAt,
|
|
&completedAt,
|
|
&leaseExpires,
|
|
&availableAt,
|
|
&heartbeatAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
item.UID = item.Source + ":" + item.ID
|
|
item.NumericID = int64PtrFromPg(numericID)
|
|
item.UUID = stringPtrFromPg(uuidValue)
|
|
item.TenantID = int64PtrFromPg(tenantID)
|
|
item.TenantName = stringPtrFromPg(tenantName)
|
|
item.WorkspaceID = int64PtrFromPg(workspaceID)
|
|
item.WorkspaceName = stringPtrFromPg(workspaceName)
|
|
item.UserID = int64PtrFromPg(userID)
|
|
item.UserName = stringPtrFromPg(userName)
|
|
item.Title = stringPtrFromPg(title)
|
|
item.Target = stringPtrFromPg(target)
|
|
item.Queue = stringPtrFromPg(queue)
|
|
item.Worker = stringPtrFromPg(worker)
|
|
item.Attempts = intPtrFromPg(attempts)
|
|
item.Priority = intPtrFromPg(priority)
|
|
item.ErrorMessage = stringPtrFromPg(errorMessage)
|
|
item.StartedAt = timePtrFromPg(startedAt)
|
|
item.CompletedAt = timePtrFromPg(completedAt)
|
|
item.LeaseExpiresAt = timePtrFromPg(leaseExpires)
|
|
item.AvailableAt = timePtrFromPg(availableAt)
|
|
item.LastHeartbeatAt = timePtrFromPg(heartbeatAt)
|
|
item.DurationSeconds = durationSeconds(item.StartedAt, item.CompletedAt, item.CreatedAt)
|
|
item.IsStuck = isJobStuck(item)
|
|
item.CanRetry = canRetryJob(item)
|
|
item.CanCancel = canCancelJob(item)
|
|
return &item, nil
|
|
}
|
|
|
|
func sortJobs(items []domain.JobSummary) {
|
|
for i := 1; i < len(items); i++ {
|
|
current := items[i]
|
|
j := i - 1
|
|
for j >= 0 && jobLess(current, items[j]) {
|
|
items[j+1] = items[j]
|
|
j--
|
|
}
|
|
items[j+1] = current
|
|
}
|
|
}
|
|
|
|
func jobLess(a, b domain.JobSummary) bool {
|
|
if !a.CreatedAt.Equal(b.CreatedAt) {
|
|
return a.CreatedAt.After(b.CreatedAt)
|
|
}
|
|
if a.Source != b.Source {
|
|
return a.Source < b.Source
|
|
}
|
|
return a.ID > b.ID
|
|
}
|
|
|
|
func buildJobCounts(items []domain.JobSummary) domain.JobCounts {
|
|
counts := domain.JobCounts{
|
|
Total: int64(len(items)),
|
|
ByPhase: map[string]int64{},
|
|
BySource: map[string]int64{},
|
|
}
|
|
for _, item := range items {
|
|
counts.ByPhase[item.Phase]++
|
|
counts.BySource[item.Source]++
|
|
}
|
|
return counts
|
|
}
|
|
|
|
func durationSeconds(startedAt, completedAt *time.Time, createdAt time.Time) *int64 {
|
|
start := createdAt
|
|
if startedAt != nil {
|
|
start = *startedAt
|
|
}
|
|
end := time.Now().UTC()
|
|
if completedAt != nil {
|
|
end = *completedAt
|
|
}
|
|
seconds := int64(end.Sub(start).Seconds())
|
|
if seconds < 0 {
|
|
seconds = 0
|
|
}
|
|
return &seconds
|
|
}
|
|
|
|
func isJobStuck(item domain.JobSummary) bool {
|
|
now := time.Now().UTC()
|
|
if item.LeaseExpiresAt != nil && item.LeaseExpiresAt.Before(now) && (item.Phase == domain.JobPhaseRunning || item.Phase == domain.JobPhaseQueued) {
|
|
return true
|
|
}
|
|
if item.Phase == domain.JobPhaseRunning && item.UpdatedAt.Before(now.Add(-30*time.Minute)) {
|
|
return true
|
|
}
|
|
if item.Phase == domain.JobPhaseQueued && item.UpdatedAt.Before(now.Add(-2*time.Hour)) {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func canRetryJob(item domain.JobSummary) bool {
|
|
switch item.Source {
|
|
case "desktop_publish":
|
|
return false
|
|
case "knowledge_parse":
|
|
return false
|
|
case "monitoring_collect":
|
|
return item.Phase == domain.JobPhaseFailed || item.Phase == domain.JobPhaseCanceled
|
|
default:
|
|
return item.Phase == domain.JobPhaseFailed || item.Phase == domain.JobPhaseCanceled || item.Phase == domain.JobPhaseUnknown
|
|
}
|
|
}
|
|
|
|
func canCancelJob(item domain.JobSummary) bool {
|
|
return item.Phase == domain.JobPhaseQueued || item.Phase == domain.JobPhaseRunning || item.Phase == domain.JobPhaseBlocked || item.Phase == domain.JobPhaseUnknown
|
|
}
|
|
|
|
func normalizeJobSource(source string) string {
|
|
source = strings.TrimSpace(source)
|
|
if source == "desktop_publish_job" {
|
|
return "desktop_publish"
|
|
}
|
|
if source == "publish_job" {
|
|
return "desktop_publish"
|
|
}
|
|
return source
|
|
}
|
|
|
|
func nullableTextValue(value string) *string {
|
|
value = strings.TrimSpace(value)
|
|
if value == "" {
|
|
return nil
|
|
}
|
|
return &value
|
|
}
|
|
|
|
func nullableKeyword(value string) *string {
|
|
value = strings.TrimSpace(value)
|
|
if value == "" {
|
|
return nil
|
|
}
|
|
return &value
|
|
}
|
|
|
|
func parseInt64ID(id string) (int64, error) {
|
|
parsed, err := strconv.ParseInt(strings.TrimSpace(id), 10, 64)
|
|
if err != nil || parsed <= 0 {
|
|
return 0, fmt.Errorf("invalid id")
|
|
}
|
|
return parsed, nil
|
|
}
|
|
|
|
func int64PtrFromPg(value pgtype.Int8) *int64 {
|
|
if !value.Valid {
|
|
return nil
|
|
}
|
|
v := value.Int64
|
|
return &v
|
|
}
|
|
|
|
func intPtrFromPg(value pgtype.Int4) *int {
|
|
if !value.Valid {
|
|
return nil
|
|
}
|
|
v := int(value.Int32)
|
|
return &v
|
|
}
|
|
|
|
func stringPtrFromPg(value pgtype.Text) *string {
|
|
if !value.Valid {
|
|
return nil
|
|
}
|
|
v := strings.TrimSpace(value.String)
|
|
if v == "" {
|
|
return nil
|
|
}
|
|
return &v
|
|
}
|
|
|
|
func timePtrFromPg(value pgtype.Timestamptz) *time.Time {
|
|
if !value.Valid {
|
|
return nil
|
|
}
|
|
v := value.Time
|
|
return &v
|
|
}
|
|
|
|
func decodeJSONValue(raw []byte) any {
|
|
if len(raw) == 0 {
|
|
return nil
|
|
}
|
|
var value any
|
|
if err := json.Unmarshal(raw, &value); err != nil {
|
|
return string(raw)
|
|
}
|
|
return redactValue(value)
|
|
}
|
|
|
|
func decodeJSONMap(raw []byte) map[string]any {
|
|
value := decodeJSONValue(raw)
|
|
if value == nil {
|
|
return nil
|
|
}
|
|
if m, ok := value.(map[string]any); ok {
|
|
return m
|
|
}
|
|
return map[string]any{"value": value}
|
|
}
|
|
|
|
func redactValue(value any) any {
|
|
switch typed := value.(type) {
|
|
case map[string]any:
|
|
out := make(map[string]any, len(typed))
|
|
for key, nested := range typed {
|
|
if isSensitiveKey(key) {
|
|
out[key] = "[REDACTED]"
|
|
continue
|
|
}
|
|
out[key] = redactValue(nested)
|
|
}
|
|
return out
|
|
case []any:
|
|
out := make([]any, len(typed))
|
|
for i := range typed {
|
|
out[i] = redactValue(typed[i])
|
|
}
|
|
return out
|
|
default:
|
|
return value
|
|
}
|
|
}
|
|
|
|
func isSensitiveKey(key string) bool {
|
|
normalized := strings.ToLower(strings.TrimSpace(key))
|
|
for _, marker := range []string{
|
|
"password",
|
|
"passwd",
|
|
"secret",
|
|
"token",
|
|
"cookie",
|
|
"authorization",
|
|
"credential",
|
|
"session",
|
|
"api_key",
|
|
"apikey",
|
|
"access_key",
|
|
"refresh",
|
|
"lease_token",
|
|
} {
|
|
if strings.Contains(normalized, marker) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func isUndefinedMonitoringRelation(err error) bool {
|
|
var pgErr *pgconn.PgError
|
|
return errors.As(err, &pgErr) && (pgErr.Code == "42P01" || pgErr.Code == "42703")
|
|
}
|