position_store

type MongoPosition struct {
	StartPosition   config.MongoPosition `json:"start_position" bson:"start_position"`
	CurrentPosition config.MongoPosition `json:"current_position" bson:"current_position"`
}

type MongoPositionRet struct {
	Version    string `json:"version" bson:"version"`
	Name       string `json:"name" bson:"name"`
	Stage      string `json:"stage" bson:"stage"`
	Value      string `json:"value" bson:"value"`
	LastUpdate string `json:"last_update" bson:"last_update"`
}

type Position struct {
	PositionMeta
	Value interface{} `bson:"-" json:"-"`
}

type PositionCacheInterface interface {
	Start() error
	Close()
	Put(position Position) error

	// Get will get a value from cache, if there is no value inside the cache
	// it will try to get it from position repo
	Get() (position Position, exist bool, err error)

	GetEncodedPersistentPosition() (position PositionMeta, v string, exist bool, err error)
	Flush() error
	Clear() error
}

type PositionEntity struct {
	Name          string `json:"name" bson:"name"`
	Stage         string `json:"stage" bson:"stage"`
	MongoPosition `json:",inline" bson:",inline"`
	LastUpdate    string `json:"last_update" bson:"last_update"`
}

type PositionMeta struct {
	// Version is the schema version of position
	Version string `bson:"version" json:"version"`
	// Name is the unique name of a pipeline
	Name       string
	Stage      config.InputMode
	UpdateTime time.Time
}

type PositionRepo interface {
	Get(pipelineName string) (PositionMeta, string, bool, error)
	Put(pipelineName string, positionMeta PositionMeta, v string) error
	Delete(pipelineName string) error
	Close() error
}

type PositionValueDecoder func(s string) (interface{}, error)

type PositionValueEncoder func(v interface{}) (string, error)

type PositionWrapper struct {
	PositionMeta
	MongoValue string `bson:"value" json:"value"`
}