registry

package
v1.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 3, 2026 License: MIT Imports: 59 Imported by: 0

Documentation

Index

Constants

View Source 
const MaxLookupCacheSize = 10000

maxLookupCacheSize bounds the number of entries kept in the in-memory lookup cache so it cannot grow without limit and leak memory.

Variables

This section is empty.

Functions

This section is empty.

Types

type DatabaseLogger 

type DatabaseLogger struct {
	// contains filtered or unexported fields
}

func NewDatabaseLogger 

func NewDatabaseLogger(parentCtx context.Context, s LogCreator, workflowID string, fallback hermod.Logger) *DatabaseLogger

func (*DatabaseLogger) Close 

func (l *DatabaseLogger) Close()

Close stops the background flusher and drains what is buffered. It is safe to call more than once.

func (*DatabaseLogger) Debug 

func (l *DatabaseLogger) Debug(msg string, keysAndValues ...any)

func (*DatabaseLogger) Error 

func (l *DatabaseLogger) Error(msg string, keysAndValues ...any)

func (*DatabaseLogger) Flush 

func (l *DatabaseLogger) Flush()

func (*DatabaseLogger) Info 

func (l *DatabaseLogger) Info(msg string, keysAndValues ...any)

func (*DatabaseLogger) UpdateStorage 

func (l *DatabaseLogger) UpdateStorage(s LogCreator)

func (*DatabaseLogger) Warn 

func (l *DatabaseLogger) Warn(msg string, keysAndValues ...any)

type DebuggerEvent 

type DebuggerEvent struct {
	WorkflowID string         `json:"workflow_id"`
	NodeID     string         `json:"node_id"`
	MsgID      string         `json:"msg_id"`
	Data       map[string]any `json:"data"`
	State      string         `json:"state"` // "paused", "resumed", "aborted"
}

type LiveMessage 

type LiveMessage struct {
	WorkflowID string         `json:"workflow_id"`
	NodeID     string         `json:"node_id"`
	Timestamp  time.Time      `json:"timestamp"`
	Data       map[string]any `json:"data"`
	IsError    bool           `json:"is_error"`
	Error      string         `json:"error,omitempty"`
}

type LogCreator 

type LogCreator interface {
	CreateLog(ctx context.Context, log storage.Log) error
	CreateLogs(ctx context.Context, logs []storage.Log) error
}

type PIIStats 

type PIIStats struct {
	Discoveries map[string]uint64 `json:"discoveries"`
	LastUpdated time.Time         `json:"last_updated"`
}

type Registry 

type Registry struct {
	// contains filtered or unexported fields
}

func NewRegistry 

func NewRegistry(s storage.Storage, ls ...storage.Storage) *Registry

func (*Registry) ApplyTransformation 

func (r *Registry) ApplyTransformation(ctx context.Context, msg hermod.Message, transType string, config map[string]any) (hermod.Message, error)

func (*Registry) BroadcastLiveMessage 

func (r *Registry) BroadcastLiveMessage(workflowID, nodeID string, msg hermod.Message, isError bool, errMsg string)

func (*Registry) BroadcastLog 

func (r *Registry) BroadcastLog(engineID, level, msg, data string)

func (*Registry) BroadcastStatus 

func (r *Registry) BroadcastStatus(update telemetry.StatusUpdate)

func (*Registry) BrowseSinkTable 

func (r *Registry) BrowseSinkTable(ctx context.Context, cfg factory.SinkConfig, table string, limit int) ([]hermod.Message, error)

func (*Registry) Close 

func (r *Registry) Close()

func (*Registry) ContextWithPipelineSnapshot 

func (r *Registry) ContextWithPipelineSnapshot(ctx context.Context) context.Context

ContextWithPipelineSnapshot returns a new context with a shared snapshot pointer for optimizing transformation pipelines by reducing redundant ToMap() calls.

func (*Registry) CreateLog 

func (r *Registry) CreateLog(ctx context.Context, l storage.Log) error

