ragit

package module
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 15 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. 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 per chunk.
  • Retrieval as two separate calls — vector and full-text — rather than one fused endpoint.
  • One resumable River job per document, checkpointed per embedding batch, so a retry resumes instead of re-billing the embedding provider.

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.

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.

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.

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.

Development

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

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.

Four 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.

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 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 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"),
	EmbeddingModel: sqlb.TextColumn[string]("embedding_model"),
	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 ErrUnscoped = errors.New("ragit: query has no tenant scope")

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

Functions

func Migrate 

func Migrate(ctx context.Context, pool *pgxpool.Pool) 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.

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) 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.

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"`
	EmbeddingModel *string         `db:"embedding_model" json:"embedding_model" sqlb:"type:text,filter"`
	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) SetEmbeddingModel 

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

SetEmbeddingModel sets embedding_model.

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 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.

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.

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) 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
}

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 embedded schema migrations.
 Package migrate applies ragit's embedded 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.