statestorage

type Manager interface {
	Start() error
	Stop() error
	Get() (map[string]*Offset, error)
	Set(
		key string, value *Offset,
	) error
	StateEncoder(
		name string, encoder encoding.BinaryMarshaler,
	) error
	StateDecoder(
		name string, decoder encoding.BinaryUnmarshaler,
	) (present bool, err error)
	SetEncodedState(
		name string, encodedState []byte,
	)
	EncodedState(
		name string,
	) (encodedState []byte, present bool)
	SnapshotContext() (*watermark.SnapshotContext, error)
	SnapshotContextTransaction(
		snapshotName string, createIfNotExists bool, transaction func(snapshotContext *watermark.SnapshotContext) error,
	) error
}

type Offset struct {
	Timestamp      time.Time   `json:"timestamp"`
	Snapshot       bool        `json:"snapshot"`
	SnapshotDone   bool        `json:"snapshot_done"`
	SnapshotName   *string     `json:"snapshot_name,omitempty"`
	SnapshotOffset int         `json:"snapshot_offset"`
	SnapshotKeyset []byte      `json:"snapshot_keyset"`
	LSN            pgtypes.LSN `json:"lsn"`
}

type StateEncoderFunc func() (data []byte, err error)

type Storage interface {
	Start() error
	Stop() error
	Save() error
	Load() error
	Get() (map[string]*Offset, error)
	Set(
		key string, value *Offset,
	) error
	StateEncoder(
		name string, encoder encoding.BinaryMarshaler,
	) error
	StateDecoder(
		name string, decoder encoding.BinaryUnmarshaler,
	) (present bool, err error)
	EncodedState(
		name string,
	) (encodedState []byte, present bool)
	SetEncodedState(
		name string, encodedState []byte,
	)
}

type StorageProvider = func(config *config.Config) (Storage, error)