eventstreams

type CheckpointsTuningConfig struct {
	Asynchronous            bool  `ffstruct:"CheckpointsConfig" json:"asynchronous"`
	UnmatchedEventThreshold int64 `ffstruct:"CheckpointsConfig" json:"unmatchedEventThreshold"`
}

type Config[CT EventStreamSpec, DT any] struct {
	TLSConfigs        map[string]*fftls.Config `ffstruct:"EventStreamConfig" json:"tlsConfigs,omitempty"`
	Retry             *retry.Retry             `ffstruct:"EventStreamConfig" json:"retry,omitempty"`
	DisablePrivateIPs bool                     `ffstruct:"EventStreamConfig" json:"disabledPrivateIPs"`
	Checkpoints       CheckpointsTuningConfig  `ffstruct:"EventStreamConfig" json:"checkpoints"`
	Defaults          EventStreamDefaults      `ffstruct:"EventStreamConfig" json:"defaults,omitempty"`

	// Allow plugging in additional types (important that the embedding code adds the FFEnum doc entry for the EventStreamType)
	AdditionalDispatchers map[EventStreamType]DispatcherFactory[CT, DT] `json:"-"`
}

type ConfigWebhookDefaults struct {
	ffresty.HTTPConfig
}

type ConfigWebsocketDefaults struct {
	DefaultDistributionMode DistributionMode `ffstruct:"EventStreamConfig" json:"distributionMode"`
}

type DBSerializable interface {
	sql.Scanner
	driver.Valuer
}

type Deliver[DT any] func(events []*Event[DT]) SourceInstruction

type DispatchStatus = fftypes.FFEnum

type Dispatcher[DT any] interface {
	AttemptDispatch(ctx context.Context, attempt int, events *EventBatch[DT]) error
}

type DispatcherFactory[CT EventStreamSpec, DT any] interface {
	Validate(ctx context.Context, conf *Config[CT, DT], spec CT, tlsConfigs map[string]*tls.Config, phase LifecyclePhase) error
	NewDispatcher(ctx context.Context, conf *Config[CT, DT], spec CT) Dispatcher[DT]
}

type DistributionMode = fftypes.FFEnum

type ErrorHandlingType = fftypes.FFEnum

type Event[DataType any] struct {
	EventCommon
	// Data can be anything to deliver for the event - must be JSON marshalable.
	// Will be flattened into the struct.
	// Can define topic and/or sequenceId, but these will overridden with EventCommon strings in the JSON serialization.
	Data *DataType `json:"-"`
}

type EventBatch[DataType any] struct {
	wsserver.BatchHeader
	Type   string             `json:"type"`   // always MessageTypeEventBatch (for consistent WebSocket flow control)
	Events []*Event[DataType] `json:"events"` // an array of events allows efficient batch acknowledgment
}

type EventCommon struct {
	Topic      string `json:"topic,omitempty"` // describes the sub-stream of events (optional) allowing sever-side event filtering (regexp)
	SequenceID string `json:"sequenceId"`      // deterministic ID for the event, that must be alpha-numerically orderable within the stream (numbers must be left-padded hex/decimal strings for ordering)
}

type EventStreamActions[CT any] interface {
	Start(ctx context.Context) error
	Stop(ctx context.Context) error
	Status(ctx context.Context) CT
}

type EventStreamCheckpoint struct {
	ID         *string         `ffstruct:"EventStreamCheckpoint" json:"id"`
	Created    *fftypes.FFTime `ffstruct:"EventStreamCheckpoint" json:"created"`
	Updated    *fftypes.FFTime `ffstruct:"EventStreamCheckpoint" json:"updated"`
	SequenceID *string         `ffstruct:"EventStreamCheckpoint" json:"sequenceId,omitempty"`
}

type EventStreamDefaults struct {
	ErrorHandling     ErrorHandlingType       `ffstruct:"EventStreamDefaults" json:"errorHandling"`
	BatchSize         int                     `ffstruct:"EventStreamDefaults" json:"batchSize"`
	BatchTimeout      fftypes.FFDuration      `ffstruct:"EventStreamDefaults" json:"batchTimeout"`
	RetryTimeout      fftypes.FFDuration      `ffstruct:"EventStreamDefaults" json:"retryTimeout"`
	BlockedRetryDelay fftypes.FFDuration      `ffstruct:"EventStreamDefaults" json:"blockedRetryDelay"`
	WebSocketDefaults ConfigWebsocketDefaults `ffstruct:"EventStreamDefaults" json:"webSockets,omitempty"`
	WebhookDefaults   ConfigWebhookDefaults   `ffstruct:"EventStreamDefaults" json:"webhooks,omitempty"`
}

type EventStreamSpec interface {
	dbsql.Resource
	SetID(s string)
	ESFields() *EventStreamSpecFields
	ESType() EventStreamType         // separated from fields to allow choice on restrictions
	WebhookConf() *WebhookConfig     // can return nil if Webhooks not supported
	WebSocketConf() *WebSocketConfig // can return nil if WebSockets not supported
	IsNil() bool                     // needed as quirk of using interfaces with generics
}

