source

type Local struct {
	// DB is the database connection used to discover event streams and read
	// events.
	DB *sql.DB

	// Signals is the bus used to receive signals that event streams have been
	// created and that events have been appended to them.
	Signals *signal.Bus

	// Logger is the target for log messages produced by the source.
	Logger *slog.Logger

	// SegmentSize is the maximum number of events delivered in one segment,
	// which is also the number of events read per query. It must be positive.
	SegmentSize int

	// BufferSize is the number of segments buffered for each call to
	// [Local.Watch]. A call that falls further behind than this ends with an
	// error that wraps [ErrConsumerTooSlow]. It must be positive.
	BufferSize int
	// contains filtered or unexported fields
}

type Memory struct {
	// SegmentSize is the maximum number of events delivered in one segment by
	// Read. Zero means no limit.
	SegmentSize int
	// contains filtered or unexported fields
}

type Remote struct {
	// Client is the client used to call the remote engine's event stream
	// consumer API.
	Client messaginggrpc.EventStreamConsumerAPIClient
}

type Segment struct {
	// StreamID is the ID of the event stream that this is a segment of.
	StreamID *uuidpb.UUID

	// Begin and End bound the segment, [Begin, End).
	Begin, End uint64

	// Envelopes are the events in the segment that match the message types for
	// which it was read, in offset order.
	//
	// It may be empty, indicating only that the range holds no events of those
	// types.
	//
	// Each envelope has the [envelopepb.EventStreamPosition] extension set.
	Envelopes []*envelopepb.Envelope
}

type Server struct {
	// EventStreams is the source of the event streams that the server
	// provides.
	EventStreams Source

	// MaxSegmentBytes is the maximum combined size, in bytes, of the events in
	// each segment that the server sends. A segment whose events are larger is
	// sent as several contiguous segments. If a single event is larger than
	// this, it is sent on its own. It must be positive.
	MaxSegmentBytes int
}

type Source interface {
	// Watch delivers events to fn as they are appended to the source's
	// streams.
	//
	// Only events of the given types are delivered. A nil set of types means
	// events of every type. It panics if the set is empty.
	//
	// The first segment of a stream that a call delivers is called the stream's
	// opening segment. It begins at the stream's end, and may be empty. Every
	// later segment of the stream begins where the previous one ended, for as
	// long as the call lasts. The opening segment of a later call may begin
	// after events that this call never delivered; a caller that needs them
	// reads them with [Source.Read].
	//
	// fn may be called concurrently for different streams, but a stream's
	// segments are delivered one at a time, in order.
	//
	// It blocks until ctx is canceled or it is unable to continue, and may be
	// called again once it returns. It returns fn's error, unchanged, if fn
	// returns one, and an error that wraps [ErrConsumerTooSlow] if the caller
	// can not keep up.
	Watch(
		ctx context.Context,
		types *uuidcollection.Set,
		fn func(Segment) error,
	) error

	// Read delivers events from a stream to fn, beginning at the given
	// offset.
	//
	// Only events of the given types are delivered, in contiguous segments,
	// until the end of the stream as it stands, at which point it returns
	// nil. A nil set of types means events of every type. It panics if the
	// set is empty.
	//
	// It delivers nothing from an offset at the end of the stream. It returns
	// an error that wraps [ErrOffsetOutOfRange] for an offset beyond the end,
	// one that wraps [ErrUnrecognizedStream] for a stream that the source does
	// not have, and fn's error, unchanged, if fn returns one.
	Read(
		ctx context.Context,
		streamID *uuidpb.UUID,
		types *uuidcollection.Set,
		begin uint64,
		fn func(Segment) error,
	) error
}