api

type Bounded interface {
	SetEofIngest(eof EOFIngest)
}

type BytesCollector interface {
	Sink
	Collect(ctx StreamContext, item RawTuple) error
}

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

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

type Closable interface {
	Close(ctx StreamContext) error
}

type Connector interface {
	Connect(ctx StreamContext, sch StatusChangeHandler) error
}

type EOFIngest func(ctx StreamContext, msg string)

type ErrorIngest func(ctx StreamContext, err error)

type Function interface {
	// Validate The argument is a list of xsql.Expr
	Validate(args []any) error
	// Exec Execute the function, return the result and if execution is successful.
	// If execution fails, return the error and false.
	Exec(ctx FunctionContext, args []any) (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 HasDynamicProps interface {
	// DynamicProps return the transformed dynamic properties (typically in sink).
	// The transformation should be done in transform op
	DynamicProps(template string) (string, bool)
	AllProps() map[string]string
}

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 LookupBytesSource interface {
	Source
	// Lookup receive multiple rows of bytes
	Lookup(ctx StreamContext, fields []string, keys []string, values []any) ([][]byte, error)
}

type LookupSource interface {
	Source
	// Lookup receive lookup values to construct the query and return query results
	Lookup(ctx StreamContext, fields []string, keys []string, values []any) ([]map[string]any, error)
}

type MessageTuple interface {
	ReadonlyMessage
}

type MessageTupleList interface {
	RangeOfTuples(f func(index int, tuple MessageTuple) bool)
	Len() int
	ToMaps() []map[string]any
}

type MetaInfo interface {
	Meta(key, table string) (any, bool)
	Created() time.Time
	AllMeta() map[string]any
}

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 PullBytesSource interface {
	Source
	Pull(ctx StreamContext, trigger time.Time, ingest BytesIngest, ingestError ErrorIngest)
}

type PullTupleSource interface {
	Source
	Pull(ctx StreamContext, trigger time.Time, ingest TupleIngest, ingestError ErrorIngest)
}

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

type ReadonlyMessage interface {
	Value(key, table string) (any, bool)
	ToMap() map[string]any
}

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 StatusChangeHandler func(status string, message string)

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)
	EnableTracer(enabled bool)
	IsTraceEnabled() bool

	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 TupleCollector interface {
	Sink
	Collect(ctx StreamContext, item MessageTuple) error
	CollectList(ctx StreamContext, items MessageTupleList) error
}

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

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