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
	GetRunId() int
	GetRootPath() string

	WithMeta(ruleId string, opId string, store Store) StreamContext
	WithInstance(instanceId int) StreamContext
	WithRun(runId 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
}