README
¶
graph-ingest
Entity and triple ingestion component for the graph subsystem.
Overview
The graph-ingest component is the entry point for all entity data flowing into the graph system. It subscribes to JetStream subjects for incoming entity events and stores them in the ENTITY_STATES KV bucket.
Architecture
┌─────────────────┐
objectstore.stored ─┤ │
│ graph-ingest ├──► ENTITY_STATES (KV)
sensor.processed ─┤ │
└─────────────────┘
Features
- Entity CRUD: Create, read, update, delete entity operations
- Triple Mutations: Add and remove triples on entities
- Hierarchy Inference: Automatically creates container entities based on 6-part entity ID structure
- At-Least-Once Delivery: JetStream subscription with proper acknowledgment
Configuration
{
"type": "processor",
"name": "graph-ingest",
"enabled": true,
"config": {
"ports": {
"inputs": [
{
"name": "objectstore_in",
"config": {"kind":"jetstream","subjects":["objectstore.stored.entity"],"interface":{"type":"storage.stored.v1"}}
},
{
"name": "sensor_in",
"config": {"kind":"jetstream","subjects":["sensor.processed.entity"],"interface":{"type":"iot.sensor.v1"}}
}
],
"outputs": [
{
"name": "entity_states",
"config": {"kind":"kv-write","bucket":"ENTITY_STATES"}
}
]
},
"enable_hierarchy": true
}
}
Configuration Options
| Option | Type | Default | Description |
|---|---|---|---|
ports |
object | required | Port configuration for inputs and outputs |
enable_hierarchy |
bool | false | Enable automatic hierarchy inference |
Ports
Inputs
| Name | Type | Subject | Description |
|---|---|---|---|
| objectstore_in | jetstream | objectstore.stored.entity | Stored document entity events |
| sensor_in | jetstream | sensor.processed.entity | Sensor entity events |
Outputs
| Name | Type | Subject | Description |
|---|---|---|---|
| entity_states | kv | ENTITY_STATES | Entity state storage |
Hierarchy Inference
When enable_hierarchy is enabled, the component automatically creates container entities based on the 6-part entity ID structure:
org.platform.domain.system.type.instance
│ │ │ │ │ │
└──────┴───────┴──────┴─────┴──────┴─► Real entity
│ │ │ │
└───────┴──────┴─────┴─► Type container (*.group)
│ │
└──────┴─► System container (*.container)
│
└─► Domain container (*.level)
This creates edges that enable:
- Efficient traversal by type, system, or domain
- Automatic grouping without explicit relationship creation
- Query-time aggregation by hierarchy level
Dependencies
Upstream
- None (entry point component)
Downstream
graph-index- watches ENTITY_STATES for index updatesgraph-embedding- watches ENTITY_STATES for embedding generationgraph-clustering- watches ENTITY_STATES for community detection triggers
Metrics
| Metric | Type | Description |
|---|---|---|
graph_ingest_entities_processed_total |
counter | Total entities processed |
graph_ingest_triples_added_total |
counter | Total triples added |
graph_ingest_errors_total |
counter | Total processing errors |
graph_ingest_processing_duration_seconds |
histogram | Processing time per entity |
Health
The component reports healthy when:
- JetStream subscription is active
- KV bucket is accessible
- No sustained error rate above threshold
Documentation
¶
Overview ¶
Package graphingest provides the graph-ingest component for entity and triple ingestion.
Package graphingest provides the graph-ingest component for entity and triple ingestion.
Overview ¶
The graph-ingest component is responsible for ingesting entities and triples into the graph system. It subscribes to JetStream subjects for incoming entity data and stores them in the ENTITY_STATES KV bucket.
Tier ¶
Tier: ALL TIERS (Tier 0, 1, 2) - Required for all deployments.
Architecture ¶
graph-ingest is a core component required for all deployment tiers (Structural, Statistical, Semantic). It serves as the entry point for all entity data flowing into the graph subsystem.
┌─────────────────┐
objectstore.stored ─┤ │
│ graph-ingest ├──► ENTITY_STATES (KV)
sensor.processed ─┤ │
└─────────────────┘
Features ¶
- Entity CRUD operations (create, read, update, delete)
- Triple mutations (add, remove)
- Hierarchy inference (optional) - creates container entities based on 6-part entity ID structure
- JetStream subscription with at-least-once delivery semantics
- KV storage with atomic updates
Configuration ¶
The component is configured via JSON with the following structure:
{
"ports": {
"inputs": [
{"name":"objectstore_in","config":{"kind":"jetstream","subjects":["objectstore.stored.entity"]}},
{"name":"sensor_in","config":{"kind":"jetstream","subjects":["sensor.processed.entity"]}}
],
"outputs": [
{"name":"entity_states","config":{"kind":"kv-write","bucket":"ENTITY_STATES"}}
]
},
"enable_hierarchy": true
}
Port Definitions ¶
Inputs:
- JetStream subscriptions for entity events (objectstore.stored.entity, sensor.processed.entity)
Outputs:
- KV bucket: ENTITY_STATES - stores entity state with triples
Hierarchy Inference ¶
When enable_hierarchy is true, the component automatically creates container entities based on the 6-part entity ID structure (org.platform.domain.system.type.instance). This creates edges for:
- Type containers (*.group)
- System containers (*.container)
- Domain containers (*.level)
Usage ¶
Register the component with the component registry:
import graphingest "github.com/c360studio/semstreams/processor/graph-ingest"
func init() {
graphingest.Register(registry)
}
Dependencies ¶
This component has no upstream graph component dependencies. It is the entry point for entity data and other graph components (graph-index, graph-embedding, etc.) watch its output KV bucket.
Package graphingest query handlers
Index ¶
- func CreateGraphIngest(rawConfig json.RawMessage, deps component.Dependencies) (component.Discoverable, error)
- func Register(registry *component.Registry) error
- type Component
- func (c *Component) ConfigSchema() component.ConfigSchema
- func (c *Component) CreateEntity(ctx context.Context, entity *graph.EntityState) error
- func (c *Component) DataFlow() component.FlowMetrics
- func (c *Component) DebugStatus() any
- func (c *Component) Health() component.HealthStatus
- func (c *Component) Initialize() error
- func (c *Component) InputPorts() []component.Port
- func (c *Component) MergeEntity(ctx context.Context, entity *graph.EntityState) error
- func (c *Component) Meta() component.Metadata
- func (c *Component) OutputPorts() []component.Port
- func (c *Component) Start(ctx context.Context) error
- func (c *Component) Stop(timeout time.Duration) error
- type Config
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func CreateGraphIngest ¶
func CreateGraphIngest(rawConfig json.RawMessage, deps component.Dependencies) (component.Discoverable, error)
CreateGraphIngest is the factory function for creating graph-ingest components
Types ¶
type Component ¶
type Component struct {
// contains filtered or unexported fields
}
Component implements the graph-ingest processor
func (*Component) ConfigSchema ¶
func (c *Component) ConfigSchema() component.ConfigSchema
ConfigSchema returns the configuration schema
func (*Component) CreateEntity ¶
CreateEntity atomically creates a new entity. Existing keys are never overwritten; callers observe natsclient.ErrKVKeyExists and decide whether that conflict is acceptable for their operation.
func (*Component) DataFlow ¶
func (c *Component) DataFlow() component.FlowMetrics
DataFlow returns current data flow metrics
func (*Component) DebugStatus ¶
DebugStatus implements component.DebugStatusProvider: the FULL poison inventory, sorted by entity ID, so a mass-poison event is enumerable in-band (design D6 — the Health message and gauge are bounded; this is not).
func (*Component) Health ¶
func (c *Component) Health() component.HealthStatus
Health returns current health status
func (*Component) Initialize ¶
Initialize validates configuration and sets up ports (no I/O)
func (*Component) InputPorts ¶
InputPorts returns input port definitions. Reads directly from config so ports are available before Initialize().
func (*Component) MergeEntity ¶
MergeEntity ingests a streaming-consumer EntityState (typically built by extractEntityFromMessage from a Graphable arriving on the JetStream input) WITHOUT clobbering pre-existing triples on the entity. First write behaves like CreateEntity (the entity didn't exist; its fields land verbatim); subsequent writes MERGE the incoming triples predicate-level (replace per (subject,predicate) via graph.MergeTriples, gh#466 — the create-time indexing profile is preserved) and refresh latest-wins metadata (MessageType, StorageRef, UpdatedAt) while monotonically bumping Version.
Closes gh#177: the prior code called CreateEntity (Put = full- replace) from handleMessage, which erased triples written through mutation request/reply. The JetStream consumer path was the lone outlier. Lifecycle-managed entities surfaced this most loudly: Manager.Create stamped the phase triple, then the first Graphable arrival via a downstream processor wiped it.
Uses entityBucket.UpdateWithRetry for atomic CAS read-modify-write, so concurrent arrivals on the same Subject converge without racing.
func (*Component) OutputPorts ¶
OutputPorts returns output port definitions. Reads directly from config so ports are available before Initialize().
type Config ¶
type Config struct {
Ports *component.PortConfig `json:"ports" schema:"type:ports,description:Port configuration,category:basic"`
EnableHierarchy bool `json:"enable_hierarchy" schema:"type:bool,description:Enable hierarchy inference,default:false,category:advanced"`
EnableTypeSiblings *bool `` /* 162-byte string literal not displayed */
// IngestLanes is the number of keyed-concurrent ingest lanes (ADR-072,
// gh#480). Messages are partitioned by entity ID (same entity → one lane →
// serial in arrival order, preserving the arrival-order merge; different
// entities → parallel), so ingest is no longer bound to a single serial
// Get+CAS round-trip chain. Default 8 (concurrent); `1` selects the prior
// fully-serial behavior.
IngestLanes int `` /* 166-byte string literal not displayed */
}
Config holds configuration for graph-ingest component
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns a valid default configuration
func (*Config) ApplyDefaults ¶
func (c *Config) ApplyDefaults()
ApplyDefaults sets default values for configuration
Source Files