ragit

package module
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 28, 2026 License: MIT Imports: 18 Imported by: 0

README

ragit

A reusable RAG pipeline for Go: extract a document, chunk it, embed it, store it in Postgres/pgvector, and retrieve it — as a library a SaaS application imports rather than a service it runs.

docs/design.md is the real documentation: the full design, and the production incidents that shaped it. examples/ is three runnable host applications, and docs/examples-plan.md is what building them found — most of the API below exists because one of them hit something. This file is the entry point.

What it does

Upload → object storage → River job → extract → chunk → embed → pgvector → retrieval
  • Extraction through an xberg sidecar when one is configured, falling back to a memory-capped child process and then to in-process parsers. The fallback fires only when an extractor was unavailable, never when it rejected the document.
  • Chunking in Go, Markdown-heading aware, because chunk metadata has to round-trip into a citation.
  • Embedding through a single OpenAI-wire-compatible client, with a provider|model|dimension fingerprint on every chunk and on the document — the identity a search filters on, so two embedding spaces can never be compared to each other.
  • Retrieval as two separate calls — vector and full-text — rather than one fused endpoint. Full-text asks for all of a query's terms, then for any of them if that found nothing, so a plain question is not answered with an empty slice.
  • One resumable River job per document, checkpointed per embedding batch, so a retry resumes instead of re-billing the embedding provider.
  • Or none of the front half: a service that extracts, chunks and embeds in one call hands the result to IngestPrepared and keeps everything above.

Wiring it up

pool, err := ragit.NewPool(ctx, os.Getenv("DATABASE_URL"))
if err != nil { return err }

if err := ragit.Migrate(ctx, pool); err != nil { return err }

processor := ragit.New(pool,
    extract.NewChain(
        extract.NewXbergExtractor(os.Getenv("XBERG_URL"), 0), // optional
        extract.NewIsolatedExtractor(),
        extract.NewLocalExtractor(),
    ),
    chunk.New(chunk.DefaultConfig()),
    embedder,
    objectStore,
)

Four things are easy to get wrong, and three of them fail quietly:

Use ragit.NewPool, or register the codec yourself. pgvector's binary codec needs the extension's OID, which only exists once the extension is installed, so it is registered per connection: cfg.AfterConnect = sqlb.RegisterVectorType. Without it embeddings still move — as text, several times slower.

Do not connect as a superuser. ragit's tables carry FORCE ROW LEVEL SECURITY, and PostgreSQL exempts superusers and BYPASSRLS roles from row-level security regardless. The stock postgres image's POSTGRES_USER is a superuser, so an application connecting as one has these policies silently doing nothing and is relying on the query predicates alone. Nothing about that is visible from inside the application, so ask:

if err := ragit.VerifyRLS(ctx, pool); err != nil {
    return err // the isolation you think you have, you do not have
}

Create the role yourself — a password is a deployment's business — and let ragit grant it what its own tables need:

CREATE ROLE myapp LOGIN NOSUPERUSER NOBYPASSRLS PASSWORD '…';

err := ragit.GrantAppRole(ctx, adminPool, "myapp") // after Migrate

GrantAppRole refuses a role that bypasses RLS rather than granting to it pointlessly, and touches only ragit's tables — a host application's own are its own to grant.

Call extract.RunIsolatedChildIfInvoked() first in main() if you use IsolatedExtractor. A library cannot re-invoke itself the way an application can — ragit does not own main(). Without it the isolation layer reports itself unavailable and the chain degrades to direct local parsing.

func main() {
    extract.RunIsolatedChildIfInvoked()
    // ... normal startup
}

Calibrate MinScore per embedding model. The band separating a relevant match from noise is a property of the model, not of retrieval. ragit ships no default beyond zero on purpose.

Retrieval is confined by construction

Every read takes a Scope, and its zero value matches no rows:

results, err := processor.VectorSearch(ctx,
    ragit.Tenant(tenantID).A(companyID).B(coachID),
    "how do I reset my password?",
    ragit.SearchOptions{TopK: 8, MinScore: 0.6},
)

A dimension nobody mentions matches only rows where it is NULL, so a corpus that never sets the scope columns works unchanged while one that does cannot leak across a boundary because a caller left a field out. Unbounded access is AnyA() / AnyB() — a separate predicate, never a magic value in the column.

Full-text search answers a question, not just a term list

FullTextSearch runs the query strictly first — every term — and retries on any of them only when that matched nothing:

results, err := processor.FullTextSearch(ctx, scope, "how do I reset my password?",
    ragit.SearchOptions{TopK: 8})

The two passes are there because of that question, which is the shape a user types. The search_vector column uses the simple text-search configuration, deliberately: a stopword list belongs to a language, and a library that does not know its corpus's language should not apply one. But simple has no stopwords and websearch_to_tsquery ANDs every term, so a strict-only search asks for a chunk containing "how" and "do" and "i", finds none, and returns an empty slice — which reads exactly like an empty corpus.

Relaxing only after a strict miss keeps precision where it was available: a query whose terms all appear together never reaches the second pass. A query carrying real search syntax — a quoted phrase, a leading minus, an explicit or — is never rewritten, because widening reset -windows term by term would turn it into "reset, or anything without windows". Set SearchOptions.RequireAllTerms where an empty result is the meaningful answer.

Filtering by your own facts

Documents carry application-supplied key/value Attributes, stored separately from the extractor's Metadata so a new xberg field can never collide with one of your keys:

processor.Ingest(ctx, ragit.DocumentInput{
    TenantID:   tenantID,
    Attributes: ragit.Attributes{"course": courseID, "kind": "recording"},
    ...
})

results, err := processor.VectorSearch(ctx, scope, query, ragit.SearchOptions{
    TopK:       8,
    Attributes: ragit.Attributes{"course": courseID},
})

Matching is JSONB containment, so a filter names only the pairs it cares about, and multiple pairs are ANDed. Attributes are denormalized onto chunks and GIN indexed, so the filter rides alongside the vector scan rather than fighting it — which is also why changing them goes through SetDocumentAttributes, which re-stamps the chunks. Reprocessing will not fix a stale copy: the resume check sees identical content and skips the rewrite.

Attributes narrow; they do not confine. An empty filter matches everything, which is the opposite of Scope's rule and deliberate — a forgotten filter should return more rows, not none. So do not use them for access control: a caller that must not see a document should be outside its scope, not merely failing to match a label.

Schema

The schema is declared in ragitschema and everything else is generated from it:

go run ./cmd/ragit-gen          # migrations/ and the *_gen.go models

Tables are prefixed ragit_ and tracked in ragit's own ragit_migrations version table, so they sit in a host application's database without colliding with its schema or its migration sequence. The embedding dimension is an argument to the declaration rather than a literal in a shipped .sql file — go run ./cmd/ragit-gen -dim 768 renders a set for a different width.

Apply a rendered set with the same runner, not your own:

//go:embed migrations/*.sql
var schema768 embed.FS

err := ragit.Migrate(ctx, pool, ragit.FromFS(schema768))

A hand-rolled goose runner has to name ragit_migrations (exported as ragit.MigrationsTable), and guessing wrong is silent — goose starts a second history in its own default table and re-applies migrations the database already has. Applying a set at one width to a database created at another is refused for the same reason: every rendered set carries the same version numbers, so goose would find them applied and report success having changed nothing.

The generated models are exported. A read ragit does not offer can be written with sqlb against ragit.Document and ragit.Chunk directly, inside ragit.WithTenant so the RLS policies resolve.

