storage

package
v1.16.2 Latest Latest
Warning

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

 Go to latest Published: Sep 29, 2026 License: MIT Imports: 11 Imported by: 0

Documentation

Index

Constants

View Source 
const (

	// TraceDedupCacheSize is how many (workflow, message, payload) facts are
	// remembered at once. A message's steps are recorded within milliseconds of
	// each other, so the useful lifetime of an entry is short and a few
	// thousand slots cover far more concurrency than the trace recorder's own
	// slot limit allows. The bound is not optional: the key contains a message
	// id, which is whatever the source supplied.
	TraceDedupCacheSize = 8192
)

Variables

View Source 
var (
	WorkflowSearchFields  = []string{"id", "name", "vhost"}
	ConnectorSearchFields = []string{"id", "name", "type", "vhost"}
	UserSearchFields      = []string{"id", "username", "full_name", "email", "role"}
	VHostSearchFields     = []string{"id", "name", "description"}
	WorkerSearchFields    = []string{"id", "name", "host", "description"}

	// LogSearchFields deliberately omits data. It is the largest column on the
	// largest table Hermod owns, a LIKE over it cannot use an index, and a log
	// search that scans every payload is a log search nobody waits for. The
	// mongo backend used to include it; this is the one place the two were
	// reconciled by taking the narrower list rather than the union.
	LogSearchFields = []string{"message", "action", "source_id", "sink_id", "workflow_id", "user_id", "username"}

	AuditLogSearchFields = []string{"id", "username", "action", "entity_id", "payload"}
)

What each list search matches.

These live here rather than in the backends because they were hand-mirrored in two packages and drifted: workflows matched name alone in SQL and _id, name and vhost in mongo; users matched role in SQL and not in mongo; workers matched description in SQL and not in mongo; logs matched action in SQL and data, user_id and username in mongo. Which backend was deployed therefore changed what the search box found, and nothing failed when the two lists disagreed — the only way to notice was to run both and compare.

One list per entity, read by both backends, is what makes that impossible rather than merely fixed. Fields are named as the SQL column; the mongo backend renames id to _id on the way past, in searchAcross, so no caller has to remember to.

View Source 
var ErrNotFound = errors.New("not found")
View Source 
var NodeConfigSourceKeys = []string{"sourceId", "sourceID"}

NodeConfigSourceKeys are the config keys through which a node names a source without being a source node. db_lookup writes "sourceId"; the enrichment SQL node (SQLConfig.tsx) reads and writes both spellings.

Functions

func DecodeTracePayload added in v1.10.0 

func DecodeTracePayload(blob []byte) ([]byte, error)

decodeTracePayload recovers the JSON behind a stored after_blob value.

func EncodeTracePayload added in v1.10.0 

func EncodeTracePayload(raw []byte) []byte

encodeTracePayload turns marshalled JSON into the bytes stored in after_blob.

func NormalizeVHost added in v1.8.1 

func NormalizeVHost(vhost string) string

NormalizeVHost maps the two spellings of "no vhost filter" onto one.

The UI sends vhost=all for the unfiltered view while GetDashboardStats treats an empty string the same way. Storing history under both spellings would split one series into two, so every read and write of a sample goes through here.

func TracePayloadHash added in v1.10.0 

func TracePayloadHash(raw []byte) []byte

tracePayloadHash identifies a payload by its content, within one message.

128 bits of SHA-256, not 64. The bytes hashed are the message's own data, which in a CDC pipeline is whatever the upstream database contained and in a webhook pipeline is whatever the caller sent. A 64-bit content address over input someone else chooses is a 2^32 search away from showing one node's payload under another node's name — and a trace viewer that can be made to lie about which node produced what is worse than a slightly larger table.

func TracePayloadKey added in v1.10.0 

func TracePayloadKey(raw []byte) string

TracePayloadKey is TracePayloadHash rendered as hex.

The document stores keep their payloads in a map, and a map key has to be a string there: a BSON field name, or a JSON object key. Keying that map by content is what makes dedup inherent for those backends rather than something a cache has to get right — writing the same payload twice writes the same key twice, which is a no-op.

func WorkflowQueriesSource added in v1.8.1 

func WorkflowQueriesSource(wf Workflow, sourceID string, delegateOf func(id string) string) bool

WorkflowQueriesSource reports whether a workflow aims SQL at this source without holding it in a source node's RefID: a db_lookup's sourceId, or the delegate of a batch_sql source it does hold.

Walking nodes for `type == "source"` alone misses both, which is how a source that a running workflow depends on could be edited or deleted out from under it. delegateOf resolves a source id to its batch_sql delegate, or "" when it has none; callers that cannot resolve sources may pass nil.

Types

type Approval 

type Approval struct {
	ID             string            `json:"id"`
	WorkflowID     string            `json:"workflow_id"`
	NodeID         string            `json:"node_id"`
	MessageID      string            `json:"message_id"`
	Payload        []byte            `json:"payload"`
	Metadata       map[string]string `json:"metadata"`
	Data           map[string]any    `json:"data"`
	FormDefinition map[string]any    `json:"form_definition,omitempty"`
	FormData       map[string]any    `json:"form_data,omitempty"`
	Status         string            `json:"status"` // pending, approved, rejected
	CreatedAt      time.Time         `json:"created_at"`
	ProcessedAt    *time.Time        `json:"processed_at,omitempty"`
	ProcessedBy    string            `json:"processed_by,omitempty"`
	Notes          string            `json:"notes,omitempty"`
}

type ApprovalFilter 

type ApprovalFilter struct {
	CommonFilter
	WorkflowID string
	Status     string
}

type AuditFilter 

type AuditFilter struct {
	CommonFilter
	UserID     string     `json:"user_id"`
	EntityType string     `json:"entity_type"`
	EntityID   string     `json:"entity_id"`
	Action     string     `json:"action"`
	From       *time.Time `json:"from" omitzero:"true"`
	To         *time.Time `json:"to" omitzero:"true"`
}

