eventbus

package module
v0.3.8 Latest Latest
Warning

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

 Go to latest Published: Jul 6, 2026 License: MIT Imports: 16 Imported by: 0

README

workflow-plugin-eventbus

✅ Verified — used in production at buymywishlist. This plugin has been validated end-to-end in a merged main-branch wfctl.yaml of an active GoCodeAlone project.

A workflow external plugin that provisions durable event-bus clusters and exposes typed pipeline steps for publish/consume operations.

v0.2.0 module-type rename: infra.eventbus* → eventbus.*. The infra. prefix is reserved for IaC modules; eventbus modules are runtime modules. If you're upgrading from v0.1.0, see MIGRATION.md.

Providers

Provider Deploy targets Notes
nats DO App Platform, AWS ECS/EKS, Kubernetes StatefulSet JetStream-backed; durable streams + consumers
pgchannel in_process Postgres LISTEN/NOTIFY + polling fallback; no broker infrastructure
kafka DO Managed Kafka, AWS MSK, Kubernetes (Strimzi) scaffold
kinesis AWS (Kinesis Data Streams) scaffold
pgchannel — Postgres-backed broker

For low-traffic deployments, the pgchannel provider eliminates the need for a NATS/Kafka broker by implementing pub/sub atop Postgres LISTEN/NOTIFY with a polling fallback for guaranteed delivery. Per-consumer advisory locks + a eventbus_event_deliveries tracking table enforce max_deliver semantics.

Tradeoffs: higher latency, lower throughput than NATS. When traffic justifies a real broker, swap provider: pgchannel → provider: nats; pipeline call sites (step.eventbus.publish / step.eventbus.consume / step.eventbus.ack / trigger.eventbus.subscribe) remain unchanged.

Reference schema migrations live at providers/pgchannel/internal/testutil/schema.sql. Consumer projects ship these directly in their own migration pipeline — the plugin does NOT embed them.

Usage

Declare a cluster (NATS)
modules:
  - name: my-events
    type: eventbus.broker
    config:
      provider: nats
      deploy_target: digitalocean.app_platform
      version: "2.10"
      replicas: 2
      jetstream:
        enabled: true
        max_storage_bytes: 53687091200  # 50 GB
Declare a cluster (pgchannel)
modules:
  - name: my-events
    type: eventbus.broker
    config:
      provider: pgchannel
      broker_target: in_process
      dsn: ${DATABASE_URL}
      poll_interval: 5s
      max_conns: 32  # size as 2*N + 4 where N = consumer module count
Declare streams and consumers
  - name: my-stream
    type: eventbus.stream
    config:
      broker_ref: my-events       # points at the broker module name above
      name: MY_EVENTS
      subjects: ["events.>"]
      retention_policy: RETENTION_POLICY_LIMITS
      max_bytes: 10737418240  # 10 GB

  - name: my-consumer
    type: eventbus.consumer
    config:
      broker_ref: my-events       # points at the broker module name above
      stream_name: MY_EVENTS
      name: my-handler
      filter_subject: "events.>"
      ack_policy: ACK_POLICY_EXPLICIT
      max_deliver: 5

broker_ref is required when multiple broker instances are registered in the same process. Single-broker deployments may omit it (the runtime falls back to the sole registered broker and logs a warning if ambiguity is detected).

Publish from a pipeline step
steps:
  - name: publish
    type: step.eventbus.publish
    config:
      subject: events.created
      payload: '{{ toJson .input }}'
Subscribe trigger
my-handler:
  trigger:
    type: trigger.eventbus.subscribe
    config:
      stream_name: MY_EVENTS
      name: my-handler
      filter_subject: "events.>"
      ack_policy: ACK_POLICY_EXPLICIT
  steps:
    - name: ack
      type: step.eventbus.ack
      config:
        ack_token: '{{ .nats.message.ack_token }}'

Upgrading from v0.1.0

See MIGRATION.md for the full guide. Highlights:

  • Rename infra.eventbus* module types to eventbus.* in YAML configs.
  • Move eventbus blocks out of infra.yaml into app.yaml (runtime, not IaC).
  • Add broker_ref to each stream and consumer config.
  • Optionally adopt the new pgchannel provider for broker-free deployments.

Development

# Regenerate proto bindings after editing proto/eventbus.proto
make proto-gen

# Build
make build

# Test
make test

Documentation

Overview

