scyllacdc

type AdvancedReaderConfig struct {
	// ConfidenceWindowSize defines a minimal age a change must have in order
	// to be read.
	//
	// Due to the eventually consistent nature of Scylla, newer writes may
	// appear in CDC log earlier than some older writes. This can cause the
	// Reader to skip the older write, therefore the need for this parameter.
	//
	// If the parameter is left as 0, the library will automatically choose
	// a default confidence window size.
	ConfidenceWindowSize time.Duration

	// The library uses select statements to fetch changes from CDC Log tables.
	// Each select fetches changes from a single table and fetches only changes
	// from a limited set of CDC streams. If such select returns one or more
	// changes then next select to this table and set of CDC streams will be
	// issued after a delay. This parameter specifies the length of the delay.
	//
	// If the parameter is left as 0, the library will automatically adjust
	// the length of the delay.
	PostNonEmptyQueryDelay time.Duration

	// The library uses select statements to fetch changes from CDC Log tables.
	// Each select fetches changes from a single table and fetches only changes
	// from a limited set of CDC streams. If such select returns no changes then
	// next select to this table and set of CDC streams will be issued after
	// a delay. This parameter specifies the length of the delay.
	//
	// If the parameter is left as 0, the library will automatically adjust
	// the length of the delay.
	PostEmptyQueryDelay time.Duration

	// If the library tries to read from the CDC log and the read operation
	// fails, it will wait some time before attempting to read again. This
	// parameter specifies the length of the delay.
	//
	// If the parameter is left as 0, the library will automatically adjust
	// the length of the delay.
	PostFailedQueryDelay time.Duration

	// Changes are queried using select statements with restriction on the time
	// those changes appeared. The restriction is bounding the time from both
	// lower and upper bounds. This parameter defines the width of the time
	// window used for the restriction.
	//
	// If the parameter is left as 0, the library will automatically adjust
	// the size of the restriction window.
	QueryTimeWindowSize time.Duration

	// When the library starts for the first time it has to start consuming
	// changes from some point in time. This parameter defines how far in the
	// past it needs to look. If the value of the parameter is set to an hour,
	// then the library will only read historical changes that are no older than
	// an hour.
	//
	// Note of caution: data in CDC Log table is automatically deleted so
	// setting this parameter to something bigger than TTL used on CDC Log won’t
	// cause changes older than this TTL to appear.
	//
	// If the parameter is left as 0, the library will automatically adjust
	// the size of the restriction window.
	ChangeAgeLimit time.Duration
}

type AtomicChange struct {
	// Value contains the scalar value of the column.
	// If the column was not changed or was deleted, it will be nil.
	//
	// Type: T.
	Value interface{}

	// IsDeleted tells if this column was set to NULL by this change.
	IsDeleted bool
}

type Change struct {
	// Corresponds to cdc$stream_id.
	StreamID StreamID

	// Corresponds to cdc$time.
	Time gocql.UUID

	// PreImage rows of the group.
	PreImage []*ChangeRow

	// Delta rows of the group.
	Delta []*ChangeRow

	// PostImage rows of the group.
	PostImage []*ChangeRow
}

type ChangeConsumer interface {
	// Processes a change from the CDC log associated with the stream of
	// the ChangeConsumer. This method is called in a sequential manner for each
	// row that appears in the stream.
	//
	// If this method returns an error, the library will stop with an error.
	Consume(ctx context.Context, change Change) error

	// Called after all rows from the stream were consumed, and the reader
	// is about to switch to a new generation, or stop execution altogether.
	//
	// If this method returns an error, the library will stop with an error.
	End() error
}

type ChangeConsumerFactory interface {
	// Creates a change consumer with given parameters.
	//
	// If this method returns an error, the library will stop with an error.
	CreateChangeConsumer(ctx context.Context, input CreateChangeConsumerInput) (ChangeConsumer, error)
}

type ChangeConsumerFunc func(ctx context.Context, tableName string, change Change) error

