ingest

package
v0.11.1 Latest Latest
Warning

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

 Go to latest Published: Aug 3, 2026 License: MIT Imports: 29 Imported by: 0

Documentation

Index

Constants

View Source 
const MetadataKeyLang = "lang"

MetadataKeyLang is the metadata key carrying the dominant *natural* language of a document, as an ISO 639-1 code ("fr", "en", ...). It is distinct from the "language" key the source-code parser injects, which names a programming language.

View Source 
const TaskTypeCleanup task.Type = "cleanup"
View Source 
const TaskTypeIndexFile task.Type = "index_file"
View Source 
const TaskTypeReindex task.Type = "reindex"

Variables

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

	// ErrCursorFilterMismatch is returned when a search cursor issued for one
	// metadata filter is replayed with a different one. The cursor anchors a
	// position inside a filtered ordering, so honouring it under another filter
	// would silently return duplicated or skipped results. Clients receiving it
	// must restart from the first page.
	ErrCursorFilterMismatch = errors.New("search cursor was issued for a different filter")
)

Functions

func CleanupTaskFactory added in v0.0.3 

func CleanupTaskFactory(id task.ID, payload []byte) (task.Task, error)

CleanupTaskFactory rebuilds a CleanupTask from its persisted payload, used by persistent task runners to resume or fetch the task.

func IndexFileTaskFactory added in v0.0.3 

func IndexFileTaskFactory(id task.ID, payload []byte) (task.Task, error)

IndexFileTaskFactory rebuilds an IndexFileTask from its persisted payload, used by persistent task runners to resume or fetch the task.

func ReindexTaskFactory added in v0.0.3 

func ReindexTaskFactory(id task.ID, payload []byte) (task.Task, error)

ReindexTaskFactory rebuilds a ReindexTask from its persisted payload, used by persistent task runners to resume or fetch the task.

Types

type BlobReferenceLister added in v0.9.0 

type BlobReferenceLister interface {
	ListReferencedBlobs(ctx context.Context, fn func(blob.Hash) error) error
}

BlobReferenceLister is the optional capability of a Store able to enumerate the blobs its documents reference without reading their content — typically from an index maintained at write time (see ingest/gorm.DocumentBlob). A store that does not implement it still works: the cleanup falls back to scanning the documents.

The enumeration must be *complete*: a reference missed here is a live blob deleted. An implementation is expected to be maintained in the same transaction as the document write, and to be covered by a differential test against blob.ScanHashes.

type CleanupHandler 

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

func NewCleanupHandler 

func NewCleanupHandler(idx index.Index, store Store, funcs ...CleanupHandlerOptionFunc) *CleanupHandler

func (*CleanupHandler) Handle 

func (h *CleanupHandler) Handle(ctx context.Context, tsk task.Task, events chan task.Event) error

Handle implements task.Handler.

type CleanupHandlerOptionFunc added in v0.9.0 

type CleanupHandlerOptionFunc func(h *CleanupHandler)

CleanupHandlerOptionFunc configures a CleanupHandler.

func WithCleanupBlobStore added in v0.9.0 

func WithCleanupBlobStore(store blob.Store) CleanupHandlerOptionFunc

WithCleanupBlobStore enables the blob garbage collection: blobs no stored document references any more are deleted. A nil store disables it.

type CleanupTask 

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

func NewCleanupTask 

func NewCleanupTask(collections []model.CollectionID) *CleanupTask

func (*CleanupTask) ID 

func (t *CleanupTask) ID() task.ID

ID implements task.Task.

func (*CleanupTask) MarshalJSON 

func (t *CleanupTask) MarshalJSON() ([]byte, error)

MarshalJSON implements task.Task.

func (*CleanupTask) Type 

func (t *CleanupTask) Type() task.Type

Type implements task.Task.

func (*CleanupTask) UnmarshalJSON 

func (t *CleanupTask) UnmarshalJSON(data []byte) error

UnmarshalJSON implements task.Task.

type CollectionUpdates 

type CollectionUpdates struct {
	Label       *string
	Description *string
}

type DocumentDigest 

type DocumentDigest struct {
	ID     model.DocumentID
	Source string
	ETag   string
}

DocumentDigest holds a minimal projection of a document used for bulk change detection.

type ImageEnricher added in v0.9.0 

type ImageEnricher interface {
	Enrich(ctx context.Context, data []byte, baseDir string, progress func(done, total int)) ([]byte, error)
}

