api

type Bounded interface {
	SetEofIngest(eof EOFIngest)
}

type BytesCollector interface {
	Sink
	Collect(ctx StreamContext, item []byte) error
}

type BytesIngest func(ctx StreamContext, payload []byte, meta map[string]any, ts time.Time)

type BytesSource interface {
	Source
	Subscribe(ctx StreamContext, ingest BytesIngest) error
}

type Closable interface {
	Close(ctx StreamContext) error
}

type Connector interface {
	Connect(ctx StreamContext) error
}

type EOFIngest func(ctx StreamContext)

type Function interface {
	// Validate The argument is a list of xsql.Expr
	Validate(args []interface{}) error
	// Exec Execute the function, return the result and if execution is successful.
	// If execution fails, return the error and false.
	Exec(args []interface{}, ctx FunctionContext) (interface{}, bool)
	// IsAggregate If this function is an aggregate function. Each parameter of an aggregate function will be a slice
	IsAggregate() bool
}

type FunctionContext interface {
	StreamContext
	GetFuncId() int
}

type Logger interface {
	Debug(args ...interface{})
	Info(args ...interface{})
	Warn(args ...interface{})
	Error(args ...interface{})
	Debugf(format string, args ...interface{})
	Infof(format string, args ...interface{})
	Warnf(format string, args ...interface{})
	Errorf(format string, args ...interface{})
}

type LookupSource interface {
	// Open creates the connection to the external data source
	Open(ctx StreamContext) error
	// Configure Called during initialization. Configure the source with the data source(e.g. topic for mqtt) and the properties
	// read from the yaml
	Configure(datasource string, props map[string]interface{}) error
	// Lookup receive lookup values to construct the query and return query results
	Lookup(ctx StreamContext, fields []string, keys []string, values []interface{}) ([]Tuple, error)
	Closable
}

type MetaInfo interface {
	Meta() ReadonlyMessage
	Timestamp() time.Time
}

type ModuleInfo struct {
	Id          string
	Description string
	New         func() Nodelet
}

type Nodelet interface {
	// Provision is called when the node is created, usually setting the configs. Do not put time-consuming operations here.
	Provision(ctx StreamContext, configs map[string]any) error
	Closable
}

type RawTuple interface {
	Raw() []byte
	MetaInfo
}

type ReadonlyMessage interface {
	Get(key string) (value any, ok bool)
	Range(f func(key string, value any) bool)
	// ToMap todo remove after eliminate map
	ToMap() map[string]any
}

type ResendSink interface {
	Sink
	// CollectResend Called when the sink cache resend is triggered
	CollectResend(ctx StreamContext, data interface{}) error
}

type Rewindable interface {
	GetOffset() (any, error)
	Rewind(offset any) error
	ResetOffset(input map[string]any) error
}

type Sink interface {
	Nodelet
	Connector
}

type Source interface {
	Nodelet
	Connector
}

type Store interface {
	SaveState(checkpointId int64, opId string, state map[string]interface{}) error
	// SaveCheckpoint saves the whole checkpoint state into storage
	SaveCheckpoint(checkpointId int64) error
	GetOpState(opId string) (*sync.Map, error)
	Clean() error
}

type StreamContext interface {
	context.Context
	GetLogger() Logger
	GetRuleId() string
	GetOpId() string
	GetInstanceId() int
	GetRootPath() string

	WithMeta(ruleId string, opId string, store Store) StreamContext
	WithInstance(instanceId int) StreamContext
	WithCancel() (StreamContext, context.CancelFunc)

	SetError(e error)
	// IncrCounter State handling
	IncrCounter(key string, amount int) error
	GetCounter(key string) (int, error)
	PutState(key string, value interface{}) error
	GetState(key string) (interface{}, error)
	DeleteState(key string) error
	// ParseTemplate parse the template string with the given data
	ParseTemplate(template string, data interface{}) (string, error)
	// ParseJsonPath parse the jsonPath string with the given data
	ParseJsonPath(jsonPath string, data interface{}) (interface{}, error)
}

type Tuple interface {
	Message() ReadonlyMessage
	MetaInfo
}

type TupleCollector interface {
	Sink
	Collect(ctx StreamContext, item Tuple) error
	CollectList(ctx StreamContext, items []Tuple) error
}

type TupleIngest func(ctx StreamContext, data any, meta map[string]any, ts time.Time)

type TupleSource interface {
	Source
	Subscribe(ctx StreamContext, ingest TupleIngest) error
}