Chunks from somewhere else

If your extraction service also chunks and embeds — xberg does both on the same /extract call — ragit does not need to redo either. IngestPrepared is ProcessDocument's sibling: same starting point, same terminal states, same events, same resume guard, only the front half differs.

documentID, err := processor.CreateDocument(ctx, in)
// ... your own extract / chunk / embed ...
err = processor.IngestPrepared(ctx, documentID, in.TenantID, ragit.PreparedDocument{
    Text:  extracted.Text,
    Space: embed.Space{Provider: "xberg", Model: "bge-base-en-v1.5", Dimension: 768},
    Chunks: []ragit.PreparedChunk{
        {Content: "...", Embedding: vec, HeadingPath: []string{"Handbook", "Accounts"}},
    },
})

A Processor for this path takes nil for the extractor, chunker and embedder — it runs none of them. embed.Space is a struct rather than a fingerprint string because retrieval filters on that fingerprint: a corpus written under a string that disagrees by one character with what the query embedder reports returns nothing, and says nothing.

Writing the chunk rows yourself with sqlb also works, and stays supported. It means owning four things that are silent when forgotten: the scope, attribute and expiry columns each chunk carries denormalized; the fingerprint; the document's terminal state; and the EventSink notification. If you go that way, ragit.ResumeChunks is the one piece not worth re-implementing — it takes your executor, reports which chunks are already stored in the space you name, and clears the document when any of them disagrees. Without it a bypassing writer re-embeds the whole document on every retry, which costs money rather than tidiness.

Runnable examples

examples/ is three host applications against one compose stack — Postgres+pgvector, xberg, MinIO — differing only in which half of the pipeline they own, and what drives it:

extraction chunking embedding driven by
extract-only/ xberg ragit EdenAI (1536-d) Ingest, inline
xberg-owned/ xberg xberg xberg, local ONNX (768-d) IngestPrepared
async/ xberg ragit EdenAI (1536-d) River jobs + retention sweep 
cd examples && make up && make verify && go run ./xberg-owned

xberg-owned needs no API key — its embedding model runs in the container — so it is the one to reach for when checking a change. The other two embed through a hosted provider, which bills on a first run and, thanks to the resume guard, on no run after it unless you pass -reset.

They are host applications rather than CLI wrappers on purpose: they carry their own table alongside ragit's, run their own migration line, and connect as an unprivileged role. Both bugs in docs/examples-plan.md that no unit test would have produced came out of that — a retention sweep that could only run once a day, and an indexing event that reported zero chunks.

Development

make test        # everything, needs Docker
make test-fast   # -short, no Docker
make generate    # regenerate migrations and models

examples/ is its own Go module with a replace back to the root, so it builds separately:

cd examples && go build ./...

Documentation

Overview

Package ragit is a reusable RAG pipeline: extract, chunk, embed, and store a document, then retrieve it. See docs/design.md for the full design and the production reference implementation it's grounded in.

Five properties are worth knowing before wiring this in.

Processor.ProcessDocument is resumable. An interrupted run picks up from whatever was already embedded in the current embedding space rather than re-billing every chunk on retry — see the jobs package for running it under River. ResumeChunks is that guard on its own, for callers whose chunks and vectors came from somewhere else.

Every read is confined by a Scope, whose zero value matches no rows. A retrieval or catalog call that forgets its confinement returns ErrUnscoped rather than another tenant's documents.

Beneath that, the tables carry FORCE ROW LEVEL SECURITY and every query runs inside a tenant-scoped transaction, so isolation is enforced by the database as well as by the query — but only if the application connects as a non-superuser role, since PostgreSQL exempts superusers from RLS. See NewPool and WithTenant.

The front half of the pipeline is optional. A caller whose extraction service also chunks and embeds hands the result to Processor.IngestPrepared and gets the same terminal states, events and resume guard as Processor.ProcessDocument.

The schema is declared in ragitschema and the models here are generated from it. They are exported deliberately: a consumer that needs a read ragit does not offer can write it with sqlb against Document and Chunk rather than being blocked by an internal package.

Index

Constants

View Source 
const (
	StatusPending         = "pending"
	StatusProcessing      = "processing"
	StatusReady           = "ready"
	StatusError           = "error"
	StatusSkippedTooLarge = "skipped_too_large"
)

Document statuses.

View Source 
const DefaultListLimit = 50

DefaultListLimit bounds a ListDocuments call that does not set one.

View Source 
const DefaultTopK = 10

DefaultTopK is used when SearchOptions.TopK is left at zero.

View Source 
const DeleteExpiredBatchSize = 500

DeleteExpiredBatchSize bounds one retention sweep pass, so a large backlog is worked through over several runs instead of one enormous transaction.

View Source 
const MaintenanceGUC = "ragit.maintenance"

MaintenanceGUC opts a transaction out of tenant scoping for reads and deletes. See WithMaintenance; it is set in exactly one place.

View Source 
const MigrationsTable = migrate.TableName

MigrationsTable is the schema-version table ragit tracks its own migration line in, deliberately not goose's default goose_db_version.

It is exported because a consumer applying a rendered migration set with their own tooling has to name it, and guessing wrong is silent: goose starts a second history in goose_db_version and re-applies migrations the database already has. Prefer Migrate with FromFS, which knows it.

View Source 
const TenantGUC = "ragit.tenant_id"

TenantGUC is the session variable the row-level security policies read to decide which rows are visible.

Variables

View Source 
var ChunkCols = chunkColumns{
	ID:                   sqlb.Typed[uuid.UUID]("id"),
	DocumentID:           sqlb.Typed[uuid.UUID]("document_id"),
	TenantID:             sqlb.Typed[uuid.UUID]("tenant_id"),
	ScopeAID:             sqlb.Typed[uuid.UUID]("scope_a_id"),
	ScopeBID:             sqlb.Typed[uuid.UUID]("scope_b_id"),
	SessionID:            sqlb.Typed[uuid.UUID]("session_id"),
	ChunkIndex:           sqlb.Typed[int32]("chunk_index"),
	HeadingPath:          sqlb.ArrayColumn[string]("heading_path"),
	Content:              sqlb.TextColumn[string]("content"),
	EmbeddingFingerprint: sqlb.TextColumn[string]("embedding_fingerprint"),
	Metadata:             sqlb.Typed[json.RawMessage]("metadata"),
	Attributes:           sqlb.Typed[json.RawMessage]("attributes"),
	ExpiresAt:            sqlb.Typed[time.Time]("expires_at"),
	CreatedAt:            sqlb.Typed[time.Time]("created_at"),
}

ChunkCols are the typed columns of ragit_chunks. Hidden columns are omitted: a predicate against one should not compile. Omitted here: embedding. Declaring LookupKey beside Hidden returns one to this facade, for the column whose own value is how the row is found. It stays off the wire either way.