ImageEnricher inserts, in the markdown source of a document, the textual description of the images it embeds — making them searchable like any other text. It is satisfied by markdown/imagetext.Enricher.

baseDir is the directory relative image paths resolve against (empty disables their resolution) and progress, when non-nil, is called as descriptions complete. Implementations must be tolerant: an image that cannot be described is left alone rather than failing the document.

type IndexFileHandler 

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

func NewIndexFileHandler 

func NewIndexFileHandler(store Store, fileConverter convert.Converter, idx index.Index, maxWordPerSection int, funcs ...IndexFileHandlerOptionFunc) *IndexFileHandler

func (*IndexFileHandler) Handle 

func (h *IndexFileHandler) Handle(ctx context.Context, tsk task.Task, events chan task.Event) error

Handle implements task.Handler.

type IndexFileHandlerOptionFunc added in v0.0.4 

type IndexFileHandlerOptionFunc func(h *IndexFileHandler)

func WithIndexFileHandlerImageEnrichment added in v0.9.0 

func WithIndexFileHandlerImageEnrichment(enricher ImageEnricher) IndexFileHandlerOptionFunc

WithIndexFileHandlerImageEnrichment describes the images embedded in the markdown source of a document before it is parsed — so it applies uniformly to native .md files and to the output of the converters (pandoc, LibreOffice, GenAI OCR). A nil enricher disables it.

func WithIndexFileHandlerSourceCode added in v0.0.4 

func WithIndexFileHandlerSourceCode(registry *sourcecode.Registry) IndexFileHandlerOptionFunc

WithIndexFileHandlerSourceCode enables source-code parsing for the file extensions registered in the registry. A nil registry disables it.

type IndexFileOptionFunc 

type IndexFileOptionFunc func(opts *IndexFileOptions)

func WithIndexFileCollections 

func WithIndexFileCollections(collections ...model.CollectionID) IndexFileOptionFunc

func WithIndexFileETag 

func WithIndexFileETag(etag string) IndexFileOptionFunc

func WithIndexFileImageBaseDir added in v0.10.1 

func WithIndexFileImageBaseDir(dir string) IndexFileOptionFunc

WithIndexFileImageBaseDir sets the directory the relative image paths embedded in the document resolve against. Callers indexing a file whose source is not its filesystem path must set it, otherwise its images cannot be found (see IndexFileTask.ImageBaseDir).

func WithIndexFileMetadata added in v0.0.3 

func WithIndexFileMetadata(metadata map[string]any) IndexFileOptionFunc

WithIndexFileMetadata attaches arbitrary metadata to the indexed document, used for metadata filtering at search time.

func WithIndexFileSource 

func WithIndexFileSource(source *url.URL) IndexFileOptionFunc

type IndexFileOptions 

type IndexFileOptions struct {
	ETag   string
	Source *url.URL
	// ImageBaseDir is the directory the relative image paths of the document
	// resolve against. Empty falls back on the directory of the source, which
	// only holds when the source is a filesystem path.
	ImageBaseDir string
	// Names of the collection to associate with the document
	Collections []model.CollectionID
	// Arbitrary document metadata used for filtering at search time.
	Metadata map[string]any
}

func NewIndexFileOptions 

func NewIndexFileOptions(funcs ...IndexFileOptionFunc) *IndexFileOptions

type IndexFileTask 

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

func NewIndexFileTask 

func NewIndexFileTask(path string, originalName string, etag string, source *url.URL, collections []model.CollectionID, metadata map[string]any, funcs ...IndexFileTaskOption) *IndexFileTask

func (*IndexFileTask) ID 

func (i *IndexFileTask) ID() task.ID

ID implements task.Task.

func (*IndexFileTask) ImageBaseDir added in v0.10.1 

func (i *IndexFileTask) ImageBaseDir() string

ImageBaseDir returns the directory relative image paths resolve against: the directory of the *original* file, not of the staged copy the handler works on.

The explicit value set by the scheduler wins, and is the only reliable one: falling back on the source assumes it carries a real filesystem path, which is not a given. An indexer may well store a logical identifier instead — the CLI's --base-dir makes sources relative to a base directory, keeping a leading slash so they stay well-formed file URLs. Such a source still looks absolute, so the fallback cannot tell it apart; it yields a directory that does not exist, every image is silently skipped (enrichment is best-effort), and the document is indexed without any of its illustrations. Schedulers that know the real location must therefore say so.

func (*IndexFileTask) MarshalJSON 

