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

package postgres

import (
	"context"
	"errors"
	"fmt"
	"time"

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

// RoutingBinding is the UI-managed knowledge base binding row. Empty ids mean
// "not bound in the UI"; callers fall back to their environment defaults.
type RoutingBinding struct {
	InternalKBID string
	ExternalKBID string
	UpdatedBy    string
}

// LoadRoutingKBBindings reads the single active binding row. A missing row
// reads back as unbound rather than an error.
func (s *Store) LoadRoutingKBBindings(ctx context.Context) (string, string, string, error) {
	var binding RoutingBinding
	err := s.db.QueryRow(ctx, `SELECT internal_kb_id,external_kb_id,updated_by FROM routing_kb_bindings WHERE id='active'`).
		Scan(&binding.InternalKBID, &binding.ExternalKBID, &binding.UpdatedBy)
	if errors.Is(err, pgx.ErrNoRows) {
		return "", "", "", nil
	}
	if err != nil {
		return "", "", "", errors.New("load routing kb bindings")
	}
	return binding.InternalKBID, binding.ExternalKBID, binding.UpdatedBy, nil
}

// RoutingBindingReplanResult reports what a binding change did to publications
// that were still waiting in 待入库.
type RoutingBindingReplanResult struct {
	// Replanned counts untouched pending publications that were re-pointed at
	// the newly bound knowledge base for their scope.
	Replanned int
	// RetiredDuplicates counts pending publications that already had a twin for
	// the new knowledge base, so the stale row was superseded instead of moved.
	RetiredDuplicates int
}

// SaveRoutingKBBindings upserts the binding row. Each side is either empty
// (unbind, fall back to env) or a knowledge base id validated by the caller.
func (s *Store) SaveRoutingKBBindings(ctx context.Context, internalID, externalID, updatedBy string) error {
	_, err := s.db.Exec(ctx, `
INSERT INTO routing_kb_bindings(id,internal_kb_id,external_kb_id,updated_by,updated_at)
VALUES('active',$1,$2,$3,NOW())
ON CONFLICT(id) DO UPDATE SET
    internal_kb_id=EXCLUDED.internal_kb_id,
    external_kb_id=EXCLUDED.external_kb_id,
    updated_by=EXCLUDED.updated_by,
    updated_at=NOW()`, internalID, externalID, updatedBy)
	if err != nil {
		return errors.New("save routing kb bindings")
	}
	return nil
}

// replanRetireDuplicatesSQL retires a pending publication whose twin for the
// new knowledge base already exists — the unique key is
// (source_item_version_id, test_kb_id), so the row cannot simply be moved.
const replanRetireDuplicatesSQL = `
UPDATE test_publications stale
SET superseded_at = $3, updated_at = $3
WHERE stale.superseded_at IS NULL
  AND stale.test_kb_id <> $1
  AND stale.processing_state = 'planned'
  AND stale.ingest_state = 'pending'
  AND stale.knowledge_id = ''
  AND stale.retrieval_state = 'unpublished'
  AND stale.target_scope = $2
  AND NOT EXISTS (SELECT 1 FROM jobs job WHERE job.type = 'test.upload' AND job.payload->>'publication_id' = stale.id)
  AND EXISTS (
      SELECT 1 FROM test_publications twin
       WHERE twin.source_item_version_id = stale.source_item_version_id
         AND twin.test_kb_id = $1
         AND twin.superseded_at IS NULL)
`

// replanMovePendingSQL re-points the untouched remainder at the new knowledge
// base. Only rows that have never been uploaded anywhere are eligible: anything
// already in flight keeps its frozen target and stays visible in 待入库.
const replanMovePendingSQL = `
UPDATE test_publications pending
SET test_kb_id = $1, updated_at = $3
WHERE pending.superseded_at IS NULL
  AND pending.test_kb_id <> $1
  AND pending.processing_state = 'planned'
  AND pending.ingest_state = 'pending'
  AND pending.knowledge_id = ''
  AND pending.retrieval_state = 'unpublished'
  AND pending.target_scope = $2
  AND NOT EXISTS (SELECT 1 FROM jobs job WHERE job.type = 'test.upload' AND job.payload->>'publication_id' = pending.id)
  AND NOT EXISTS (
      SELECT 1 FROM test_publications twin
       WHERE twin.source_item_version_id = pending.source_item_version_id
         AND twin.test_kb_id = $1
         AND twin.superseded_at IS NULL)
`

// SaveRoutingKBBindingsWithReplan saves the binding and, in the SAME
// transaction, re-points 待入库 items that were still waiting for the old
// knowledge base. Without this, changing the binding left every pending
// publication aimed at the previous (possibly deleted) KB, and re-marking the
// file created a SECOND publication instead of moving the first.
//
// effectiveInternal/effectiveExternal are the ids the platform actually routes
// to (the saved binding, or the deployment default when a side is unbound);
// an empty effective id means that scope cannot be routed and is skipped.
func (s *Store) SaveRoutingKBBindingsWithReplan(ctx context.Context, internalID, externalID, effectiveInternal, effectiveExternal, updatedBy string, now time.Time) (RoutingBindingReplanResult, error) {
	var result RoutingBindingReplanResult
	if now.IsZero() {
		return result, errors.New("binding replan requires a timestamp")
	}
	tx, err := s.db.Begin(ctx)
	if err != nil {
		return result, fmt.Errorf("begin routing binding save: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()
	if _, err := tx.Exec(ctx, `
INSERT INTO routing_kb_bindings(id,internal_kb_id,external_kb_id,updated_by,updated_at)
VALUES('active',$1,$2,$3,$4)
ON CONFLICT(id) DO UPDATE SET
    internal_kb_id=EXCLUDED.internal_kb_id,
    external_kb_id=EXCLUDED.external_kb_id,
    updated_by=EXCLUDED.updated_by,
    updated_at=EXCLUDED.updated_at`, internalID, externalID, updatedBy, now); err != nil {
		return result, errors.New("save routing kb bindings")
	}
	for _, scope := range []struct{ name, kbID string }{
		{"internal", effectiveInternal},
		{"external", effectiveExternal},
	} {
		if scope.kbID == "" {
			continue
		}
		retired, err := tx.Exec(ctx, replanRetireDuplicatesSQL, scope.kbID, scope.name, now)
		if err != nil {
			return result, fmt.Errorf("retire duplicate pending publications: %w", err)
		}
		moved, err := tx.Exec(ctx, replanMovePendingSQL, scope.kbID, scope.name, now)
		if err != nil {
			return result, fmt.Errorf("replan pending publications: %w", err)
		}
		result.RetiredDuplicates += int(retired.RowsAffected())
		result.Replanned += int(moved.RowsAffected())
	}
	if err := tx.Commit(ctx); err != nil {
		return result, fmt.Errorf("commit routing binding save: %w", err)
	}
	return result, nil
}