姚章浩
1c80b07f2c
build / build (push) Successful in 1m59s
Dockerfile 运行阶段补齐 dws_dingtalk_scan（F12），并新增 runtime-scripts 门禁 做脚本依赖闭包正负例；就绪探测每次解析当前端点/绑定/凭据并区分 404、未配置、 不可达（F13）；路由写入来源路径与父目录，历史空值仅按同连接同 item 的意图读时 补齐（F14）；目标筛选显示短 ID 与当前绑定/历史并提供快捷筛选（F15）；刷新终态 移除后台运行、展示失败时间/旧快照/脱敏原因/作业 ID 与重试（F16）。 验证：make verify 17 项全过（identity 本地按设计 skipped）；一次性 schema 的 Monitor 目录/路由集成用例通过；8088 部署与真实 UI 验收仍由父会话执行。
164 lines
11 KiB
Go
164 lines
11 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
)
|
|
|
|
type MonitorTaskFilter struct {
|
|
Search, Source, Scope, Action, Status, Stage, ID, Target, Since, Until string
|
|
Limit, Offset int
|
|
}
|
|
type MonitorTaskEvent struct {
|
|
At string `json:"at"`
|
|
Kind string `json:"kind"`
|
|
Detail string `json:"detail"`
|
|
}
|
|
type MonitorTask struct {
|
|
History []MonitorTaskEvent `json:"history"`
|
|
ID string `json:"id"`
|
|
Title string `json:"title"`
|
|
Path string `json:"path"`
|
|
// NodeID/ParentNodeID let the UI return to the exact source directory;
|
|
// both come from the catalog snapshot bound to THIS source item.
|
|
NodeID string `json:"node_id"`
|
|
ParentNodeID string `json:"parent_node_id"`
|
|
SourceID string `json:"source_id"`
|
|
SourceName string `json:"source_name"`
|
|
TargetScope string `json:"target_scope"`
|
|
TargetKB string `json:"target_kb"`
|
|
Action string `json:"action"`
|
|
Stage string `json:"stage"`
|
|
Status string `json:"status"`
|
|
Detail string `json:"detail"`
|
|
UpdatedAt string `json:"updated_at"`
|
|
RetryKind string `json:"retry_kind"`
|
|
FailureID string `json:"failure_id"`
|
|
RetryCount int `json:"retry_count"`
|
|
VersionID string `json:"version_id"`
|
|
}
|
|
type MonitorTaskPage struct {
|
|
Items []MonitorTask `json:"items"`
|
|
Total int `json:"total"`
|
|
Limit int `json:"limit"`
|
|
Offset int `json:"offset"`
|
|
Sources []MonitorOption `json:"sources"`
|
|
Targets []MonitorOption `json:"targets"`
|
|
Retryable int `json:"retryable"`
|
|
// Bindings is the effective (persisted, env-fallback aware) target for each
|
|
// scope. The UI marks rows/options against it; counts stay this page's totals.
|
|
Bindings MonitorBindings `json:"bindings"`
|
|
}
|
|
type MonitorOption struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
Scope string `json:"scope,omitempty"`
|
|
}
|
|
type MonitorBindings struct {
|
|
Internal string `json:"internal"`
|
|
External string `json:"external"`
|
|
}
|
|
|
|
// Reduce publication versions before filtering; totals and pages share one snapshot.
|
|
const monitorTasksCTE = `WITH history_publications AS (
|
|
SELECT hp.* FROM test_publications hp JOIN source_item_versions hv ON hv.id=hp.source_item_version_id
|
|
WHERE $7<>'' AND (hv.source_item_id,hp.test_kb_id)=(SELECT v.source_item_id,p.test_kb_id FROM test_publications p JOIN source_item_versions v ON v.id=p.source_item_version_id WHERE p.id=$7)
|
|
), history_events AS (
|
|
SELECT updated_at AS at,'版本状态' AS kind,source_item_version_id || ' · ' || processing_state || ' / ' || retrieval_state AS detail FROM history_publications
|
|
UNION ALL SELECT f.created_at,'失败记录',f.job_type || ' · ' || f.error_code || ' · 尝试 ' || f.attempt FROM job_failures f LEFT JOIN jobs fj ON fj.id=f.job_id WHERE f.publication_id IN(SELECT id FROM history_publications) OR (f.job_type='artifact.fetch' AND COALESCE(NULLIF(f.business_version_id,''),fj.payload->>'version_id') IN(SELECT source_item_version_id FROM history_publications))
|
|
UNION ALL SELECT r.created_at,'重试记录',r.reason || ' · ' || r.outcome FROM job_retry_operations r WHERE r.publication_id IN(SELECT id FROM history_publications)
|
|
), latest AS (
|
|
SELECT DISTINCT ON(v.source_item_id,p.test_kb_id) p.*,v.source_item_id,v.content_object_id
|
|
FROM test_publications p JOIN source_item_versions v ON v.id=p.source_item_version_id
|
|
WHERE p.superseded_at IS NULL
|
|
ORDER BY v.source_item_id,p.test_kb_id,p.created_at DESC,p.id DESC
|
|
), base AS (
|
|
SELECT p.id,si.title,
|
|
COALESCE(NULLIF(si.normalized_path,''),ci.relative_path,'') AS path,
|
|
COALESCE(NULLIF(si.parent_external_id,''),cn.parent_node_id,'') AS parent_node_id,
|
|
COALESCE(ci.node_id,'') AS node_id,
|
|
c.id AS source_id,c.name AS source_name,
|
|
p.target_scope,p.test_kb_id AS target_kb,p.source_item_version_id AS version_id,p.retry_count,
|
|
GREATEST(p.updated_at,j.updated_at,f.created_at) AS updated_at,
|
|
p.processing_state,p.ingest_state,p.retrieval_state,p.last_error,p.knowledge_id,
|
|
co.content_state,co.local_spool_ref,si.policy_state,
|
|
CASE WHEN p.retrieval_state IN ('disabling','disabled') OR (p.knowledge_id<>'' AND si.policy_state='blocked') OR j.type='test.remove' THEN 'remove'
|
|
WHEN EXISTS(SELECT 1 FROM test_publications prev JOIN source_item_versions pv ON pv.id=prev.source_item_version_id WHERE pv.source_item_id=si.id AND prev.test_kb_id=p.test_kb_id AND prev.id<>p.id AND prev.created_at<p.created_at) THEN 'update' ELSE 'add' END AS action,
|
|
COALESCE(j.type,'') AS job_type,COALESCE(j.state,'') AS job_state,COALESCE(j.attempt,0) AS job_attempt,j.available_at,
|
|
COALESCE(f.id,'') AS failure_id,COALESCE(f.job_type,'') AS failure_type,COALESCE(NULLIF(f.last_error,''),f.error_code,'') AS failure_error
|
|
FROM latest p JOIN source_items si ON si.id=p.source_item_id JOIN source_connections c ON c.id=si.source_connection_id
|
|
-- 历史行路径为空时，只用「同一连接 + 同一 item」的取件意图补齐：不按文件名猜配。
|
|
LEFT JOIN LATERAL(SELECT intent.relative_path,intent.node_id FROM catalog_import_intents intent
|
|
WHERE intent.source_item_id=si.id AND intent.source_connection_id=si.source_connection_id
|
|
ORDER BY intent.updated_at DESC,intent.node_id LIMIT 1) ci ON TRUE
|
|
LEFT JOIN LATERAL(SELECT node.parent_node_id FROM source_catalog_heads head
|
|
JOIN source_catalog_nodes node ON node.snapshot_id=head.snapshot_id
|
|
WHERE head.source_connection_id=si.source_connection_id AND node.node_id=ci.node_id LIMIT 1) cn ON TRUE
|
|
LEFT JOIN content_objects co ON co.id=p.content_object_id
|
|
LEFT JOIN LATERAL(SELECT j.type,j.state,j.attempt,j.available_at,j.updated_at FROM jobs j WHERE
|
|
(j.payload->>'publication_id'=p.id OR (j.type='artifact.fetch' AND j.payload->>'version_id'=p.source_item_version_id))
|
|
AND j.type IN ('artifact.fetch','test.upload','test.poll','test.remove')
|
|
ORDER BY CASE WHEN j.state IN ('pending','running','retry') THEN 0 ELSE 1 END,j.updated_at DESC,j.id DESC LIMIT 1) j ON TRUE
|
|
LEFT JOIN LATERAL(SELECT f.id,f.job_type,f.error_code,f.last_error,f.created_at FROM job_failures f LEFT JOIN jobs fj ON fj.id=f.job_id WHERE f.state='open' AND (f.publication_id=p.id OR (f.job_type='artifact.fetch' AND COALESCE(NULLIF(f.business_version_id,''),fj.payload->>'version_id')=p.source_item_version_id)) ORDER BY f.created_at DESC,f.id DESC LIMIT 1) f ON TRUE
|
|
), classified AS (
|
|
SELECT *,
|
|
CASE WHEN action='remove' AND retrieval_state='disabled' THEN 'completed'
|
|
WHEN action<>'remove' AND processing_state='completed' THEN 'completed'
|
|
WHEN job_state IN ('pending','retry') AND job_attempt>0 THEN 'retrying'
|
|
WHEN job_state='running' THEN 'running'
|
|
WHEN job_state IN ('pending','retry') THEN 'waiting'
|
|
WHEN job_state='failed' OR failure_id<>'' OR processing_state IN ('failed','stalled') OR ingest_state='failed' OR retrieval_state='failed' THEN 'attention'
|
|
WHEN processing_state='cancelled' THEN 'cancelled'
|
|
WHEN action='remove' THEN 'waiting'
|
|
WHEN processing_state IN ('processing','finalizing') THEN 'running'
|
|
ELSE 'waiting' END AS status,
|
|
CASE WHEN action='remove' AND retrieval_state='disabled' THEN 'finished'
|
|
WHEN action='remove' THEN 'removal'
|
|
WHEN processing_state IN ('completed','cancelled') THEN 'finished'
|
|
WHEN job_type='artifact.fetch' AND job_state IN ('pending','running','retry') THEN 'fetch'
|
|
WHEN processing_state='finalizing' THEN 'indexing'
|
|
WHEN processing_state='processing' OR job_type='test.poll' THEN 'parsing'
|
|
WHEN ingest_state='uploading' OR job_type='test.upload' THEN 'upload'
|
|
WHEN failure_type='artifact.fetch' OR COALESCE(content_state,'')<>'verified' THEN 'fetch'
|
|
ELSE 'waiting' END AS stage
|
|
FROM base
|
|
), projected AS (
|
|
SELECT id,title,path,parent_node_id,node_id,source_id,source_name,target_scope,target_kb,version_id,retry_count,updated_at,action,stage,status,failure_id,
|
|
CASE WHEN status<>'attention' THEN ''
|
|
WHEN action='remove' THEN CASE WHEN job_type='test.remove' AND job_state='failed' AND retrieval_state IN ('failed','disabling') AND policy_state='blocked' THEN 'remove' ELSE '' END
|
|
WHEN failure_type='artifact.fetch' THEN 'fetch'
|
|
WHEN content_state='verified' AND local_spool_ref<>'' AND processing_state IN ('failed','stalled','pending') THEN 'upload' ELSE '' END AS retry_kind,
|
|
CASE WHEN status='completed' THEN CASE WHEN action='remove' THEN '已从测试库移除；来源文件保留' ELSE '测试库处理完成；未代表已发布生产' END
|
|
WHEN status='cancelled' THEN '处理已取消，记录保留'
|
|
WHEN status='retrying' THEN '等待自动重试；下次调度 ' || COALESCE(to_char(available_at AT TIME ZONE 'Asia/Shanghai','YYYY-MM-DD HH24:MI:SS'),'未提供')
|
|
WHEN status='attention' THEN COALESCE(NULLIF(last_error,''),NULLIF(failure_error,''),'处理异常，请查看详情')
|
|
WHEN stage='removal' THEN CASE WHEN status='running' THEN '正在移除测试库知识' ELSE '等待测试库移除任务' END
|
|
WHEN stage='fetch' THEN '等待或正在获取源文件'
|
|
WHEN stage='waiting' THEN '文件已就绪，等待后台处理调度'
|
|
WHEN stage='upload' THEN '等待或正在上传测试知识库'
|
|
WHEN stage='indexing' THEN '正在索引与增强'
|
|
ELSE '正在解析处理' END AS detail
|
|
FROM classified
|
|
), filtered AS (
|
|
SELECT * FROM projected WHERE ($1='' OR strpos(lower(title || ' ' || path),lower($1))>0)
|
|
AND ($2='' OR source_id=$2) AND ($3='' OR target_scope=$3) AND ($4='' OR action=$4)
|
|
AND ($5='' OR ($5='active' AND status NOT IN ('completed','cancelled')) OR status=$5)
|
|
AND ($6='' OR stage=$6) AND ($7='' OR id=$7) AND ($10='' OR target_kb=$10)
|
|
AND ($11='' OR updated_at>=NULLIF($11,'')::timestamptz) AND ($12='' OR updated_at<=NULLIF($12,'')::timestamptz)
|
|
) `
|
|
|
|
func (s *Store) ListMonitorTasks(ctx context.Context, f MonitorTaskFilter) (MonitorTaskPage, error) {
|
|
if f.Limit < 1 || f.Limit > 100 || f.Offset < 0 {
|
|
return MonitorTaskPage{}, errors.New("invalid monitor pagination")
|
|
}
|
|
var raw []byte
|
|
err := s.db.QueryRow(ctx, monitorTasksCTE+`SELECT jsonb_build_object('items',COALESCE((SELECT jsonb_agg(to_jsonb(x)||jsonb_build_object('history',COALESCE((SELECT jsonb_agg(e ORDER BY e.at DESC,e.kind,e.detail) FROM (SELECT * FROM history_events ORDER BY at DESC,kind,detail LIMIT 50)e),'[]'::jsonb)) ORDER BY x.updated_at DESC,x.id) FROM (SELECT * FROM filtered ORDER BY updated_at DESC,id LIMIT $8 OFFSET $9)x),'[]'::jsonb),'total',(SELECT count(*) FROM filtered),'limit',$8::int,'offset',$9::int,'retryable',(SELECT count(*) FROM filtered WHERE retry_kind<>''),'sources',COALESCE((SELECT jsonb_agg(x ORDER BY x.name) FROM (SELECT DISTINCT source_id AS id,source_name AS name FROM projected)x),'[]'::jsonb),'targets',COALESCE((SELECT jsonb_agg(x ORDER BY x.scope,x.id) FROM (SELECT target_kb AS id,MIN(target_scope) AS scope FROM projected GROUP BY target_kb)x),'[]'::jsonb))`, f.Search, f.Source, f.Scope, f.Action, f.Status, f.Stage, f.ID, f.Limit, f.Offset, f.Target, f.Since, f.Until).Scan(&raw)
|
|
if err != nil {
|
|
return MonitorTaskPage{}, err
|
|
}
|
|
var result MonitorTaskPage
|
|
err = json.Unmarshal(raw, &result)
|
|
return result, err
|
|
}
|