View Source 
var DocumentCols = documentColumns{
	ID:                   sqlb.Typed[uuid.UUID]("id"),
	TenantID:             sqlb.Typed[uuid.UUID]("tenant_id"),
	ScopeAID:             sqlb.Typed[uuid.UUID]("scope_a_id"),
	ScopeBID:             sqlb.Typed[uuid.UUID]("scope_b_id"),
	SessionID:            sqlb.Typed[uuid.UUID]("session_id"),
	SourceURI:            sqlb.TextColumn[string]("source_uri"),
	Filename:             sqlb.TextColumn[string]("filename"),
	MimeType:             sqlb.TextColumn[string]("mime_type"),
	Status:               sqlb.TextColumn[string]("status"),
	Error:                sqlb.TextColumn[string]("error"),
	TextContent:          sqlb.TextColumn[string]("text_content"),
	Metadata:             sqlb.Typed[json.RawMessage]("metadata"),
	Attributes:           sqlb.Typed[json.RawMessage]("attributes"),
	ChunkCount:           sqlb.Typed[int32]("chunk_count"),
	EmbeddingFingerprint: sqlb.TextColumn[string]("embedding_fingerprint"),
	ProcessedAt:          sqlb.Typed[time.Time]("processed_at"),
	ExpiresAt:            sqlb.Typed[time.Time]("expires_at"),
	CreatedAt:            sqlb.Typed[time.Time]("created_at"),
	UpdatedAt:            sqlb.Typed[time.Time]("updated_at"),
}

DocumentCols are the typed columns of ragit_documents.

View Source 
var ErrNotFound = errors.New("ragit: document not found")

ErrNotFound is returned when a document does not exist, or is not visible to the scope that asked for it. The two are deliberately indistinguishable: telling a caller that a document exists but belongs to someone else is itself a disclosure.

View Source 
var ErrRLSNotEnforced = errors.New("ragit: row-level security is not enforced for this connection")

ErrRLSNotEnforced reports that ragit's row-level security is not actually confining the role a pool connects as. See VerifyRLS.

View Source 
var ErrUnscoped = errors.New("ragit: query has no tenant scope")

ErrUnscoped is returned when a query is attempted without a tenant.

Functions

func GrantAppRole added in v0.3.0 

func GrantAppRole(ctx context.Context, pool *pgxpool.Pool, role string) error

GrantAppRole grants role what an application needs on ragit's tables, and nothing else.

Run it as the role that owns the tables, after Migrate: 

if err := ragit.Migrate(ctx, adminPool); err != nil { return err }
if err := ragit.GrantAppRole(ctx, adminPool, "myapp"); err != nil { return err }

Creating the role is deliberately not part of this. A role needs a password, and where that comes from is a deployment's business, not a library's. What the role must not have is SUPERUSER or BYPASSRLS — PostgreSQL exempts both from row-level security entirely, FORCE or not, so an application connecting as one has every policy here silently doing nothing. Granting to such a role is refused rather than quietly wasted: 

CREATE ROLE myapp LOGIN NOSUPERUSER NOBYPASSRLS PASSWORD '…';

This grants on ragit's own tables only. A host application's tables are its own to grant, and a library handing out privileges across a schema it does not own would be a poor guest. ragit's version table is not included either: migrations run as the owner, and an application role has no business writing schema history.

Call it again after any Migrate that adds a table. GRANT is not a standing rule — it applies to what exists when it runs.

func Migrate 

func Migrate(ctx context.Context, pool *pgxpool.Pool, opts ...MigrateOption) error

Migrate brings ragit's own tables up to the schema this build expects.

ragit owns its migration line rather than shipping loose .sql files for a host application to vendor into its own sequence: the migrations are embedded in the binary and tracked in a ragit_migrations version table, so upgrading the library upgrades its schema, and a host app's own migration tool never has to know these tables exist. This mirrors how River manages its river_* tables.

It is safe to call on every startup, and touches nothing outside the ragit_-prefixed tables.

FromFS points it at a set rendered at another embedding dimension, so a consumer at a width other than the default does not have to re-implement the runner — or remember MigrationsTable.

The connecting role must not be a superuser if the row-level security policies are to have any effect — PostgreSQL exempts superusers from RLS entirely, FORCE or not. Migrating as an admin role and then running the application as an ordinary one is the intended split.

func MigrateDown 

func MigrateDown(ctx context.Context, pool *pgxpool.Pool, opts ...MigrateOption) error

MigrateDown rolls back the most recent migration. Intended for development and tests; rolling back a populated vector index rarely is what you want in production.

func NewPool 

func NewPool(ctx context.Context, dsn string) (*pgxpool.Pool, error)

NewPool opens a pool wired the way ragit needs.

It exists because two pieces of setup are easy to omit and fail unhelpfully when they are. pgvector's binary codec needs the extension's OID, which only exists once the extension is installed, so it is registered per connection — without it embeddings still move, as text, several times slower. A pool built by hand works too; it just has to do this: 

cfg.AfterConnect = sqlb.RegisterVectorType

The role this connects as matters as much as the codec: PostgreSQL exempts superusers and BYPASSRLS roles from row-level security entirely, so a pool connected as one has ragit's tenant policies silently doing nothing. Connect as an ordinary role, and have VerifyRLS confirm it rather than assuming — nothing else about that failure is visible from inside the application.

func ResumeChunks added in v0.3.0 

func ResumeChunks(ctx context.Context, db sqlb.Executor, tenantID, documentID uuid.UUID, contents []string, fingerprint string) ([]bool, error)

ResumeChunks reports which of a document's chunks are already persisted in the embedding space named by fingerprint, so a caller embeds only what is missing instead of paying for the whole document again.

contents is the document's freshly chunked text, positionally: contents[i] is the content of chunk index i. The returned slice is parallel to it, and true means that chunk is already stored under fingerprint with byte- identical content — skip it.

A stored chunk survives only if its fingerprint matches AND its content matches contents at the same index. On the first disagreement — a different embedder, re-chunked text, an index past the end of the fresh set — every chunk for the document is deleted and the result is all false. That is deliberate and total: two embedding spaces inside one document produce cosine distances that are not a weaker signal but a meaningless one, and a partial wipe is exactly how that state arises.

It runs on the caller's executor rather than opening its own transaction, so a caller writing chunks by hand — the sqlb escape hatch this package's doc comment describes — can put the guard and its own inserts in one transaction, and needs no Processor to reach it: 

err := ragit.WithTenant(ctx, pool, tenantID, func(db sqlb.Executor) error {
	reusable, err := ragit.ResumeChunks(ctx, db, tenantID, docID, contents, fp)
	if err != nil {
		return err
	}
	// ... embed and insert every index whose entry is false ...
})

Processor.ProcessDocument calls it too. This is the guard itself, not a second copy of the rule.

func VerifyRLS added in v0.3.0 

func VerifyRLS(ctx context.Context, pool *pgxpool.Pool) error

VerifyRLS reports whether tenant isolation on this connection is the database's doing or only the query's.

It answers one question — would ragit's policies actually stop a query that forgot its confinement? — and it is worth asking at startup, because every way of getting this wrong is silent. A superuser or BYPASSRLS role reads every tenant's rows while every test that goes through Scope still passes, since the predicates alone are doing the work. Tables missing FORCE ROW LEVEL SECURITY leak to their owner in the same way.

It is the counterpart of Processor.CountMisalignedChunks: a state that cannot be noticed by using the library normally, made detectable deliberately. 

if err := ragit.VerifyRLS(ctx, pool); err != nil {
    return err // refuse to serve rather than serve unconfined
}

Errors wrap ErrRLSNotEnforced. Run it on the application's own pool: it reports on the role that pool connects as, so a migration pool answering as a superuser is expected and says nothing about the application's.

func WithMaintenance 

func WithMaintenance(ctx context.Context, pool *pgxpool.Pool, fn func(sqlb.Executor) error) error

WithMaintenance runs fn in a transaction that can read and delete across every tenant.