Package eventbus implements the workflow-plugin-eventbus plugin. It provides eventbus.broker, eventbus.stream, and eventbus.consumer module types plus step and trigger types for durable event-bus integration.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func DefaultBusConn deprecated 

func DefaultBusConn() (*nats.Conn, error)

DefaultBusConn returns a live NATS connection for the lexicographically first registered eventbus.broker module. Sorting ensures deterministic selection across invocations and concurrent goroutines, even when multiple buses are registered. For multi-bus workflows, use GetOrDialNATSConn(instanceName) directly.

Deprecated: superseded by LookupRuntimeWithFallback, which routes through providers.RuntimeBroker and works across nats / pgchannel / future providers. Retained for source-compat with callers that still hold a direct *nats.Conn.

func GetBusURI 

func GetBusURI(instanceName string) (string, bool)

GetBusURI returns the broker URI for instanceName.

func GetCluster 

func GetCluster(instanceName string) (*eventbusv1.ClusterConfig, bool)

GetCluster looks up a ClusterConfig by instance name.

func GetConsumer 

func GetConsumer(instanceName string) (*eventbusv1.ConsumerConfig, bool)

GetConsumer looks up a ConsumerConfig by instance name.

func GetConsumerByName 

func GetConsumerByName(durableName string) (*eventbusv1.ConsumerConfig, bool)

GetConsumerByName looks up a ConsumerConfig by its durable consumer name (cfg.name), iterating all registered instances. This is used by step.eventbus.consume to resolve the consumer config from the durable name supplied in ConsumeRequest.consumer.

func GetNATSConn deprecated 

func GetNATSConn(instanceName string) (*nats.Conn, bool)

GetNATSConn returns the cached *nats.Conn for instanceName, or false if absent.

Deprecated: see RegisterNATSConn.

func GetOrDialNATSConn deprecated 

func GetOrDialNATSConn(instanceName string) (*nats.Conn, error)

GetOrDialNATSConn returns the cached NATS connection for instanceName, dialing a new one (via natsDialFn) if no live connection is cached. Returns an error if no URI is registered for instanceName or the dial fails.

Lock ordering: connCacheMu and urlMu (held inside GetBusURI) are never held simultaneously. The URI lookup happens between the fast-path unlock and the slow-path re-lock so that no nested acquisition is possible.

Deprecated: new code should use LookupRuntime / LookupRuntimeWithFallback, which dispatch through providers.RuntimeBroker. Retained for source-compat with external consumers that still hold a direct *nats.Conn.

func GetStream 

func GetStream(instanceName string) (*eventbusv1.StreamConfig, bool)

GetStream looks up a StreamConfig by instance name.

func LookupBrokerInstance 

func LookupBrokerInstance(name string) (*clusterModule, bool)

LookupBrokerInstance returns the *clusterModule registered under name, or false when absent. Primarily for tests; production callers should use LookupRuntime, which also validates that Start has run.

func LookupRuntime 

func LookupRuntime(name string) (providers.RuntimeBroker, providers.Connection, error)

LookupRuntime returns the RuntimeBroker + cached Connection for the named broker. Used by stream/consumer modules' Start (Group E) and by step factories + the trigger module (Group F) to dispatch through the provider abstraction.

Returns an error when:

  • no broker module is registered under name (Init never ran or wrong name);
  • the broker module is registered but Start has not yet completed (runtime/conn are still nil).

func LookupRuntimeWithFallback 

func LookupRuntimeWithFallback(brokerRef string) (providers.RuntimeBroker, providers.Connection, error)

LookupRuntimeWithFallback resolves a RuntimeBroker + Connection given an optional brokerRef. When brokerRef is non-empty, behaves like LookupRuntime. When brokerRef is empty AND exactly one broker is registered, returns that broker (the legacy single-bus fallback that mirrors DefaultBusConn).

Used by step factories + the trigger so configs predating the broker_ref field continue to work in single-bus deployments. Returns a clear error when the fallback is ambiguous (multiple brokers registered) or impossible (no brokers registered), guiding the caller to set broker_ref explicitly.

func NewClusterModule 

func NewClusterModule(instanceName string, cfg *eventbusv1.ClusterConfig) (sdk.ModuleInstance, error)

NewClusterModule creates a clusterModule from a typed ClusterConfig proto.

