Documentation
¶
Overview ¶
dna_batch.go implements the optional DNA batch edge-count capability on the local SQLite graph store. It is deliberately separate from store.go so the traversal work owned by other tasks stays untouched; the method is an additive, read-only hydration helper with per-ID semantics identical to CountEdgesByObservation.
Package graph implements the SQLite graph store for Cortex.
It provides knowledge graph operations for creating, querying, and deleting edges between observations. The store implements the domain.GraphRepository interface.
Index ¶
- type Store
- func (s *Store) CountAllEdges(ctx context.Context) (int, error)
- func (s *Store) CountEdgesByObservation(ctx context.Context, obsID int64) (int, error)
- func (s *Store) CountEdgesByObservationIDs(ctx context.Context, obsIDs []int64) (map[int64]int, error)
- func (s *Store) CreateEdge(ctx context.Context, edge *domain.Edge) error
- func (s *Store) CreateEdgeInTx(ctx context.Context, edge *domain.Edge) error
- func (s *Store) CurrentEdgeByPairInTx(ctx context.Context, fromObsID, toObsID int64, relationType string) (*domain.Edge, error)
- func (s *Store) CurrentEdges(ctx context.Context, obsID int64) ([]*domain.Edge, error)
- func (s *Store) DeleteEdge(ctx context.Context, id int64) error
- func (s *Store) EdgesAsOf(ctx context.Context, obsID int64, validAt, systemAt time.Time) ([]*domain.Edge, error)
- func (s *Store) GetContradictions(ctx context.Context, from, to time.Time) ([]*domain.Edge, error)
- func (s *Store) GetEdge(ctx context.Context, id int64) (*domain.Edge, error)
- func (s *Store) GetEdgesForObservation(ctx context.Context, obsID int64) ([]*domain.Edge, error)
- func (s *Store) GetEdgesValidAt(ctx context.Context, obsID int64, at time.Time) ([]*domain.Edge, error)
- func (s *Store) GetEvolutionChain(ctx context.Context, fromObsID, toObsID int64) ([]*domain.Edge, error)
- func (s *Store) GetLevelNeighborObservations(ctx context.Context, frontier []int64) (map[int64][]*domain.Observation, error)
- func (s *Store) GetRelated(ctx context.Context, obsID int64, depth int) ([]*domain.Observation, error)
- func (s *Store) GetRelatedScoped(ctx context.Context, obsID int64, opts domain.GraphTraversalOptions) ([]*domain.Observation, error)
- func (s *Store) UpdateEdge(ctx context.Context, edge *domain.Edge) error
- func (s *Store) WithinTx(ctx context.Context, handle any, fn func(context.Context) error) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store implements the SQLite graph store.
func (*Store) CountAllEdges ¶
CountAllEdges counts all edges in the system.
func (*Store) CountEdgesByObservation ¶
CountEdgesByObservation counts edges connected to a specific observation.
func (*Store) CountEdgesByObservationIDs ¶
func (s *Store) CountEdgesByObservationIDs(ctx context.Context, obsIDs []int64) (map[int64]int, error)
CountEdgesByObservationIDs counts edges connected to each requested observation in one statement per maxDNABatchIDs chunk. The statement unions two index-driven GROUP BY scans — one over edge sources using idx_edges_from, one over non-self-loop edge targets using idx_edges_to — and the two row sets are merged in Go. Counts match CountEdgesByObservation exactly: an edge contributes once per listed endpoint, and a self-loop counts once because only the source scan sees it (the target scan excludes from_obs_id = to_obs_id rows). IDs with no edges (and an empty input) yield no entry — callers treat absence as zero.
func (*Store) CreateEdge ¶
CreateEdge creates a relationship between two observations. Returns domain.ErrAlreadyExists if an edge with the same (from, to, relation_type) exists.
func (*Store) CreateEdgeInTx ¶
CreateEdgeInTx is the explicit atomic-save seam; the transaction is supplied by WithinTx and therefore predecessor closure and successor insert cannot commit independently.
func (*Store) CurrentEdgeByPairInTx ¶
func (s *Store) CurrentEdgeByPairInTx(ctx context.Context, fromObsID, toObsID int64, relationType string) (*domain.Edge, error)
CurrentEdgeByPairInTx returns the single current durable edge for the exact (from, to, relationType) triple. It reads through the shared transaction enlisted via WithinTx when one is active in ctx, so the caller observes its own uncommitted state rather than a separate connection snapshot. Returns domain.NotFoundError when no current fact exists for the triple.
func (*Store) CurrentEdges ¶
CurrentEdges returns current facts only. Historical, deprecated and superseded edges remain queryable through GetEdge/EvolutionChain.
func (*Store) DeleteEdge ¶
DeleteEdge removes a relationship between observations.
func (*Store) EdgesAsOf ¶
func (s *Store) EdgesAsOf(ctx context.Context, obsID int64, validAt, systemAt time.Time) ([]*domain.Edge, error)
EdgesAsOf reconstructs a graph at a valid-time and system-time point.
func (*Store) GetContradictions ¶
GetContradictions retrieves contradiction edges created in a time range.
func (*Store) GetEdgesForObservation ¶
GetEdgesForObservation retrieves all edges where the observation is either source or target.
func (*Store) GetEdgesValidAt ¶
func (s *Store) GetEdgesValidAt(ctx context.Context, obsID int64, at time.Time) ([]*domain.Edge, error)
GetEdgesValidAt retrieves edges for an observation that were valid at the given time. An edge is valid at time `at` if: (valid_from IS NULL OR valid_from <= at) AND (invalid_at IS NULL OR invalid_at > at).
func (*Store) GetEvolutionChain ¶
func (s *Store) GetEvolutionChain(ctx context.Context, fromObsID, toObsID int64) ([]*domain.Edge, error)
GetEvolutionChain retrieves all edges that share the same endpoints.
func (*Store) GetLevelNeighborObservations ¶
func (s *Store) GetLevelNeighborObservations(ctx context.Context, frontier []int64) (map[int64][]*domain.Observation, error)
GetLevelNeighborObservations resolves one-hop adjacency for an entire BFS frontier in one SQL statement per chunk (domain/graph.LevelNeighborBatcher). It returns hydrated neighbor observations for every requested frontier ID, deduplicated and ordered by ascending observation ID, following edges in both directions. Soft-deleted endpoints are excluded; on the v2 temporal schema only current facts (open validity, not deprecated/superseded) are followed.
func (*Store) GetRelated ¶
func (s *Store) GetRelated(ctx context.Context, obsID int64, depth int) ([]*domain.Observation, error)
GetRelated retrieves observations related to the given observation ID, up to the specified depth using a recursive CTE.
func (*Store) GetRelatedScoped ¶
func (s *Store) GetRelatedScoped(ctx context.Context, obsID int64, opts domain.GraphTraversalOptions) ([]*domain.Observation, error)
GetRelatedScoped performs bounded bidirectional traversal with cycle safety and tenant/workspace/project isolation. The visited set is maintained by an ID path predicate, so UNION ALL recursion cannot revisit a node indefinitely.
func (*Store) UpdateEdge ¶
UpdateEdge updates mutable edge fields by ID.
Source Files