This exists for one caller — the retention sweep — and the reason it needs an escape at all is that the work is inherently cross-tenant: finding expired rows means reading rows whose owning tenants cannot be enumerated beforehand, and enumerating them would itself be the cross-tenant read.

It widens reads and deletes only. The policies' WITH CHECK clause stays tenant-scoped, so nothing reached from here can write a row into, or move a row between, tenants.

Do not reach for this to make an ordinary query simpler. Every use is a place where isolation rests on the surrounding code being correct rather than on the database, which is what WithTenant exists to avoid.

func WithTenant 

func WithTenant(ctx context.Context, pool *pgxpool.Pool, tenantID uuid.UUID, fn func(sqlb.Executor) error) error

WithTenant runs fn inside a transaction scoped to one tenant.

The scoping is the GUC the row-level security policies read. That is a second layer beneath the confinement predicates ragit's own queries carry: the predicates constrain what ragit asks for, and RLS constrains what the database will answer regardless of who asks — a raw pgx call, a psql session, a query written later by code that never heard of Scope.

With FORCE ROW LEVEL SECURITY enabled, a query run outside such a transaction sees zero rows rather than every row: the policies fail closed.

The caveat worth knowing at deployment time: PostgreSQL exempts superusers (and BYPASSRLS roles) from row-level security, FORCE or not. The stock postgres image's POSTGRES_USER is a superuser, so an application connecting as one has these policies silently inert and is relying on the predicates alone. See NewPool.

Types

type Attributes 

type Attributes map[string]string

Attributes are the host application's own key/value pairs on a document.

ragit stores and filters them without interpreting them: they are the seam for narrowing a search by facts ragit does not model — a course id, a language, a document kind, a visibility label the application understands.

They are kept separate from Document.Metadata, which holds whatever the extractor produced (page count, detected language, table warnings). Merging the two would let a new xberg field collide with an application key.

Attributes are not a security boundary

Scope is. An attribute filter narrows a result set that confinement has already bounded, and an *empty* filter narrows nothing — the opposite of Scope's rule, and deliberately so, because a forgotten attribute filter should return more rows rather than none.

So do not use attributes for access control. A caller that must not see a document should be outside its scope, not merely failing to match a label; otherwise the day someone forgets the filter is the day the document leaks.

func DocumentAttributes 

func DocumentAttributes(doc *Document) (Attributes, error)

DocumentAttributes decodes a document's stored attributes.

type Chunk 

type Chunk struct {
	ID                   uuid.UUID       `db:"id" json:"id" sqlb:"type:uuid,pk,default,filter,readonly"`
	DocumentID           uuid.UUID       `db:"document_id" json:"document_id" sqlb:"type:uuid"`
	TenantID             uuid.UUID       `db:"tenant_id" json:"tenant_id" sqlb:"type:uuid,filter,readonly,scope"`
	ScopeAID             *uuid.UUID      `db:"scope_a_id" json:"scope_a_id" sqlb:"type:uuid,filter"`
	ScopeBID             *uuid.UUID      `db:"scope_b_id" json:"scope_b_id" sqlb:"type:uuid,filter"`
	SessionID            *uuid.UUID      `db:"session_id" json:"session_id" sqlb:"type:uuid,filter"`
	ChunkIndex           int32           `db:"chunk_index" json:"chunk_index" sqlb:"type:int,filter,sort"`
	HeadingPath          []string        `db:"heading_path" json:"heading_path" sqlb:"type:text"`
	Content              string          `db:"content" json:"content" sqlb:"type:text"`
	Embedding            *sqlb.Vector    `db:"embedding" json:"-" sqlb:"type:vector,hidden"`
	EmbeddingFingerprint *string         `db:"embedding_fingerprint" json:"embedding_fingerprint" sqlb:"type:text,filter"`
	Metadata             json.RawMessage `db:"metadata" json:"metadata" sqlb:"type:jsonb,default"`
	Attributes           json.RawMessage `db:"attributes" json:"attributes" sqlb:"type:jsonb,default"`
	ExpiresAt            *time.Time      `db:"expires_at" json:"expires_at" sqlb:"type:timestamptz,filter"`
	CreatedAt            time.Time       `db:"created_at" json:"created_at" sqlb:"type:timestamptz,default"`
}

Chunk one retrieval-sized piece of a document, with its embedding.

func (Chunk) TableName 

func (Chunk) TableName() string

TableName is the table Chunk maps to.

type ChunkUpdate 

type ChunkUpdate struct {
	// contains filtered or unexported fields
}

ChunkUpdate is a typed update statement for ragit_chunks.

func UpdateChunk 

func UpdateChunk() *ChunkUpdate

UpdateChunk starts a typed update.

func (*ChunkUpdate) SetAttributes 

func (u *ChunkUpdate) SetAttributes(v json.RawMessage) *ChunkUpdate

SetAttributes sets attributes.

func (*ChunkUpdate) SetChunkIndex 

func (u *ChunkUpdate) SetChunkIndex(v int32) *ChunkUpdate

SetChunkIndex sets chunk_index.

func (*ChunkUpdate) SetContent 

func (u *ChunkUpdate) SetContent(v string) *ChunkUpdate

SetContent sets content.

func (*ChunkUpdate) SetCreatedAt 

func (u *ChunkUpdate) SetCreatedAt(v time.Time) *ChunkUpdate

SetCreatedAt sets created_at.

func (*ChunkUpdate) SetDocumentID 

func (u *ChunkUpdate) SetDocumentID(v uuid.UUID) *ChunkUpdate

SetDocumentID sets document_id.

func (*ChunkUpdate) SetEmbedding 

func (u *ChunkUpdate) SetEmbedding(v *sqlb.Vector) *ChunkUpdate

SetEmbedding sets embedding.

func (*ChunkUpdate) SetEmbeddingFingerprint 

func (u *ChunkUpdate) SetEmbeddingFingerprint(v *string) *ChunkUpdate

SetEmbeddingFingerprint sets embedding_fingerprint.

func (*ChunkUpdate) SetExpiresAt 

func (u *ChunkUpdate) SetExpiresAt(v *time.Time) *ChunkUpdate

SetExpiresAt sets expires_at.

func (*ChunkUpdate) SetHeadingPath 

func (u *ChunkUpdate) SetHeadingPath(v []string) *ChunkUpdate

SetHeadingPath sets heading_path.

func (*ChunkUpdate) SetMetadata 

func (u *ChunkUpdate) SetMetadata(v json.RawMessage) *ChunkUpdate

SetMetadata sets metadata.

func (*ChunkUpdate) SetScopeAID 

func (u *ChunkUpdate) SetScopeAID(v *uuid.UUID) *ChunkUpdate

SetScopeAID sets scope_a_id.

func (*ChunkUpdate) SetScopeBID 

func (u *ChunkUpdate) SetScopeBID(v *uuid.UUID) *ChunkUpdate

SetScopeBID sets scope_b_id.

func (*ChunkUpdate) SetSessionID 

func (u *ChunkUpdate) SetSessionID(v *uuid.UUID) *ChunkUpdate

SetSessionID sets session_id.

func (*ChunkUpdate) SetTenantID 

func (u *ChunkUpdate) SetTenantID(v uuid.UUID) *ChunkUpdate

SetTenantID sets tenant_id.

func (*ChunkUpdate) Stmt 

func (u *ChunkUpdate) Stmt() *sqlb.Update[Chunk]