Validation is per-provider because the configuration shape diverges:

  • pgchannel runs in-process against an existing Postgres database. It does not deploy a broker, so deploy_target is meaningless; instead broker_target=in_process is required (the only supported mode in the in-process runtime) along with cfg.dsn carrying the Postgres connection string.
  • nats, kafka, kinesis each deploy a managed/self-hosted broker onto a cloud target, so deploy_target is required and must be in the supported matrix (providers.ValidateProviderTarget).

Any provider not in the {pgchannel, nats, kafka, kinesis} set is rejected here. The previous implementation rejected any empty deploy_target uniformly; the relaxation lands as part of design §1.7 to enable the pg-backed-provider flow.

func NewConsumerModule 

func NewConsumerModule(instanceName string, cfg *eventbusv1.ConsumerConfig) (sdk.ModuleInstance, error)

NewConsumerModule creates a consumerModule from a typed ConsumerConfig proto.

Returns an error if:

  • config.name is empty
  • config.stream_name is empty

func NewStreamModule 

func NewStreamModule(instanceName string, cfg *eventbusv1.StreamConfig) (sdk.ModuleInstance, error)

NewStreamModule creates a streamModule from a typed StreamConfig proto.

Returns an error if:

  • config.name is empty
  • config.subjects contains no entries

func NewSubscribeTrigger 

func NewSubscribeTrigger(instanceName string, cfg *eventbusv1.ConsumerConfig, cb sdk.TriggerCallback) (sdk.ModuleInstance, error)

NewSubscribeTrigger creates a subscribeTrigger from a typed ConsumerConfig proto.

Returns an error if:

  • config.name is empty
  • config.stream_name is empty

func RegisterBrokerInstance 

func RegisterBrokerInstance(name string, m *clusterModule)

RegisterBrokerInstance stores m under name. Exported so integration tests can pre-seed the registry with a hand-built clusterModule.

func RegisterBusURI 

func RegisterBusURI(instanceName, uri string)

RegisterBusURI stores a broker URI under instanceName.

func RegisterCluster 

func RegisterCluster(instanceName string, cfg *eventbusv1.ClusterConfig)

RegisterCluster stores a ClusterConfig in the global registry under instanceName.

func RegisterConsumer 

func RegisterConsumer(instanceName string, cfg *eventbusv1.ConsumerConfig)

RegisterConsumer stores a ConsumerConfig in the global registry under instanceName.

func RegisterNATSConn deprecated 

func RegisterNATSConn(instanceName string, conn *nats.Conn)

RegisterNATSConn stores a live connection under instanceName. Exported so that integration tests and the trigger can pre-populate the cache.

Deprecated: this helper predates the providers.RuntimeBroker abstraction. New code should construct an eventbus.broker module + call Init/Start, which publishes the runtime + connection through LookupRuntime. Kept for legacy callers and the bounded lifecycle teardown path in clusterModule.Stop.

func RegisterStream 

func RegisterStream(instanceName string, cfg *eventbusv1.StreamConfig)

RegisterStream stores a StreamConfig in the global registry under instanceName.

func UnregisterBrokerInstance 

func UnregisterBrokerInstance(name string)

UnregisterBrokerInstance removes the entry for name. Idempotent.

func UnregisterBusURI 

func UnregisterBusURI(instanceName string)

UnregisterBusURI removes the URI entry for instanceName.

func UnregisterCluster 

func UnregisterCluster(instanceName string)

UnregisterCluster removes a ClusterConfig from the registry.

func UnregisterConsumer 

func UnregisterConsumer(instanceName string)

UnregisterConsumer removes a ConsumerConfig from the registry.

func UnregisterNATSConn deprecated 

func UnregisterNATSConn(instanceName string)

UnregisterNATSConn removes the cached connection entry for instanceName without closing the connection. Use this in tests that manage the connection's lifetime separately (e.g., via nc.Close() + embedded-server shutdown).

Deprecated: see RegisterNATSConn.

func UnregisterStream 

func UnregisterStream(instanceName string)

UnregisterStream removes a StreamConfig from the registry.

Types

type ClusterModuleFactory 

type ClusterModuleFactory struct{}

ClusterModuleFactory implements sdk.TypedModuleProvider for the eventbus.broker module type. The plugin wires this factory into CreateTypedModule.

func (*ClusterModuleFactory) CreateTypedModule 

func (f *ClusterModuleFactory) CreateTypedModule(typeName, name string, config *anypb.Any) (sdk.ModuleInstance, error)

CreateTypedModule unpacks the typed proto config and delegates to NewClusterModule.

