mongo

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

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. A document with no tenant_id field at all counts too.

func (*Store) CreateChunkBatch ¶

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

CreateChunkBatch persists a batch of chunks. Each chunk's document must belong to the context tenant and sit in the chunk's collection; that is checked for the whole batch before the insert, as CreateDocument checks its collection.

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

CreateDocument persists a new document. Its collection must belong to the context tenant, else ErrCollectionNotFound. The check runs before the insert, outside a transaction: a collection's tenant never changes, and Mongo has no cascade, so a race with a delete can only leave this tenant's document without its collection.

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

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 (collections and indexes) 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.

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 a multi-document transaction, so it needs a replica set (or mongos); on a standalone server it fails with an error that says so. The session's WithTransaction reruns the whole step when the server reports a transient conflict, which is how two replaces of one source that collide end up one after the other. The document read and the chunk ID check happen inside the transaction, and every refusal comes before the first write.

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 filtered UpdateOne, so of several claims of one stalled document only the first matches.

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