flow

type Block interface {
	// Process handles one message. A source receives msg == nil and may return
	// zero or more messages. A transform receives exactly one non-nil message
	// and may return zero or more. A sink returns nil.
	Process(msg *Message) ([]*Message, error)
	// Name returns a human-readable label for metrics and logging.
	Name() string
}

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

type Config struct {
	// PipelineID is the Manager map key for this instance (realm/flowName).
	// Historical field name; value is InstanceID, not the pipeline recipe name.
	PipelineID string
	// Blocks is the ordered list of blocks forming the processing graph.
	Blocks []Block
	// RouterCfg is applied to the underlying Router.
	RouterCfg RouterConfig
	// Registerer receives Prometheus collectors; nil leaves them
	// unregistered.
	Registerer prometheus.Registerer
}

type Connection struct {
	// From is the name of the source block.
	From string `json:"from"`
	// FromPort is the output port on the source block (empty means default).
	FromPort string `json:"from_port,omitempty"`
	// To is the name of the target block.
	To string `json:"to"`
	// ToPort is the input port on the target block (empty means default).
	ToPort string `json:"to_port,omitempty"`
}

type Constructor func(name string, config map[string]any, deps Deps) (Block, error)

type DataType uint8

type Deps struct {
	// Bus is the live event bus (required by astarte_source).
	Bus *stream.Bus
	// Realm is the tenant the pipeline runs for; sources may default to it
	// when their config omits realm.
	Realm string
	// FlowName is the durable/named flow instance name (optional). Used by
	// container labels and similar operator metadata.
	FlowName string
	// NotifyFatal, when set, is called by blocks that die at runtime (e.g. a
	// container block whose container exits unexpectedly) so the service layer
	// can fail the flow and schedule a restart. Must be safe to call from any
	// goroutine; blocks must not call it after their Stopper.Stop() ran.
	NotifyFatal func(block string, cause error)
	// Ingest, when set, lets a block land data as if a registered device had
	// produced it (virtual_device_pool, issue #84): full validation plus a
	// datastream/property row, no MQTT delivery. Required by
	// virtual_device_pool; every other block ignores it.
	Ingest IngestFunc
	// Register, when set, lets a block register a first-seen virtual device
	// through the pairing door (virtual_device_pool auto_register, #84).
	// Required by virtual_device_pool with auto_register; every other block
	// ignores it.
	Register RegisterFunc
}

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

type IngestFunc func(ctx context.Context, realm, deviceID, ifaceName, path string, payload json.RawMessage, ts *time.Time) error

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

type Message struct {
	// Key identifies the stream this message belongs to. It must be non-empty.
	Key string
	// Metadata is an optional string→string map carried alongside the payload.
	Metadata map[string]string
	// Type is the base data type of the payload.
	Type DataType
	// Subtype is an optional MIME hint (meaningful when Type is TypeBinary).
	Subtype string
	// Timestamp is the event-time in microseconds since Unix epoch.
	Timestamp int64
	// Data is the payload; its concrete type must match Type (int64 for
	// TypeInteger, float64 for TypeReal, bool for TypeBoolean, time.Time for
	// TypeDatetime, []byte for TypeBinary, string for TypeString, map[string]any
	// for TypeMap).
	Data any
	// FieldTypes holds per-field types when Type is TypeMap.
	FieldTypes map[string]DataType
	// FieldSubtypes holds per-field subtypes when Type is TypeMap.
	FieldSubtypes map[string]string
}

type OverflowPolicy uint8

type Pipeline struct {
	// ID is a unique identifier for this pipeline.
	ID string `json:"id"`
	// Name is a human-readable label.
	Name string `json:"name"`
	// Blocks is the set of nodes in the graph.
	Blocks []PipelineNode `json:"blocks"`
	// Connections is the set of edges linking block output ports to input ports.
	Connections []Connection `json:"connections"`
}

type PipelineNode struct {
	// Name is a unique identifier for this node within the pipeline.
	Name string `json:"name"`
	// BlockType identifies which block implementation to use.
	BlockType string `json:"block_type"`
	// Config holds block-specific parameters.
	Config map[string]any `json:"config,omitempty"`
}

type RegisterFunc func(ctx context.Context, realm, deviceID string) error

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

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

type RouterConfig struct {
	// Lanes is the number of processing lanes (default DefaultLaneCount).
	Lanes int
	// LaneCapacity is the per-lane channel capacity (default
	// DefaultLaneCapacity).
	LaneCapacity int
	// QoS0Overflow is the policy when a QoS 0 message targets a full lane.
	QoS0Overflow OverflowPolicy
	// QoS1Overflow is the policy when a QoS ≥ 1 message targets a full lane.
	QoS1Overflow OverflowPolicy
	// Registerer receives Prometheus collectors; nil leaves them
	// unregistered.
	Registerer prometheus.Registerer
	// Logger receives router logs (default slog.Default()).
	Logger *slog.Logger
}

type SinkFunc func(msg *Message) error

type Source interface {
	Block
	// Emit returns newly available messages. Implementations may block until
	// at least one message is ready or ctx is cancelled.
	Emit(ctx context.Context) ([]*Message, error)
}

type SourceFunc func() ([]*Message, error)

type Status uint8

type Stopper interface {
	Stop()
}

type TransformFunc func(msg *Message) ([]*Message, error)

type UserBlock struct {
	Name         string
	BlockType    string // producer | consumer | producer_consumer; stored, not enforced here
	Source       []byte // a pipeline definition body: {"blocks":[…],"connections":[…]}
	ConfigSchema json.RawMessage
}