type EventStreamSpecFields struct {
	Name              *string            `ffstruct:"eventstream" json:"name,omitempty"`
	Status            *EventStreamStatus `ffstruct:"eventstream" json:"status,omitempty"`
	InitialSequenceID *string            `ffstruct:"eventstream" json:"initialSequenceId,omitempty"`
	TopicFilter       *string            `ffstruct:"eventstream" json:"topicFilter,omitempty"`

	ErrorHandling     *ErrorHandlingType  `ffstruct:"eventstream" json:"errorHandling"`
	BatchSize         *int                `ffstruct:"eventstream" json:"batchSize"`
	BatchTimeout      *fftypes.FFDuration `ffstruct:"eventstream" json:"batchTimeout"`
	RetryTimeout      *fftypes.FFDuration `ffstruct:"eventstream" json:"retryTimeout"`
	BlockedRetryDelay *fftypes.FFDuration `ffstruct:"eventstream" json:"blockedRetryDelay"`
	// contains filtered or unexported fields
}

type EventStreamStatistics struct {
	StartTime            *fftypes.FFTime `ffstruct:"EventStreamStatistics" json:"startTime"`
	LastDispatchTime     *fftypes.FFTime `ffstruct:"EventStreamStatistics" json:"lastDispatchTime"`
	LastDispatchAttempts int             `ffstruct:"EventStreamStatistics" json:"lastDispatchAttempts,omitempty"`
	LastDispatchFailure  string          `ffstruct:"EventStreamStatistics" json:"lastDispatchFailure,omitempty"`
	LastDispatchStatus   DispatchStatus  `ffstruct:"EventStreamStatistics" json:"lastDispatchComplete"`
	HighestDetected      string          `ffstruct:"EventStreamStatistics" json:"highestDetected"`
	HighestDispatched    string          `ffstruct:"EventStreamStatistics" json:"highestDispatched"`
	Checkpoint           string          `ffstruct:"EventStreamStatistics" json:"checkpoint"`
}

type EventStreamStatus = fftypes.FFEnum

type EventStreamType = fftypes.FFEnum

type GenericEventStream struct {
	dbsql.ResourceBase
	Type *EventStreamType `ffstruct:"eventstream" json:"type,omitempty" ffenum:"estype"`
	EventStreamSpecFields
	Webhook    *WebhookConfig         `ffstruct:"eventstream" json:"webhook,omitempty"`
	WebSocket  *WebSocketConfig       `ffstruct:"eventstream" json:"websocket,omitempty"`
	Statistics *EventStreamStatistics `ffstruct:"EventStream" json:"statistics,omitempty"`
}

type IDValidator func(ctx context.Context, idStr string) error

type LifecyclePhase int

type Manager[CT EventStreamSpec] interface {
	UpsertStream(ctx context.Context, nameOrID string, esSpec CT) (bool, error)
	GetStreamByID(ctx context.Context, id string, opts ...dbsql.GetOption) (CT, error)
	GetStreamByNameOrID(ctx context.Context, nameOrID string, opts ...dbsql.GetOption) (CT, error)
	ListStreams(ctx context.Context, filter ffapi.Filter) ([]CT, *ffapi.FilterResult, error)
	StopStream(ctx context.Context, nameOrID string) error
	StartStream(ctx context.Context, nameOrID string) error
	ResetStream(ctx context.Context, nameOrID string, sequenceID *string, preStartCallbacks ...func(ctx context.Context, spec CT) error) error
	DeleteStream(ctx context.Context, nameOrID string) error
	Close(ctx context.Context)
}

type Persistence[CT EventStreamSpec] interface {
	EventStreams() dbsql.CRUD[CT]
	Checkpoints() dbsql.CRUD[*EventStreamCheckpoint]
	IDValidator() IDValidator
	Close()
}

type Runtime[ConfigType EventStreamSpec, DataType any] interface {
	// Generate a new unique resource ID (such as a UUID)
	NewID() string
	// Return a COPY of the config object with runtime status and statistics (runtime enrichment)
	WithRuntimeStatus(spec ConfigType, status EventStreamStatus, stats *EventStreamStatistics) ConfigType
	// Type specific config validation goes here
	Validate(ctx context.Context, config ConfigType) error
	// The run function should execute in a loop detecting events until instructed to stop:
	// - The Run function should block when no events are available
	//   - Must detect if the context is closed (see below)
	// - The Deliver function will block if the stream is blocked:
	//   - Blocked means the previous batch is being processed, and the current batch is full
	// - If the stream stops, the Exit instruction will be returned from deliver
	// - The supplied context will be cancelled as well on exit, so should be used:
	//   1. In any blocking i/o functions
	//   2. To wake any sleeps early, such as batch polling scenarios
	// - If the function returns without an Exit instruction, it will be restarted from the last checkpoint
	Run(ctx context.Context, spec ConfigType, checkpointSequenceID string, deliver Deliver[DataType]) error
}

type SourceInstruction int

type WebSocketConfig struct {
	DistributionMode *DistributionMode `ffstruct:"wsconfig" json:"distributionMode,omitempty"`
}

type WebhookConfig struct {
	URL           *string             `ffstruct:"whconfig" json:"url,omitempty"`
	Method        *string             `ffstruct:"whconfig" json:"method,omitempty"`
	TLSConfigName *string             `ffstruct:"whconfig" json:"tlsConfigName,omitempty"`
	HTTP          *ffresty.HTTPConfig `ffstruct:"whconfig" json:"http,omitempty"`
	// contains filtered or unexported fields
}