func (*Registry) CreateLogs 

func (r *Registry) CreateLogs(ctx context.Context, logs []storage.Log) error

func (*Registry) CreateSink 

func (r *Registry) CreateSink(ctx context.Context, cfg factory.SinkConfig) (hermod.Sink, error)

func (*Registry) CreateSource 

func (r *Registry) CreateSource(ctx context.Context, cfg factory.SourceConfig) (hermod.Source, error)

func (*Registry) DebuggerCommand 

func (r *Registry) DebuggerCommand(workflowID, msgID, action string)

func (*Registry) DeleteLogs 

func (r *Registry) DeleteLogs(ctx context.Context, filter storage.LogFilter) error

func (*Registry) DiscoverDatabases 

func (r *Registry) DiscoverDatabases(ctx context.Context, cfg factory.SourceConfig) ([]string, error)

func (*Registry) DiscoverPublications 

func (r *Registry) DiscoverPublications(ctx context.Context, cfg factory.SourceConfig) ([]hermod.PublicationInfo, error)

func (*Registry) DiscoverReplicationSlots 

func (r *Registry) DiscoverReplicationSlots(ctx context.Context, cfg factory.SourceConfig) ([]hermod.ReplicationSlotInfo, error)

func (*Registry) DiscoverSinkColumns 

func (r *Registry) DiscoverSinkColumns(ctx context.Context, cfg factory.SinkConfig, table string) ([]hermod.ColumnInfo, error)

func (*Registry) DiscoverSinkDatabases 

func (r *Registry) DiscoverSinkDatabases(ctx context.Context, cfg factory.SinkConfig) ([]string, error)

func (*Registry) DiscoverSinkTables 

func (r *Registry) DiscoverSinkTables(ctx context.Context, cfg factory.SinkConfig) ([]string, error)

func (*Registry) DiscoverSourceColumns 

func (r *Registry) DiscoverSourceColumns(ctx context.Context, cfg factory.SourceConfig, table string) ([]hermod.ColumnInfo, error)

func (*Registry) DiscoverTables 

func (r *Registry) DiscoverTables(ctx context.Context, cfg factory.SourceConfig) ([]string, error)

func (*Registry) DrainWorkflowDLQ 

func (r *Registry) DrainWorkflowDLQ(ctx context.Context, id string) error

func (*Registry) EvaluateConditions 

func (r *Registry) EvaluateConditions(msg hermod.Message, conditions []map[string]any) bool

func (*Registry) ExecSinkStatement 

func (r *Registry) ExecSinkStatement(ctx context.Context, cfg factory.SinkConfig, stmt string) error

func (*Registry) ExecuteSQL 

func (r *Registry) ExecuteSQL(ctx context.Context, cfg factory.SourceConfig, query string, userSample map[string]any) ([]map[string]any, error)

func (*Registry) ExecuteSinkSQL 

func (r *Registry) ExecuteSinkSQL(ctx context.Context, cfg factory.SinkConfig, query string, userSample map[string]any) ([]map[string]any, error)

func (*Registry) GetAllStatuses 

func (r *Registry) GetAllStatuses() []telemetry.StatusUpdate

func (*Registry) GetDB 

func (r *Registry) GetDB(ctx context.Context, typeName string, config map[string]string) (*sql.DB, error)

func (*Registry) GetDQScorer 

func (r *Registry) GetDQScorer() *governance.Scorer

func (*Registry) GetDashboardStats 

func (r *Registry) GetDashboardStats(ctx context.Context, vhost string) (storage.DashboardStats, error)

func (*Registry) GetEngine 

func (r *Registry) GetEngine(id string) (*pkgengine.Engine, bool)

func (*Registry) GetLogStorage 

func (r *Registry) GetLogStorage() storage.Storage

func (*Registry) GetLogger 

func (r *Registry) GetLogger() hermod.Logger

func (*Registry) GetLookupCache 

