姚章浩
2c7265efc4
build / build (push) Successful in 58s
主人要求：去掉入口，统一成 dws 导入与 PP 导入；其他来源连数据一起下架，之后用 dws 重新获取。
界面（web）：
- 「数据源配置」只剩两块：**dws 导入（钉钉）**（按当前钉钉账号列出可访问的知识库与钉盘空间）
与 **PP 导入**（PP 资料下载中心）。上一步加的「显示 AList／钉钉知识库来源」折叠开关删除。
- 来源卡片新增管理员可见的「去掉来源」（带确认弹窗，说明是归档下架、数据保留可恢复）。
- 侧栏文案改成"只留两个入口"；发现面板标题改为「dws 导入（钉钉）」。
后端（归档语义，可逆）：
- 迁移 000047：source_connections 增 archived_at / archived_by + 只覆盖未归档行的部分索引。
- store：ArchiveSourceConnection / RestoreSourceConnection（幂等；归档同时停用并清空
next_refresh_at，避免被手工改回后立刻带着旧调度跑）。列表、目录、计划刷新（claimDue）、
手动扫描（triggerBlock/inspect）四类查询一律排除归档行。
- 注册路径（dws 重新获取）：RegisterDingtalkSpace 的 ON CONFLICT 清掉归档标记 ——
重新登记同一个空间**原地复活同一行**，不产生重复，历史快照仍属于同一行。
- HTTP：POST /api/v1/source-connections/{id}/archive|restore（管理员专属，写意图头
source-archive / source-restore + Idempotency-Key；404 区分"不存在"与"已是目标状态"）；
openapi 同步两条契约。
验证：
- 集成测试（真 postgres）：归档从列表消失但 source_items 一行不少、重复归档不漂移归档时间与人、
未知 ID 报 ErrSourceConnectionNotFound、归档来源不能再被扫描、恢复后回到列表但保持停用；
dws 重新登记同一空间复活同一行。
- httpapi：写意图/幂等键门禁、200/403/404/500、错误体不回显内部细节、classifyRole 管理员专属。
- 迁移护栏：000047 结构 + 四类 SQL 必须含 archived_at is null（漏一条就是入口没去掉）。
- make verify 15/15。
416 lines
23 KiB
Go
416 lines
23 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 connection.archived_at IS NULL 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
|
|
WHERE connection.archived_at IS NULL
|
|
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
|
|
}
|