Stmt exposes the underlying statement for what the wrapper does not cover, such as Everything, SetExpr, Exec and One.

func (*ChunkUpdate) Where 

func (u *ChunkUpdate) Where(preds ...sqlb.Pred) *ChunkUpdate

Where narrows the affected rows.

type Document 

type Document struct {
	ID                   uuid.UUID       `db:"id" json:"id" sqlb:"type:uuid,pk,default,filter,readonly"`
	TenantID             uuid.UUID       `db:"tenant_id" json:"tenant_id" sqlb:"type:uuid,filter,readonly,scope"`
	ScopeAID             *uuid.UUID      `db:"scope_a_id" json:"scope_a_id" sqlb:"type:uuid,filter"`
	ScopeBID             *uuid.UUID      `db:"scope_b_id" json:"scope_b_id" sqlb:"type:uuid,filter"`
	SessionID            *uuid.UUID      `db:"session_id" json:"session_id" sqlb:"type:uuid,filter"`
	SourceURI            *string         `db:"source_uri" json:"source_uri" sqlb:"type:text"`
	Filename             string          `db:"filename" json:"filename" sqlb:"type:text,filter,sort"`
	MimeType             string          `db:"mime_type" json:"mime_type" sqlb:"type:text,filter"`
	Status               string          `db:"status" json:"status" sqlb:"type:text,default,filter,sort"` // pending|processing|ready|error|skipped_too_large
	Error                *string         `db:"error" json:"error" sqlb:"type:text"`
	TextContent          *string         `db:"text_content" json:"text_content" sqlb:"type:text"`
	Metadata             json.RawMessage `db:"metadata" json:"metadata" sqlb:"type:jsonb,default"`     // the extractor's own structured output: page count, language, detected tables
	Attributes           json.RawMessage `db:"attributes" json:"attributes" sqlb:"type:jsonb,default"` // application-supplied key/value pairs, filterable by containment
	ChunkCount           *int32          `db:"chunk_count" json:"chunk_count" sqlb:"type:int,sort"`
	EmbeddingFingerprint *string         `db:"embedding_fingerprint" json:"embedding_fingerprint" sqlb:"type:text,filter"` // provider|model|dimension of the space this document's chunks live in
	ProcessedAt          *time.Time      `db:"processed_at" json:"processed_at" sqlb:"type:timestamptz,sort"`
	ExpiresAt            *time.Time      `db:"expires_at" json:"expires_at" sqlb:"type:timestamptz,filter"`
	CreatedAt            time.Time       `db:"created_at" json:"created_at" sqlb:"type:timestamptz,default,sort,readonly"`
	UpdatedAt            time.Time       `db:"updated_at" json:"updated_at" sqlb:"type:timestamptz,default,sort,readonly"`
}

Document a source document ingested by ragit.

func (Document) TableName 

func (Document) TableName() string

TableName is the table Document maps to.

type DocumentInput 

type DocumentInput struct {
	// TenantID is required; it is the security boundary.
	TenantID uuid.UUID
	// ScopeA and ScopeB file the document under ragit's two generic scope
	// dimensions. ragit does not know what they mean — a host application maps
	// its own domain onto them, and searches confine with the matching
	// [Scope].
	ScopeA *uuid.UUID
	ScopeB *uuid.UUID
	// SessionID marks the document as an ephemeral attachment belonging to one
	// conversation or agent session. Such documents are invisible to ordinary
	// library search unless a caller names that session.
	SessionID *uuid.UUID
	// Attributes are the application's own key/value pairs, stored on the
	// document and denormalized onto its chunks so searches can filter by
	// them. They narrow a result set; they do not confine it — see
	// [Attributes].
	Attributes Attributes
	// ExpiresAt sets a retention clock on the document and its chunks. Nil
	// keeps it until explicitly deleted.
	ExpiresAt *time.Time
	Filename  string
	MimeType  string
	Data      []byte
}

DocumentInput describes a document to ingest.

type DocumentUpdate 

type DocumentUpdate struct {
	// contains filtered or unexported fields
}

DocumentUpdate is a typed update statement for ragit_documents.

func UpdateDocument 

func UpdateDocument() *DocumentUpdate

UpdateDocument starts a typed update.

func (*DocumentUpdate) SetAttributes 

func (u *DocumentUpdate) SetAttributes(v json.RawMessage) *DocumentUpdate

SetAttributes sets attributes.

func (*DocumentUpdate) SetChunkCount 

func (u *DocumentUpdate) SetChunkCount(v *int32) *DocumentUpdate

SetChunkCount sets chunk_count.

func (*DocumentUpdate) SetCreatedAt 

func (u *DocumentUpdate) SetCreatedAt(v time.Time) *DocumentUpdate

SetCreatedAt sets created_at.

func (*DocumentUpdate) SetEmbeddingFingerprint added in v0.3.0 

func (u *DocumentUpdate) SetEmbeddingFingerprint(v *string) *DocumentUpdate

SetEmbeddingFingerprint sets embedding_fingerprint.

func (*DocumentUpdate) SetError 

func (u *DocumentUpdate) SetError(v *string) *DocumentUpdate

SetError sets error.

func (*DocumentUpdate) SetExpiresAt 

func (u *DocumentUpdate) SetExpiresAt(v *time.Time) *DocumentUpdate

SetExpiresAt sets expires_at.

func (*DocumentUpdate) SetFilename 

func (u *DocumentUpdate) SetFilename(v string) *DocumentUpdate

SetFilename sets filename.

func (*DocumentUpdate) SetMetadata 

func (u *DocumentUpdate) SetMetadata(v json.RawMessage) *DocumentUpdate

SetMetadata sets metadata.

func (*DocumentUpdate) SetMimeType 

func (u *DocumentUpdate) SetMimeType(v string) *DocumentUpdate

SetMimeType sets mime_type.

func (*DocumentUpdate) SetProcessedAt 

func (u *DocumentUpdate) SetProcessedAt(v *time.Time) *DocumentUpdate

SetProcessedAt sets processed_at.

func (*DocumentUpdate) SetScopeAID 

func (u *DocumentUpdate) SetScopeAID(v *uuid.UUID) *DocumentUpdate

SetScopeAID sets scope_a_id.

func (*DocumentUpdate) SetScopeBID 

func (u *DocumentUpdate) SetScopeBID(v *uuid.UUID) *DocumentUpdate

SetScopeBID sets scope_b_id.

func (*DocumentUpdate) SetSessionID 

func (u *DocumentUpdate) SetSessionID(v *uuid.UUID) *DocumentUpdate

SetSessionID sets session_id.

func (*DocumentUpdate) SetSourceURI 

func (u *DocumentUpdate) SetSourceURI(v *string) *DocumentUpdate

SetSourceURI sets source_uri.

func (*DocumentUpdate) SetStatus 

func (u *DocumentUpdate) SetStatus(v string) *DocumentUpdate

SetStatus sets status.

func (*DocumentUpdate) SetTenantID 

func (u *DocumentUpdate) SetTenantID(v uuid.UUID) *DocumentUpdate

SetTenantID sets tenant_id.

func (*DocumentUpdate) SetTextContent 

func (u *DocumentUpdate) SetTextContent(v *string) *DocumentUpdate

SetTextContent sets text_content.

func (*DocumentUpdate) SetUpdatedAt 

func (u *DocumentUpdate) SetUpdatedAt(v time.Time) *DocumentUpdate