func (r *Registry) GetLookupCache(key string) (any, bool)

func (*Registry) GetLookupCacheSize 

func (r *Registry) GetLookupCacheSize() (int, int)

func (*Registry) GetMeshManager 

func (r *Registry) GetMeshManager() *mesh.Manager

func (*Registry) GetNodeState 

func (r *Registry) GetNodeState(key string) (any, bool)

func (*Registry) GetNodeStates 

func (r *Registry) GetNodeStates(ctx context.Context, workflowID string) (map[string]any, error)

func (*Registry) GetOrOpenDB 

func (r *Registry) GetOrOpenDB(src storage.Source) (*sql.DB, error)

func (*Registry) GetOrOpenDBByID 

func (r *Registry) GetOrOpenDBByID(ctx context.Context, id string) (*sql.DB, string, error)

func (*Registry) GetPIIStats 

func (r *Registry) GetPIIStats() map[string]*PIIStats

func (*Registry) GetSink 

func (r *Registry) GetSink(workflowID, nodeID string) (hermod.Sink, bool)

func (*Registry) GetSinkConfig 

func (r *Registry) GetSinkConfig(ctx context.Context, id string) (storage.Sink, error)

func (*Registry) GetSinkConfigs 

func (r *Registry) GetSinkConfigs(id string) ([]factory.SinkConfig, bool)

func (*Registry) GetSource 

func (r *Registry) GetSource(ctx context.Context, id string) (storage.Source, error)

func (*Registry) GetSourceConfig 

func (r *Registry) GetSourceConfig(ctx context.Context, id string) (storage.Source, error)

func (*Registry) GetSourceConfigs 

func (r *Registry) GetSourceConfigs(id string) ([]factory.SourceConfig, bool)

func (*Registry) GetSourceFactoryConfig 

func (r *Registry) GetSourceFactoryConfig(ctx context.Context, id string) (factory.SourceConfig, error)

func (*Registry) GetSourceFormSamples 

func (r *Registry) GetSourceFormSamples(ctx context.Context, path string, limit int) ([]hermod.Message, error)

func (*Registry) GetStorage 

func (r *Registry) GetStorage() storage.Storage

func (*Registry) GetWorkflowConfig 

func (r *Registry) GetWorkflowConfig(id string) (storage.Workflow, bool)

func (*Registry) GetWorkflowHealth 

func (r *Registry) GetWorkflowHealth(id string) (storage.WorkflowHealth, bool)

GetWorkflowHealth returns a real-time health summary for a running workflow.

func (*Registry) GetWorkflowStatus 

func (r *Registry) GetWorkflowStatus(id string) (telemetry.StatusUpdate, bool)

func (*Registry) IsDebuggerAttached 

func (r *Registry) IsDebuggerAttached(workflowID string) bool

func (*Registry) IsEngineRunning 

func (r *Registry) IsEngineRunning(id string) bool

func (*Registry) IsResourceInUse 

func (r *Registry) IsResourceInUse(ctx context.Context, resourceID string, excludeID string, isSource bool) bool

func (*Registry) Logger 

func (r *Registry) Logger() hermod.Logger

func (*Registry) PauseForDebugger 

func (r *Registry) PauseForDebugger(workflowID, nodeID string, msg hermod.Message)

func (*Registry) PurgeLogs 

func (r *Registry) PurgeLogs(ctx context.Context, before time.Time) error

func (*Registry) RebuildWorkflow 

func (r *Registry) RebuildWorkflow(ctx context.Context, workflowID string, fromOffset int64) error

func (*Registry) RecordCircuitBreakerFailure 

func (r *Registry) RecordCircuitBreakerFailure(workflowID, breakerNodeID string)

RecordCircuitBreakerFailure counts a downstream failure against a circuit breaker node.

The breaker reads this count and opens once it passes its threshold. Nothing incremented it before, so a node offered in the editor as "Stop flow on failure threshold" always reported success however broken the downstream was — a control that cannot fire, which is worse than no control because someone believes it is there.