func (i *IndexFileTask) MarshalJSON() ([]byte, error)

MarshalJSON implements task.Task.

func (*IndexFileTask) Type 

func (i *IndexFileTask) Type() task.Type

Type implements task.Task.

func (*IndexFileTask) UnmarshalJSON 

func (i *IndexFileTask) UnmarshalJSON(data []byte) error

UnmarshalJSON implements task.Task.

type IndexFileTaskOption added in v0.10.1 

type IndexFileTaskOption func(*IndexFileTask)

IndexFileTaskOption configures an optional aspect of an IndexFileTask, kept out of the constructor's positional arguments.

func WithIndexFileTaskImageBaseDir added in v0.10.1 

func WithIndexFileTaskImageBaseDir(dir string) IndexFileTaskOption

WithIndexFileTaskImageBaseDir sets the directory the relative image paths of the document resolve against, overriding the one derived from the source.

type Manager 

type Manager struct {
	Store
	// contains filtered or unexported fields
}

Manager orchestrates the ingestion pipeline: file conversion, parsing, storage and indexing, scheduled through a task runner.

func NewManager 

func NewManager(store Store, idx index.Index, taskRunner task.Runner, funcs ...ManagerOptionFunc) *Manager

func (*Manager) CleanupIndex 

func (m *Manager) CleanupIndex(ctx context.Context, collections ...model.CollectionID) (task.ID, error)

func (*Manager) CleanupTempDir added in v0.0.3 

func (m *Manager) CleanupTempDir() error

CleanupTempDir removes this manager's staging directory and everything left in it. It is a best-effort, idempotent operation typically called on shutdown, once in-flight indexing tasks have drained. It is a no-op when a stable staging directory is configured, since its files may back pending, resumable tasks that must survive the restart.

func (*Manager) IndexFile 

func (m *Manager) IndexFile(ctx context.Context, filename string, r io.Reader, funcs ...IndexFileOptionFunc) (task.ID, error)

IndexFile copies the file to a temporary location then schedules an asynchronous indexing task. The returned task.ID can be used to track progress through the task runner.

func (*Manager) RegisterHandlers 

func (m *Manager) RegisterHandlers(runner task.Runner)

RegisterHandlers registers the ingestion task handlers on the given runner. When the runner is a task.PersistentRunner, the matching deserialization factories are registered too so pending tasks can be rebuilt and resumed after a restart.

func (*Manager) Reindex 

func (m *Manager) Reindex(ctx context.Context) (task.ID, error)

Reindex rebuilds the index for the whole store.

func (*Manager) ReindexCollection 

func (m *Manager) ReindexCollection(ctx context.Context, collectionID model.CollectionID) (task.ID, error)

ReindexCollection rebuilds the index for a single collection.

func (*Manager) Search 

func (m *Manager) Search(ctx context.Context, query string, funcs ...SearchOptionFunc) (*SearchResults, error)

func (*Manager) SupportedExtensions 

func (m *Manager) SupportedExtensions() []string

type ManagerOptionFunc 

type ManagerOptionFunc func(opts *ManagerOptions)

func WithManagerBlobStore added in v0.9.0 

func WithManagerBlobStore(store blob.Store) ManagerOptionFunc

WithManagerBlobStore declares the blob store holding the images referenced by the documents, so the cleanup task can collect the ones no document references any more.

func WithManagerFileConverter 

func WithManagerFileConverter(fileConverter convert.Converter) ManagerOptionFunc

func WithManagerImageEnrichment added in v0.9.0 

func WithManagerImageEnrichment(enricher ImageEnricher) ManagerOptionFunc

WithManagerImageEnrichment describes the images embedded in the markdown source of a document (native .md as well as converter output) before it is parsed, so their descriptions are indexed as ordinary text.

func WithManagerMaxWordPerSection 

func WithManagerMaxWordPerSection(maxWordPerSection int) ManagerOptionFunc

func WithManagerReranker added in v0.0.3 

func WithManagerReranker(reranker Reranker) ManagerOptionFunc

WithManagerReranker plugs a reranker into the search pipeline: it reorders the fused (and filtered) candidates before pagination.

func WithManagerSourceCode added in v0.0.4 

func WithManagerSourceCode(registry *sourcecode.Registry) ManagerOptionFunc

WithManagerSourceCode enables source-code indexing: files whose extension is registered in the registry are parsed into declaration-level sections instead of going through the converter and markdown pipeline.

func WithManagerStagingDir added in v0.0.3 

