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

package postgres

import (
	"context"
	"fmt"

	"github.com/jackc/pgx/v5/pgxpool"

	ppconnector "knowledge-governance/internal/connectors/pp"
)

type LegacyImportReport struct {
	Connections    int
	SourceItems    int
	ContentObjects int
	Versions       int
	Publications   int
}

func LoadPPCurrentItems(ctx context.Context, pool *pgxpool.Pool, connectionID string) (map[string]ppconnector.CurrentItem, error) {
	rows, err := pool.Query(ctx, `
		SELECT stable_external_id, revision
		FROM source_items
		WHERE source_connection_id = $1
		ORDER BY stable_external_id`, connectionID)
	if err != nil {
		return nil, fmt.Errorf("query current PP items: %w", err)
	}
	defer rows.Close()

	items := make(map[string]ppconnector.CurrentItem)
	for rows.Next() {
		var sourceHash string
		var item ppconnector.CurrentItem
		if err := rows.Scan(&sourceHash, &item.Revision); err != nil {
			return nil, fmt.Errorf("scan current PP item: %w", err)
		}
		items[sourceHash] = item
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate current PP items: %w", err)
	}
	return items, nil
}

func ImportPPLegacy(ctx context.Context, pool *pgxpool.Pool, plan ppconnector.ImportPlan) (LegacyImportReport, error) {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return LegacyImportReport{}, fmt.Errorf("begin PP legacy import: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()

	report := LegacyImportReport{}
	tag, err := tx.Exec(ctx, `INSERT INTO source_connections (
		id, source_type, name, base_url, secret_ref, enabled, schedule, last_health
	) VALUES ($1, $2, $3, $4, $5, $6, '{}'::jsonb, '{}'::jsonb)
	ON CONFLICT DO NOTHING`,
		plan.Connection.ID,
		plan.Connection.Type,
		plan.Connection.Name,
		plan.Connection.BaseURL,
		plan.Connection.SecretRef,
		plan.Connection.Enabled,
	)
	if err != nil {
		return LegacyImportReport{}, fmt.Errorf("insert PP source connection: %w", err)
	}
	report.Connections = int(tag.RowsAffected())

	for _, content := range plan.ContentObjects {
		tag, err = tx.Exec(ctx, `INSERT INTO content_objects (
			id, sha256, size_bytes, mime, validated_format, local_spool_ref, content_state
		) VALUES ($1, $2, $3, $4, $5, $6, $7)
		ON CONFLICT DO NOTHING`,
			content.ID,
			content.SHA256,
			content.SizeBytes,
			content.MIME,
			content.ValidatedFormat,
			content.LocalSpoolRef,
			content.State,
		)
		if err != nil {
			return LegacyImportReport{}, fmt.Errorf("insert PP content object: %w", err)
		}
		report.ContentObjects += int(tag.RowsAffected())
	}

	for _, item := range plan.SourceItems {
		tag, err = tx.Exec(ctx, `INSERT INTO source_items (
			id, source_connection_id, stable_external_id, canonical_uri_hash,
			parent_external_id, normalized_path, title, item_type, size_bytes,
			mime, revision, visibility, visibility_origin, last_seen_scan_id,
			source_change, absence_streak, identity_confidence
		) VALUES (
			$1, $2, $3, $4, $5, $6, $7, $8, $9,
			$10, $11, $12, $13, $14, $15, $16, 'stable_id'
		)
		ON CONFLICT DO NOTHING`,
			item.ID,
			item.SourceConnectionID,
			item.StableExternalID,
			item.CanonicalURIHash,
			item.ParentExternalID,
			item.NormalizedPath,
			item.Title,
			item.ItemType,
			item.SizeBytes,
			item.MIME,
			item.Revision,
			item.Visibility,
			item.VisibilityOrigin,
			item.LastSeenScanID,
			item.SourceChange,
			item.AbsenceStreak,
		)
		if err != nil {
			return LegacyImportReport{}, fmt.Errorf("insert PP source item: %w", err)
		}
		report.SourceItems += int(tag.RowsAffected())
	}

	for _, version := range plan.Versions {
		tag, err = tx.Exec(ctx, `INSERT INTO source_item_versions (
			id, source_item_id, revision, content_object_id, content_state,
			download_error, raw_record_summary, source_updated_at
		) VALUES ($1, $2, $3, $4, $5, $6, CAST($7 AS JSONB), $8)
		ON CONFLICT DO NOTHING`,
			version.ID,
			version.SourceItemID,
			version.Revision,
			version.ContentObjectID,
			version.ContentState,
			version.DownloadError,
			string(version.RawRecordSummary),
			version.SourceUpdatedAt,
		)
		if err != nil {
			return LegacyImportReport{}, fmt.Errorf("insert PP source item version: %w", err)
		}
		report.Versions += int(tag.RowsAffected())
	}

	for _, publication := range plan.Publications {
		tag, err = tx.Exec(ctx, `INSERT INTO test_publications (
			id, source_item_version_id, test_kb_id, knowledge_id,
			ingest_state, processing_state, retrieval_state,
			parser_generation, index_generation, last_error, retry_count
		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
		ON CONFLICT DO NOTHING`,
			publication.ID,
			publication.SourceItemVersionID,
			publication.TestKBID,
			publication.KnowledgeID,
			publication.IngestState,
			publication.ProcessingState,
			publication.RetrievalState,
			publication.ParserGeneration,
			publication.IndexGeneration,
			publication.LastError,
			publication.RetryCount,
		)
		if err != nil {
			return LegacyImportReport{}, fmt.Errorf("insert PP test publication: %w", err)
		}
		report.Publications += int(tag.RowsAffected())
	}

	if err := tx.Commit(ctx); err != nil {
		return LegacyImportReport{}, fmt.Errorf("commit PP legacy import: %w", err)
	}
	return report, nil
}