type AuditLog 

type AuditLog struct {
	ID         string    `json:"id"`
	Timestamp  time.Time `json:"timestamp"`
	UserID     string    `json:"user_id"`
	Username   string    `json:"username"`
	Action     string    `json:"action"`      // e.g., "CREATE", "UPDATE", "DELETE", "START", "STOP"
	EntityType string    `json:"entity_type"` // e.g., "workflow", "source", "sink", "user"
	EntityID   string    `json:"entity_id"`
	Payload    string    `json:"payload,omitempty"` // Details about the change (JSON)
	IP         string    `json:"ip,omitempty"`
}

type CommonFilter 

type CommonFilter struct {
	Page        int
	Limit       int
	Search      string
	VHost       string
	WorkspaceID string
	// WithoutWorkspace limits matches to rows with no workspace — the view you
	// need to find what is left to organise. Mirrors LogFilter.WithoutWorkflow.
	// It takes precedence over WorkspaceID, which is never set at the same time.
	WithoutWorkspace bool
	// Since and Until bound time-based queries (e.g., logs). Zero value means not set.
	Since time.Time `json:"since" omitzero:"true"`
	Until time.Time `json:"until" omitzero:"true"`

	// Workflow filters
	WorkerID string `json:"worker_id,omitempty"`
	OwnerID  string `json:"owner_id,omitempty"`
	Active   *bool  `json:"active,omitempty"`
}

type DashboardSample added in v1.8.1 

type DashboardSample struct {
	Timestamp       time.Time `json:"timestamp"`
	VHost           string    `json:"vhost"`
	Throughput      float64   `json:"throughput"`
	TotalProcessed  uint64    `json:"total_processed"`
	TotalErrors     uint64    `json:"total_errors"`
	TotalLag        uint64    `json:"total_lag"`
	ErrorRate       float64   `json:"error_rate"`
	AvgLatencyMs    float64   `json:"avg_latency_ms"`
	ActiveWorkflows int       `json:"active_workflows"`
	ActiveWorkers   int       `json:"active_workers"`
}

DashboardSample is one point of dashboard history.

The throughput chart used to be built entirely in the browser: the page held the last thirty WebSocket readings in React state and threw them away on reload. That makes a reload indistinguishable from an outage — the chart restarts at zero either way — and means nobody can answer "was it like this an hour ago?". Persisting a sample per interval is what turns that sliver into a series the chart can be rebuilt from.

It is deliberately a narrow projection of DashboardStats rather than the whole struct: configuration counts (TotalSources, TotalWorkflows) describe what exists rather than what happened, and storing them once per interval would be a lot of rows to say nothing. Only the fields that move over time are kept, and none of them are message payloads, so no customer data enters this table.

type DashboardStats 

type DashboardStats struct {
	ActiveSources   int     `json:"active_sources"`
	ActiveSinks     int     `json:"active_sinks"`
	ActiveWorkflows int     `json:"active_workflows"`
	TotalProcessed  uint64  `json:"total_processed"`
	TotalErrors     uint64  `json:"total_errors"`
	TotalLag        uint64  `json:"total_lag"`
	FailedWorkflows int     `json:"failed_workflows"`
	Uptime          int64   `json:"uptime"`
	ActiveWorkers   int     `json:"active_workers"`
	TotalWorkflows  int     `json:"total_workflows"`
	TotalSources    int     `json:"total_sources"`
	TotalSinks      int     `json:"total_sinks"`
	Throughput      float64 `json:"throughput"` // Messages per second

	// Health signals. These answer "is the data moving well?", which the
	// counters above cannot: a pipeline with a rising TotalProcessed and a
	// wedged sink looks identical to a healthy one until someone reads the
	// logs.
	//
	// There is deliberately no DeadLetterCount here. TotalErrors is already
	// exactly that — flushStatsToStorage persists baseErrors +
	// StatusUpdate.DeadLetterCount into total_errors — and showing the same
	// number twice under two names is how a dashboard loses the reader's
	// trust.
	//
	// ErrorRate is derived from the persisted counters and so is cumulative
	// over the workflow's life. The other three are point-in-time readings of
	// engines running on this node, so they reset when an engine restarts and
	// read zero when nothing is running.
	ErrorRate           float64 `json:"error_rate"`     // Dead-lettered share of attempted messages, 0..1
	AvgLatencyMs        float64 `json:"avg_latency_ms"` // Mean end-to-end latency across running engines
	Backpressure        float64 `json:"backpressure"`   // Worst sink buffer fill, 0..1
	CircuitBreakersOpen int     `json:"circuit_breakers_open"`

	// Cluster capacity and utilisation, over exactly the workers ActiveWorkers
	// counts — those that reported a heartbeat inside the last two minutes. A
	// worker that stopped reporting has already been excluded from the count,
	// and counting its cores and disk would describe machines that are no
	// longer there.
	//
	// Totalled rather than averaged, because the question is "how much machine
	// is behind this platform, and how much of it is gone". An average hides a
	// node that is full next to one that is idle, and the full one is the
	// reason anybody opened the page.
	//
	// Workers that report no capacity contribute nothing rather than dragging a
	// total down — see WorkerResources. And one worker per host is assumed:
	// two workers on one machine would count that machine's memory and disk
	// twice. Hermod's topology is a worker per node, and the alternative,
	// deduplicating on a self-reported hostname, is wrong in exactly the
	// container case it would exist to handle.
	CPUCores          int     `json:"cpu_cores"`
	CPUUsage          float64 `json:"cpu_usage"` // 0..1, weighted by each worker's core count
	MemoryTotalBytes  int64   `json:"memory_total_bytes"`
	MemoryUsedBytes   int64   `json:"memory_used_bytes"`
	StorageTotalBytes int64   `json:"storage_total_bytes"`
	StorageUsedBytes  int64   `json:"storage_used_bytes"`
}

type FormSubmission 

