internal/search/enqueue.go

Rama main · réplica de lectura del repositorio original.

// Package search is the lexical index layer of SPEC-0022: the queue write paths
// enqueue into, the indexer that drains it, and the backfill that rebuilds
// everything. The query side — ranking, snippets, the REST/MCP/CLI search
// surface — is the follow-up story and is deliberately absent here.
//
// The shape is ADR-0028's: Postgres full-text search over a `search_documents`
// table that lives and dies with its artifact by ON DELETE CASCADE, fed by a
// `search_queue` table that write paths populate AFTER their transaction
// commits. Indexing is therefore never on a create's critical path: an artifact
// is durable and served the moment its create commits, and it becomes
// searchable when a worker gets to it. With the indexer stopped, creates keep
// succeeding and the queue simply grows; on resume the backlog drains (SPEC-0022
// REQ-2).
//
// Text comes from two places, and only two: the artifact's stored body, which
// ADR-0023's ingest scanner already processed (SPEC-0022 REQ-8 — the indexer
// reads stored text, it never re-scans an original upload), and the artifact's
// live comments in Postgres. Binaries, images, webhook captures and trace span
// payloads are never indexed; a trace contributes its title and tags like any
// other artifact and nothing more.
//
// Governing: ADR-0028 (Search and Semantic Search), SPEC-0022 REQ-2 (indexed
// content), REQ-8 (redaction before indexing), REQ-10 (the index is deleted
// with the artifact), REQ-11 (reindex), "Concurrency Safety", "Database
// Operation Standards"; ADR-0008 (bodies live in object storage), ADR-0023 /
// SPEC-0017 (ingest redaction).
package search

