sql

type AggregateProjector struct {
	sync.Mutex
	// contains filtered or unexported fields
}

type AggregateProjectorStorage interface {
	ProjectorStorage

	LoadOutOfSync(ctx context.Context, conn Queryer) (*sql.Rows, error)

	PersistFailure(conn Execer, notification *ProjectionNotification) error
}

type EventStreamLoader func(ctx context.Context, conn *sql.Conn, notification *ProjectionNotification, position int64) (goengine.EventStream, error)

type Execer interface {
	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
}

type Listener interface {
	// Listen starts listening to the event stream and call the trigger when a event was appended
	Listen(ctx context.Context, trigger ProjectionTrigger) error
}

type MessageFactory interface {
	// CreateEventStream reconstructs the message from the provided rows
	CreateEventStream(rows *sql.Rows) (goengine.EventStream, error)
}

type Metrics interface {
	// ReceivedNotification sends the metric to keep count of notifications received by goengine
	ReceivedNotification(isNotification bool)
	// QueueNotification is called when a notification is queued.
	// It saves start time for an event on aggregate when it's queued
	QueueNotification(notification *ProjectionNotification)
	// StartNotificationProcessing is called when a notification processing is started
	// It saves start time for an event on aggregate when it's picked to be processed by background processor
	StartNotificationProcessing(notification *ProjectionNotification)
	// FinishNotificationProcessing is called when a notification processing is finished
	// It actually sends metrics calculating duration for which a notification spends in queue and then processed by background processor
	FinishNotificationProcessing(notification *ProjectionNotification, success bool)
}

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

type NotificationQueuer interface {
	Open() func()

	Empty() bool
	Next(context.Context) (*ProjectionNotification, bool)

	Queue(context.Context, *ProjectionNotification) error
	ReQueue(context.Context, *ProjectionNotification) error
}

type PersistenceStrategy interface {
	CreateSchema(tableName string) []string
	// EventColumnsNames represent the event store columns selected from the event stream table. Used by PrepareSearch
	EventColumnNames() []string
	// InsertColumnNames represent the ordered event store columns that are used to insert data into the event stream.
	InsertColumnNames() []string
	PrepareData([]goengine.Message) ([]interface{}, error)
	PrepareSearch(metadata.Matcher) ([]byte, []interface{})
	GenerateTableName(streamName goengine.StreamName) (string, error)
}

type ProcessHandler func(context.Context, *ProjectionNotification, ProjectionTrigger) error

type ProjectionErrorAction int

type ProjectionErrorCallback func(err error, notification *ProjectionNotification) ProjectionErrorAction

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

type ProjectionNotification struct {
	No          int64     `json:"no"`
	AggregateID string    `json:"aggregate_id"`
	ValidAfter  time.Time `json:"valid_after"`
}

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

type ProjectionRawState struct {
	Position        int64
	ProjectionState []byte
}

type ProjectionState struct {
	Position        int64
	ProjectionState interface{}
}

type ProjectionStateSerialization interface {
	// Init initializes the state
	Init(ctx context.Context) (interface{}, error)

	// DecodeState reconstitute the projection state based on the provided state data
	DecodeState(data []byte) (interface{}, error)

	// EncodeState encode the given object for storage
	EncodeState(obj interface{}) ([]byte, error)
}

type ProjectionTrigger func(ctx context.Context, notification *ProjectionNotification) error

type ProjectorStorage interface {
	// Acquire this function is used to acquire the projection and it's projectionState
	// A projection can only be acquired once and must be released using the returned func
	Acquire(ctx context.Context, conn *sql.Conn, notification *ProjectionNotification) (ProjectorTransaction, int64, error)
}

type ProjectorTransaction interface {
	AcquireState(ctx context.Context) (ProjectionState, error)
	CommitState(ProjectionState) error

	Close() error
}

type Queryer interface {
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
}

type ReadOnlyEventStore interface {
	// LoadWithConnection returns a eventstream based on the provided constraints using the provided Queryer
	LoadWithConnection(ctx context.Context, conn Queryer, streamName goengine.StreamName, fromNumber int64, count *uint, metadataMatcher metadata.Matcher) (goengine.EventStream, error)
}

type StreamProjector struct {
	sync.Mutex
	// contains filtered or unexported fields
}

type StreamProjectorStorage interface {
	ProjectorStorage

	CreateProjection(ctx context.Context, conn Execer) error
}