type FormSubmission struct {
	ID        string    `json:"id"`
	Timestamp time.Time `json:"timestamp" omitzero:"true"`
	Path      string    `json:"path"`
	Data      []byte    `json:"data"`
	Status    string    `json:"status"` // pending, processing, completed, failed
}

type FormSubmissionFilter 

type FormSubmissionFilter struct {
	CommonFilter
	Path   string `json:"path"`
	Status string `json:"status"`
}

type LineageEdge 

type LineageEdge struct {
	SourceID     string `json:"source_id"`
	SourceName   string `json:"source_name"`
	SourceType   string `json:"source_type"`
	SinkID       string `json:"sink_id"`
	SinkName     string `json:"sink_name"`
	SinkType     string `json:"sink_type"`
	WorkflowID   string `json:"workflow_id"`
	WorkflowName string `json:"workflow_name"`
}

type Log 

type Log struct {
	ID         string    `json:"id"`
	Timestamp  time.Time `json:"timestamp"`
	Level      string    `json:"level"`
	Message    string    `json:"message"`
	Action     string    `json:"action,omitempty"`
	SourceID   string    `json:"source_id,omitempty"`
	SinkID     string    `json:"sink_id,omitempty"`
	WorkflowID string    `json:"workflow_id,omitempty"`
	UserID     string    `json:"user_id,omitempty"`
	Username   string    `json:"username,omitempty"`
	Data       string    `json:"data,omitempty"`
}

type LogFilter 

type LogFilter struct {
	CommonFilter
	SourceID   string
	SinkID     string
	WorkflowID string
	Level      string
	Action     string
	// WithoutWorkflow limits matches to logs without a workflow_id (NULL).
	WithoutWorkflow bool
}

type MessageTrace 

type MessageTrace struct {
	ID         string             `json:"id"`
	MessageID  string             `json:"message_id"`
	WorkflowID string             `json:"workflow_id"`
	Steps      []hermod.TraceStep `json:"steps"`
	CreatedAt  time.Time          `json:"created_at"`

	// Summary fields, carried on the parent row so the trace list never has to
	// touch the step table. Listing used to aggregate every step a workflow had
	// ever produced — measured on PostgreSQL 17, a sequential scan of 250k rows
	// and ~390 MB of I/O to return 25 of them — because "when did this message
	// start" was only derivable as MIN(timestamp) over its steps. It is now
	// written once, when the steps are.
	StepCount  int   `json:"step_count"`
	ErrorCount int   `json:"error_count"`
	DurationMs int64 `json:"duration_ms"`
}

type MissingRef added in v1.8.1 

type MissingRef struct {
	Kind   string `json:"kind"` // "source" or "sink"
	ID     string `json:"id"`
	NodeID string `json:"node_id,omitempty"`
}

MissingRef is one dependency an export could not resolve.

type OutboxItem 

type OutboxItem = hermod.OutboxItem

type Plugin 

type Plugin struct {
	ID          string     `json:"id"`
	Name        string     `json:"name"`
	Description string     `json:"description"`
	Author      string     `json:"author"`
	Stars       int        `json:"stars"`
	Category    string     `json:"category"`
	Certified   bool       `json:"certified"`
	Type        string     `json:"type"` // WASM, Connector, Transformer
	WasmURL     string     `json:"wasm_url,omitempty"`
	Installed   bool       `json:"installed"`
	InstalledAt *time.Time `json:"installed_at" omitzero:"true"`
	// CreatedAt is the sort key the plugin list pages on. Storage stamps it when a
	// caller leaves it zero and never rewrites it on update. See Workflow.CreatedAt.
	CreatedAt time.Time `json:"created_at" bson:"created_at"`
}

type Role 

type Role string
const (
	RoleAdministrator Role = "Administrator"
	RoleEditor        Role = "Editor"
	RoleViewer        Role = "Viewer"
)

type Schema 

type Schema struct {
	ID        string    `json:"id"`
	Name      string    `json:"name"`
	Version   int       `json:"version"`
	Type      string    `json:"type"`
	Content   string    `json:"content"`
	CreatedAt time.Time `json:"created_at" omitzero:"true"`
}

type Sink 

type Sink struct {
	ID          string           `json:"id"`
	Name        string           `json:"name"`
	Type        string           `json:"type"`
	VHost       string           `json:"vhost"`
	Active      bool             `json:"active"`
	Status      string           `json:"status,omitempty"`
	WorkerID    string           `json:"worker_id"`
	WorkspaceID string           `json:"workspace_id,omitempty"`
	Config      hermod.StringMap `json:"config"`
	// CreatedAt is the sort key the sink list pages on. Storage stamps it
	// when a caller leaves it zero and never rewrites it on update, so it stays
	// the creation time rather than the last-touched time. See Workflow.CreatedAt.
	CreatedAt time.Time `json:"created_at" bson:"created_at"`
}

type Source 

type Source struct {
	ID          string            `json:"id"`
	Name        string            `json:"name"`
	Type        string            `json:"type"`
	VHost       string            `json:"vhost"`
	Active      bool              `json:"active"`
	Status      string            `json:"status,omitempty"`
	WorkerID    string            `json:"worker_id"`
	WorkspaceID string            `json:"workspace_id,omitempty"`
	Config      hermod.StringMap  `json:"config"`
	Sample      string            `json:"sample,omitempty"`
	State       map[string]string `json:"state" omitzero:"true"`
	// CreatedAt is the sort key the source list pages on. Storage stamps it
	// when a caller leaves it zero and never rewrites it on update, so it stays
	// the creation time rather than the last-touched time. See Workflow.CreatedAt.
	CreatedAt time.Time `json:"created_at" bson:"created_at"`
}

type Storage 