The registry is the natural place for this: it already holds node state and is what the traversal has a handle on.

func (*Registry) RecordStep 

func (r *Registry) RecordStep(ctx context.Context, workflowID, messageID string, step hermod.TraceStep)

func (*Registry) ResumeApproval 

func (r *Registry) ResumeApproval(ctx context.Context, app storage.Approval, branch string) error

ResumeApproval resumes a halted workflow at an approval node with the specified decision branch ("approved" or "rejected").

func (*Registry) RunWorkflowNode 

func (r *Registry) RunWorkflowNode(workflowID string, node *storage.WorkflowNode, msg hermod.Message) ([]hermod.Message, string, error)

func (*Registry) SampleSinkTable 

func (r *Registry) SampleSinkTable(ctx context.Context, cfg factory.SinkConfig, table string) (hermod.Message, error)

func (*Registry) SampleTable 

func (r *Registry) SampleTable(ctx context.Context, cfg factory.SourceConfig, table string) (hermod.Message, error)

func (*Registry) SetConfig 

func (r *Registry) SetConfig(cfg config.Config)

func (*Registry) SetFactories 

func (r *Registry) SetFactories(sourceFactory SourceFactory, sinkFactory SinkFactory)

func (*Registry) SetLogStorage 

func (r *Registry) SetLogStorage(s storage.Storage)

func (*Registry) SetLogger 

func (r *Registry) SetLogger(logger hermod.Logger)

func (*Registry) SetLookupCache 

func (r *Registry) SetLookupCache(key string, value any, ttl time.Duration)

SetLookupCache stores a value with an optional TTL. Expired entries are reclaimed lazily on read and during a bounded sweep on write, so no per-key goroutine is spawned (which previously leaked goroutines and memory under high lookup throughput).

func (*Registry) SetNodeState 

func (r *Registry) SetNodeState(key string, val any)

func (*Registry) SetSecretManager 

func (r *Registry) SetSecretManager(mgr secrets.Manager)

func (*Registry) SetSinkFactory 

func (r *Registry) SetSinkFactory(f SinkFactory)

func (*Registry) SetSourceFactory 

func (r *Registry) SetSourceFactory(f SourceFactory)

func (*Registry) SetStateStore 

func (r *Registry) SetStateStore(ss hermod.StateStore)

func (*Registry) SetStorage 

func (r *Registry) SetStorage(s storage.Storage)

func (*Registry) StartWorkflow 

func (r *Registry) StartWorkflow(id string, wf storage.Workflow) error

StartWorkflow creates and starts a workflow engine for the given workflow configuration.

func (*Registry) StateStore 

func (r *Registry) StateStore() hermod.StateStore

func (*Registry) StatusSubscriberCount 

func (r *Registry) StatusSubscriberCount() int

StatusSubscriberCount returns the number of active status subscribers. It is primarily useful for observability and tests that assert subscribers are released when a client disconnects.

func (*Registry) StopAll 

func (r *Registry) StopAll()

func (*Registry) StopEngine 

func (r *Registry) StopEngine(ctx context.Context, id string) error

StopEngine stops a workflow on an operator's instruction. Unlike StopEngineWithoutUpdate — which the supervisor and the worker's own reconciliation use — this marks the end of a stall episode, so the workflow's automatic-restart budget starts fresh the next time it runs.

func (*Registry) StopEngineWithoutUpdate 

func (r *Registry) StopEngineWithoutUpdate(ctx context.Context, id string) error

func (*Registry) Storage 

func (r *Registry) Storage() interfaces.RegistryStorage

func (*Registry) SubscribeDashboardStats 

func (r *Registry) SubscribeDashboardStats(vhost string) chan storage.DashboardStats

func (*Registry) SubscribeDebugger 

func (r *Registry) SubscribeDebugger(workflowID string) chan DebuggerEvent

func (*Registry) SubscribeLiveMessages 

