姚章浩
1283f65a95
用户反馈：加了钉盘来源后「来源与目录」里看不到。根因不是丢数据，而是列表只列"已经有快照" 的来源 —— 首次扫描要几分钟到几十分钟（钉盘空间 2900+ 个文件夹），这段时间页面上什么都没有。 - 新 Store 方法 ListCatalogScanProgress：已启用 + 还没有 catalog head + 有在跑/排队的 source.catalog.refresh（三条件同时成立）。进度直接取 worker 写在任务载荷里的同一份计数 （progress/message/scanned_directories/discovered_directories/scanned_files），不另立口径。 - /api/v1/source-catalog/sources 在既有列表后追加这些行（snapshot_status="scanning" + scan_progress/scan_message）。走窄接口按需接线：未接线时安静跳过，不会把旧列表弄坏。 - 前端：左列显示「首次扫描中 44%」，选中后是进度面板（百分比 + 进度条 + 当前文案 + 已发现文件数），并说明"扫完自动出现目录"；有来源在扫描时每 3 秒自动刷新列表， 扫完那一行自动变成「本地只读快照」。扫描中不拉目录节点、也不显示标注/格式入口。 - 解析器保持严格：snapshot_status 只接受 ready/scanning，scanning 必须带 scan_progress， 否则整包拒绝 —— 不给"假装 0%"的机会。 测试：postgres 集成测试（真库）覆盖四种不该出现的来源（停用、已有快照、任务类型不符、 历史上已完成的那次任务）；前端 parser + 渲染各一条；Go 全绿、前端 192 条 + tsc 干净。 面板实测：左列「公司级营销资料管理 · 首次扫描中 44%」，主区「首次扫描中 · 44% / 正在扫描目录：已完成 1540 个文件夹 / 已发现约 4934 份文件」。
415 lines
22 KiB
Go
415 lines
22 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
|
|
"knowledge-governance/internal/governance"
|
|
)
|
|
|
|
var ErrCatalogNodeNotFound = errors.New("catalog node not found")
|
|
|
|
// ListCatalogScanProgress 列出「已启用、但还没有快照」的来源的首次扫描进度。
|
|
// 进度取自 worker 写在任务载荷里的同一份计数（jobs.payload），不另立一套口径：
|
|
// 新来源在首次扫描完成前也能出现在「来源与目录」里，而不是等快照落地才出现。
|
|
func (s *Store) ListCatalogScanProgress(ctx context.Context) ([]governance.CatalogScanProgress, error) {
|
|
rows, err := s.db.Query(ctx, `
|
|
SELECT connection.id, connection.source_type, connection.name,
|
|
COALESCE(NULLIF(job.payload->>'progress','')::integer,0),
|
|
COALESCE(job.payload->>'message',''),
|
|
COALESCE(NULLIF(job.payload->>'scanned_directories','')::integer,0),
|
|
COALESCE(NULLIF(job.payload->>'discovered_directories','')::integer,0),
|
|
COALESCE(NULLIF(job.payload->>'scanned_files','')::integer,0),
|
|
job.created_at
|
|
FROM source_connections connection
|
|
JOIN LATERAL (
|
|
SELECT candidate.*
|
|
FROM jobs candidate
|
|
WHERE candidate.type='source.catalog.refresh'
|
|
AND candidate.state IN ('pending','running','retry')
|
|
AND candidate.payload->>'connection_id'=connection.id
|
|
ORDER BY candidate.created_at DESC
|
|
LIMIT 1
|
|
) job ON TRUE
|
|
LEFT JOIN source_catalog_heads head ON head.source_connection_id=connection.id
|
|
WHERE connection.enabled AND head.source_connection_id IS NULL
|
|
ORDER BY job.created_at DESC, connection.id`)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list catalog scan progress: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]governance.CatalogScanProgress, 0)
|
|
for rows.Next() {
|
|
var item governance.CatalogScanProgress
|
|
var sourceType string
|
|
if err := rows.Scan(&item.ConnectionID, &sourceType, &item.Name, &item.Progress, &item.Message,
|
|
&item.ScannedDirectories, &item.DiscoveredDirectories, &item.ScannedFiles, &item.StartedAt); err != nil {
|
|
return nil, fmt.Errorf("scan catalog scan progress: %w", err)
|
|
}
|
|
item.SourceType = governance.SourceType(sourceType)
|
|
if !item.SourceType.Valid() {
|
|
return nil, errors.New("invalid catalog scan progress projection")
|
|
}
|
|
if item.Progress < 0 {
|
|
item.Progress = 0
|
|
}
|
|
if item.Progress > 100 {
|
|
item.Progress = 100
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *Store) ListCatalogSources(ctx context.Context) ([]governance.CatalogSource, error) {
|
|
rows, err := s.db.Query(ctx, `
|
|
SELECT connection.id, connection.source_type, snapshot.source_label, root.node_id,
|
|
snapshot.folder_count - 1,
|
|
(SELECT COUNT(*)::integer FROM source_catalog_nodes file
|
|
LEFT JOIN LATERAL catalog_filter_context(connection.id) ctx ON TRUE
|
|
WHERE file.snapshot_id=snapshot.id AND file.node_type='file'
|
|
AND (file.ext=''
|
|
OR (coalesce(ctx.has_dir,FALSE) AND NOT filename_is_filtered(connection.id,file.relative_path,file.name))
|
|
OR (NOT coalesce(ctx.has_dir,FALSE)
|
|
AND (ctx.src_allow IS NULL OR file.ext = ANY(ctx.src_allow))
|
|
AND NOT file.ext = ANY(coalesce(ctx.src_excl,'{}'::text[]))))),
|
|
head.activated_at,
|
|
'ready', COALESCE(refresh.duration_ms,0)
|
|
FROM source_catalog_heads head
|
|
JOIN source_catalog_snapshots snapshot ON snapshot.id=head.snapshot_id AND snapshot.source_connection_id=head.source_connection_id
|
|
JOIN source_connections connection ON connection.id=head.source_connection_id
|
|
JOIN source_catalog_nodes root ON root.snapshot_id=snapshot.id AND root.parent_node_id IS NULL
|
|
LEFT JOIN LATERAL (
|
|
SELECT (EXTRACT(EPOCH FROM (job.completed_at-COALESCE(job.started_at,job.created_at)))*1000)::BIGINT AS duration_ms
|
|
FROM jobs job
|
|
WHERE job.type='source.catalog.refresh' AND job.state='completed'
|
|
AND job.payload->>'connection_id'=connection.id
|
|
ORDER BY job.completed_at DESC
|
|
LIMIT 1
|
|
) refresh ON TRUE
|
|
ORDER BY CASE connection.source_type WHEN 'dingtalk' THEN 1 WHEN 'pp' THEN 2 WHEN 'alist' THEN 3 ELSE 4 END,
|
|
snapshot.source_label, connection.id`)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list catalog sources: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]governance.CatalogSource, 0)
|
|
for rows.Next() {
|
|
var item governance.CatalogSource
|
|
var sourceType string
|
|
if err := rows.Scan(&item.ConnectionID, &sourceType, &item.Name, &item.RootNodeID, &item.FolderCount, &item.FileCount, &item.ObservedAt, &item.SnapshotStatus, &item.LastRefreshDurationMS); err != nil {
|
|
return nil, fmt.Errorf("scan catalog source: %w", err)
|
|
}
|
|
item.SourceType = governance.SourceType(sourceType)
|
|
item.SnapshotOnly = true
|
|
if !item.SourceType.Valid() || item.FolderCount < 0 || item.FileCount < 0 {
|
|
return nil, errors.New("invalid catalog source projection")
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate catalog sources: %w", err)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Store) ListCatalogNodes(ctx context.Context, filter governance.CatalogNodeFilter) (governance.CatalogNodePage, error) {
|
|
if filter.ConnectionID == "" || len(filter.ConnectionID) > 36 || filter.ParentNodeID == "" || len(filter.ParentNodeID) > 36 || filter.Limit < 1 || filter.Limit > 100 || filter.Offset < 0 || filter.Offset > 1000000 {
|
|
return governance.CatalogNodePage{}, errors.New("invalid catalog node filter")
|
|
}
|
|
page := governance.CatalogNodePage{Items: []governance.CatalogNode{}, Limit: filter.Limit, Offset: filter.Offset}
|
|
var parentID *string
|
|
err := s.db.QueryRow(ctx, `
|
|
WITH filter_ctx AS MATERIALIZED (SELECT * FROM catalog_filter_context($1))
|
|
SELECT node.node_id,node.parent_node_id,node.node_type,node.name,
|
|
(SELECT COUNT(*)::integer FROM source_catalog_nodes child
|
|
LEFT JOIN filter_ctx ctx ON TRUE
|
|
WHERE child.snapshot_id=node.snapshot_id AND child.parent_node_id=node.node_id
|
|
AND (child.node_type='folder' OR child.ext=''
|
|
OR (ctx.has_dir AND NOT filename_is_filtered($1,child.relative_path,child.name))
|
|
OR (NOT ctx.has_dir AND (ctx.src_allow IS NULL OR child.ext = ANY(ctx.src_allow)) AND NOT child.ext = ANY(ctx.src_excl)))),
|
|
(SELECT COUNT(*)::integer FROM source_catalog_nodes file
|
|
LEFT JOIN filter_ctx ctx ON TRUE
|
|
WHERE file.snapshot_id=node.snapshot_id AND file.node_type='file'
|
|
AND (node.relative_path='' OR file.relative_path LIKE node.relative_path||'/%')
|
|
AND (file.ext=''
|
|
OR (ctx.has_dir AND NOT filename_is_filtered($1,file.relative_path,file.name))
|
|
OR (NOT ctx.has_dir AND (ctx.src_allow IS NULL OR file.ext = ANY(ctx.src_allow)) AND NOT file.ext = ANY(ctx.src_excl)))),
|
|
(SELECT COALESCE(SUM(COALESCE(file.size_bytes,0)),0)::bigint FROM source_catalog_nodes file
|
|
LEFT JOIN filter_ctx ctx ON TRUE
|
|
WHERE file.snapshot_id=node.snapshot_id AND file.node_type='file'
|
|
AND (node.relative_path='' OR file.relative_path LIKE node.relative_path||'/%')
|
|
AND (file.ext=''
|
|
OR (ctx.has_dir AND NOT filename_is_filtered($1,file.relative_path,file.name))
|
|
OR (NOT ctx.has_dir AND (ctx.src_allow IS NULL OR file.ext = ANY(ctx.src_allow)) AND NOT file.ext = ANY(ctx.src_excl)))),
|
|
node.size_bytes,node.modified_at
|
|
FROM source_catalog_heads head
|
|
JOIN source_catalog_nodes node ON node.snapshot_id=head.snapshot_id
|
|
WHERE head.source_connection_id=$1 AND node.node_id=$2 AND node.node_type='folder'`, filter.ConnectionID, filter.ParentNodeID).Scan(
|
|
&page.Current.NodeID, &parentID, &page.Current.NodeType, &page.Current.Name, &page.Current.DirectChildCount,
|
|
&page.Current.SubtreeFileCount, &page.Current.SubtreeBytes, &page.Current.SizeBytes, &page.Current.ModifiedAt)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return governance.CatalogNodePage{}, ErrCatalogNodeNotFound
|
|
}
|
|
if err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("get catalog parent: %w", err)
|
|
}
|
|
if parentID != nil {
|
|
page.Current.ParentNodeID = *parentID
|
|
}
|
|
page.Current.Visibility = "unknown"
|
|
|
|
breadcrumbs, err := s.db.Query(ctx, `
|
|
WITH RECURSIVE trail AS (
|
|
SELECT node.snapshot_id,node.node_id,node.parent_node_id,node.name,0 AS depth
|
|
FROM source_catalog_heads head
|
|
JOIN source_catalog_nodes node ON node.snapshot_id=head.snapshot_id
|
|
WHERE head.source_connection_id=$1 AND node.node_id=$2
|
|
UNION ALL
|
|
SELECT parent.snapshot_id,parent.node_id,parent.parent_node_id,parent.name,trail.depth+1
|
|
FROM trail JOIN source_catalog_nodes parent
|
|
ON parent.snapshot_id=trail.snapshot_id AND parent.node_id=trail.parent_node_id
|
|
)
|
|
SELECT node_id,name FROM trail ORDER BY depth DESC`, filter.ConnectionID, filter.ParentNodeID)
|
|
if err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("list catalog breadcrumbs: %w", err)
|
|
}
|
|
for breadcrumbs.Next() {
|
|
var item governance.CatalogBreadcrumb
|
|
if err := breadcrumbs.Scan(&item.NodeID, &item.Name); err != nil {
|
|
breadcrumbs.Close()
|
|
return governance.CatalogNodePage{}, fmt.Errorf("scan catalog breadcrumb: %w", err)
|
|
}
|
|
page.Breadcrumbs = append(page.Breadcrumbs, item)
|
|
}
|
|
if err := breadcrumbs.Err(); err != nil {
|
|
breadcrumbs.Close()
|
|
return governance.CatalogNodePage{}, fmt.Errorf("iterate catalog breadcrumbs: %w", err)
|
|
}
|
|
breadcrumbs.Close()
|
|
|
|
if err := s.db.QueryRow(ctx, `
|
|
WITH filter_ctx AS MATERIALIZED (SELECT * FROM catalog_filter_context($1))
|
|
SELECT COUNT(*)::integer FROM source_catalog_heads head
|
|
JOIN source_catalog_nodes node ON node.snapshot_id=head.snapshot_id
|
|
LEFT JOIN filter_ctx ctx ON TRUE
|
|
WHERE head.source_connection_id=$1 AND node.parent_node_id=$2
|
|
AND (node.node_type='folder' OR node.ext=''
|
|
OR (ctx.has_dir AND NOT filename_is_filtered($1,node.relative_path,node.name))
|
|
OR (NOT ctx.has_dir AND (ctx.src_allow IS NULL OR node.ext = ANY(ctx.src_allow)) AND NOT node.ext = ANY(ctx.src_excl)))`, filter.ConnectionID, filter.ParentNodeID).Scan(&page.Total); err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("count catalog nodes: %w", err)
|
|
}
|
|
|
|
rows, err := s.db.Query(ctx, `
|
|
WITH RECURSIVE filter_ctx AS MATERIALIZED (
|
|
SELECT * FROM catalog_filter_context($1)
|
|
), current_head AS (
|
|
SELECT snapshot_id FROM source_catalog_heads WHERE source_connection_id=$1
|
|
), selected AS MATERIALIZED (
|
|
SELECT node.* FROM current_head head JOIN source_catalog_nodes node ON node.snapshot_id=head.snapshot_id
|
|
LEFT JOIN filter_ctx ctx ON TRUE
|
|
WHERE node.parent_node_id=$2
|
|
AND (node.node_type='folder' OR node.ext=''
|
|
OR (ctx.has_dir AND NOT filename_is_filtered($1,node.relative_path,node.name))
|
|
OR (NOT ctx.has_dir AND (ctx.src_allow IS NULL OR node.ext = ANY(ctx.src_allow)) AND NOT node.ext = ANY(ctx.src_excl)))
|
|
ORDER BY CASE node.node_type WHEN 'folder' THEN 0 ELSE 1 END,node.sort_order,lower(node.name),node.node_id
|
|
LIMIT $3 OFFSET $4
|
|
), descendants(ancestor_id,snapshot_id,node_id) AS (
|
|
SELECT node_id,snapshot_id,node_id FROM selected
|
|
UNION ALL
|
|
SELECT descendants.ancestor_id,child.snapshot_id,child.node_id
|
|
FROM descendants JOIN source_catalog_nodes child
|
|
ON child.snapshot_id=descendants.snapshot_id AND child.parent_node_id=descendants.node_id
|
|
), file_state AS (
|
|
SELECT descendants.ancestor_id,leaf.source_item_id,
|
|
COALESCE(item_scope.visibility,directory_scope.visibility,item.visibility) AS effective_visibility,
|
|
publication.id AS publication_id,publication.ingest_state,publication.processing_state,
|
|
current_version.content_state AS version_content_state,content.content_state AS object_content_state,
|
|
content.local_spool_ref,leaf.size_bytes
|
|
FROM descendants
|
|
CROSS JOIN filter_ctx ctx
|
|
JOIN source_catalog_nodes leaf ON leaf.snapshot_id=descendants.snapshot_id AND leaf.node_id=descendants.node_id AND leaf.node_type='file'
|
|
AND (leaf.ext=''
|
|
OR (ctx.has_dir AND NOT filename_is_filtered($1,leaf.relative_path,leaf.name))
|
|
OR (NOT ctx.has_dir AND (ctx.src_allow IS NULL OR leaf.ext = ANY(ctx.src_allow)) AND NOT leaf.ext = ANY(ctx.src_excl)))
|
|
LEFT JOIN source_items item ON item.id=leaf.source_item_id
|
|
LEFT JOIN source_item_versions current_version ON current_version.source_item_id=item.id AND current_version.revision=item.revision
|
|
LEFT JOIN content_objects content ON content.id=current_version.content_object_id
|
|
LEFT JOIN source_item_scope_overrides item_scope ON item_scope.source_item_id=item.id
|
|
LEFT JOIN LATERAL (
|
|
SELECT rule.visibility FROM directory_scope_rules rule
|
|
WHERE rule.source_connection_id=$1
|
|
AND (rule.directory_path='' OR leaf.relative_path=rule.directory_path
|
|
OR LEFT(leaf.relative_path,LENGTH(rule.directory_path)+1)=rule.directory_path||'/')
|
|
ORDER BY LENGTH(rule.directory_path) DESC LIMIT 1
|
|
) directory_scope ON TRUE
|
|
LEFT JOIN LATERAL (
|
|
SELECT publication.id,publication.ingest_state,publication.processing_state
|
|
FROM test_publications publication
|
|
JOIN source_item_versions version ON version.id=publication.source_item_version_id
|
|
WHERE version.source_item_id=item.id
|
|
ORDER BY publication.updated_at DESC,publication.id LIMIT 1
|
|
) publication ON TRUE
|
|
), stats AS (
|
|
SELECT ancestor_id,
|
|
COUNT(*)::integer AS eligible_file_count,
|
|
COALESCE(SUM(COALESCE(size_bytes,0)),0)::bigint AS eligible_bytes,
|
|
COUNT(DISTINCT source_item_id) FILTER (WHERE publication_id IS NOT NULL)::integer AS imported_count,
|
|
COUNT(DISTINCT source_item_id) FILTER (WHERE publication_id IS NOT NULL AND processing_state='completed')::integer AS parsed_count,
|
|
COUNT(DISTINCT source_item_id) FILTER (WHERE publication_id IS NOT NULL AND (ingest_state='uploading' OR processing_state IN ('processing','finalizing')))::integer AS active_count,
|
|
COUNT(DISTINCT source_item_id) FILTER (WHERE publication_id IS NOT NULL AND (ingest_state='failed' OR processing_state IN ('failed','stalled')))::integer AS failed_count,
|
|
COUNT(DISTINCT source_item_id) FILTER (WHERE version_content_state='verified' AND object_content_state='verified' AND local_spool_ref<>'')::integer AS routable_count,
|
|
CASE WHEN COUNT(DISTINCT effective_visibility) FILTER (WHERE effective_visibility IS NOT NULL)=0 THEN 'unknown'
|
|
WHEN COUNT(DISTINCT effective_visibility) FILTER (WHERE effective_visibility IS NOT NULL)=1 THEN MIN(effective_visibility) FILTER (WHERE effective_visibility IS NOT NULL)
|
|
ELSE 'mixed' END AS visibility
|
|
FROM file_state GROUP BY ancestor_id
|
|
)
|
|
SELECT selected.node_id,selected.parent_node_id,selected.node_type,selected.name,
|
|
CASE WHEN selected.node_type='file' THEN 0 ELSE (
|
|
SELECT COUNT(*)::integer FROM source_catalog_nodes child
|
|
LEFT JOIN filter_ctx ctx2 ON TRUE
|
|
WHERE child.snapshot_id=selected.snapshot_id AND child.parent_node_id=selected.node_id
|
|
AND (child.node_type='folder' OR child.ext=''
|
|
OR (ctx2.has_dir AND NOT filename_is_filtered($1,child.relative_path,child.name))
|
|
OR (NOT ctx2.has_dir AND (ctx2.src_allow IS NULL OR child.ext = ANY(ctx2.src_allow)) AND NOT child.ext = ANY(ctx2.src_excl)))
|
|
) END,
|
|
COALESCE(stats.eligible_file_count,0),COALESCE(stats.eligible_bytes,0),
|
|
selected.size_bytes,selected.modified_at,selected.content_sha256,COALESCE(stats.visibility,'unknown'),
|
|
COALESCE(stats.imported_count,0),COALESCE(stats.parsed_count,0),
|
|
COALESCE(stats.active_count,0),COALESCE(stats.failed_count,0),COALESCE(stats.routable_count,0)
|
|
FROM selected LEFT JOIN stats ON stats.ancestor_id=selected.node_id
|
|
ORDER BY CASE selected.node_type WHEN 'folder' THEN 0 ELSE 1 END,selected.sort_order,lower(selected.name),selected.node_id`, filter.ConnectionID, filter.ParentNodeID, filter.Limit, filter.Offset)
|
|
if err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("list catalog nodes: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var item governance.CatalogNode
|
|
var parent *string
|
|
if err := rows.Scan(&item.NodeID, &parent, &item.NodeType, &item.Name, &item.DirectChildCount, &item.SubtreeFileCount, &item.SubtreeBytes, &item.SizeBytes, &item.ModifiedAt, &item.ContentSHA256, &item.Visibility, &item.ImportedCount, &item.ParsedCount, &item.ActiveCount, &item.FailedCount, &item.RoutableCount); err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("scan catalog node: %w", err)
|
|
}
|
|
if parent != nil {
|
|
item.ParentNodeID = *parent
|
|
}
|
|
page.Items = append(page.Items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("iterate catalog nodes: %w", err)
|
|
}
|
|
return page, nil
|
|
}
|
|
|
|
var ErrCatalogSnapshotNotFound = errors.New("source catalog snapshot not found")
|
|
|
|
// ListCatalogSnapshots returns the persisted snapshot history for one configured
|
|
// connection. The head snapshot is returned first (current view), followed by
|
|
// older snapshots newest-first. It never returns locator hashes or snapshot hashes.
|
|
func (s *Store) ListCatalogSnapshots(ctx context.Context, connectionID string, limit, offset int) (governance.CatalogSnapshotPage, error) {
|
|
if connectionID == "" || len(connectionID) > 36 || limit < 1 || limit > 100 || offset < 0 || offset > 1000000 {
|
|
return governance.CatalogSnapshotPage{}, errors.New("invalid catalog snapshot filter")
|
|
}
|
|
var exists bool
|
|
if err := s.db.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM source_connections WHERE id=$1)`, connectionID).Scan(&exists); err != nil {
|
|
return governance.CatalogSnapshotPage{}, fmt.Errorf("inspect snapshot connection: %w", err)
|
|
}
|
|
if !exists {
|
|
return governance.CatalogSnapshotPage{}, ErrCatalogSnapshotNotFound
|
|
}
|
|
page := governance.CatalogSnapshotPage{Items: []governance.CatalogSnapshot{}, Limit: limit, Offset: offset}
|
|
if err := s.db.QueryRow(ctx, `SELECT COUNT(*)::integer FROM source_catalog_snapshots WHERE source_connection_id=$1`, connectionID).Scan(&page.Total); err != nil {
|
|
return governance.CatalogSnapshotPage{}, fmt.Errorf("count catalog snapshots: %w", err)
|
|
}
|
|
rows, err := s.db.Query(ctx, `
|
|
SELECT snapshot.id,snapshot.source_connection_id,root.node_id,snapshot.source_label,snapshot.node_count,
|
|
snapshot.folder_count,snapshot.file_count,snapshot.observed_at,snapshot.created_at,
|
|
(head.snapshot_id IS NOT NULL) AS is_head
|
|
FROM source_catalog_snapshots snapshot
|
|
JOIN source_catalog_nodes root ON root.snapshot_id=snapshot.id AND root.parent_node_id IS NULL
|
|
LEFT JOIN source_catalog_heads head ON head.source_connection_id=snapshot.source_connection_id AND head.snapshot_id=snapshot.id
|
|
WHERE snapshot.source_connection_id=$1
|
|
ORDER BY is_head DESC,snapshot.observed_at DESC,snapshot.id DESC
|
|
LIMIT $2 OFFSET $3`, connectionID, limit, offset)
|
|
if err != nil {
|
|
return governance.CatalogSnapshotPage{}, fmt.Errorf("list catalog snapshots: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var item governance.CatalogSnapshot
|
|
if err := rows.Scan(&item.SnapshotID, &item.ConnectionID, &item.RootNodeID, &item.SourceName, &item.NodeCount, &item.FolderCount, &item.FileCount, &item.ObservedAt, &item.CreatedAt, &item.IsHead); err != nil {
|
|
return governance.CatalogSnapshotPage{}, fmt.Errorf("scan catalog snapshot: %w", err)
|
|
}
|
|
page.Items = append(page.Items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return governance.CatalogSnapshotPage{}, fmt.Errorf("iterate catalog snapshots: %w", err)
|
|
}
|
|
return page, nil
|
|
}
|
|
|
|
// ListCatalogNodesInSnapshot is a read-only view of one persisted snapshot. It never
|
|
// mutates the current head. content_sha256 is returned only when actually recorded;
|
|
// unknown files return null so the caller can render "内容Hash未获取".
|
|
func (s *Store) ListCatalogNodesInSnapshot(ctx context.Context, filter governance.CatalogSnapshotNodeFilter) (governance.CatalogNodePage, error) {
|
|
if filter.ConnectionID == "" || len(filter.ConnectionID) > 36 || filter.SnapshotID == "" || len(filter.SnapshotID) > 36 ||
|
|
filter.ParentNodeID == "" || len(filter.ParentNodeID) > 36 || filter.Limit < 1 || filter.Limit > 100 || filter.Offset < 0 || filter.Offset > 1000000 {
|
|
return governance.CatalogNodePage{}, errors.New("invalid catalog snapshot node filter")
|
|
}
|
|
var parent governance.CatalogNode
|
|
var parentID *string
|
|
err := s.db.QueryRow(ctx, `
|
|
SELECT node.node_id,node.parent_node_id,node.node_type,node.name,node.direct_child_count,
|
|
node.subtree_file_count,node.subtree_bytes,node.size_bytes,node.modified_at,node.content_sha256
|
|
FROM source_catalog_snapshots snapshot
|
|
JOIN source_catalog_nodes node ON node.snapshot_id=snapshot.id
|
|
WHERE snapshot.id=$1 AND snapshot.source_connection_id=$2 AND node.node_id=$3 AND node.node_type='folder'`,
|
|
filter.SnapshotID, filter.ConnectionID, filter.ParentNodeID).Scan(
|
|
&parent.NodeID, &parentID, &parent.NodeType, &parent.Name, &parent.DirectChildCount,
|
|
&parent.SubtreeFileCount, &parent.SubtreeBytes, &parent.SizeBytes, &parent.ModifiedAt, &parent.ContentSHA256)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return governance.CatalogNodePage{}, ErrCatalogSnapshotNotFound
|
|
}
|
|
if err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("get snapshot parent: %w", err)
|
|
}
|
|
page := governance.CatalogNodePage{Current: parent, Limit: filter.Limit, Offset: filter.Offset}
|
|
if parentID != nil {
|
|
page.Current.ParentNodeID = *parentID
|
|
}
|
|
page.Current.Visibility = "unknown"
|
|
if err := s.db.QueryRow(ctx, `
|
|
SELECT COUNT(*)::integer FROM source_catalog_nodes
|
|
WHERE snapshot_id=$1 AND parent_node_id=$2`, filter.SnapshotID, filter.ParentNodeID).Scan(&page.Total); err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("count snapshot nodes: %w", err)
|
|
}
|
|
rows, err := s.db.Query(ctx, `
|
|
SELECT node_id,parent_node_id,node_type,name,direct_child_count,subtree_file_count,subtree_bytes,size_bytes,modified_at,content_sha256
|
|
FROM source_catalog_nodes
|
|
WHERE snapshot_id=$1 AND parent_node_id=$2
|
|
ORDER BY CASE node_type WHEN 'folder' THEN 0 ELSE 1 END,sort_order,lower(name),node_id
|
|
LIMIT $3 OFFSET $4`, filter.SnapshotID, filter.ParentNodeID, filter.Limit, filter.Offset)
|
|
if err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("list snapshot nodes: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var item governance.CatalogNode
|
|
var parentPtr *string
|
|
if err := rows.Scan(&item.NodeID, &parentPtr, &item.NodeType, &item.Name, &item.DirectChildCount, &item.SubtreeFileCount, &item.SubtreeBytes, &item.SizeBytes, &item.ModifiedAt, &item.ContentSHA256); err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("scan snapshot node: %w", err)
|
|
}
|
|
if parentPtr != nil {
|
|
item.ParentNodeID = *parentPtr
|
|
}
|
|
item.Visibility = "unknown"
|
|
page.Items = append(page.Items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return governance.CatalogNodePage{}, fmt.Errorf("iterate snapshot nodes: %w", err)
|
|
}
|
|
return page, nil
|
|
}
|