type ChangeOrEmptyNotificationConsumer interface {
	ChangeConsumer

	// Invoked upon empty results from the CDC log associated with the stream of
	// the ChangeConsumer. This method is called to acknowledge a query window
	// has been executed against the stream and the CDC log is to be considered
	// completed as of 'ackTime' param passed.
	//
	// If this method returns an error, the library will stop with an error.
	Empty(ctx context.Context, ackTime gocql.UUID) error
}

type ChangeRow struct {
	// contains filtered or unexported fields
}

type CreateChangeConsumerInput struct {
	// Name of the table from which the new ChangeConsumer will receive changes.
	TableName string

	// ID of the stream from which the new ChangeConsumer will receive changes.
	StreamID StreamID

	ProgressReporter *ProgressReporter
}

type ListChange struct {
	// AppendedElements contains values appended to the list in the form
	// of map from cell timestamps to values.
	//
	// For more information about how to interpret it, see "Advanced column"
	// types" in the CDC documentation.
	//
	// Type: map[gocql.UUID]T
	AppendedElements interface{}

	// RemovedElements contains indices of the removed elements.
	//
	// For more information about how to interpret it, see "Advanced column"
	// types" in the CDC documentation.
	//
	// Type: []gocql.UUID
	RemovedElements []gocql.UUID

	// IsReset tells if the list value was overwritten instead of being
	// appended to or removed from. If it's true, than AppendedValue will
	// contain the new state of the list (which can be NULL).
	IsReset bool
}

type Logger interface {
	Printf(format string, v ...interface{})
}

type MapChange struct {
	// AddedElements contains a map of elements which were added to the map
	// by the operation.
	//
	// Type: map[K]V.
	AddedElements interface{}

	// RemovedElements contains a slice of keys which were removed from the map
	// by the operation.
	// Please note that if the operation overwrote the old value of the map
	// instead of adding/removing elements, this field _will be nil_.
	// Instead, IsReset field will be set, and AddedValues will contain
	// the new state of the map.
	//
	// Type: []K
	RemovedElements interface{}

	// IsReset tells if the map value was overwritten instead of being
	// appended to or removed from. If it's true, than AddedElements will
	// contain the new state of the map (which can be NULL).
	IsReset bool
}

type OperationType int8

type PeriodicProgressReporter struct {
	// contains filtered or unexported fields
}

type Progress struct {
	// LastProcessedRecordTime represents the value of the cdc$time column
	// of the last processed record in the stream.
	LastProcessedRecordTime gocql.UUID
}

type ProgressManager interface {
	// GetCurrentGeneration returns the time of the generation that was
	// last saved by StartGeneration. The library will call this function
	// at the beginning in order to determine from which generation it should
	// start reading first.
	//
	// If there is no information available about the time of the generation
	// from which reading should start, GetCurrentGeneration can return
	// a zero time value. In that case, reading will start from the point
	// determined by AdvancedReaderConfig.ChangeAgeLimit.
	//
	// If this function returns an error, the library will stop with an error.
	GetCurrentGeneration(ctx context.Context) (time.Time, error)

	// StartGeneration is called after all changes have been read from the
	// previous generation and the library is about to start processing
	// the next one. The ProgressManager should save this information so that
	// GetCurrentGeneration will return it after the library is restarted.
	//
	// If this function returns an error, the library will stop with an error.
	StartGeneration(ctx context.Context, gen time.Time) error

	// GetProgress retrieves information about the progress of given stream,
	// in a given table. If there was no progress saved for this stream
	// during this generation, GetProgress can return a zero time value
	// and the library will start processing changes from the stream
	// starting from the beginning of the generation.
	//
	// This method needs to be thread-safe, as the library is allowed to
	// call it concurrently for different combinations of `table` and `streamID`.
	// The library won't issue concurrent calls to this method with the same
	// `table` and `streamID` parameters.
	//
	// If this function returns an error, the library will stop with an error.
	GetProgress(ctx context.Context, gen time.Time, table string, streamID StreamID) (Progress, error)

	// SaveProgress stores information about the last cdc log record which was
	// processed successfully. If the reader is restarted, it should resume
	// work for this stream starting from the row _after_ the last saved
	// timestamp.
	//
	// This method is only called by ChangeConsumers, indirectly through
	// the ProgressReporter struct. Within a generation, ChangeConsumers
	// are run concurrently, therefore SaveProgress should be safe to call
	// concurrently.
	//
	// Contrary to other methods, an error returned does not immediately
	// result in the library stopping with an error. The error is propagated
	// to the ChangeConsumer, and it can decide what to do with the error next.
	SaveProgress(ctx context.Context, gen time.Time, table string, streamID StreamID, progress Progress) error
}