SetUpdatedAt sets updated_at.

func (*DocumentUpdate) Stmt 

func (u *DocumentUpdate) Stmt() *sqlb.Update[Document]

Stmt exposes the underlying statement for what the wrapper does not cover, such as Everything, SetExpr, Exec and One.

func (*DocumentUpdate) Where 

func (u *DocumentUpdate) Where(preds ...sqlb.Pred) *DocumentUpdate

Where narrows the affected rows.

type Event 

type Event struct {
	DocumentID uuid.UUID
	TenantID   uuid.UUID
	ScopeA     *uuid.UUID
	ScopeB     *uuid.UUID
	SessionID  *uuid.UUID
	Filename   string
	// Status is one of StatusReady, StatusError or StatusSkippedTooLarge.
	Status string
	// Error carries the failure message for StatusError and the reason for
	// StatusSkippedTooLarge. Empty for StatusReady.
	Error string
	// ChunkCount is the number of chunks indexed. Zero unless Status is
	// StatusReady.
	ChunkCount int
	At         time.Time
}

Event reports that a document reached a terminal state.

func (Event) Succeeded 

func (e Event) Succeeded() bool

Succeeded reports whether the document is now searchable.

type EventSink 

type EventSink interface {
	DocumentProcessed(ctx context.Context, event Event)
}

EventSink observes documents reaching a terminal state.

Two properties this contract commits to, because both matter to what a subscriber can be built on:

It fires on **every** terminal state, not only success. A document that ended in error or was skipped as too large is precisely the case a user who uploaded it needs told about, so a success-only callback would leave the interesting half unreported. Check Event.Succeeded.

It fires **after** the chunks are committed, and its error is ignored. A subscriber that fails must not roll back or retry the indexing: the indexing already happened, the work is already paid for, and re-running it would re-bill the embedding provider to satisfy a notification. Handle and log failures inside the sink.

A sink that blocks holds up the job that called it, so a slow subscriber should hand off to its own queue.

If a transactional guarantee is wanted later, a durable outbox table written in the same transaction as the chunks is the shape to reach for; this interface is the seam it would be implemented behind.

type EventSinkFunc 

type EventSinkFunc func(ctx context.Context, event Event)

EventSinkFunc adapts a function to EventSink.

func (EventSinkFunc) DocumentProcessed 

func (f EventSinkFunc) DocumentProcessed(ctx context.Context, event Event)

DocumentProcessed implements EventSink.

type ListFilter 

type ListFilter struct {
	// Status restricts to documents in the given states. Empty means every
	// state, which is the useful default for "what has been uploaded".
	Status []string
	// Attributes restricts to documents carrying all of these key/value
	// pairs. Empty narrows nothing — like Status, and unlike Scope, this is a
	// filter rather than a boundary. See [Attributes].
	Attributes Attributes
	// Limit caps the result. Zero means DefaultListLimit.
	Limit int
	// Offset pages through results.
	Offset int
}

ListFilter narrows a catalog listing. Confinement is the Scope argument, not a field here: a catalog read is as much a boundary as a retrieval, and the same rule applies — the zero value must not widen anything.

type MigrateOption added in v0.3.0 

type MigrateOption func(*migrateConfig)

MigrateOption configures where Migrate and MigrateDown read their SQL.

func FromFS added in v0.3.0 

func FromFS(fsys fs.FS) MigrateOption

FromFS applies a migration set rendered by cmd/ragit-gen instead of the one embedded in this build.

This is the path for a corpus at a different embedding dimension. A vector column's width is part of its type, so an embedder that is not 1536-wide needs its own rendered schema: 

//go:embed *.sql
var migrationsFS embed.FS

err := ragit.Migrate(ctx, pool, ragit.FromFS(migrationsFS))

generated by: 

go run github.com/mind-vm/ragit/cmd/ragit-gen -dim 768 -migrations ./migrations -skip-models

The models need no such treatment — Chunk.Embedding is a *sqlb.Vector whatever the width — so only the SQL differs.

Applying a set at one width to a database created at another is refused rather than skipped: every rendered set carries the same version numbers, so goose would otherwise find them already applied and report success having changed nothing.

type PreparedChunk added in v0.3.0 

type PreparedChunk struct {
	// Content is the chunk's text, and is what a later run compares against
	// to decide whether the chunk can be reused. See [ResumeChunks].
	Content string
	// Embedding is the vector for Content, in the space named by
	// [PreparedDocument.Space]. Its length must match that space's Dimension.
	Embedding embed.Vector
	// HeadingPath is the chunk's heading trail, for citations. Optional, but
	// a citation UI has nothing to show without it.
	HeadingPath []string
	// Metadata is whatever the producing pipeline recorded about this chunk —
	// page spans, byte offsets, a table flag. Stored as-is; nil becomes {}.
	Metadata json.RawMessage
}

PreparedChunk is one chunk of a document that was chunked and embedded outside ragit.

type PreparedDocument added in v0.3.0 

type PreparedDocument struct {
	// Text is the document's full extracted text, stored on the document row.
	Text string
	// Metadata is the extractor's document-level metadata. Nil becomes {}.
	Metadata json.RawMessage
	// Space identifies where Chunks' vectors live. Retrieval filters on its
	// fingerprint, so it must be the same space the query embedder reports.
	Space embed.Space
	// Chunks are the document's chunks in order: Chunks[i] is chunk index i.
	Chunks []PreparedChunk
}

PreparedDocument is a front half of the pipeline — extract, chunk, embed — that ragit did not run.

type Processor 

type Processor struct {
	// contains filtered or unexported fields
}

Processor wires extraction, chunking, embedding, storage, and retrieval into one pipeline.

func New 

func New(pool *pgxpool.Pool, extractor extract.Extractor, chunker *chunk.Chunker, embedder embed.Embedder, st store.Store) *Processor

New builds a Processor. The caller owns pool/store's lifecycle.

The extractor, chunker and embedder may be nil for a Processor that never runs the front half of the pipeline itself — one that creates documents, indexes them with Processor.IngestPrepared, and searches by text. What they are needed for says so: Processor.ProcessDocument needs all three, and Processor.VectorSearch needs the embedder to embed the query. Both report a missing dependency rather than panicking on it.

func (*Processor) CountDocuments 

func (p *Processor) CountDocuments(ctx context.Context, scope Scope, filter ListFilter) (int64, error)

CountDocuments returns how many documents match, ignoring paging.

func (*Processor) CountMisalignedChunks 

func (p *Processor) CountMisalignedChunks(ctx context.Context, scope Scope) (int64, error)

CountMisalignedChunks reports how many of a tenant's embedded chunks were produced by an embedder other than the active one.

A non-zero count means the corpus straddles two embedding spaces and that Processor.VectorSearch is silently ignoring part of it. Call it at startup and decide what it means for your deployment — block queries, or log loudly and schedule a re-embed. It is reported rather than acted on because "refuse to serve" and "serve a degraded corpus" are both defensible, and which is right is the host application's call.

func (*Processor) CreateDocument 

func (p *Processor) CreateDocument(ctx context.Context, in DocumentInput) (uuid.UUID, error)

CreateDocument stores the bytes and inserts a pending row. Fast and synchronous — meant to be called from an upload handler, before a ProcessDocument job is enqueued.

func (*Processor) DeleteDocument 

func (p *Processor) DeleteDocument(ctx context.Context, tenantID, documentID uuid.UUID) error