import (
	"context"
	"fmt"

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

// Reasons a document is queued. They are stored in search_queue.reason, whose
// CHECK constraint keeps this list and the schema's in step.
const (
	ReasonCreated  = "created"
	ReasonComment  = "comment"
	ReasonMetadata = "metadata"
	ReasonReindex  = "reindex"
)

// execer is the Exec surface of *pgxpool.Pool and pgx.Tx, so an enqueue runs
// either on the connection a write path just committed on or inside its
// transaction when a caller wants the two to be atomic.
type execer interface {
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
}

// enqueueSQL puts one artifact in the queue by internal id.
//
// The insert is guarded by EXISTS on artifacts: an artifact deleted between the
// write path's commit and this enqueue must be a no-op rather than a foreign
// key error, because the enqueue runs after commit and can no longer be rolled
// back with the write.
//
// Folding into an existing row keeps the earliest not_before (so a queued item
// that is due is not pushed back by a second event) and preserves the attempt
// counter, so an artifact that keeps failing keeps backing off no matter how
// often it is touched.
const enqueueSQL = `
	INSERT INTO search_queue (artifact_id, reason, attempts, not_before, last_error)
	SELECT $1, $2, 0, now(), ''
	 WHERE EXISTS (SELECT 1 FROM artifacts WHERE id = $1)
	ON CONFLICT (artifact_id) DO UPDATE
	    SET reason     = EXCLUDED.reason,
	        not_before = LEAST(search_queue.not_before, EXCLUDED.not_before),
	        last_error = ''`

// enqueueByPublicIDSQL resolves the handle a transport adapter has (a public
// id, never the internal key — ADR-0005) and enqueues it.
const enqueueByPublicIDSQL = `
	INSERT INTO search_queue (artifact_id, reason, attempts, not_before, last_error)
	SELECT id, $2, 0, now(), ''
	  FROM artifacts
	 WHERE public_id = $1
	ON CONFLICT (artifact_id) DO UPDATE
	    SET reason     = EXCLUDED.reason,
	        not_before = LEAST(search_queue.not_before, EXCLUDED.not_before),
	        last_error = ''`

// EnqueueByID queues one artifact for (re)indexing. Call it AFTER the writing
// transaction commits: SPEC-0022 REQ-2 requires indexing to follow the commit,
// never to join it, so a queue write racing an uncommitted row could otherwise
// claim a document that does not exist yet.
//
// A missing artifact is not an error: the queue has nothing to index, and the
// caller's write already succeeded.
func EnqueueByID(ctx context.Context, ex execer, artifactID int64, reason string) error {
	if _, err := ex.Exec(ctx, enqueueSQL, artifactID, reason); err != nil {
		return fmt.Errorf("search: enqueue artifact %d: %w", artifactID, err)
	}
	return nil
}

// requeueSQL puts a failed artifact back with its attempt count bumped and its
// next claim delayed by the caller's backoff. The EXISTS guard is the same
// delete-during-index no-op: an artifact whose row is gone has no index to
// build and no queue row worth keeping (SPEC-0022 REQ-10).
const requeueSQL = `
	INSERT INTO search_queue (artifact_id, reason, attempts, not_before, last_error)
	SELECT $1, $2, $3, now() + make_interval(secs => $4::int), left($5, 512)
	 WHERE EXISTS (SELECT 1 FROM artifacts WHERE id = $1)
	ON CONFLICT (artifact_id) DO UPDATE
	    SET reason     = EXCLUDED.reason,
	        attempts   = EXCLUDED.attempts,
	        not_before = EXCLUDED.not_before,
	        last_error = EXCLUDED.last_error`

func requeue(ctx context.Context, ex execer, artifactID int64, reason string, attempts int, delaySecs int, lastError string) error {
	if _, err := ex.Exec(ctx, requeueSQL, artifactID, reason, attempts, delaySecs, lastError); err != nil {
		return fmt.Errorf("search: requeue artifact %d: %w", artifactID, err)
	}
	return nil
}

// Enqueuer is the slice of the indexer the write paths need. The store and the
// annotation service depend on this, never on *Indexer, so a server started
// without an indexer passes nil and behaves exactly as it did before the index
// existed.
type Enqueuer interface {
	EnqueueArtifact(ctx context.Context, artifactID int64, reason string) error
}

// EnqueueArtifact implements Enqueuer.
func (x *Indexer) EnqueueArtifact(ctx context.Context, artifactID int64, reason string) error {
	return EnqueueByID(ctx, x.pool, artifactID, reason)
}

// enqueueLiveQueue is the backfill's enqueue: every LIVE artifact, optionally
// narrowed to one owner, into the queue with reason 'reindex'.
//
// Expired artifacts are excluded on purpose (SPEC-0022 REQ-11: "live artifacts
// only"). They are already invisible and the reaper will remove them; indexing
// one would put text into the index that no query may return.
const enqueueLiveQueue = `
	INSERT INTO search_queue (artifact_id, reason, attempts, not_before, last_error)
	SELECT id, 'reindex', 0, now(), ''
	  FROM artifacts
	 WHERE expires_at > now()
	   AND ($1::text = '' OR COALESCE(owner_user_id::text, '') = $1 OR COALESCE(owner_team_id::text, '') = $1)
	ON CONFLICT (artifact_id) DO UPDATE
	    SET reason     = EXCLUDED.reason,
	        not_before = LEAST(search_queue.not_before, EXCLUDED.not_before),
	        last_error = ''`

// EnqueueLive queues every live artifact (or, when owner is non-empty, every
// live artifact whose workspace is that user or team id) for a rebuild. It is
// the enqueue half of `cairn search reindex`; the drain half is Indexer.Drain.
// It returns how many rows it touched.
func EnqueueLive(ctx context.Context, ex execer, owner string) (int64, error) {
	tag, err := ex.Exec(ctx, enqueueLiveQueue, owner)
	if err != nil {
		return 0, fmt.Errorf("search: enqueue live artifacts for reindex: %w", err)
	}
	return tag.RowsAffected(), nil
}

// compile-time proof that the indexer satisfies the write paths' seam.
var _ Enqueuer = (*Indexer)(nil)