model

type BaseRecord struct {
	// CheckpointID is the ID of the record.
	CheckpointID int64 `json:"checkpointId"`
	// BeginMessage.CommitTime.UnixNano(), 16 bytes smaller than time.Time
	CommitTimeNano int64 `json:"commitTimeNano"`
	// TransactionID is the `XID` corresponding to the transaction that committed this record.
	TransactionID uint64 `json:"transactionId"`
}

type CDCFlowSignal int32

type CDCStream[T Items] struct {

	// Schema changes from slot
	SchemaDeltas []*protos.TableSchemaDelta
	// contains filtered or unexported fields
}

type CdcCheckpoint struct {
	Text string
	ID   int64
}

type CdcColumnNumericTruncator struct {
	Stat *qvalue.NumericStat
}

type CdcTableNumericTruncator struct {
	TruncatorsByColumn map[string]CdcColumnNumericTruncator
	DestinationTable   string
}

type DeleteRecord[T Items] struct {
	// Items is a map of column name to value.
	Items T
	// unchanged toast columns, filled from latest UpdateRecord
	UnchangedToastColumns map[string]struct{}
	// Name of the source table
	SourceTableName string
	// Name of the destination table
	DestinationTableName string
	BaseRecord
}

type HeaderProvider interface {
	GetHeaders(ctx context.Context) (http.Header, error)
}

type InsertRecord[T Items] struct {
	// Items is a map of column name to value.
	Items T
	// Name of the source table
	SourceTableName string
	// Name of the destination table
	DestinationTableName string
	// CommitID is the ID of the commit corresponding to this record.
	CommitID int64
	BaseRecord
}

type Items interface {
	json.Marshaler
	UpdateIfNotExists(Items) []string
	UpdateWithBaseRecord(BaseRecord)
	GetBytesByColName(string) ([]byte, error)
	ToJSONWithOptions(ToJSONOptions) (string, error)
	DeleteColName(string)
}

type MessageRecord[T Items] struct {
	Prefix  string
	Content string
	BaseRecord
}

type NameAndExclude struct {
	Exclude map[string]struct{}
	Name    string
}

type NormalizeRecordsRequest struct {
	Env                    map[string]string
	TableNameSchemaMapping map[string]*protos.TableSchema
	FlowJobName            string
	SoftDeleteColName      string
	SyncedAtColName        string
	TableMappings          []*protos.TableMapping
	SyncBatchID            int64
	Version                uint32
}

type NormalizeResponse struct {
	StartBatchID int64
	EndBatchID   int64
}

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

type Object struct {
	URL  string
	Size int64
}

type PgItems struct {
	ColToVal map[string][]byte
}

type PullRecordsRequest[T Items] struct {
	// record batch for pushing changes into
	RecordStream *CDCStream[T]
	// ConsumedOffset can be reported as committed to reduce slot size
	ConsumedOffset *atomic.Int64
	// FlowJobName is the name of the flow job.
	FlowJobName string
	// relId to name Mapping
	SrcTableIDNameMapping map[uint32]string
	// source to destination table name mapping
	TableNameMapping map[string]NameAndExclude
	// tablename to schema mapping
	TableNameSchemaMapping map[string]*protos.TableSchema
	// overrides dynamic configuration
	Env map[string]string
	// override publication name
	OverridePublicationName string
	// override replication slot name
	OverrideReplicationSlotName string
	// LastOffset is the latest LSN that was synced.
	LastOffset CdcCheckpoint
	// MaxBatchSize is the max number of records to fetch.
	MaxBatchSize uint32
	// peerdb versioning to prevent breaking changes
	InternalVersion uint32
	// IdleTimeout is the timeout to wait for new records.
	IdleTimeout time.Duration
}

type QObjectStream struct {
	Objects chan *Object
	// contains filtered or unexported fields
}

type QObjectStreamFormat string

type QRecordAvroChunkSizeTracker struct {
	TrackUncompressed bool
	Bytes             atomic.Int64
}

type QRecordAvroConverter struct {
	Schema                   *QRecordAvroSchemaDefinition
	ColNames                 []string
	TargetDWH                protos.DBType
	UnboundedNumericAsString bool
	NullMismatchTracker      *NullMismatchTracker
	// contains filtered or unexported fields
}

