姚章浩
394 lines
18 KiB
Go
394 lines
18 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
|
|
"knowledge-governance/internal/governance"
|
|
)
|
|
|
|
// fulfillCatalogIntentsTx is the policy-driven ingestion core: saving an
|
|
// annotation in 来源与目录 records one durable intent per selected file and,
|
|
// for importable policies, materializes a versioned source item, enqueues an
|
|
// artifact.fetch job (bytes come from the live source; a matching local copy
|
|
// short-circuits the download) and creates one 'planned' publication per routed
|
|
// knowledge base. Planned publications are what 解析监控 lists as 待入库;
|
|
// 开始处理 promotes them within the configured concurrency budget. Blocked
|
|
// intents retire any planned publication and flip the item policy so the
|
|
// knowledge-base diff demands removal of anything still live.
|
|
func fulfillCatalogIntentsTx(ctx context.Context, tx pgx.Tx, request governance.CatalogRouteRequest, files []governance.CatalogRouteFile, now time.Time) (governance.CatalogRouteResult, error) {
|
|
targetKnowledgeBaseIDs := request.TargetKnowledgeBaseIDs
|
|
if len(targetKnowledgeBaseIDs) == 0 && request.TargetKnowledgeBaseID != "" {
|
|
targetKnowledgeBaseIDs = []string{request.TargetKnowledgeBaseID}
|
|
}
|
|
result := governance.CatalogRouteResult{SelectedFiles: len(files)}
|
|
for _, file := range files {
|
|
recorded, fetchQueued, planned, blocked, err := fulfillCatalogIntent(ctx, tx, request, file, targetKnowledgeBaseIDs, now)
|
|
if err != nil {
|
|
return governance.CatalogRouteResult{}, err
|
|
}
|
|
if !recorded {
|
|
result.UnboundFiles++
|
|
continue
|
|
}
|
|
result.IntentsRecorded++
|
|
if blocked {
|
|
result.BlockedItems++
|
|
continue
|
|
}
|
|
if fetchQueued {
|
|
result.FetchQueued++
|
|
}
|
|
result.PlannedPublications += planned
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
// fulfillCatalogIntent applies one file decision. It returns recorded=false
|
|
// only when the node vanished mid-transaction; fetchQueued reports a newly
|
|
// enqueued artifact.fetch job and planned the number of publications created.
|
|
func fulfillCatalogIntent(ctx context.Context, tx pgx.Tx, request governance.CatalogRouteRequest, file governance.CatalogRouteFile, targetKnowledgeBaseIDs []string, now time.Time) (bool, bool, int, bool, error) {
|
|
policy := string(request.Visibility)
|
|
if _, err := tx.Exec(ctx, `INSERT INTO catalog_import_intents(
|
|
source_connection_id,node_id,visibility,decided_by,reason,relative_path,name,size_bytes,modified_at,content_sha256,
|
|
locator_hash,fetch_state,last_error,created_at,updated_at)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,'pending','',NOW(),NOW())
|
|
ON CONFLICT(source_connection_id,node_id) DO UPDATE SET
|
|
visibility=EXCLUDED.visibility,decided_by=EXCLUDED.decided_by,reason=EXCLUDED.reason,
|
|
relative_path=EXCLUDED.relative_path,name=EXCLUDED.name,size_bytes=EXCLUDED.size_bytes,
|
|
modified_at=EXCLUDED.modified_at,content_sha256=EXCLUDED.content_sha256,
|
|
locator_hash=EXCLUDED.locator_hash,
|
|
fetch_state=CASE WHEN catalog_import_intents.content_sha256 IS DISTINCT FROM EXCLUDED.content_sha256
|
|
OR catalog_import_intents.size_bytes IS DISTINCT FROM EXCLUDED.size_bytes
|
|
OR catalog_import_intents.modified_at IS DISTINCT FROM EXCLUDED.modified_at
|
|
THEN 'pending' ELSE catalog_import_intents.fetch_state END,
|
|
last_error=CASE WHEN catalog_import_intents.content_sha256 IS DISTINCT FROM EXCLUDED.content_sha256
|
|
OR catalog_import_intents.size_bytes IS DISTINCT FROM EXCLUDED.size_bytes
|
|
OR catalog_import_intents.modified_at IS DISTINCT FROM EXCLUDED.modified_at
|
|
THEN '' ELSE catalog_import_intents.last_error END,
|
|
updated_at=NOW()`,
|
|
request.ConnectionID, file.NodeID, policy, request.OperatorRef, request.Reason,
|
|
file.RelativePath, file.Name, file.SizeBytes, file.ModifiedAt, file.ContentSHA256, file.LocatorHash); err != nil {
|
|
return false, false, 0, false, fmt.Errorf("record catalog intent: %w", err)
|
|
}
|
|
|
|
var existingItemID *string
|
|
err := tx.QueryRow(ctx, `SELECT source_item_id FROM catalog_import_intents
|
|
WHERE source_connection_id=$1 AND node_id=$2 FOR UPDATE`, request.ConnectionID, file.NodeID).Scan(&existingItemID)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return false, false, 0, false, nil
|
|
}
|
|
if err != nil {
|
|
return false, false, 0, false, fmt.Errorf("read catalog intent: %w", err)
|
|
}
|
|
if existingItemID == nil {
|
|
// The node itself may already carry an item binding from an earlier
|
|
// scanner generation (or a sibling intent); reuse it instead of
|
|
// materializing a duplicate item and refetching identical bytes.
|
|
var nodeItemID *string
|
|
lookup := tx.QueryRow(ctx, `SELECT node.source_item_id
|
|
FROM source_catalog_heads head JOIN source_catalog_nodes node
|
|
ON node.snapshot_id=head.snapshot_id AND node.node_type='file'
|
|
WHERE head.source_connection_id=$1 AND node.node_id=$2
|
|
AND node.source_item_id IS NOT NULL
|
|
ORDER BY head.activated_at DESC NULLS LAST LIMIT 1 FOR UPDATE OF node`, request.ConnectionID, file.NodeID).Scan(&nodeItemID)
|
|
if lookup == nil {
|
|
existingItemID = nodeItemID
|
|
} else if !errors.Is(lookup, pgx.ErrNoRows) {
|
|
return false, false, 0, false, fmt.Errorf("read catalog node item: %w", lookup)
|
|
}
|
|
}
|
|
|
|
itemID := ""
|
|
if existingItemID != nil {
|
|
itemID = *existingItemID
|
|
}
|
|
if itemID == "" {
|
|
newItemID, err := newOpaqueID("si-")
|
|
if err != nil {
|
|
return false, false, 0, false, err
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO source_items(
|
|
id,source_connection_id,stable_external_id,canonical_uri_hash,title,item_type,revision,
|
|
visibility,visibility_origin,source_change,absence_streak,identity_confidence,
|
|
policy_state,created_at,updated_at)
|
|
VALUES($1,$2,$3,$4,$5,'file','',$7,'configured_scope','unchanged',0,'object_id',$6,NOW(),NOW())`,
|
|
newItemID, request.ConnectionID, "cat:"+file.NodeID,
|
|
sha256Hex("policy:"+request.ConnectionID+"\x00"+file.NodeID),
|
|
file.Name, policy, itemVisibilityFor(request.Visibility)); err != nil {
|
|
return false, false, 0, false, fmt.Errorf("materialize policy source item: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE catalog_import_intents SET source_item_id=$3 WHERE source_connection_id=$1 AND node_id=$2`,
|
|
request.ConnectionID, file.NodeID, newItemID); err != nil {
|
|
return false, false, 0, false, fmt.Errorf("bind catalog intent item: %w", err)
|
|
}
|
|
itemID = newItemID
|
|
}
|
|
// Bind the snapshot nodes too (new or reused item): the browser's
|
|
// 已导入/已解析 counters join source_catalog_nodes.source_item_id; without
|
|
// this write every catalog-only source keeps showing "尚未导入" even after
|
|
// the document finished parsing. node_id is deterministic per path, so we
|
|
// bind across all snapshots of this connection, current and future heads.
|
|
if _, err := tx.Exec(ctx, `UPDATE source_catalog_nodes node SET source_item_id=$3
|
|
FROM source_catalog_snapshots snap
|
|
WHERE snap.source_connection_id=$1 AND node.snapshot_id=snap.id
|
|
AND node.node_type='file' AND node.node_id=$2 AND node.source_item_id IS NULL`,
|
|
request.ConnectionID, file.NodeID, itemID); err != nil {
|
|
return false, false, 0, false, fmt.Errorf("bind catalog node item: %w", err)
|
|
}
|
|
|
|
if request.Visibility == governance.VisibilityBlocked {
|
|
if _, err := tx.Exec(ctx, `UPDATE source_items SET policy_state='blocked',updated_at=NOW() WHERE id=$1`, itemID); err != nil {
|
|
return false, false, 0, false, fmt.Errorf("block policy item: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE test_publications SET superseded_at=$2,updated_at=$2
|
|
WHERE source_item_version_id IN (SELECT id FROM source_item_versions WHERE source_item_id=$1)
|
|
AND superseded_at IS NULL AND processing_state='planned'`, itemID, now); err != nil {
|
|
return false, false, 0, false, fmt.Errorf("retire blocked planned publications: %w", err)
|
|
}
|
|
return true, false, 0, true, nil
|
|
}
|
|
|
|
revision := "locator:" + file.LocatorHash
|
|
if file.ContentSHA256 != nil && *file.ContentSHA256 != "" {
|
|
revision = "sha:" + *file.ContentSHA256
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE source_items
|
|
SET policy_state=$2,visibility=$3,visibility_origin='configured_scope',updated_at=NOW() WHERE id=$1`,
|
|
itemID, policy, itemVisibilityFor(request.Visibility)); err != nil {
|
|
return false, false, 0, false, fmt.Errorf("update policy item: %w", err)
|
|
}
|
|
|
|
// Resolve the version this intent points at. A bound item whose current
|
|
// version already carries verified bytes is reused without any fetch — the
|
|
// 本地已有先比对 requirement. Only when the node proves different content
|
|
// (scan-reported hash mismatch) or bytes were never fetched do we create a
|
|
// revision and enqueue artifact.fetch.
|
|
versionID := ""
|
|
contentState := ""
|
|
var curID, curSHA string
|
|
err = tx.QueryRow(ctx, `SELECT version.id,COALESCE(content.sha256,''),version.content_state
|
|
FROM source_items item
|
|
JOIN source_item_versions version ON version.source_item_id=item.id AND version.revision=item.revision
|
|
LEFT JOIN content_objects content ON content.id=version.content_object_id
|
|
WHERE item.id=$1`, itemID).Scan(&curID, &curSHA, &contentState)
|
|
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
|
return false, false, 0, false, fmt.Errorf("inspect current item version: %w", err)
|
|
}
|
|
nodeSHA := ""
|
|
if file.ContentSHA256 != nil {
|
|
nodeSHA = *file.ContentSHA256
|
|
}
|
|
reuse := err == nil && contentState == string(governance.ContentVerified) && (nodeSHA == "" || nodeSHA == curSHA)
|
|
if !reuse {
|
|
contentState = "unknown"
|
|
err = tx.QueryRow(ctx, `SELECT version.id,version.content_state
|
|
FROM source_item_versions version WHERE version.source_item_id=$1 AND version.revision=$2`,
|
|
itemID, revision).Scan(&versionID, &contentState)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
newVersionID, err := newOpaqueID("sv-")
|
|
if err != nil {
|
|
return false, false, 0, false, err
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO source_item_versions(
|
|
id,source_item_id,revision,content_object_id,content_state,raw_record_summary,source_updated_at,created_at,updated_at)
|
|
VALUES($1,$2,$3,NULL,'unknown','{}'::jsonb,$4,NOW(),NOW())`, newVersionID, itemID, revision, file.ModifiedAt); err != nil {
|
|
return false, false, 0, false, fmt.Errorf("create policy version: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE source_items SET revision=$2,updated_at=NOW() WHERE id=$1`, itemID, revision); err != nil {
|
|
return false, false, 0, false, fmt.Errorf("advance policy item revision: %w", err)
|
|
}
|
|
versionID = newVersionID
|
|
} else if err != nil {
|
|
return false, false, 0, false, fmt.Errorf("lookup policy version: %w", err)
|
|
}
|
|
} else {
|
|
versionID = curID
|
|
}
|
|
|
|
planned := 0
|
|
for _, kbID := range targetKnowledgeBaseIDs {
|
|
scope := kbScopeFor(request, kbID)
|
|
publicationID, err := newOpaqueID("tp-")
|
|
if err != nil {
|
|
return false, false, 0, false, err
|
|
}
|
|
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,target_scope,created_at,updated_at)
|
|
VALUES($1,$2,$3,'','pending','planned','unpublished','','','',0,$4,NOW(),NOW())
|
|
ON CONFLICT (source_item_version_id,test_kb_id) DO NOTHING`,
|
|
publicationID, versionID, kbID, scope)
|
|
if err != nil {
|
|
return false, false, 0, false, fmt.Errorf("plan publication: %w", err)
|
|
}
|
|
planned += int(tag.RowsAffected())
|
|
}
|
|
|
|
fetchQueued := false
|
|
if contentState != string(governance.ContentVerified) {
|
|
jobID, err := newOpaqueID("job-")
|
|
if err != nil {
|
|
return false, false, 0, false, err
|
|
}
|
|
payload, err := json.Marshal(map[string]string{"source_connection_id": request.ConnectionID, "node_id": file.NodeID, "version_id": versionID})
|
|
if err != nil {
|
|
return false, false, 0, false, errors.New("encode artifact fetch job")
|
|
}
|
|
tag, err := tx.Exec(ctx, `INSERT INTO jobs(id,type,payload,state,priority,available_at,max_attempts,idempotency_key,created_at,updated_at)
|
|
VALUES($1,'artifact.fetch',$2::jsonb,'pending',20,$3,8,$4,$3,$3) ON CONFLICT (idempotency_key) DO NOTHING`,
|
|
jobID, payload, now, "artifact-fetch:"+versionID)
|
|
if err != nil {
|
|
return false, false, 0, false, fmt.Errorf("enqueue artifact fetch: %w", err)
|
|
}
|
|
fetchQueued = tag.RowsAffected() > 0
|
|
if fetchQueued {
|
|
if _, err := tx.Exec(ctx, `UPDATE catalog_import_intents SET fetch_state='pending',last_error='',updated_at=NOW()
|
|
WHERE source_connection_id=$1 AND node_id=$2 AND fetch_state IN ('failed','missing_local','fetching')`, request.ConnectionID, file.NodeID); err != nil {
|
|
return false, false, 0, false, fmt.Errorf("reset catalog intent fetch state: %w", err)
|
|
}
|
|
}
|
|
}
|
|
return true, fetchQueued, planned, false, nil
|
|
}
|
|
|
|
// kbScopeFor maps a routed knowledge base id to its diff scope label.
|
|
func kbScopeFor(request governance.CatalogRouteRequest, knowledgeBaseID string) string {
|
|
if request.Visibility == governance.VisibilityExternal {
|
|
return "external"
|
|
}
|
|
if request.Visibility == governance.VisibilityBoth && len(request.TargetKnowledgeBaseIDs) > 1 &&
|
|
knowledgeBaseID == request.TargetKnowledgeBaseIDs[len(request.TargetKnowledgeBaseIDs)-1] {
|
|
return "external"
|
|
}
|
|
return "internal"
|
|
}
|
|
|
|
// itemVisibilityFor mirrors source_items' two-value visibility CHECK: both and
|
|
// blocked decisions are governed through policy_state; the legacy visibility
|
|
// column keeps a valid value and is widened to external for both.
|
|
func itemVisibilityFor(visibility governance.Visibility) string {
|
|
if visibility == governance.VisibilityInternal {
|
|
return string(governance.VisibilityInternal)
|
|
}
|
|
return string(governance.VisibilityExternal)
|
|
}
|
|
|
|
func sha256Hex(value string) string {
|
|
sum := sha256.Sum256([]byte(value))
|
|
return hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
// MarkArtifactFetched finalizes one artifact.fetch success: it (re)uses the
|
|
// content_objects row for the verified hash (content-level dedupe), attaches it
|
|
// to the pending version and flips the intent to ready. matchedLocal records
|
|
// that the bytes already existed locally and no source download happened.
|
|
func (s *Store) MarkArtifactFetched(ctx context.Context, versionID, sha256Value string, sizeBytes int64, spoolRef, validatedFormat string, matchedLocal bool, now time.Time) error {
|
|
if len(sha256Value) != 64 || spoolRef == "" || sizeBytes < 0 || validatedFormat == "" {
|
|
return errors.New("invalid artifact fetch result")
|
|
}
|
|
var tx pgx.Tx
|
|
ownsTransaction := false
|
|
if existing, ok := s.db.(pgx.Tx); ok {
|
|
tx = existing
|
|
} else {
|
|
beginner, ok := s.db.(interface {
|
|
Begin(context.Context) (pgx.Tx, error)
|
|
})
|
|
if !ok {
|
|
return errors.New("artifact fetch transactions are unavailable")
|
|
}
|
|
var err error
|
|
tx, err = beginner.Begin(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("begin artifact fetch: %w", err)
|
|
}
|
|
ownsTransaction = true
|
|
defer tx.Rollback(ctx)
|
|
}
|
|
var connectionID, nodeID string
|
|
err := tx.QueryRow(ctx, `SELECT intent.source_connection_id,intent.node_id
|
|
FROM catalog_import_intents intent
|
|
JOIN source_item_versions version ON version.source_item_id=intent.source_item_id
|
|
WHERE version.id=$1 FOR UPDATE OF intent`, versionID).Scan(&connectionID, &nodeID)
|
|
if err != nil {
|
|
return fmt.Errorf("locate intent for fetched artifact: %w", err)
|
|
}
|
|
freshID, err := newOpaqueID("co-")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
contentID := ""
|
|
err = tx.QueryRow(ctx, `INSERT INTO content_objects(id,sha256,size_bytes,mime,validated_format,local_spool_ref,content_state,created_at,updated_at)
|
|
VALUES($1,$2,$3,'',$5,$4,'verified',NOW(),NOW())
|
|
ON CONFLICT(sha256) DO UPDATE SET
|
|
size_bytes=EXCLUDED.size_bytes,
|
|
validated_format=CASE WHEN content_objects.validated_format='' THEN EXCLUDED.validated_format ELSE content_objects.validated_format END,
|
|
local_spool_ref=CASE WHEN content_objects.local_spool_ref='' THEN EXCLUDED.local_spool_ref ELSE content_objects.local_spool_ref END,
|
|
content_state='verified',updated_at=NOW()
|
|
RETURNING id`,
|
|
freshID, sha256Value, sizeBytes, spoolRef, validatedFormat).Scan(&contentID)
|
|
if err != nil {
|
|
return fmt.Errorf("record fetched content: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE source_item_versions SET
|
|
content_object_id=$2,content_state='verified',updated_at=$4,
|
|
raw_record_summary=jsonb_set(raw_record_summary,'{matched_local}',$3::jsonb)
|
|
WHERE id=$1`,
|
|
versionID, contentID, mustJSONBool(matchedLocal), now.UTC()); err != nil {
|
|
return fmt.Errorf("bind fetched content: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE catalog_import_intents SET fetch_state='ready',last_error='',updated_at=NOW()
|
|
WHERE source_connection_id=$1 AND node_id=$2`, connectionID, nodeID); err != nil {
|
|
return fmt.Errorf("mark intent fetched: %w", err)
|
|
}
|
|
if ownsTransaction {
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return fmt.Errorf("commit artifact fetch: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func mustJSONBool(value bool) string {
|
|
if value {
|
|
return "true"
|
|
}
|
|
return "false"
|
|
}
|
|
|
|
// MarkArtifactFetchMissing parks an intent whose source-side file disappeared
|
|
// (a 404 on download, not a transient outage) so the UI can ask for a rescan.
|
|
func (s *Store) MarkArtifactFetchMissing(ctx context.Context, connectionID, nodeID string, now time.Time) error {
|
|
if _, err := s.db.Exec(ctx, `UPDATE catalog_import_intents SET fetch_state='missing_local',updated_at=$3
|
|
WHERE source_connection_id=$1 AND node_id=$2 AND fetch_state<>'ready'`, connectionID, nodeID, now.UTC()); err != nil {
|
|
return fmt.Errorf("record artifact fetch miss: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// MarkArtifactFetchFailed records a fetch failure for the intent so 解析监控
|
|
// can show why an item is not yet 待入库.
|
|
func (s *Store) MarkArtifactFetchFailed(ctx context.Context, connectionID, nodeID, message string, now time.Time) error {
|
|
if _, err := s.db.Exec(ctx, `UPDATE catalog_import_intents SET fetch_state='failed',last_error=$3,updated_at=$4
|
|
WHERE source_connection_id=$1 AND node_id=$2 AND fetch_state<>'ready'`, connectionID, nodeID, truncateMonitorText(message), now.UTC()); err != nil {
|
|
return fmt.Errorf("record artifact fetch failure: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func truncateMonitorText(value string) string {
|
|
runes := []rune(value)
|
|
if len(runes) <= 500 {
|
|
return value
|
|
}
|
|
return string(runes[:500])
|
|
}
|