type Storage interface {
	// Init performs storage initialization/migrations and is safe to call multiple times.
	Init(ctx context.Context) error

	// Ping checks the connection to the storage.
	Ping(ctx context.Context) error

	ListSources(ctx context.Context, filter CommonFilter) ([]Source, int, error)
	CreateSource(ctx context.Context, src Source) error
	UpdateSource(ctx context.Context, src Source) error
	UpdateSourceStatus(ctx context.Context, id string, status string) error
	UpdateSourceState(ctx context.Context, id string, state map[string]string) error
	// UpdateSourceSample writes only the stored preview payload. Storing a
	// sample through UpdateSource carried every other column along with it,
	// from whatever copy of the source the caller happened to hold.
	UpdateSourceSample(ctx context.Context, id string, sample string) error
	DeleteSource(ctx context.Context, id string) error
	GetSource(ctx context.Context, id string) (Source, error)

	ListSinks(ctx context.Context, filter CommonFilter) ([]Sink, int, error)
	CreateSink(ctx context.Context, snk Sink) error
	UpdateSink(ctx context.Context, snk Sink) error

	// ReEncryptSecrets rewrites every stored credential under newKey. Key
	// rotation must call this *before* installing the new key: rotating
	// without it leaves every password and API key encrypted under a key the
	// process no longer has.
	ReEncryptSecrets(ctx context.Context, newKey string) error
	UpdateSinkStatus(ctx context.Context, id string, status string) error
	DeleteSink(ctx context.Context, id string) error
	GetSink(ctx context.Context, id string) (Sink, error)

	ListUsers(ctx context.Context, filter CommonFilter) ([]User, int, error)
	CreateUser(ctx context.Context, user User) error
	UpdateUser(ctx context.Context, user User) error
	DeleteUser(ctx context.Context, id string) error
	GetUser(ctx context.Context, id string) (User, error)
	GetUserByUsername(ctx context.Context, username string) (User, error)
	GetUserByEmail(ctx context.Context, email string) (User, error)

	ListVHosts(ctx context.Context, filter CommonFilter) ([]VHost, int, error)
	CreateVHost(ctx context.Context, vhost VHost) error
	UpdateVHost(ctx context.Context, vhost VHost) error
	DeleteVHost(ctx context.Context, id string) error
	GetVHost(ctx context.Context, id string) (VHost, error)

	ListWorkflows(ctx context.Context, filter CommonFilter) ([]Workflow, int, error)
	ListWorkspaces(ctx context.Context) ([]Workspace, error)
	CreateWorkspace(ctx context.Context, ws Workspace) error
	UpdateWorkspace(ctx context.Context, ws Workspace) error
	GetWorkspace(ctx context.Context, id string) (Workspace, error)
	DeleteWorkspace(ctx context.Context, id string) error
	// ClearWorkspaceAssignments removes every workflow, source and sink from
	// the given workspace and reports how many rows it touched. Deleting a
	// workspace without this leaves members pointing at an id that no longer
	// resolves: the list renders a raw UUID, no filter matches them, and the
	// quota checks stop applying because GetWorkspace errors and they bail.
	ClearWorkspaceAssignments(ctx context.Context, workspaceID string) (int, error)
	CreateWorkflow(ctx context.Context, wf Workflow) error
	UpdateWorkflow(ctx context.Context, wf Workflow) error
	UpdateWorkflowStatus(ctx context.Context, id string, status string) error
	UpdateWorkflowStats(ctx context.Context, id string, processed, errors, lag uint64) error
	DeleteWorkflow(ctx context.Context, id string) error
	GetWorkflow(ctx context.Context, id string) (Workflow, error)

	// Lease-based ownership for workflows
	// AcquireWorkflowLease attempts to set owner_id and extend lease_until atomically with TTL seconds.
	// Returns true if the lease was acquired.
	AcquireWorkflowLease(ctx context.Context, workflowID, ownerID string, ttlSeconds int) (bool, error)
	// RenewWorkflowLease extends lease_until for an existing owner if not expired yet. Returns true if renewed.
	RenewWorkflowLease(ctx context.Context, workflowID, ownerID string, ttlSeconds int) (bool, error)
	// ReleaseWorkflowLease clears ownership if owned by ownerID.
	ReleaseWorkflowLease(ctx context.Context, workflowID, ownerID string) error

	ListWorkers(ctx context.Context, filter CommonFilter) ([]Worker, int, error)
	CreateWorker(ctx context.Context, worker Worker) error
	UpdateWorker(ctx context.Context, worker Worker) error
	UpdateWorkerHeartbeat(ctx context.Context, id string, res WorkerResources) error
	DeleteWorker(ctx context.Context, id string) error
	GetWorker(ctx context.Context, id string) (Worker, error)

	ListLogs(ctx context.Context, filter LogFilter) ([]Log, int, error)
	CreateLog(ctx context.Context, log Log) error
	CreateLogs(ctx context.Context, logs []Log) error
	DeleteLogs(ctx context.Context, filter LogFilter) error
	PurgeLogs(ctx context.Context, before time.Time) error

	ListAuditLogs(ctx context.Context, filter AuditFilter) ([]AuditLog, int, error)
	CreateAuditLog(ctx context.Context, log AuditLog) error
	PurgeAuditLogs(ctx context.Context, before time.Time) error
	PurgeMessageTraces(ctx context.Context, retention TraceRetention) error
	DeleteWorkflowMessageTraces(ctx context.Context, workflowID string) error

	ListWebhookRequests(ctx context.Context, filter WebhookRequestFilter) ([]WebhookRequest, int, error)
	CreateWebhookRequest(ctx context.Context, req WebhookRequest) error
	GetWebhookRequest(ctx context.Context, id string) (WebhookRequest, error)
	DeleteWebhookRequests(ctx context.Context, filter WebhookRequestFilter) error

	CreateFormSubmission(ctx context.Context, sub FormSubmission) error
	ListFormSubmissions(ctx context.Context, filter FormSubmissionFilter) ([]FormSubmission, int, error)
	GetFormSubmission(ctx context.Context, id string) (FormSubmission, error)
	UpdateFormSubmissionStatus(ctx context.Context, id string, status string) error
	DeleteFormSubmissions(ctx context.Context, filter FormSubmissionFilter) error

	GetSetting(ctx context.Context, key string) (string, error)
	SaveSetting(ctx context.Context, key string, value string) error

	// Node State Management
	UpdateNodeState(ctx context.Context, workflowID, nodeID string, state any) error
	GetNodeStates(ctx context.Context, workflowID string) (map[string]any, error)

	// Schema Registry
	ListSchemas(ctx context.Context, name string) ([]Schema, error)
	ListAllSchemas(ctx context.Context) ([]Schema, error)
	GetSchema(ctx context.Context, name string, version int) (Schema, error)
	GetLatestSchema(ctx context.Context, name string) (Schema, error)
	CreateSchema(ctx context.Context, schema Schema) error

	// Message Tracing
	RecordTraceStep(ctx context.Context, workflowID, messageID string, step hermod.TraceStep) error
	GetMessageTrace(ctx context.Context, workflowID, messageID string) (MessageTrace, error)
	ListMessageTraces(ctx context.Context, workflowID string, filter TraceFilter) ([]MessageTrace, error)

	// Workflow Versioning
	CreateWorkflowVersion(ctx context.Context, version WorkflowVersion) error
	ListWorkflowVersions(ctx context.Context, workflowID string) ([]WorkflowVersion, error)
	GetWorkflowVersion(ctx context.Context, workflowID string, version int) (WorkflowVersion, error)

	// Transactional Outbox
	CreateOutboxItem(ctx context.Context, item OutboxItem) error
	ListOutboxItems(ctx context.Context, status string, limit int) ([]OutboxItem, error)
	DeleteOutboxItem(ctx context.Context, id string) error
	UpdateOutboxItem(ctx context.Context, item OutboxItem) error

	// Lineage
	GetLineage(ctx context.Context) ([]LineageEdge, error)

	// Marketplace
	ListPlugins(ctx context.Context) ([]Plugin, error)
	GetPlugin(ctx context.Context, id string) (Plugin, error)
	InstallPlugin(ctx context.Context, id string) error
	UninstallPlugin(ctx context.Context, id string) error

	// Approvals
	ListApprovals(ctx context.Context, filter ApprovalFilter) ([]Approval, int, error)
	CreateApproval(ctx context.Context, app Approval) error
	GetApproval(ctx context.Context, id string) (Approval, error)
	UpdateApprovalStatus(ctx context.Context, id string, status string, processedBy string, notes string, formData map[string]any) error
	DeleteApproval(ctx context.Context, id string) error

	// Suspended Messages
	CreateSuspendedMessage(ctx context.Context, m SuspendedMessage) error
	ListSuspendedMessages(ctx context.Context, workflowID string, before time.Time) ([]SuspendedMessage, error)
	DeleteSuspendedMessage(ctx context.Context, id string) error

	// Aggregated Dashboard Stats
	GetDashboardStats(ctx context.Context, vhost string) (DashboardStats, error)

	// Dashboard history. Samples are appended on an interval so the dashboard
	// chart survives a page reload; PurgeDashboardHistory bounds the table.
	RecordDashboardSample(ctx context.Context, sample DashboardSample) error
	GetDashboardHistory(ctx context.Context, vhost string, since time.Time, limit int) ([]DashboardSample, error)
	PurgeDashboardHistory(ctx context.Context, before time.Time) error
}