DeleteDocument removes a document, its chunks (cascaded via the FK), and the original bytes in object storage.

The database row goes first. If the object-storage delete then fails, the result is an orphaned object rather than a document that still answers searches but whose bytes have vanished — the cheaper of the two inconsistencies, and the one a storage lifecycle rule can mop up. The error is still returned so the caller knows it happened.

func (*Processor) DeleteExpired 

func (p *Processor) DeleteExpired(ctx context.Context) (*RetentionResult, error)

DeleteExpired removes documents and chunks whose retention clock has run out, across every tenant, along with their stored bytes.

It is cross-tenant, which is why it runs under WithMaintenance rather than a tenant scope — finding expired rows means reading rows whose owning tenants cannot be enumerated beforehand, and enumerating them would itself be the cross-tenant read. It processes at most DeleteExpiredBatchSize documents per call and is safe to run on a schedule; see the jobs package for a River worker.

func (*Processor) FullTextSearch 

func (p *Processor) FullTextSearch(ctx context.Context, scope Scope, query string, opts SearchOptions) ([]SearchResult, error)

FullTextSearch returns chunks matching query via Postgres full-text search, ranked by ts_rank.

It is a separate call from Processor.VectorSearch rather than fused with it. Fusing the two (reciprocal rank fusion or similar) means committing to one blend of the rankings for every caller, and the blend that suits a citation UI is rarely the one that suits an agent's tool call. A caller who wants fusion can run both and combine them.

The query goes through websearch_to_tsquery, so a caller can pass what a user typed — quoted phrases, OR, leading minus — without sanitising it into tsquery syntax, and without malformed input raising an error the way to_tsquery would.

A plain question matches on any of its terms rather than all of them

websearch_to_tsquery ANDs every term it is given, and the 'simple' configuration this column is built with has no stopword dictionary — chosen deliberately, since a stopword list is a language's and this library does not know the corpus's language. Together those mean "how do I reset my password?" asks for a chunk containing "how" AND "do" AND "i", finds none, and returns an empty slice indistinguishable from an empty corpus. That is the exact shape of a question a user types.

So a query that matched nothing on all of its terms is retried on any of them, ranked, which puts the chunk matching the most of them first. The relaxation is skipped when the caller wrote real search syntax — a quoted phrase, a leading minus, an explicit "or" — because rewriting those would change what they asked rather than widen it. SearchOptions.RequireAllTerms turns it off entirely.

Widening only when the strict query found nothing keeps precision where precision was available: a query whose terms all appear somewhere never reaches the second pass.

func (*Processor) GetDocument 

func (p *Processor) GetDocument(ctx context.Context, scope Scope, documentID uuid.UUID) (*Document, error)

GetDocument reads one document by id, confined to scope.

A document that exists but is outside the scope returns ErrNotFound, the same as one that does not exist. Distinguishing the two would tell a caller that a document id is real and belongs to someone else, which is itself a disclosure.

func (*Processor) Ingest 

func (p *Processor) Ingest(ctx context.Context, in DocumentInput) (*Document, error)

Ingest is a synchronous convenience wrapper around CreateDocument + ProcessDocument, for callers that don't need async job processing. On failure it still returns the underlying error; the Document reflects the persisted state either way.

func (*Processor) IngestPrepared added in v0.3.0 

func (p *Processor) IngestPrepared(ctx context.Context, documentID, tenantID uuid.UUID, prepared PreparedDocument) error

IngestPrepared indexes a document whose extraction, chunking and embedding already happened somewhere else — an extraction service that returns chunks and vectors from one call, a batch job, another pipeline entirely.

It is Processor.ProcessDocument's sibling: same starting point (a document created by Processor.CreateDocument), same terminal states, same events, same resume guard — only the front half of the pipeline differs. 

documentID, err := p.CreateDocument(ctx, in)
// ... run your own extract/chunk/embed ...
err = p.IngestPrepared(ctx, documentID, in.TenantID, prepared)

This exists because the alternative is writing chunk rows by hand, and four of the things ragit does on that path fail *silently* when they are forgotten. IngestPrepared owns all four: the scope, attribute and expiry columns each chunk carries denormalized (miss one and searches quietly return the wrong rows, or none); the embedding fingerprint retrieval filters on; the document's terminal state; and the EventSink notification a host application's own catalog depends on.

The whole write is one transaction — resume guard, chunk rows and terminal state together — because unlike ProcessDocument there is no provider call in the middle to keep it open. A prepared corpus is durable or it is absent, never half-written.

It needs no extractor, chunker or embedder, so a Processor for this path can be built with nil for all three. A malformed PreparedDocument is reported without touching the document: a bad call is a caller's bug, not a failed document.

func (*Processor) ListChunks 

func (p *Processor) ListChunks(ctx context.Context, scope Scope, documentID uuid.UUID) ([]Chunk, error)

ListChunks returns a document's chunks in order, confined to scope. Useful for showing what was indexed, and for debugging a chunker change.

func (*Processor) ListDocuments 

func (p *Processor) ListDocuments(ctx context.Context, scope Scope, filter ListFilter) ([]Document, error)

ListDocuments returns the documents visible to scope, newest first.

This is the catalog read a host application needs to answer "what has been indexed here", "is this upload still processing", and "why did it fail" — the last of which is why Document.Error is on the returned row rather than being reachable only through a failing call.

func (*Processor) MoveDocumentScope 

func (p *Processor) MoveDocumentScope(ctx context.Context, tenantID, documentID uuid.UUID, scopeA, scopeB, sessionID *uuid.UUID) error

MoveDocumentScope reassigns a document's scope dimensions and re-stamps its chunks to match.

The resync is why this method exists rather than callers updating the row themselves: chunks carry denormalized copies of the scope columns so that retrieval never needs a join, and those copies do not self-heal. Reprocessing does not fix them either — the resume check sees identical content, skips the rewrite, and leaves the chunks answering searches for their old scope.

func (*Processor) ProcessDocument 

func (p *Processor) ProcessDocument(ctx context.Context, documentID, tenantID uuid.UUID) error

ProcessDocument runs extract→chunk→embed→store for an existing document, resuming from whatever was already embedded in the current embedding space rather than re-billing chunks a prior attempt already paid for.

The document always ends in status ready, error, or skipped_too_large. ProcessDocument still returns the underlying error on failure — callers need it to decide whether the failure is worth retrying; it is not swallowed into a nil-error result.

The work is split across several short transactions rather than held in one. That is deliberate: a single transaction spanning the extractor's and embedding provider's HTTP calls would hold a connection open for the whole run and — worse — make the per-batch checkpointing meaningless, since nothing would be durable until the final commit.

func (*Processor) SetDocumentAttributes 

func (p *Processor) SetDocumentAttributes(ctx context.Context, tenantID, documentID uuid.UUID, attrs Attributes) error

SetDocumentAttributes replaces a document's attributes and re-stamps its chunks to match.

The resync is why this exists rather than callers updating the row: chunks carry a denormalized copy so retrieval can filter without a join, and that copy does not self-heal. Reprocessing will not fix it either — the resume check sees identical content and skips the rewrite, leaving chunks matching the labels they used to have. Same obligation as Processor.MoveDocumentScope, for the same reason.

func (*Processor) VectorSearch 

func (p *Processor) VectorSearch(ctx context.Context, scope Scope, query string, opts SearchOptions) ([]SearchResult, error)

VectorSearch returns the chunks nearest to query by cosine similarity.

