Documentation
¶
Overview ¶
Package resourceindex is the managed-resource consumer of the shared indexjobs framework (#1012). It registers a Source/Sink pair under source_kind = "resources" so uploaded reference material is embedded off the request path: a newly uploaded resource (whose vector is still NULL), one whose metadata was edited (the request-path Update clears the vector), and one left stale by a provider model swap all self-heal on the next sweep.
Like the prompt, memory, and portal-asset consumers, resources store their vectors inline on the resources table (one embedding per row), not in a dedicated vector table. So this package's Store reads and writes the embedding / embedding_model / embedding_text_hash columns of resources directly: a resource IS its own indexing unit. SourceID is the resource id; each unit yields exactly one Item whose text is resource.IndexText.
What makes this consumer different from assets ¶
An asset's indexable text lives entirely in Postgres; a resource's does not. The bytes a human uploaded live in S3, and they are the whole point of indexing a resource — a data dictionary has to be findable by a column name that appears only inside the file. So the Source additionally reads the blob through the same reader contract and bucket the resources/read middleware uses, extracts a bounded text prefix from text-family content (classified by pkg/contenttype, never by ad-hoc MIME prefix matching), and denormalizes it onto the row's content_text column, which the lexical FTS index covers.
Blob reads fail in two distinguishable ways and the consumer treats them differently. A transient failure (S3 unreachable, credentials rejected) degrades to metadata-only indexing while KEEPING whatever content_text was already extracted, and leaves content_indexed_at NULL. That NULL is what makes the retry real: the metadata embed succeeds either way, so if the row's only gap signal were its embedding, the job would succeed, the row would stop being a gap, and the file's contents would never be indexed — while coverage reported it done. A confirmed missing object (resource.IsObjectNotFound) means the content is permanently gone, so the stale extracted text is cleared and the row settles. Neither case deletes the resource row: pruning an orphan stays the read path's decision (it prunes on a real read, where the caller sees the failure), because a background sweep against a misconfigured bucket must never be able to delete a user's uploads.
Deleting a resource needs no work here: the index lives on the row, so the DELETE removes the index entry with it, and a job left over for a deleted resource reports indexjobs.ErrSourceGone so the worker resolves it cleanly.
Index ¶
- Constants
- type BlobReader
- type Row
- type Sink
- func (s *Sink) Coverage(ctx context.Context) (indexjobs.Coverage, error)
- func (s *Sink) FindGaps(ctx context.Context) ([]string, error)
- func (*Sink) Kind() string
- func (s *Sink) ListExisting(ctx context.Context, key indexjobs.Key) (map[string]indexjobs.Vector, error)
- func (*Sink) StampExpected(context.Context, indexjobs.Key, int) error
- func (s *Sink) Upsert(ctx context.Context, key indexjobs.Key, rows []indexjobs.Vector) error
- func (s *Sink) UpsertBatch(ctx context.Context, key indexjobs.Key, rows []indexjobs.Vector) error
- type Source
- type Store
- func (s *Store) Coverage(ctx context.Context) (indexed, expected int, err error)
- func (s *Store) FindGaps(ctx context.Context, currentModel string) ([]string, error)
- func (s *Store) ListVectors(ctx context.Context, id string) (map[string]indexjobs.Vector, error)
- func (s *Store) Load(ctx context.Context, id string) (Row, error)
- func (s *Store) SetContentText(ctx context.Context, id, text string) error
- func (s *Store) UpsertVectors(ctx context.Context, id string, rows []indexjobs.Vector) error
Constants ¶
const SourceKind = "resources"
SourceKind is the indexjobs source_kind this package serves.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BlobReader ¶
type BlobReader interface {
GetObject(ctx context.Context, bucket, key string) (body []byte, contentType string, err error)
}
BlobReader fetches resource content from blob storage. It is the same contract the resources/read middleware uses (resource.S3Client's read half), declared narrowly here so the consumer takes only the capability it needs.
type Row ¶
type Row struct {
Resource resource.Resource
ContentText string
// ContentSettled reports whether a previous pass already settled the content
// question for this row (content_indexed_at is set). It is what lets the
// Source skip a redundant write while still guaranteeing that a row which has
// never settled gets stamped, including when the extracted text is empty.
ContentSettled bool
}
Row is the resource state the Source needs to compose the indexed text: Resource carries the indexed metadata fields plus the location and size of the blob whose text is extracted (only the columns this consumer selects are populated), and ContentText is the text a previous pass extracted (kept when a blob read fails transiently).
type Sink ¶
type Sink struct {
// contains filtered or unexported fields
}
Sink implements indexjobs.Sink for the resources kind over the embedding columns of the resources table. currentModel is the provider model the gap query diffs stored rows against, so a model swap re-embeds rows stamped with the previous model.
func NewSink ¶
NewSink returns a Sink backed by the given store. currentModel is the embedding provider's model identifier (embedding.ModelName); pass "" on a deployment whose provider does not name its model, in which case every row matches "" and only NULL-embedding rows are treated as gaps.
func (*Sink) Coverage ¶
Coverage reports the resources kind's indexed-vs-expected totals (resources with an embedding vs all resources). ExpectedKnown is true: every resource is expected to converge to one vector.
func (*Sink) FindGaps ¶
FindGaps returns the resource ids whose embedding is missing or was produced by a model other than the current one.
func (*Sink) ListExisting ¶
func (s *Sink) ListExisting(ctx context.Context, key indexjobs.Key) (map[string]indexjobs.Vector, error)
ListExisting returns the resource's persisted vector keyed by item id for the worker's dedup pass.
func (*Sink) StampExpected ¶
StampExpected is a no-op for resources. Gap detection is condition-based (embedding IS NULL OR model mismatch), not count-based, so there is no expected count to record per unit.
type Source ¶
type Source struct {
// contains filtered or unexported fields
}
Source implements indexjobs.Source for the resources kind. A unit is one managed resource (SourceID = resource id) and yields exactly one item: the resource's composed index text (metadata plus, for text-family content, a bounded prefix extracted from its blob). The worker embeds it and the Sink writes the vector back onto the same row.
func NewSource ¶
func NewSource(store *Store, blobs BlobReader, bucket string) *Source
NewSource returns a Source backed by the given store. blobs and bucket locate the resource content; a nil reader (a deployment with no S3 connection configured for resources) indexes metadata only.
func (*Source) LoadItems ¶
LoadItems returns the resource's single embeddable item. A resource deleted between enqueue and claim reports indexjobs.ErrSourceGone, so the worker clears the unit and resolves the job instead of recording a failure.
func (*Source) OnSucceeded ¶
OnSucceeded is a no-op: the ranked search reads embeddings from the resources table directly on every query, so there is no in-memory cache to refresh after a backfill writes a vector.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store reads and writes resource index state on the resources table for the indexjobs resources consumer. It is intentionally separate from resource.Store: it touches only the index columns (content_text, embedding, embedding_model, embedding_text_hash) plus the metadata the indexed text is composed from, and is scoped to the backfill path, so it does not widen the request-path store contract. The request-path Update clears the embedding columns when a resource's indexed metadata changes; this Store writes them back.
func (*Store) Coverage ¶
Coverage returns the number of fully indexed resources (a vector AND a settled content pass) and the total number of resources. It counts the same condition FindGaps excludes, so a resource whose content extraction is still owed reports as a gap rather than as covered — the failure mode this consumer is most likely to hit is a blob read that fails while the metadata embed succeeds, and coverage must not report that as done.
func (*Store) FindGaps ¶
FindGaps returns the ids of resources whose index state is incomplete: the embedding is missing, it was produced by a model other than the current provider's, or the content pass has never settled (content_indexed_at IS NULL). Missing embeddings cover a freshly uploaded resource and a metadata edit (the request-path Update clears the embedding); the model mismatch covers a provider model swap.
The content clause is what makes a failed extraction recoverable. A blob read that fails transiently still yields a valid metadata-only embedding, so without it the job would succeed, the row would stop being a gap, and the file's contents would never be indexed — while Coverage reported the resource fully indexed. The consumer stamps content_indexed_at only when it has either extracted the text or established there is nothing to extract, so a row whose content is still owed keeps coming back until it settles.
func (*Store) ListVectors ¶
ListVectors returns the resource's persisted embedding keyed by item id (the resource id), for the worker's text-hash + model dedup pass. A resource with no embedding yields an empty map, so the worker embeds it.
func (*Store) Load ¶
Load returns the indexable state of one resource. A resource deleted between enqueue and claim yields errGone.
func (*Store) SetContentText ¶
SetContentText writes the text extracted from a resource's blob onto the row so the lexical index covers it, and stamps content_indexed_at to record that the content question is settled for this row. The two are written together because they must not diverge: content_indexed_at is what takes the row OUT of the gap query, so stamping it without the text (or leaving it NULL after writing the text) either strands the content unindexed or re-reads the blob forever.
updated_at is deliberately left untouched: a background extraction is not a user-visible edit, so the resource's "last modified" timestamp must not move. A row deleted concurrently updates nothing, which is not an error.
func (*Store) UpsertVectors ¶
UpsertVectors writes the embedding back onto the resource. The resource unit holds exactly one item; a missing or empty row set is a no-op. updated_at is deliberately left untouched (see SetContentText).
Source Files