type StoredStep added in v1.10.0 

type StoredStep struct {
	NodeID    string        `bson:"node_id" json:"node_id"`
	Timestamp time.Time     `bson:"timestamp" json:"timestamp"`
	Duration  time.Duration `bson:"duration,omitempty" json:"duration,omitempty"`
	AfterKey  string        `bson:"after_key,omitempty" json:"after_key,omitempty"`
	Error     string        `bson:"error,omitempty" json:"error,omitempty"`
	Lineage   string        `bson:"lineage,omitempty" json:"lineage,omitempty"`

	// Documents written before payload dedup carry the maps inline, under the
	// field names the driver derived from hermod.TraceStep's own fields. They
	// are read and never written, so an upgrade needs no backfill.
	LegacyBefore map[string]any `bson:"before,omitempty" json:"before,omitempty"`
	LegacyAfter  map[string]any `bson:"after,omitempty" json:"after,omitempty"`
}

StoredStep is one node's step. It names its payload instead of carrying it.

func PackTraceStep added in v1.10.0 

func PackTraceStep(step hermod.TraceStep) (StoredStep, string, []byte)

PackTraceStep turns a step into its stored form plus the payload to file under the returned key. An empty key means the step carried no payload.

type StoredTrace added in v1.10.0 

type StoredTrace struct {
	ID         string    `bson:"id" json:"id"`
	WorkflowID string    `bson:"workflow_id" json:"workflow_id"`
	MessageID  string    `bson:"message_id" json:"message_id"`
	CreatedAt  time.Time `bson:"created_at" json:"created_at"`

	Steps []StoredStep `bson:"steps" json:"steps"`

	// Payloads maps a content key to the encoded payload that produced it.
	// Compressed, and self-describing about how — see EncodeTracePayload.
	Payloads map[string][]byte `bson:"payloads,omitempty" json:"payloads,omitempty"`
}

StoredTrace is one trace as written to a document store.

func (StoredTrace) ErrorCount added in v1.10.0 

func (t StoredTrace) ErrorCount() int

ErrorCount is how many steps recorded an error, without decoding a payload.

The trace list needs it and nothing else about the steps; decompressing every payload of every trace to count errors would be the whole-table read that the SQL backends added a parent row to avoid.

func (StoredTrace) UnpackTrace added in v1.10.0 

func (t StoredTrace) UnpackTrace() MessageTrace

UnpackTrace rebuilds the trace a caller sees from the document that was stored.

type SuspendedMessage 

