Skip to main content

internal/infra/pg/embeddingbuild.go

internal/infra/pg · 441 lines · 9 declarations · source

Declarations​

type EmbeddingBatch​

type EmbeddingBatch func(context.Context, []string) ([][]float32, error)

EmbeddingBatch is supplied by the worker after checking the configured model identity. Source text leaves PostgreSQL only for this bounded call; model output cannot choose ownership or IDs.

source

type embeddingSource​

type embeddingSource struct {
	source, chunk, content string
	cursor                 ChunkRecoveryCursor
	digest                 []byte
}

source

type EmbeddingBuildPage​

type EmbeddingBuildPage struct {
	Examined int                 `json:"examined"`
	Stored   int                 `json:"stored"`
	Complete bool                `json:"complete"`
	After    ChunkRecoveryCursor `json:"after"`
}

source

method MessageEmbeddingStore.BuildPage​

func (s *MessageEmbeddingStore) BuildPage(ctx context.Context, scope, id, actor string, model EmbeddingModel, limit int, embed EmbeddingBatch) (EmbeddingBuildPage, error)

BuildPage keeps a session lock across the provider call but no database transaction. A persisted lease fences a disconnected worker: holding an old result never grants permission to publish it.

source

method MessageEmbeddingStore.buildPage​

func (s *MessageEmbeddingStore) buildPage(ctx context.Context, scope, id, actor string, model EmbeddingModel, limit int, embed EmbeddingBatch, activeOnly bool) (EmbeddingBuildPage, error)

source

method MessageEmbeddingStore.embeddingInputs​

func (s *MessageEmbeddingStore) embeddingInputs(ctx context.Context, conn interface {
	Query(context.Context, string, ...any) (pgx.Rows, error)
}, g EmbeddingGeneration, limit int) ([]embeddingSource, bool, error)

Metadata is read first so even a malformed oversized source cannot allocate an unbounded payload. The second query clips to the checked length; a concurrent edit causes a refusal, not truncation.

source

func validateEmbeddingBatch​

func validateEmbeddingBatch(vectors [][]float32, count, dimensions int) error

source

method MessageEmbeddingStore.commitEmbeddingInputs​

func (s *MessageEmbeddingStore) commitEmbeddingInputs(ctx context.Context, tx pgx.Tx, g EmbeddingGeneration, entries []embeddingSource, vectors [][]float32) (int, error)

All surviving inputs are locked and checked in one query. The provider's result supplies only vector coordinates; every relationship is taken from the retained observation and message.

source

method MessageEmbeddingStore.Repair​

func (s *MessageEmbeddingStore) Repair(ctx context.Context, scope, id, actor string) error

Repair replays the same captured source range and model. It cannot silently extend coverage or replace a compatible value. A concurrent in-flight page loses its lease before any write commits.

source