README
¶
Flow diagram persistence
The flowstore package stores author-authored diagrams in the
semstreams_flows NATS KV bucket. A saved flow contains identity, display
metadata, nodes, connections, version, and audit metadata. It does not contain
deployment or runtime lifecycle state.
manager, err := flowstore.NewManager(ctx, natsClient)
flow := &flowstore.Flow{
ID: "example",
Name: "Example",
Nodes: []flowstore.FlowNode{
{
ID: "input-node",
Component: "udp",
Type: types.ComponentTypeInput,
Name: "udp-input",
Config: map[string]any{"port": 5000},
},
},
}
err = manager.Create(ctx, flow)
Create, Get, List, Update, and Delete affect diagram state only.
Updates use optimistic concurrency: the caller supplies the version it read,
and a stale update is rejected. A separate explicit service operation may
validate, compile, and publish component-config candidates for a later boot.
go test ./flowstore
Documentation
¶
Overview ¶
Package flowstore provides flow persistence and management.
Package flowstore persists author-authored flow diagrams in NATS KV.
Flows contain diagram identity, nodes, connections, version, and audit metadata. They deliberately contain no deployment or runtime lifecycle state. Create, Get, List, Update, and Delete operate only on this saved authoring state.
Updates use optimistic concurrency: callers supply the version they read, and Manager rejects a stale version rather than overwriting a concurrent author. A separate explicit service operation may validate and compile a saved diagram into component-configuration candidates. Diagram persistence alone never changes the running component set.
Index ¶
- type Flow
- type FlowConnection
- type FlowNode
- type Manager
- func (s *Manager) Create(ctx context.Context, flow *Flow) error
- func (s *Manager) Delete(ctx context.Context, id string) error
- func (s *Manager) Get(ctx context.Context, id string) (*Flow, error)
- func (s *Manager) List(ctx context.Context) ([]*Flow, error)
- func (s *Manager) Update(ctx context.Context, flow *Flow) error
- func (s *Manager) Watch(ctx context.Context, pattern string) (jetstream.KeyWatcher, error)
- type Position
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Flow ¶
type Flow struct {
// Identity
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description,omitempty"`
// Version for optimistic concurrency control
Version int64 `json:"version"`
// Canvas layout
Nodes []FlowNode `json:"nodes"`
Connections []FlowConnection `json:"connections"`
// Audit
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
CreatedBy string `json:"created_by,omitempty"`
LastModified time.Time `json:"last_modified"`
}
Flow represents a visual flow definition with metadata and canvas layout
func FromComponentConfigs ¶
FromComponentConfigs creates a Flow from component configurations. This bridges static config files to the FlowStore, making headless configs visible in the UI.
The conversion:
- Each ComponentConfig becomes a FlowNode
- Node.ID = config key (e.g., "udp-input")
- Node.Component = cfg.Name (factory name, e.g., "udp")
- Node.Type = cfg.Type (category, e.g., "input", "processor")
- Node.Config = component config as map[string]any
- Positions are auto-calculated using grid layout
Connections are left empty. A caller may use the flow validator's detached port discovery to suggest connections, or users can author them directly.
func FromComponentConfigsWithConnections ¶
func FromComponentConfigsWithConnections( name string, configs map[string]types.ComponentConfig, connections []FlowConnection, ) (*Flow, error)
FromComponentConfigsWithConnections creates a Flow with connection inference. This variant accepts pre-computed connections from FlowGraph analysis. Use this when you have access to instantiated components for port matching.
type FlowConnection ¶
type FlowConnection struct {
ID string `json:"id"`
SourceNodeID string `json:"source_node_id"`
SourcePort string `json:"source_port"`
TargetNodeID string `json:"target_node_id"`
TargetPort string `json:"target_port"`
}
FlowConnection represents a connection between two component ports
type FlowNode ¶
type FlowNode struct {
ID string `json:"id"` // Unique instance ID
Component string `json:"component"` // Component factory name (e.g., "udp", "graph-processor")
Type types.ComponentType `json:"type"` // Component category (input/processor/output/storage/gateway)
Name string `json:"name"` // Instance name
Position Position `json:"position"` // Canvas coordinates
Config map[string]any `json:"config"` // Component configuration
}
FlowNode represents a component instance on the canvas
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager provides persistence for Flow entities using NATS KV. Pattern-B CRUD surface per ADR-029. Named Manager (not Store) so the name matches the other Pattern-B types (rule.ConfigManager, persona.Manager, flowtemplate.Manager). Methods preserve the optimistic-concurrency split (Create + Update + Version) that flow definitions need; the ADR's canonical "Save" collapses into the existing Create/Update pair here.
func NewManager ¶
func NewManager(natsClient *natsclient.Client) (*Manager, error)
NewManager creates a new flow store
func (*Manager) List ¶
List returns the Flows currently saved in the bucket.
It reads CURRENT STATE. Keys are enumerated through KVStore.Keys, so an empty bucket is a successful non-nil empty result, and a key whose read reports typed absence — errors.Is(err, natsclient.ErrKVKeyNotFound), which KVStore.Get returns for a never-created and for a tombstoned key — is omitted rather than failing the list. One client deleting a Flow between another client's enumeration and its read is ordinary churn, not that caller's error.
Every other per-key failure (transport, permission, deadline or cancellation, a stored record that does not decode) aborts the list with a nil result — never a partial list reported as success — and is returned carrying the classification Get assigned it. The wrap is a plain %w for exactly that reason: errs.IsFatal and errs.IsTransient resolve the FIRST classified error in the chain, so an outer errs.Wrap* here would re-stamp a fatal decode failure as transient. No message text is inspected anywhere on this path.
A context that is done when the enumeration returns aborts the list even when the enumeration itself reported nothing, so a cancellation that raced the key watcher can never be reported as an authoritative empty result.
The result is in whatever order the bucket enumerated; List promises none.
func (*Manager) Update ¶
Update updates an existing flow with optimistic concurrency control.
The server owns the audit fields: the persisted record keeps the stored CreatedAt, takes the stored version plus one, and carries one server-observed instant in both UpdatedAt and LastModified, whatever the request supplied. CreatedBy is persisted exactly as the caller sent it. The request's Version is a precondition, never a stored value.
The write is revision-fenced against the revision the stored record was read at, so concurrent Updates through any number of Managers over one bucket commit exactly once. A stale request version and a lost fence are the same typed conflict: a classified invalid error carrying the ADR-060 revision_mismatch code, so callers branch with errors.Is(err, errs.ErrRevisionMismatch) rather than on message text.
flow is left untouched on every failure path and is assigned the committed record only after the fenced write succeeds.
Source Files