type SuspendedMessage struct {
	ID         string            `json:"id"`
	WorkflowID string            `json:"workflow_id"`
	NodeID     string            `json:"node_id"`
	Payload    []byte            `json:"payload"`
	Metadata   map[string]string `json:"metadata"`
	Data       map[string]any    `json:"data"`
	ResumeAt   time.Time         `json:"resume_at"`
	CreatedAt  time.Time         `json:"created_at"`
}

type TraceDedupCache added in v1.10.0 

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

traceDedupCache remembers which payloads already have their bytes in the table, so the steps that follow can store a reference instead of a copy.

It is deliberately a cache and not an index, and it is consulted pessimistically. An entry is added only *after* an insert has actually succeeded, and "not in the cache" means "store the bytes". So a cold start, a second control-plane replica, an eviction under load or a race between two steps of the same message all cost a duplicate copy of a payload — never a reference with nothing behind it. Getting that direction the wrong way round would turn a full table into a blank trace viewer.

func NewTraceDedupCache added in v1.10.0 

func NewTraceDedupCache(capacity int) *TraceDedupCache

func (*TraceDedupCache) AlreadyStored added in v1.10.0 

func (c *TraceDedupCache) AlreadyStored(workflowID, messageID string, hash []byte) bool

func (*TraceDedupCache) Len added in v1.10.0 

func (c *TraceDedupCache) Len() int

func (*TraceDedupCache) MarkStored added in v1.10.0 

func (c *TraceDedupCache) MarkStored(workflowID, messageID string, hash []byte)

type TraceFilter added in v1.8.1 

type TraceFilter struct {
	Before time.Time
	Limit  int
	Offset int
}

TraceFilter selects a page of message traces.

Before is a keyset cursor: pass the CreatedAt of the last row you saw and the next page is an index seek, not a scan-and-discard. Offset remains for callers that page by number, and is cheap now that it walks one row per message rather than one per step — but it still reads and throws away everything it skips, so Before is the one to reach for.

type TraceRetention added in v1.10.0 

type TraceRetention struct {
	// Keep maps a workflow id to the oldest trace step it keeps.
	//
	// A workflow with no entry here keeps everything it has. Either no window
	// is configured or the configured one could not be parsed, and neither is
	// a licence to delete: an operator-supplied window has no safe default,
	// because a short one deletes traces nobody asked to lose and none at all
	// restores the growth bug this mechanism exists to prevent.
	Keep map[string]time.Time

	// Live is every workflow that exists, including those absent from Keep.
	//
	// Traces of a workflow that is not here are unreachable — the viewer
	// reaches a trace through its workflow, and ListMessageTraces takes a
	// workflow id — so they are swept regardless of age. Nothing else removes
	// them: DeleteWorkflow deletes one row.
	Live map[string]struct{}

	// LiveIsComplete reports whether Live really is every workflow there is.
	//
	// Not inferred from Live being non-empty, and not a detail. The caller
	// reads workflows through a paged list; on a deployment with more
	// workflows than it asked for, every workflow past the first page would
	// look like a deleted one and have its traces swept. When this is false
	// the orphan sweep is skipped and everything else still runs.
	LiveIsComplete bool
}

TraceRetention describes one sweep of the message trace tables.

A cutoff *per workflow*, where this used to be a single time.Time. Retention is a per-workflow setting, but the purge it drove was `DELETE FROM message_trace_steps WHERE timestamp < ?` with no workflow predicate, called from inside a loop over workflows. So the shortest window in the deployment decided what every workflow kept: one set to 7d deleted the traces of one set to 365d, and on PostgreSQL it dropped whole day partitions out from under it, which no VACUUM undoes. That same full-table delete also ran once per workflow per hour.

func (TraceRetention) Orphans added in v1.10.0 

func (r TraceRetention) Orphans(present []string) ([]string, bool)

Orphans returns the workflow ids in present that no longer exist, and whether the caller is entitled to act on the answer.

func (TraceRetention) PartitionFloor added in v1.10.0 

func (r TraceRetention) PartitionFloor() (time.Time, bool)

PartitionFloor is the oldest moment no live workflow needs any more, and whether one exists at all.

Only whole days below this may be reclaimed by dropping a partition, because a drop takes every workflow's rows for that day at once. A single live workflow that keeps everything therefore blocks partition dropping for the whole deployment — correctly: the alternative is destroying its traces to save space on someone else's.

type TraceStep 

type TraceStep = hermod.TraceStep

type Transformation 

type Transformation struct {
	Type   string         `json:"type"`
	Config map[string]any `json:"config"`
}

type UnitTest 

type UnitTest struct {
	Name           string         `json:"name"`
	Input          map[string]any `json:"input"`
	ExpectedOutput map[string]any `json:"expected_output"`
	Description    string         `json:"description,omitempty"`
}

type User 

type User struct {
	ID               string   `json:"id"`
	Username         string   `json:"username"`
	Password         string   `json:"password,omitempty"`
	FullName         string   `json:"full_name"`
	Email            string   `json:"email"`
	Role             Role     `json:"role"`
	VHosts           []string `json:"vhosts"`
	TwoFactorEnabled bool     `json:"two_factor_enabled"`
	TwoFactorSecret  string   `json:"two_factor_secret,omitempty"`
	// CreatedAt is the sort key the user list pages on. Storage stamps it when a
	// caller leaves it zero and never rewrites it on update. See Workflow.CreatedAt.
	CreatedAt time.Time `json:"created_at" bson:"created_at"`
}

type VHost 

type VHost struct {
	ID          string `json:"id"`
	Name        string `json:"name"`
	Description string `json:"description"`
	// CreatedAt is the sort key the vhost list pages on. Storage stamps it when a
	// caller leaves it zero and never rewrites it on update. See Workflow.CreatedAt.
	CreatedAt time.Time `json:"created_at" bson:"created_at"`
}

type WebhookRequest 

type WebhookRequest struct {
	ID        string            `json:"id"`
	Timestamp time.Time         `json:"timestamp" omitzero:"true"`
	Path      string            `json:"path"`
	Method    string            `json:"method"`
	Headers   map[string]string `json:"headers"`
	Body      []byte            `json:"body"`
}

