postgres

func (*Store) Close ¶

func (s *Store) Close() error

Close closes the database connection.

func (*Store) CompleteDocument ¶

func (s *Store) CompleteDocument(ctx context.Context, docID id.DocumentID, chunkCount int) error

CompleteDocument moves a processing document to ready with one guarded UPDATE.

func (*Store) CountChunks ¶

func (s *Store) CountChunks(ctx context.Context, filter *chunk.CountFilter) (int64, error)

func (*Store) CountCollections ¶

func (s *Store) CountCollections(ctx context.Context, filter *collection.CountFilter) (int64, error)

func (*Store) CountDocuments ¶

func (s *Store) CountDocuments(ctx context.Context, filter *document.CountFilter) (int64, error)

func (*Store) CountStalledUntenanted ¶

func (s *Store) CountStalledUntenanted(ctx context.Context, before time.Time) (int64, error)

CountStalledUntenanted counts on the same index as ListStalledDocuments.

func (*Store) CreateChunkBatch ¶

func (s *Store) CreateChunkBatch(ctx context.Context, chunks []*chunk.Chunk) error

func (*Store) CreateCollection ¶

func (s *Store) CreateCollection(ctx context.Context, col *collection.Collection) error

func (*Store) CreateDocument ¶

func (s *Store) CreateDocument(ctx context.Context, doc *document.Document) error

func (*Store) DeleteChunksByCollection ¶

func (s *Store) DeleteChunksByCollection(ctx context.Context, colID id.CollectionID) error

func (*Store) DeleteChunksByDocument ¶

func (s *Store) DeleteChunksByDocument(ctx context.Context, docID id.DocumentID) error

func (*Store) DeleteCollection ¶

func (s *Store) DeleteCollection(ctx context.Context, colID id.CollectionID) error

func (*Store) DeleteDocument ¶

func (s *Store) DeleteDocument(ctx context.Context, docID id.DocumentID) error

func (*Store) DeleteDocumentsByCollection ¶

func (s *Store) DeleteDocumentsByCollection(ctx context.Context, colID id.CollectionID) error

func (*Store) FailDocument ¶

func (s *Store) FailDocument(ctx context.Context, docID id.DocumentID, errText string, staleBefore time.Time) error

FailDocument marks a pending or processing document failed with one guarded UPDATE.

func (*Store) GetChunk ¶

func (s *Store) GetChunk(ctx context.Context, chunkID id.ChunkID) (*chunk.Chunk, error)

func (*Store) GetCollection ¶

func (s *Store) GetCollection(ctx context.Context, colID id.CollectionID) (*collection.Collection, error)

func (*Store) GetCollectionByName ¶

func (s *Store) GetCollectionByName(ctx context.Context, tenantID, name string) (*collection.Collection, error)

func (*Store) GetDocument ¶

func (s *Store) GetDocument(ctx context.Context, docID id.DocumentID) (*document.Document, error)

func (*Store) ListChunks ¶

func (s *Store) ListChunks(ctx context.Context, filter *chunk.ListFilter) ([]*chunk.Chunk, error)

func (*Store) ListChunksByDocument ¶

func (s *Store) ListChunksByDocument(ctx context.Context, docID id.DocumentID) ([]*chunk.Chunk, error)

func (*Store) ListCollections ¶

func (s *Store) ListCollections(ctx context.Context, filter *collection.ListFilter) ([]*collection.Collection, error)

func (*Store) ListDocuments ¶

func (s *Store) ListDocuments(ctx context.Context, filter *document.ListFilter) ([]*document.Document, error)

func (*Store) ListStalledDocuments ¶

func (s *Store) ListStalledDocuments(ctx context.Context, before time.Time, limit int) ([]*document.Document, error)

ListStalledDocuments reads across tenants on the (state, updated_at) index.

func (*Store) Migrate ¶

func (s *Store) Migrate(ctx context.Context) error

Migrate applies the pending migrations in Migrations through grove's orchestrator, which records each version in grove_migrations. Every Up is idempotent, so an install that ran them untracked is adopted in place.

Grove v1.10.0 creates grove_migrations and grove_migration_locks before it takes the migration lock, so replicas booting together on a database without those tables race on the CREATE, and the losers fail with duplicate_table or duplicate_object. Migrate retries that failure a few times with a short jittered pause. By the next attempt the tables exist and grove's lock serialises the replicas.

func (*Store) Ping ¶

func (s *Store) Ping(ctx context.Context) error

Ping verifies the database connection.

func (*Store) ReplaceSourceDocument ¶

func (s *Store) ReplaceSourceDocument(ctx context.Context, doc *document.Document, chunks []*chunk.Chunk) ([]id.DocumentID, error)

ReplaceSourceDocument runs in one transaction. It first takes a transaction-scoped advisory lock on the tenant, collection and source key, so concurrent replaces of one source wait for each other instead of colliding on the one-ready-document index, then reads doc FOR UPDATE. Every refusal happens before the first write; any later failure rolls the whole replace back.

func (*Store) TouchDocument ¶

func (s *Store) TouchDocument(ctx context.Context, docID id.DocumentID, staleBefore time.Time) error

TouchDocument records an ingest's heartbeat with one guarded UPDATE, so of several claims of one stalled document only the first matches a row.

func (*Store) UpdateCollection ¶

func (s *Store) UpdateCollection(ctx context.Context, col *collection.Collection) error

func (*Store) UpdateDocument ¶

func (s *Store) UpdateDocument(ctx context.Context, doc *document.Document) error