func (*ClusterModuleFactory) TypedModuleTypes 

func (f *ClusterModuleFactory) TypedModuleTypes() []string

TypedModuleTypes returns the single module type served by this factory.

type ConsumerModuleFactory 

type ConsumerModuleFactory struct{}

ConsumerModuleFactory implements sdk.TypedModuleProvider for the eventbus.consumer module type.

func (*ConsumerModuleFactory) CreateTypedModule 

func (f *ConsumerModuleFactory) CreateTypedModule(typeName, name string, config *anypb.Any) (sdk.ModuleInstance, error)

CreateTypedModule unpacks the typed proto config and delegates to NewConsumerModule.

func (*ConsumerModuleFactory) TypedModuleTypes 

func (f *ConsumerModuleFactory) TypedModuleTypes() []string

TypedModuleTypes returns the single module type served by this factory.

type StreamModuleFactory 

type StreamModuleFactory struct{}

StreamModuleFactory implements sdk.TypedModuleProvider for the eventbus.stream module type.

func (*StreamModuleFactory) CreateTypedModule 

func (f *StreamModuleFactory) CreateTypedModule(typeName, name string, config *anypb.Any) (sdk.ModuleInstance, error)

CreateTypedModule unpacks the typed proto config and delegates to NewStreamModule.

func (*StreamModuleFactory) TypedModuleTypes 

func (f *StreamModuleFactory) TypedModuleTypes() []string

TypedModuleTypes returns the single module type served by this factory.

type SubscribeTriggerModuleFactory 

type SubscribeTriggerModuleFactory struct{}

SubscribeTriggerModuleFactory implements sdk.TypedModuleProvider for the trigger.eventbus.subscribe module type. The external plugin adapter calls CreateTypedModule with the trigger type name to instantiate triggers over gRPC.

func (*SubscribeTriggerModuleFactory) CreateTypedModule 

func (f *SubscribeTriggerModuleFactory) CreateTypedModule(typeName, name string, config *anypb.Any) (sdk.ModuleInstance, error)

CreateTypedModule unpacks the typed proto config and delegates to NewSubscribeTrigger. cb is always nil in the external gRPC subprocess path (the callback client is never wired in production SDK code); triggers that receive cb=nil behave as no-ops on Start, which is correct for IaC-only and plan/apply workflows.

func (*SubscribeTriggerModuleFactory) TypedModuleTypes 

func (f *SubscribeTriggerModuleFactory) TypedModuleTypes() []string

TypedModuleTypes returns the single trigger module type served by this factory.

Source Files

View all Source files

Directories

Path Synopsis
cmd
workflow-plugin-eventbus command
Command workflow-plugin-eventbus is a workflow engine external plugin that provisions durable event-bus clusters (NATS / Kafka / Kinesis) as IaC and exposes typed pipeline steps for publish / consume operations.
 Command workflow-plugin-eventbus is a workflow engine external plugin that provisions durable event-bus clusters (NATS / Kafka / Kinesis) as IaC and exposes typed pipeline steps for publish / consume operations.
Package iac defines typed IaC primitives emitted by eventbus Providers.
 Package iac defines typed IaC primitives emitted by eventbus Providers.
Package providers defines the Provider interface and DeployTarget compatibility matrix for workflow-plugin-eventbus.
 Package providers defines the Provider interface and DeployTarget compatibility matrix for workflow-plugin-eventbus.
kafka
Package kafka provides a stub implementation of the kafka event-bus Provider.
 Package kafka provides a stub implementation of the kafka event-bus Provider.
kinesis
Package kinesis provides a stub implementation of the kinesis event-bus Provider.
 Package kinesis provides a stub implementation of the kinesis event-bus Provider.
nats
Package nats provides the NATS event-bus Provider implementation.
 Package nats provides the NATS event-bus Provider implementation.
pgchannel
Package pgchannel — see subject_match.go for the package doc.
 Package pgchannel — see subject_match.go for the package doc.
pgchannel/internal/testutil
Package testutil — testcontainers helpers for the pgchannel provider tests.
 Package testutil — testcontainers helpers for the pgchannel provider tests.
Package steps provides typed step handlers for the workflow-plugin-eventbus plugin: publish, consume, and ack operations dispatched through the providers.RuntimeBroker abstraction.
 Package steps provides typed step handlers for the workflow-plugin-eventbus plugin: publish, consume, and ack operations dispatched through the providers.RuntimeBroker abstraction.

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.