type WebhookRequestFilter 

type WebhookRequestFilter struct {
	CommonFilter
	Path string `json:"path"`
}

type Worker 

type Worker struct {
	ID          string     `json:"id"`
	Name        string     `json:"name" bson:"name"`
	Host        string     `json:"host" bson:"host"`
	Port        int        `json:"port" bson:"port"`
	Description string     `json:"description" bson:"description"`
	Token       string     `json:"token" bson:"token"`
	LastSeen    *time.Time `json:"last_seen" omitzero:"true" bson:"last_seen"`

	// Resources are inlined rather than nested so the JSON and BSON shapes are
	// unchanged for the two fields that were already here.
	//
	// The bson tags on the fields above are load-bearing and were missing. The
	// driver's default is the lowercased Go name, so this struct decoded
	// "lastseen", "cpuusage" and "memoryusage" while UpdateWorkerHeartbeat wrote
	// "last_seen", "cpu_usage" and "memory_usage" — meaning that on the MongoDB
	// backend every worker read back with a nil LastSeen and no usage at all,
	// which the UI renders as permanently Offline.
	WorkerResources `bson:",inline"`

	// Draining is a transient flag (never persisted) used to signal a running
	// worker that the platform has requested a graceful shutdown. It is set on
	// API responses by the platform when an administrator triggers a shutdown.
	Draining bool `json:"draining,omitempty" bson:"-"`
	// CreatedAt is the sort key the worker list pages on. Storage stamps it when a
	// caller leaves it zero and never rewrites it on update. See Workflow.CreatedAt.
	CreatedAt time.Time `json:"created_at" bson:"created_at"`
}

type WorkerResources added in v1.12.0 

type WorkerResources struct {
	// CPUUsage and MemoryUsage are fractions in 0..1, not percentages. They
	// predate the rest of this struct and keep their column and JSON names.
	CPUUsage    float64 `json:"cpu_usage,omitempty" bson:"cpu_usage,omitempty"`
	MemoryUsage float64 `json:"memory_usage,omitempty" bson:"memory_usage,omitempty"`

	// CPUCores is the number of logical CPUs visible to the worker process.
	CPUCores int `json:"cpu_cores,omitempty" bson:"cpu_cores,omitempty"`

	// MemoryTotalBytes and MemoryUsedBytes describe physical memory.
	MemoryTotalBytes int64 `json:"memory_total_bytes,omitempty" bson:"memory_total_bytes,omitempty"`
	MemoryUsedBytes  int64 `json:"memory_used_bytes,omitempty" bson:"memory_used_bytes,omitempty"`

	// StorageTotalBytes and StorageUsedBytes describe the filesystem holding
	// the worker's data directory — where SQLite metadata, WAL files and trace
	// payloads are written — rather than every disk on the box. That is the one
	// a full disk stops Hermod on, and the one an operator can act on.
	StorageTotalBytes int64 `json:"storage_total_bytes,omitempty" bson:"storage_total_bytes,omitempty"`
	StorageUsedBytes  int64 `json:"storage_used_bytes,omitempty" bson:"storage_used_bytes,omitempty"`
}

WorkerResources is what one worker reports about the machine it runs on: how much of it there is, and how much of it is gone.

Capacity and utilisation travel together deliberately. A usage fraction on its own does not say how much work a worker can still take — 80% of two cores and 80% of sixty-four are the same number and not the same machine — and until this struct existed the platform stored only the fractions. The scheduler in internal/engine/worker/sharding.go picks between workers on exactly this basis, and the workers page and dashboard report it.

Sizes are bytes because bytes are what the operating system reports; converting to gigabytes is a presentation decision and belongs in the one place that presents it. They are signed int64 rather than uint64 because BSON has no unsigned 64-bit type and the SQL drivers reject one outright — the same reason dashboardSampleDoc gives — and no host has eight exabytes of RAM.

Every field is optional and zero means "not reported", not "none". A worker running a release from before capacity reporting existed sends nothing here, and a cluster total that treated its silence as a machine with no CPU would read worse the more old workers were still running. Readers must skip a worker whose capacity is zero rather than average it in.

func (WorkerResources) Capacity added in v1.12.0 

func (r WorkerResources) Capacity() WorkerResources

Capacity is this reading with the load fractions dropped: what the machine is, without a claim about how busy it is.

Registration reports capacity only, and deliberately. The two fractions are scheduler inputs — calculateWeight scores peers on them and admission control sheds load on them — and a worker's own view of itself comes from its local reading, which is empty until the first health check. Persisting a real load figure at registration therefore makes every worker see itself as idle and its peers as loaded, and each one grabs everything it can. Capacity has no such reader: nothing schedules on cores or bytes, and the dashboard wants them immediately.

func (WorkerResources) ReportsCapacity added in v1.12.0 

func (r WorkerResources) ReportsCapacity() bool

ReportsCapacity says whether this worker told us how big it is.

Aggregates use it to skip the workers that did not, so an old worker abstains from the cluster totals instead of voting zero into them.

func (WorkerResources) StorageUsage added in v1.12.0 

func (r WorkerResources) StorageUsage() float64

StorageUsage is the used share of the data filesystem, 0..1.

Derived rather than stored: CPUUsage and MemoryUsage are persisted because they existed before the byte counts did and the scheduler reads them, but a third fraction alongside the two numbers it is computed from would be the same figure under two names.

type Workflow 