type QRecordAvroField struct {
	Type any    `json:"type"`
	Name string `json:"name"`
}

type QRecordAvroSchema struct {
	Type   string             `json:"type"`
	Name   string             `json:"name"`
	Fields []QRecordAvroField `json:"fields"`
}

type QRecordAvroSchemaDefinition struct {
	Schema *avro.RecordSchema
	Fields []types.QField
}

type QRecordBatch struct {
	Schema  types.QRecordSchema
	Records [][]types.QValue
}

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

type QRecordStream struct {
	Records chan []types.QValue
	// contains filtered or unexported fields
}

type Record[T Items] interface {
	Kind() string
	GetCheckpointID() int64
	GetCommitTime() time.Time
	GetTransactionID() uint64
	GetDestinationTableName() string
	GetSourceTableName() string
	// get columns and values for the record
	GetItems() T
	PopulateCountMap(mapOfCounts map[string]*RecordTypeCounts)
}

type RecordItems struct {
	ColToVal               map[string]types.QValue
	TruncateThresholdBytes int
}

type RecordTypeCounts struct {
	InsertCount atomic.Int32
	UpdateCount atomic.Int32
	DeleteCount atomic.Int32
}

type RecordsToStreamRequest[T Items] struct {
	TableMapping             map[string]*RecordTypeCounts
	BatchID                  int64
	UnboundedNumericAsString bool
	TargetDWH                protos.DBType
	// contains filtered or unexported fields
}

type RelationMessageMapping map[uint32]*pglogrepl.RelationMessage

type RelationRecord[T Items] struct {
	TableSchemaDelta *protos.TableSchemaDelta `json:"tableSchemaDelta"`
	BaseRecord
}

type RemoveFlowDetailsFromCatalogRequest struct {
	FlowName string
	Resync   bool
}

type SetupReplicationResult struct {
	Conn             interface{ Close(context.Context) error }
	SlotName         string
	SnapshotName     string
	SupportsTIDScans bool
}

type SnapshotTableNumericTruncator []qvalue.NumericStat

type StreamNumericTruncator map[string]CdcTableNumericTruncator

type SyncCompositeResponse struct {
	SyncResponse   *SyncResponse
	NeedsNormalize bool
}

type SyncRecordsRequest[T Items] struct {
	Records *CDCStream[T]
	// ConsumedOffset allows destination to confirm lsn for slot
	ConsumedOffset *atomic.Int64
	// FlowJobName is the name of the flow job.
	FlowJobName string
	// destination table name -> schema mapping
	TableNameSchemaMapping map[string]*protos.TableSchema
	Env                    map[string]string
	// Staging path for AVRO files in CDC
	StagingPath string
	// Lua script
	Script string
	// source:destination mappings
	TableMappings []*protos.TableMapping
	SyncBatchID   int64
	Version       uint32
}

type SyncResponse struct {
	// TableNameRowsMapping tells how many records need to be synced to each destination table.
	TableNameRowsMapping map[string]*RecordTypeCounts
	// to be carried to parent workflow
	TableSchemaDeltas []*protos.TableSchemaDelta
	// LastSyncedCheckpoint is the last state (eg LSN, GTID) that was synced.
	LastSyncedCheckpoint CdcCheckpoint
	// NumRecordsSynced is the number of records that were synced.
	NumRecordsSynced   int64
	CurrentSyncBatchID int64
	Warnings           shared.QRepWarnings
}

type TableWithPkey struct {
	TableName string
	// SHA256 hash of the primary key columns
	PkeyColVal [32]byte
}

type ToJSONOptions struct {
	UnnestColumns map[string]struct{}
	HStoreAsJSON  bool
}

type TypedReceiveChannel[T any] struct {
	Chan workflow.ReceiveChannel
}

type TypedSignal[T any] struct {
	Name string
}

type UpdateRecord[T Items] struct {
	// OldItems is a map of column name to value.
	OldItems T
	// NewItems is a map of column name to value.
	NewItems T
	// unchanged toast columns
	UnchangedToastColumns map[string]struct{}
	// Name of the source table
	SourceTableName string
	// Name of the destination table
	DestinationTableName string
	BaseRecord
}