func WithManagerStagingDir(dir string) ManagerOptionFunc

WithManagerStagingDir pins the ingestion staging directory to a stable location. Use it together with a persistent task runner so that files staged by IndexFile survive a restart and their resumed indexing tasks can find them. When set, the directory is not removed on shutdown (CleanupTempDir becomes a no-op).

type ManagerOptions 

type ManagerOptions struct {
	MaxWordPerSection int
	FileConverter     convert.Converter
	Reranker          Reranker
	// SourceCode, when set, enables source-code indexing for the file
	// extensions registered in the registry.
	SourceCode *sourcecode.Registry
	// ImageEnricher, when set, describes the images embedded in the markdown
	// source of a document before it is parsed.
	ImageEnricher ImageEnricher
	// Blobs, when set, is the blob store holding the images referenced by the
	// indexed documents; the cleanup task collects the unreferenced ones.
	Blobs blob.Store
	// StagingDir, when set, pins the directory where files awaiting indexing are
	// staged to a stable location instead of a per-process temporary directory.
	// It is required for a persistent task runner: a resumed IndexFile task must
	// still find its staged file after a restart.
	StagingDir string
}

func NewManagerOptions 

func NewManagerOptions(funcs ...ManagerOptionFunc) *ManagerOptions

type MetadataProvider added in v0.0.3 

type MetadataProvider interface {
	// GetDocumentsMetadataBySources returns, for each of the given source URLs
	// that maps to a stored document, the document's metadata. Sources without
	// a document (or without metadata) may be omitted from the result.
	GetDocumentsMetadataBySources(ctx context.Context, sources []string) (map[string]map[string]any, error)
}

MetadataProvider is an optional capability a Store may implement to expose document metadata keyed by source URL. The ingestion Manager uses it to apply metadata filters at search time. A Store that does not implement it cannot be used together with a search metadata filter.

type QueryCollectionsOptions 

type QueryCollectionsOptions struct {
	Page  *int
	Limit *int

	// Do not retrieve associations
	HeaderOnly bool

	// Collections with these ids
	IDs []model.CollectionID
}

type QueryDocumentsOptions 

type QueryDocumentsOptions struct {
	// Page is the 0-based page index: the query skips Page*Limit documents.
	Page  *int
	Limit *int

	// Do not retrieve associations
	HeaderOnly bool

	// Documents matching the given source
	MatchingSource *url.URL

	// Documents without parent collection
	Orphaned *bool

	// Documents matching the given source pattern (LIKE %pattern%)
	SourcePattern *string

	// Column to sort by: "source" or "created_at" (default)
	SortBy *string

	// Sort direction: "asc" or "desc" (default)
	SortOrder *string
}

type ReindexHandler 

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

func NewReindexHandler 

func NewReindexHandler(store Store, idx index.Index, maxWordPerSection int) *ReindexHandler

func (*ReindexHandler) Handle 

func (h *ReindexHandler) Handle(ctx context.Context, tsk task.Task, events chan task.Event) error

Handle implements task.Handler.

type ReindexTask 

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

ReindexTask rebuilds the index from the stored documents. An empty CollectionID reindexes the whole store.

func NewReindexTask 

func NewReindexTask(collectionID model.CollectionID) *ReindexTask

NewReindexTask creates a task reindexing a single collection, or the whole store if collectionID is empty.

func (*ReindexTask) CollectionID 

func (t *ReindexTask) CollectionID() model.CollectionID

CollectionID returns the collection ID to reindex (empty means all).

func (*ReindexTask) ID 

func (t *ReindexTask) ID() task.ID

ID implements task.Task.

func (*ReindexTask) MarshalJSON 

func (t *ReindexTask) MarshalJSON() ([]byte, error)

MarshalJSON implements task.Task.

func (*ReindexTask) Type 

func (t *ReindexTask) Type() task.Type

Type implements task.Task.

func (*ReindexTask) UnmarshalJSON 

func (t *ReindexTask) UnmarshalJSON(data []byte) error

UnmarshalJSON implements task.Task.

type Reranker added in v0.0.3 

type Reranker interface {
	Rerank(ctx context.Context, query string, results []*index.SearchResult) ([]*index.SearchResult, error)
}

Reranker reorders search results by relevance to the query, refining the initial retrieval/fusion ranking. It is an optional component plugged into the search pipeline (see Manager and WithManagerReranker). Implementations may reorder results and their sections and adjust scores, but must not fabricate new results. It runs after metadata filtering and before pagination, so the reranked order is the one exposed to callers and encoded in pagination cursors.