func (r *Registry) SubscribeLiveMessages() chan LiveMessage

func (*Registry) SubscribeLogs 

func (r *Registry) SubscribeLogs() chan storage.Log

func (*Registry) SubscribeStatus 

func (r *Registry) SubscribeStatus() chan telemetry.StatusUpdate

func (*Registry) SubscribeWorkflowLiveMessages 

func (r *Registry) SubscribeWorkflowLiveMessages(workflowID string) chan LiveMessage

func (*Registry) SubscribeWorkflowLogs 

func (r *Registry) SubscribeWorkflowLogs(workflowID string) chan storage.Log

func (*Registry) SubscribeWorkflowStatus 

func (r *Registry) SubscribeWorkflowStatus(workflowID string) chan telemetry.StatusUpdate

func (*Registry) TestSink 

func (r *Registry) TestSink(ctx context.Context, cfg factory.SinkConfig) error

func (*Registry) TestSource 

func (r *Registry) TestSource(ctx context.Context, cfg factory.SourceConfig) error

func (*Registry) TestTransformationPipeline 

func (r *Registry) TestTransformationPipeline(ctx context.Context, transformations []storage.Transformation, msg hermod.Message) ([]hermod.Message, error)

func (*Registry) TestWorkflow 

func (r *Registry) TestWorkflow(ctx context.Context, wf storage.Workflow, msg hermod.Message) ([]WorkflowStepResult, error)

func (*Registry) TriggerSnapshot 

func (r *Registry) TriggerSnapshot(ctx context.Context, sourceID string, tables ...string) error

func (*Registry) UnsubscribeDashboardStats 

func (r *Registry) UnsubscribeDashboardStats(ch chan storage.DashboardStats)

func (*Registry) UnsubscribeDebugger 

func (r *Registry) UnsubscribeDebugger(workflowID string, ch chan DebuggerEvent)

func (*Registry) UnsubscribeLiveMessages 

func (r *Registry) UnsubscribeLiveMessages(ch chan LiveMessage)

func (*Registry) UnsubscribeLogs 

func (r *Registry) UnsubscribeLogs(ch chan storage.Log)

func (*Registry) UnsubscribeStatus 

func (r *Registry) UnsubscribeStatus(ch chan telemetry.StatusUpdate)

func (*Registry) UpdateNodeState 

func (r *Registry) UpdateNodeState(ctx context.Context, workflowID, nodeID string, state any) error

func (*Registry) UpdateSink 

func (r *Registry) UpdateSink(ctx context.Context, snk storage.Sink) error

func (*Registry) UpdateSinkStatus 

func (r *Registry) UpdateSinkStatus(ctx context.Context, id string, status string) error

func (*Registry) UpdateSource 

func (r *Registry) UpdateSource(ctx context.Context, src storage.Source) error

func (*Registry) UpdateSourceStatus 

func (r *Registry) UpdateSourceStatus(ctx context.Context, id string, status string) error

func (*Registry) ValidateWorkflow 

func (r *Registry) ValidateWorkflow(ctx context.Context, wf storage.Workflow) error

type SinkFactory 

type SinkFactory func(factory.SinkConfig) (hermod.Sink, error)

type SourceFactory 

type SourceFactory func(factory.SourceConfig) (hermod.Source, error)

type WorkflowStepResult 

type WorkflowStepResult struct {
	NodeID   string            `json:"node_id"`
	NodeType string            `json:"node_type"`
	Payload  map[string]any    `json:"payload,omitempty"`
	Metadata map[string]string `json:"metadata,omitempty"`
	Error    string            `json:"error,omitempty"`
	Filtered bool              `json:"filtered,omitempty"`
	Branch   string            `json:"branch,omitempty"`
}

Source Files

View all Source files

Directories

Path Synopsis
Package nodes is the one import that makes every workflow node type available.
 Package nodes is the one import that makes every workflow node type available.
control
core
flow
reliability
util

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.