type ProgressManagerWithStartTime interface {
	ProgressManager

	// GetApplicationReadStartTime returns the timestamp from which
	// the application started reading data. The library uses this timestamp
	// as a lower bound to determine where it should start reading. For example,
	// if there is no generation saved or there is no progress information
	// saved for a stream, reading will be restarted from the given timestamp
	// (or higher if the generation timestamp is higher).
	//
	// If this function returns a zero timeuuid, the library will start reading
	// from `time.Now() - AdvancedReaderConfig.ChangeAgeLimit`.
	// If this function returns an error, the library will stop with an error.
	GetApplicationReadStartTime(ctx context.Context) (time.Time, error)

	// SaveApplicationReadStartTime stores information about the timestamp
	// from which the application originally started reading data.
	// It is called by the library if there was no start timestamp saved.
	//
	// If this function returns an error, the library will stop with an error.
	SaveApplicationReadStartTime(ctx context.Context, startTime time.Time) error
}

type ProgressReporter struct {
	// contains filtered or unexported fields
}

type Reader struct {
	// contains filtered or unexported fields
}

type ReaderConfig struct {
	// An active gocql session to the cluster.
	Session *gocql.Session

	// Names of the tables for which to read changes. This should be the name
	// of the base table, not the cdc log table.
	// Can be prefixed with keyspace name.
	TableNames []string

	// Consistency to use when querying CDC log.
	// If not specified, QUORUM consistency will be used.
	Consistency gocql.Consistency

	// Creates ChangeProcessors, which process information fetched from the CDC log.
	// A callback which processes information fetched from the CDC log.
	ChangeConsumerFactory ChangeConsumerFactory

	// An object which allows the reader to read and write information about
	// current progress.
	ProgressManager ProgressManager

	// A logger. If set, it will receive log messages useful for debugging of the library.
	Logger Logger

	// Advanced parameters.
	Advanced AdvancedReaderConfig
}

type SetChange struct {
	// AddedElements contains a slice of values which were added to the set
	// by the operation. If there were any values added, it will contain
	// a slice of form []T, where T is gocql's representation of the element
	// type.
	//
	// Type: []T
	AddedElements interface{}

	// RemovedElements contains a slice of values which were removed from the set
	// by the operation. Like AddedValues, it's either a slice or a nil
	// interface.
	//
	// Please note that if the operation overwrote the old value of the set
	// instead of adding/removing elements, this field _will be nil_.
	// Instead, IsReset field will be set, and AddedValues will contain
	// the new state of the set.
	//
	// Type: []T
	RemovedElements interface{}

	// IsReset tells if the set value was overwritten instead of being
	// appended to or removed from. If it's true, than AddedElements will
	// contain the new state of the set (which can be NULL).
	IsReset bool
}

type StreamID []byte

type TableBackedProgressManager struct {
	// contains filtered or unexported fields
}

type UDTChange struct {
	// AddedFields contains a map of fields. Non-null value of a field
	// indicate that the field was written to, otherwise it was not written.
	AddedFields map[string]interface{}

	// RemovedFields contains names of fields which were set to null
	// by this operation.
	RemovedFields []string

	// RemovedFieldsIndices contains indices of tields which were set to null
	// by this operation.
	RemovedFieldsIndices []int16

	// IsReset tells if the UDT was overwritten instead of only some fields
	// being overwritten. If this flag is true, then nil fields in AddedFields
	// will mean that those fields should be set to null.
	IsReset bool
}