type SearchOptionFunc 

type SearchOptionFunc func(opts *SearchOptions)

func WithSearchCandidatePoolSize added in v0.0.3 

func WithSearchCandidatePoolSize(size int) SearchOptionFunc

WithSearchCandidatePoolSize overrides the number of fused candidates fetched before filtering, reranking and pagination.

func WithSearchCollections 

func WithSearchCollections(collections ...model.CollectionID) SearchOptionFunc

func WithSearchCursor added in v0.0.3 

func WithSearchCursor(cursor string) SearchOptionFunc

WithSearchCursor resumes pagination after the given opaque cursor (the NextCursor of a previous SearchResults).

func WithSearchFilter added in v0.0.3 

func WithSearchFilter(filter index.Filter) SearchOptionFunc

WithSearchFilter restricts results to documents whose metadata satisfies the given filter. It requires the configured Store to implement MetadataProvider.

func WithSearchMaxResults 

func WithSearchMaxResults(max int) SearchOptionFunc

type SearchOptions 

type SearchOptions struct {
	// MaxResults is the page size (number of results returned per call).
	MaxResults int
	// Names of the collection the query will be restricted to
	Collections []model.CollectionID
	// Filter restricts results to documents whose metadata matches every
	// condition. It requires the Store to implement MetadataProvider.
	Filter index.Filter
	// Cursor resumes pagination after a previous page. Empty means the first
	// page. Use the NextCursor returned by the previous call.
	Cursor string
	// CandidatePoolSize pins how many fused candidates are fetched before
	// filtering, reranking and pagination, disabling the adaptive sizing. 0
	// (the default) lets the Manager size the window from the requested page,
	// widening it while a filter leaves too few survivors.
	CandidatePoolSize int
}

func NewSearchOptions 

func NewSearchOptions(funcs ...SearchOptionFunc) *SearchOptions

type SearchResults added in v0.0.3 

type SearchResults struct {
	Results    []*index.SearchResult
	NextCursor string
}

SearchResults holds a page of search results together with the cursor needed to fetch the next page (empty when the last page has been reached).

type Store 

type Store interface {
	// ListDocumentDigests returns (Source, ETag) pairs for documents whose source URL
	// starts with sourcePrefix. Results are paginated; pageSize=0 defaults to 500.
	ListDocumentDigests(ctx context.Context, sourcePrefix string, page int, pageSize int) ([]DocumentDigest, error)

	GetDocumentByID(ctx context.Context, id model.DocumentID) (model.PersistedDocument, error)
	SaveDocuments(ctx context.Context, documents ...model.Document) error
	DeleteDocumentBySource(ctx context.Context, source *url.URL) error
	DeleteDocumentByID(ctx context.Context, ids ...model.DocumentID) error
	QueryDocuments(ctx context.Context, opts QueryDocumentsOptions) ([]model.PersistedDocument, int64, error)

	// QueryDocumentsByCollectionID retrieves all documents belonging to a specific collection.
	QueryDocumentsByCollectionID(ctx context.Context, collectionID model.CollectionID, opts QueryDocumentsOptions) ([]model.PersistedDocument, int64, error)

	GetSectionByID(ctx context.Context, id model.SectionID) (model.Section, error)
	GetSectionsByIDs(ctx context.Context, ids []model.SectionID) (map[model.SectionID]model.Section, error)
	SectionExists(ctx context.Context, id model.SectionID) (bool, error)
	// SectionsExist checks the existence of multiple sections in a single query.
	SectionsExist(ctx context.Context, ids []model.SectionID) (map[model.SectionID]bool, error)

	GetCollectionByID(ctx context.Context, id model.CollectionID, full bool) (model.PersistedCollection, error)
	QueryCollections(ctx context.Context, opts QueryCollectionsOptions) ([]model.PersistedCollection, error)
	CreateCollection(ctx context.Context, label string) (model.PersistedCollection, error)
	UpdateCollection(ctx context.Context, id model.CollectionID, updates CollectionUpdates) (model.PersistedCollection, error)
	GetCollectionStats(ctx context.Context, id model.CollectionID) (*model.CollectionStats, error)
	DeleteCollection(ctx context.Context, id model.CollectionID) error
}

Store persists documents, sections and collections backing the ingestion pipeline.

Source Files

View all Source files

Directories

Path Synopsis

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.