type Workflow struct {
	ID                string         `json:"id"`
	Name              string         `json:"name"`
	VHost             string         `json:"vhost"`
	Active            bool           `json:"active"`
	Status            string         `json:"status,omitempty"`
	WorkerID          string         `json:"worker_id"`
	OwnerID           string         `json:"owner_id,omitempty"`
	LeaseUntil        *time.Time     `json:"lease_until" omitzero:"true"`
	Nodes             []WorkflowNode `json:"nodes"`
	Edges             []WorkflowEdge `json:"edges"`
	DeadLetterSinkID  string         `json:"dead_letter_sink_id,omitempty"`
	PrioritizeDLQ     bool           `json:"prioritize_dlq,omitempty"`
	MaxRetries        int            `json:"max_retries,omitempty"`
	RetryInterval     string         `json:"retry_interval,omitempty"`
	ReconnectInterval string         `json:"reconnect_interval,omitempty"`
	DryRun            bool           `json:"dry_run,omitempty"`
	IdleTimeout       string         `json:"idle_timeout,omitempty"` // e.g. "5m"
	Tier              WorkflowTier   `json:"tier,omitempty"`
	// RetentionDays optionally overrides global log retention for this workflow.
	// nil means inherit the global setting; 0 means keep forever; >0 means days to retain.
	RetentionDays     *int     `json:"retention_days,omitempty"`
	SchemaType        string   `json:"schema_type,omitempty"`
	Schema            string   `json:"schema,omitempty"`
	Cron              string   `json:"cron,omitempty"`
	DLQThreshold      int      `json:"dlq_threshold,omitempty"`
	TraceSampleRate   float64  `json:"trace_sample_rate,omitempty"`
	Tags              []string `json:"tags" omitzero:"true"`
	TraceRetention    string   `json:"trace_retention,omitempty"` // e.g. "7d", "30d"
	AuditRetention    string   `json:"audit_retention,omitempty"` // e.g. "30d", "365d"
	WorkspaceID       string   `json:"workspace_id,omitempty"`
	CPURequest        float64  `json:"cpu_request,omitempty"`
	MemoryRequest     float64  `json:"memory_request,omitempty"`
	ThroughputRequest int      `json:"throughput_request,omitempty"`
	TotalProcessed    uint64   `json:"total_processed,omitempty"`
	TotalErrors       uint64   `json:"total_errors,omitempty"`
	TotalLag          uint64   `json:"total_lag,omitempty"`
	// CreatedAt is the sort key the workflow list pages on. Storage stamps it
	// when a caller leaves it zero and never rewrites it on update, so it stays
	// the creation time rather than the last-touched time. The bson tag is here
	// because the mongo backend writes snake_case keys explicitly.
	CreatedAt time.Time `json:"created_at" bson:"created_at"`
}

type WorkflowEdge 

type WorkflowEdge struct {
	ID           string         `json:"id"`
	SourceID     string         `json:"source_id"`
	TargetID     string         `json:"target_id"`
	SourceHandle string         `json:"source_handle,omitempty"`
	TargetHandle string         `json:"target_handle,omitempty"`
	Config       map[string]any `json:"config" omitzero:"true"`
}

type WorkflowExportBundle 

type WorkflowExportBundle struct {
	Workflow Workflow `json:"workflow"`
	Sources  []Source `json:"sources,omitempty" omitzero:"true"`
	Sinks    []Sink   `json:"sinks,omitempty" omitzero:"true"`

	// MissingRefs names the dependencies the workflow points at that no longer
	// exist on the exporting instance. They were previously dropped from the
	// bundle in silence, so an export of a workflow with a dangling reference
	// looked complete and imported into a workflow that could not start.
	MissingRefs []MissingRef `json:"missing_refs,omitempty" omitzero:"true"`
}

WorkflowExportBundle packages a workflow along with its referenced dependencies (like sources and sinks) for easy export and import between Hermod instances.

type WorkflowHealth 

type WorkflowHealth struct {
	WorkflowID string        `json:"workflow_id"`
	Status     string        `json:"status"` // "healthy", "degraded", "error"
	Issues     []string      `json:"issues" omitzero:"true"`
	Uptime     time.Duration `json:"uptime"`
	Processed  uint64        `json:"processed"`
	Errors     uint64        `json:"errors"`
	Lag        uint64        `json:"lag"`
	MPS        float64       `json:"mps"`
	Latency    time.Duration `json:"latency"`
}

type WorkflowNode 

type WorkflowNode struct {
	ID        string         `json:"id"`
	Type      string         `json:"type"`             // source, sink, transformer, condition, etc.
	RefID     string         `json:"ref_id,omitempty"` // ID of the source, sink, or transformation
	Config    map[string]any `json:"config" omitzero:"true"`
	X         float64        `json:"x"`
	Y         float64        `json:"y"`
	UnitTests []UnitTest     `json:"unit_tests" omitzero:"true"`
}

type WorkflowTier 

type WorkflowTier string
const (
	WorkflowTierHot  WorkflowTier = "Hot"
	WorkflowTierCold WorkflowTier = "Cold"
)

type WorkflowVersion 

type WorkflowVersion struct {
	ID             string         `json:"id"`
	WorkflowID     string         `json:"workflow_id"`
	Version        int            `json:"version"`
	Nodes          []WorkflowNode `json:"nodes"`
	Edges          []WorkflowEdge `json:"edges"`
	TraceRetention string         `json:"trace_retention,omitempty"`
	AuditRetention string         `json:"audit_retention,omitempty"`
	Config         string         `json:"config"` // JSON of other workflow settings
	CreatedAt      time.Time      `json:"created_at"`
	CreatedBy      string         `json:"created_by"`
	Message        string         `json:"message"` // commit message for the version
}

type Workspace 

type Workspace struct {
	ID            string    `json:"id"`
	Name          string    `json:"name"`
	Description   string    `json:"description"`
	MaxWorkflows  int       `json:"max_workflows"`
	MaxCPU        float64   `json:"max_cpu"`
	MaxMemory     float64   `json:"max_memory"`
	MaxThroughput int       `json:"max_throughput"` // messages per second
	CreatedAt     time.Time `json:"created_at"`
}

Source Files

View all Source files

Directories

Path Synopsis
Package configsecrets decides which connector configuration values are credentials, and encrypts them at rest.
 Package configsecrets decides which connector configuration values are credentials, and encrypts them at rest.

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.