providers

type Connection interface {
	// Close releases the underlying broker connection / pool. Idempotent.
	Close() error
	// Provider returns the provider identifier this connection belongs to
	// (e.g. "nats", "pgchannel"). Used for diagnostics + telemetry.
	Provider() string
}

type DeployTarget string

type HealthCheck struct {
	// URI is the address that was probed.
	URI string
	// Status is the typed health state.
	Status HealthStatus
	// Err is non-nil when Status is HealthStatusDegraded or HealthStatusUnreachable.
	Err error
}

type HealthStatus string

type MessageHandler func(ctx context.Context, msg *eventbusv1.Message) error

type Provider interface {
	// Name returns the provider identifier: "nats", "kafka", or "kinesis".
	Name() string

	// Resources returns the IaC resource declarations required to provision
	// the event-bus cluster described by cfg on the given deploy target.
	// cfg is passed as a pointer because proto messages embed sync.Mutex via
	// protoimpl.MessageState and must not be copied by value.
	// Returns an error if the provider × target combination is unsupported
	// or if cfg is invalid for this provider.
	Resources(cfg *eventbusv1.ClusterConfig, target DeployTarget) ([]iac.Resource, error)

	// ConnectionString derives the broker connection string from provisioned state.
	// env selects environment-specific outputs (e.g. "prod", "staging").
	ConnectionString(state iac.State, env string) (string, error)

	// StreamResources returns the IaC resource declarations required to
	// declare the given streams against an already-provisioned cluster
	// (represented by state). Streams are pointers for the same reason as cfg.
	StreamResources(streams []*eventbusv1.StreamConfig, state iac.State) ([]iac.Resource, error)

	// Probe probes the event-bus cluster at uri and returns its health state.
	Probe(uri string) HealthCheck
}

type RuntimeBroker interface {
	// Connect opens a connection to the broker described by cfg. The returned
	// Connection is opaque to callers and is passed back into subsequent
	// EnsureStream / EnsureConsumer / Publish / Subscribe / Ack calls.
	Connect(ctx context.Context, cfg *eventbusv1.ClusterConfig) (Connection, error)

	// EnsureStream creates or updates the stream described by cfg. Implementations
	// MUST be idempotent: calling EnsureStream twice with the same cfg is a no-op.
	EnsureStream(ctx context.Context, conn Connection, cfg *eventbusv1.StreamConfig) error

	// EnsureConsumer creates or updates the consumer described by cfg on the named
	// stream. Implementations MUST be idempotent.
	EnsureConsumer(ctx context.Context, conn Connection, streamName string, cfg *eventbusv1.ConsumerConfig) error

	// Publish publishes a single message via the broker. The returned
	// PublishResponse carries provider-assigned sequence / timestamp metadata.
	Publish(ctx context.Context, conn Connection, req *eventbusv1.PublishRequest) (*eventbusv1.PublishResponse, error)

	// Subscribe attaches handler to streamName / consumerName and blocks until
	// ctx is cancelled or an unrecoverable error occurs. Returning nil from
	// handler acks the message; returning an error naks it (delivery counted
	// against max_deliver per ConsumerConfig).
	//
	// Subscribe is the long-lived "push" path used by trigger.eventbus.subscribe.
	// For bounded-batch pull semantics (step.eventbus.consume — return up to
	// batch_size messages bounded by max_wait, then return) use Consume instead.
	Subscribe(ctx context.Context, conn Connection, streamName, consumerName string, handler MessageHandler) error

	// Consume returns up to req.batch_size already-delivered messages from the
	// durable consumer named req.consumer on the named stream, blocking at
	// most req.max_wait for the batch to fill. Returns an empty
	// ConsumeResponse when no messages are available within max_wait (NOT an
	// error).
	//
	// Each returned Message.ack_token is opaque to callers and must be passed
	// back through Ack (typically via step.eventbus.ack) to acknowledge the
	// message. Implementations choose the token format (NATS reply subject,
	// "<stream>:<consumer>:<id>" for pgchannel, etc.).
	//
	// streamName is passed positionally (rather than embedded in req) because
	// the proto ConsumeRequest does not carry stream identity; callers
	// (typically step.eventbus.consume) resolve it from the registered
	// ConsumerConfig before dispatching here, mirroring the EnsureConsumer
	// shape.
	//
	// Consume is the bounded-pull counterpart to Subscribe: it does not
	// register a long-lived handler — each call sets up + tears down whatever
	// fetch infrastructure the provider needs, returns a single batch, and
	// exits. This separation per Group A's TODO cleanly splits pull
	// (step.eventbus.consume) from push (trigger.eventbus.subscribe).
	Consume(ctx context.Context, conn Connection, streamName string, req *eventbusv1.ConsumeRequest) (*eventbusv1.ConsumeResponse, error)

	// Ack acknowledges a previously delivered message identified by ackToken
	// (Message.ack_token). Used by step.eventbus.ack for explicit-ack flows.
	Ack(ctx context.Context, conn Connection, ackToken string) error
}