Documentation
¶
Overview ¶
Package eventsync provides dependency-based event ordering for real-time sync. It ensures events are published in the correct order for frontend rendering (TanStack DB requires entities to exist before things that reference them).
Index ¶
- Constants
- Variables
- func AddSync[Topic any, Payload any](batch *EventBatch, kind EventKind, subOrder int, ...)
- func AddSyncSimple[Topic any, Payload any](batch *EventBatch, kind EventKind, ...)
- func AddSyncTransform[T any, Topic any, Payload any](batch *EventBatch, kind EventKind, subOrder int, ...)
- func AddSyncTransformSimple[T any, Topic any, Payload any](batch *EventBatch, kind EventKind, ...)
- func ComputeNodeOrder(nodes []mflow.Node, edges []mflow.Edge) []idwrap.IDWrap
- func GetEventPriority(kind EventKind) int
- func GetNodeKindPriority(kind mflow.NodeKind) int
- func SortEventKinds(kinds []EventKind)
- func SortNodesByOrder(nodes []mflow.Node, edges []mflow.Edge)
- func ValidateDependencies() error
- type EventBatch
- func (b *EventBatch) Add(kind EventKind, subOrder int, publish ...PublishFunc)
- func (b *EventBatch) AddSimple(kind EventKind, publish ...PublishFunc)
- func (b *EventBatch) Clear()
- func (b *EventBatch) GetOrderedKinds() []EventKind
- func (b *EventBatch) Len() int
- func (b *EventBatch) Publish(ctx context.Context) error
- type EventEntry
- type EventKind
- type NodeOrderInfo
- type PublishFunc
Constants ¶
const ( PriorityManualStart = 0 // Entry point - always first PriorityFor = 100 // Loop container PriorityForEach = 100 // Loop container PriorityCondition = 100 // Branch container PriorityRequest = 200 // Leaf node PriorityJS = 200 // Leaf node PriorityUnspecified = 999 // Unknown - last )
NodePriority defines the base priority for each node kind. Lower values = higher priority (published first). Container nodes have lower priority so they exist before children.
Variables ¶
var Dependencies = map[EventKind][]EventKind{ KindFlow: {}, KindEnvironment: {}, KindFlowFile: {KindFlow}, KindFolder: {KindFlowFile}, KindNode: {KindFlow}, KindEdge: {KindNode}, KindHTTP: {KindNode}, KindHTTPFile: {KindHTTP}, KindHTTPHeader: {KindHTTP}, KindHTTPParam: {KindHTTP}, KindHTTPBodyForm: {KindHTTP}, KindHTTPBodyURL: {KindHTTP}, KindHTTPBodyRaw: {KindHTTP}, KindHTTPAssert: {KindHTTP}, KindEnvVariable: {KindEnvironment}, }
Dependencies defines what each event kind depends on. This is the single source of truth for event ordering. The topological sort uses this to compute the correct publish order.
Frontend TanStack DB requires: - Flow must exist before FlowFile (FlowFile.contentId references Flow.id) - Flow must exist before Node (Node.flowId references Flow.id) - Node must exist before Edge (Edge.sourceId/targetId reference Node.id) - HTTP must exist before its children (headers, params, etc.)
var NodeKindPriority = map[mflow.NodeKind]int{ mflow.NODE_KIND_MANUAL_START: PriorityManualStart, mflow.NODE_KIND_FOR: PriorityFor, mflow.NODE_KIND_FOR_EACH: PriorityForEach, mflow.NODE_KIND_CONDITION: PriorityCondition, mflow.NODE_KIND_REQUEST: PriorityRequest, mflow.NODE_KIND_JS: PriorityJS, mflow.NODE_KIND_UNSPECIFIED: PriorityUnspecified, }
NodeKindPriority maps node kinds to their base priority.
Functions ¶
func AddSync ¶
func AddSync[Topic any, Payload any]( batch *EventBatch, kind EventKind, subOrder int, streamer eventstream.SyncStreamer[Topic, Payload], topic Topic, payloads ...Payload, )
AddSync adds multiple payloads to the batch for a specific streamer and topic. This handles the closure wrapping and capture logic automatically.
func AddSyncSimple ¶
func AddSyncSimple[Topic any, Payload any]( batch *EventBatch, kind EventKind, streamer eventstream.SyncStreamer[Topic, Payload], topic Topic, payloads ...Payload, )
AddSyncSimple is AddSync with default subOrder (0).
func AddSyncTransform ¶
func AddSyncTransform[T any, Topic any, Payload any]( batch *EventBatch, kind EventKind, subOrder int, streamer eventstream.SyncStreamer[Topic, Payload], topic Topic, items []T, transform func(T) Payload, )
AddSyncTransform adds multiple items to the batch after transforming them into payloads. This is ideal for converting internal models to API events during bulk operations.
func AddSyncTransformSimple ¶
func AddSyncTransformSimple[T any, Topic any, Payload any]( batch *EventBatch, kind EventKind, streamer eventstream.SyncStreamer[Topic, Payload], topic Topic, items []T, transform func(T) Payload, )
AddSyncTransformSimple is AddSyncTransform with default subOrder (0).
func ComputeNodeOrder ¶
ComputeNodeOrder computes the publishing order for nodes based on: 1. Graph topology (BFS level from start node) 2. Node kind priority (containers before leaves)
Returns node IDs in the order they should be published.
func GetEventPriority ¶
GetEventPriority returns the priority (lower = earlier) for an event kind. Returns -1 if the kind is unknown.
func GetNodeKindPriority ¶
GetNodeKindPriority returns the base priority for a node kind.
func SortEventKinds ¶
func SortEventKinds(kinds []EventKind)
SortEventKinds sorts a slice of event kinds by their publish order.
func SortNodesByOrder ¶
SortNodesByOrder sorts nodes in-place according to ComputeNodeOrder.
func ValidateDependencies ¶
func ValidateDependencies() error
ValidateDependencies checks that the Dependencies map has no cycles. This is useful for testing configuration.
Types ¶
type EventBatch ¶
type EventBatch struct {
// contains filtered or unexported fields
}
EventBatch collects events and publishes them in dependency order. Events can be added in any order - the batch sorts them before publishing.
Usage:
batch := eventsync.NewEventBatch()
batch.Add(eventsync.KindFlow, 0, func() { publishFlow(...) })
batch.Add(eventsync.KindNode, nodeLevel, func() { publishNode(...) })
batch.Add(eventsync.KindFlowFile, 0, func() { publishFlowFile(...) })
batch.Publish() // Publishes in correct order: Flow, FlowFile, Node, ...
func NewEventBatch ¶
func NewEventBatch() *EventBatch
NewEventBatch creates a new empty event batch.
func (*EventBatch) Add ¶
func (b *EventBatch) Add(kind EventKind, subOrder int, publish ...PublishFunc)
Add queues one or more events for later publishing. subOrder is used for ordering within the same event kind (e.g., node graph level). Lower subOrder values are published first.
func (*EventBatch) AddSimple ¶
func (b *EventBatch) AddSimple(kind EventKind, publish ...PublishFunc)
AddSimple queues one or more events with default subOrder (0).
func (*EventBatch) Clear ¶
func (b *EventBatch) Clear()
Clear removes all queued events without publishing.
func (*EventBatch) GetOrderedKinds ¶
func (b *EventBatch) GetOrderedKinds() []EventKind
GetOrderedKinds returns the event kinds in the batch in sorted order. Useful for debugging and testing.
func (*EventBatch) Publish ¶
func (b *EventBatch) Publish(ctx context.Context) error
Publish executes all events in dependency-sorted order. After publishing, the batch is cleared.
The order is determined by: 1. EventKind priority (from topological sort of Dependencies) 2. SubOrder (lower values first, for ordering within same kind)
type EventEntry ¶
type EventEntry struct {
Kind EventKind
SubOrder int // For ordering within the same kind (e.g., node graph level)
Publish PublishFunc
}
EventEntry represents a single event to be published.
type EventKind ¶
type EventKind string
EventKind represents a type of sync event. The ordering of events is determined by the Dependencies map.
const ( // Core entities (roots) KindFlow EventKind = "flow" KindEnvironment EventKind = "environment" KindFolder EventKind = "folder" // Flow-related (depend on Flow) KindFlowFile EventKind = "flow_file" KindNode EventKind = "node" // Graph structure (depend on Node) KindEdge EventKind = "edge" // HTTP entities (depend on Node for request nodes) KindHTTP EventKind = "http" // HTTP children (depend on HTTP) KindHTTPFile EventKind = "http_file" KindHTTPHeader EventKind = "http_header" KindHTTPParam EventKind = "http_param" KindHTTPBodyForm EventKind = "http_body_form" KindHTTPBodyURL EventKind = "http_body_url" KindHTTPBodyRaw EventKind = "http_body_raw" KindHTTPAssert EventKind = "http_assert" // Environment children (depend on Environment) KindEnvVariable EventKind = "env_variable" )
func ComputeEventOrder ¶
func ComputeEventOrder() []EventKind
ComputeEventOrder returns the pre-computed event order based on Dependencies. This is cached at package init time for efficiency.
func GetEventOrder ¶
func GetEventOrder() []EventKind
GetEventOrder returns the pre-computed event publishing order.
func TopologicalSort ¶
TopologicalSort computes the order in which event kinds should be published. It uses Kahn's algorithm to compute a topological ordering of the dependency graph. Returns an error if there's a cycle in the dependencies (should never happen with valid config).
Source Files