knowledge-governance/internal/platform/postgres/source_assets.go

package postgres

import (
	"context"
	"fmt"
	"strings"
	"unicode"

	"knowledge-governance/internal/governance"
)

func (s *Store) ListSourceAssets(ctx context.Context, filter governance.SourceAssetFilter) (governance.SourceAssetPage, error) {
	var page governance.SourceAssetPage
	if err := s.db.QueryRow(ctx, `
		SELECT
			COUNT(*),
			COUNT(*) FILTER (WHERE v.raw_record_summary->>'legacy_status' = 'completed'),
			COUNT(*) FILTER (WHERE v.raw_record_summary->>'legacy_status' = 'content_alias'),
			COUNT(*) FILTER (WHERE v.raw_record_summary->>'legacy_status' = 'over_limit')
		FROM source_items i
		JOIN source_item_versions v ON v.source_item_id = i.id AND v.revision = i.revision`).Scan(
		&page.Summary.Total,
		&page.Summary.Completed,
		&page.Summary.ContentAlias,
		&page.Summary.OverLimit,
	); err != nil {
		return governance.SourceAssetPage{}, fmt.Errorf("summarize source assets: %w", err)
	}

	rows, err := s.db.Query(ctx, `
		SELECT
			i.id,
			i.title,
			COALESCE(i.normalized_path, ''),
			i.source_connection_id,
			CASE
				WHEN item_scope.source_item_id IS NOT NULL THEN 'file_override'
				WHEN directory_scope.source_connection_id IS NOT NULL THEN 'directory_rule'
				ELSE i.visibility_origin
			END,
			COALESCE(directory_scope.directory_path, ''),
			COALESCE(p.ingest_state, ''),
			COALESCE(NULLIF(p.last_error, ''), v.download_error, ''),
			COALESCE(p.id, ''),
			COALESCE(((p.processing_state IN ('failed','stalled') OR (p.ingest_state = 'failed' AND p.processing_state = 'pending'))
			 AND co.content_state = 'verified' AND co.local_spool_ref <> ''), false),
			c.source_type,
			c.name,
			COALESCE(i.size_bytes, 0),
			COALESCE(item_scope.visibility, directory_scope.visibility, i.visibility),
			i.source_change,
			v.content_state,
			COALESCE(v.raw_record_summary->>'legacy_status', ''),
			p.id IS NOT NULL,
			COALESCE(p.processing_state, ''),
			COALESCE(p.retrieval_state, ''),
			GREATEST(i.updated_at, v.updated_at, COALESCE(p.updated_at, i.updated_at))
		FROM source_items i
		JOIN source_connections c ON c.id = i.source_connection_id
		JOIN source_item_versions v ON v.source_item_id = i.id AND v.revision = i.revision
		LEFT JOIN LATERAL (
			SELECT id, ingest_state, processing_state, retrieval_state, last_error, updated_at
			FROM test_publications
			WHERE source_item_version_id = v.id
			ORDER BY updated_at DESC, id
			LIMIT 1
		) p ON TRUE
		LEFT JOIN content_objects co ON co.id = v.content_object_id
		LEFT JOIN source_item_scope_overrides item_scope ON item_scope.source_item_id = i.id
		LEFT JOIN LATERAL (
			SELECT rule.source_connection_id, rule.directory_path, rule.visibility
			FROM directory_scope_rules rule
			WHERE rule.source_connection_id = i.source_connection_id
			  AND (
				rule.directory_path = ''
				OR i.normalized_path = rule.directory_path
				OR LEFT(i.normalized_path, LENGTH(rule.directory_path) + 1) = rule.directory_path || '/'
			  )
			ORDER BY LENGTH(rule.directory_path) DESC
			LIMIT 1
		) directory_scope ON TRUE
		WHERE ($1 = '' OR v.raw_record_summary->>'legacy_status' = $1)
		ORDER BY i.updated_at DESC, i.id
		LIMIT $2 OFFSET $3`, filter.State, filter.Limit, filter.Offset)
	if err != nil {
		return governance.SourceAssetPage{}, fmt.Errorf("list source assets: %w", err)
	}
	defer rows.Close()

	page.Items = make([]governance.SourceAsset, 0)
	for rows.Next() {
		var item governance.SourceAsset
		var sourceType, visibility, sourceChange, contentState string
		if err := rows.Scan(
			&item.ID,
			&item.Title,
			&item.DirectoryPath,
			&item.ConnectionID,
			&item.VisibilityOrigin,
			&item.ScopeRulePath,
			&item.IngestState,
			&item.LastError,
			&item.PublicationID,
			&item.CanRetry,
			&sourceType,
			&item.ConnectionName,
			&item.SizeBytes,
			&visibility,
			&sourceChange,
			&contentState,
			&item.LegacyStatus,
			&item.HasPublication,
			&item.ProcessingState,
			&item.RetrievalState,
			&item.UpdatedAt,
		); err != nil {
			return governance.SourceAssetPage{}, fmt.Errorf("scan source asset: %w", err)
		}
		item.SourceType = governance.SourceType(sourceType)
		item.DirectoryPath = safeDocumentDirectory(item.DirectoryPath)
		item.Visibility = governance.Visibility(visibility)
		item.SourceChange = governance.SourceChange(sourceChange)
		item.ContentState = governance.ContentState(contentState)
		page.Items = append(page.Items, item)
	}
	if err := rows.Err(); err != nil {
		return governance.SourceAssetPage{}, fmt.Errorf("iterate source assets: %w", err)
	}
	return page, nil
}

// Expose only relative source labels, never host paths or signed URLs.
func safeDocumentDirectory(value string) string {
	if strings.HasPrefix(value, "/") || strings.ContainsAny(value, ":\\") || len(value) > 2048 {
		return ""
	}
	for _, part := range strings.Split(value, "/") {
		if part == ".." || part == "." {
			return ""
		}
	}
	for _, r := range value {
		if unicode.IsControl(r) {
			return ""
		}
	}
	return value
}