Only chunks embedded by the active embedder are considered. Cosine distance between vectors from different models is not a weaker signal, it is a meaningless one, so chunks from another embedding space are excluded rather than ranked. If a provider or model changed without the corpus being re-embedded, this returns fewer results (or none) instead of confidently wrong ones — use Processor.CountMisalignedChunks to detect that state deliberately.

func (*Processor) WithEventSink 

func (p *Processor) WithEventSink(sink EventSink) *Processor

WithEventSink attaches a sink notified when a document reaches a terminal state. See EventSink for what is and is not guaranteed.

func (*Processor) WithMaxChunksPerDocument 

func (p *Processor) WithMaxChunksPerDocument(n int) *Processor

WithMaxChunksPerDocument sets the per-document chunk cap (0 = no cap) and returns the Processor for chaining. Above the cap, embedding is skipped and the document is flagged skipped_too_large instead of consuming the embedding budget.

type RetentionResult 

type RetentionResult struct {
	Documents int
	Chunks    int
	// ObjectErrors holds failures to purge object storage. They do not fail
	// the sweep: the rows are already gone, so a later pass will never revisit
	// these objects, and surfacing them here is the only way a caller learns
	// about the orphans.
	ObjectErrors []error
}

RetentionResult reports what one DeleteExpired pass removed.

type Scope 

type Scope struct {
	// contains filtered or unexported fields
}

Scope confines a read to the rows a caller may see. It is required by every retrieval and catalog call, and **its zero value matches no rows**.

That is the point of the type existing rather than the arguments being passed loose. A confinement expressed as optional parameters is one a caller can forget, and forgetting it returns everything — the failure is silent, looks like a working feature, and is only visible to whoever received rows they should not have. Here, forgetting produces ErrUnscoped.

Every dimension is restrictive by default

A dimension nobody mentioned matches only rows where it is NULL: 

ragit.Tenant(t)                   // tenant t, unscoped documents only
ragit.Tenant(t).A("acme")         // …in scope A "acme"
ragit.Tenant(t).AnyA()            // …in any scope A, said explicitly

So a corpus that never sets the scope columns works unchanged — every row has NULL in them — while a corpus that does use them cannot leak across a boundary because a caller left a field out. Widening is always a thing you can see in the call.

Unbounded access is a separate predicate, not a magic value

A caller who may see every scope says so with Scope.AnyA / Scope.AnyB. There is deliberately no "all scopes" id to put in the column, because a sentinel is one careless equality away from being treated as a real scope, and that failure is silent.

func Tenant 

func Tenant(tenantID uuid.UUID) Scope

Tenant begins a scope confined to one tenant. Every other dimension starts restrictive; widen explicitly.

func (Scope) A 

func (s Scope) A(ids ...uuid.UUID) Scope

A restricts to the given scope-A values.

Passing no values matches nothing rather than everything, so a caller that computes a permitted set and finds it empty gets an empty result rather than the whole tenant.

func (Scope) AnyA 

func (s Scope) AnyA() Scope

AnyA widens scope A to every value, including rows that have none. This is the "may see everything in this dimension" case, said out loud.

func (Scope) AnyB 

func (s Scope) AnyB() Scope

AnyB widens scope B to every value. See Scope.AnyA.

func (Scope) B 

func (s Scope) B(ids ...uuid.UUID) Scope

B restricts to the given scope-B values. See Scope.A.

func (Scope) Session 

func (s Scope) Session(id uuid.UUID) Scope

Session opts one ephemeral session's rows into the result, alongside the durable library. Without it, no session-scoped row is visible at all — an attachment uploaded into one conversation does not surface in another caller's search because a filter was forgotten.

func (Scope) TenantID 

func (s Scope) TenantID() uuid.UUID

TenantID returns the tenant this scope is confined to.

func (Scope) Validate 

func (s Scope) Validate() error

Validate reports whether the scope can be used. A scope with no tenant is the zero value, or close enough to it to be a bug.

type SearchOptions 

type SearchOptions struct {
	// TopK caps the number of results. Zero means DefaultTopK.
	TopK int

	// MinScore drops results below a cosine-similarity cutoff. It applies to
	// vector search only, and there is deliberately no default beyond zero:
	// the band separating a relevant match from noise is a property of the
	// embedding model, not of retrieval in general (Gemini's relevant matches
	// sit around 0.5–0.7, OpenAI's much higher), so a value baked in here
	// would be wrong for most models. Calibrate it per embedder.
	MinScore float64

	// Attributes narrows to chunks whose document carries all of these
	// key/value pairs. Empty narrows nothing — this filters a result set that
	// Scope has already confined, and is not itself a boundary. See
	// [Attributes].
	Attributes Attributes

	// RequireAllTerms keeps [Processor.FullTextSearch] strict: a query whose
	// terms do not all appear in one chunk returns nothing, instead of being
	// retried on any of its terms. Applies to full-text search only.
	//
	// The default is the relaxed form because the strict one answers a
	// user's plain question with an empty slice — see FullTextSearch. Set this
	// where a caller composed the query itself and an empty result is the
	// meaningful answer.
	RequireAllTerms bool
}

SearchOptions tunes a search. Confinement is not here: it is the Scope argument, which is required and cannot be defaulted away.

type SearchResult 

type SearchResult struct {
	ChunkID    uuid.UUID
	DocumentID uuid.UUID
	Filename   string
	ChunkIndex int32
	// HeadingPath is the chunk's trail of Markdown headings, e.g.
	// {"Chapter 2", "Section 2.1"} — the raw material for a citation.
	HeadingPath []string
	Content     string
	Metadata    json.RawMessage
	// Score is cosine similarity (1 = identical) for vector search, and a
	// ts_rank value for full-text search. The two are not comparable, and
	// neither has a meaningful absolute scale across models — see MinScore.
	Score float64
}

SearchResult is one retrieved chunk, carrying enough context to cite it.

Source Files

View all Source files

Directories

Path Synopsis
Package chunk splits extracted Markdown into retrieval-sized pieces.
 Package chunk splits extracted Markdown into retrieval-sized pieces.
cmd
ragit-gen command
Command ragit-gen regenerates ragit's migrations and models from the schema declaration in ragitschema.
 Command ragit-gen regenerates ragit's migrations and models from the schema declaration in ragitschema.
Package embed turns text into vectors via a single client speaking the OpenAI embeddings wire format — not per-provider adapters.
 Package embed turns text into vectors via a single client speaking the OpenAI embeddings wire format — not per-provider adapters.
Package extract turns raw document bytes into extracted text.
 Package extract turns raw document bytes into extracted text.
internal
migrate
Package migrate applies ragit's schema migrations.
 Package migrate applies ragit's schema migrations.
testutil
Package testutil boots a real, migrated Postgres for integration tests.
 Package testutil boots a real, migrated Postgres for integration tests.
Package jobs wires ragit's Processor into a River job queue.
 Package jobs wires ragit's Processor into a River job queue.
Package migrations embeds ragit's schema migrations so a host application never has to vendor the SQL into its own migration sequence.
 Package migrations embeds ragit's schema migrations so a host application never has to vendor the SQL into its own migration sequence.
Package ragitschema is ragit's schema declaration: the single source of truth from which its migrations and models are generated.
 Package ragitschema is ragit's schema declaration: the single source of truth from which its migrations and models are generated.
Package store puts and gets original document bytes in object storage.
 Package store puts and gets original document bytes in object storage.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.