publication

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

type AObjectMapping struct {
	DownstreamStats objectio.ObjectStats
	IsTombstone     bool
	DBName          string
	TableName       string
	// RowOffsetMap maps original rowoffset to new rowoffset after sorting
	// Key: original rowoffset, Value: new rowoffset
	RowOffsetMap map[uint32]uint32
}

type AObjectMappingJSON struct {
	DownstreamStats string `json:"downstream_stats"` // ObjectStats as base64-encoded string
	IsTombstone     bool   `json:"is_tombstone"`
	DBName          string `json:"db_name"`
	TableName       string `json:"table_name"`
}

type ActiveRoutine struct {
	sync.Mutex
	Pause  chan struct{}
	Cancel chan struct{}
}

type AlterInplaceType int

type BackoffStrategy interface {
	// Next returns the wait duration before the given attempt (1-indexed).
	Next(attempt int) time.Duration
}

type CCPRConfig struct {

	// PublicationWorkerThread is the number of threads for publication worker pool
	PublicationWorkerThread int `toml:"publication-worker-thread"`
	// FilterObjectWorkerThread is the number of threads for filter object worker pool
	FilterObjectWorkerThread int `toml:"filter-object-worker-thread"`
	// GetChunkWorkerThread is the number of threads for get chunk worker pool
	GetChunkWorkerThread int `toml:"get-chunk-worker-thread"`
	// WriteObjectWorkerThread is the number of threads for write object worker pool
	WriteObjectWorkerThread int `toml:"write-object-worker-thread"`

	// GetChunkMaxMemory is the maximum memory for concurrent GetChunkJob operations
	GetChunkMaxMemory int64 `toml:"get-chunk-max-memory"`
	// GetChunkSize is the size of each chunk
	GetChunkSize int64 `toml:"get-chunk-size"`

	// GCInterval is the interval for garbage collection
	GCInterval toml.Duration `toml:"gc-interval"`
	// GCTTL is the time-to-live for garbage collection items
	GCTTL toml.Duration `toml:"gc-ttl"`
	// SyncTaskInterval is the interval for sync task execution
	SyncTaskInterval toml.Duration `toml:"sync-task-interval"`
	// RetryTimes is the number of retry attempts for failed operations
	RetryTimes int `toml:"retry-times"`
	// RetryInterval is the interval between retries
	RetryInterval toml.Duration `toml:"retry-interval"`

	// SyncProtectionTTLDuration is the TTL duration for sync protection
	SyncProtectionTTLDuration toml.Duration `toml:"sync-protection-ttl-duration"`
	// SyncProtectionRenewInterval is the interval for renewing sync protection
	SyncProtectionRenewInterval toml.Duration `toml:"sync-protection-renew-interval"`

	// BloomFilterExpectedItems is the expected number of items in bloom filter
	BloomFilterExpectedItems int `toml:"bloom-filter-expected-items"`
	// BloomFilterFalsePositiveRate is the false positive rate for bloom filter
	BloomFilterFalsePositiveRate float64 `toml:"bloom-filter-false-positive-rate"`

	// RetryThreshold is the maximum number of retries before stopping
	RetryThreshold int `toml:"retry-threshold"`
}

type CCPRTxnCacheWriter interface {
	// WriteObject checks if an object needs to be written and registers it in the cache.
	// Does NOT write the file - caller should write the file when isNewFile=true.
	// Returns isNewFile (true if file needs to be written) and any error.
	WriteObject(ctx context.Context, objectName string, txnID []byte) (isNewFile bool, err error)
	// OnFileWritten is called after the file has been successfully written to fileservice.
	OnFileWritten(objectName string)
}

type CommitErrorClassifier struct{}

type DatabaseMetadata struct {
	DatID        uint64
	DatName      string
	DatCreateSQL sql.NullString
	AccountID    uint32
}

type DefaultClassifier struct{}

type DownstreamCommitClassifier struct {
	MultiClassifier
}

type DownstreamConnectionClassifier struct {
	MultiClassifier
}

type ErrorClassifier interface {
	// IsRetryable returns true if the error is transient and worth retrying.
	IsRetryable(error) bool
}

type ErrorMetadata struct {
	IsRetryable bool      // Whether this error is retryable
	RetryCount  int       // Number of retry attempts
	FirstSeen   time.Time // When first seen
	LastSeen    time.Time // When last seen
	Message     string    // Error message
}

type ExecutorRetryOption struct {
	RetryTimes    int           // Maximum number of retries (-1 for infinite)
	RetryInterval time.Duration // Base interval between retries
	RetryDuration time.Duration // Maximum total retry duration
}

type ExponentialBackoff struct {
	// Base delay before the first retry attempt.
	Base time.Duration
	// Factor to multiply delays by each attempt (>1).
	Factor float64
	// Max caps the computed delay (optional).
	Max time.Duration
	// Jitter adds randomization in range [0, Jitter] (optional).
	Jitter time.Duration
	// contains filtered or unexported fields
}

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

type FilterObjectJobResult struct {
	Err              error
	HasMappingUpdate bool
	UpstreamAObjUUID *objectio.ObjectId
	PreviousStats    objectio.ObjectStats
	CurrentStats     objectio.ObjectStats
	// DownstreamStats holds the stats for non-appendable objects that were written to fileservice
	DownstreamStats objectio.ObjectStats
	// RowOffsetMap maps original rowoffset to new rowoffset after sorting
	// Key: original rowoffset, Value: new rowoffset
	RowOffsetMap map[uint32]uint32
}

type FilterObjectResult struct {
	HasMappingUpdate bool
	UpstreamAObjUUID *objectio.ObjectId
	PreviousStats    objectio.ObjectStats
	CurrentStats     objectio.ObjectStats
	// DownstreamStats holds the stats for non-appendable objects that were written to fileservice
	DownstreamStats objectio.ObjectStats
	// RowOffsetMap maps original rowoffset to new rowoffset after sorting
	// Key: original rowoffset, Value: new rowoffset
	RowOffsetMap map[uint32]uint32
}

type FilterObjectWorker interface {
	SubmitFilterObject(job Job) error
	Stop()
}

type GCRunningClassifier struct{}

type GCStatus struct {
	Running     bool  `json:"running"`
	Protections int   `json:"protections"`
	TS          int64 `json:"ts"`
}

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

type GetChunkJobDuration struct {
	ObjectName string
	ChunkIndex int64
	Duration   time.Duration
}

type GetChunkJobInfo interface {
	GetObjectName() string
	GetChunkIndex() int64
}

type GetChunkJobResult struct {
	ChunkData  []byte
	ChunkIndex int64
	Err        error
}

type GetChunkWorker interface {
	SubmitGetChunk(job Job) error
	Stop()
}

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

type GetMetaJobResult struct {
	MetadataData []byte
	TotalSize    int64
	ChunkIndex   int64
	TotalChunks  int64
	IsComplete   bool
	Err          error
}

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

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

type IterationContext struct {
	// Task identification
	TaskID                  string
	SubscriptionName        string
	SubscriptionAccountName string // The subscription account name for FROM clause
	SrcInfo                 SrcInfo

	// Upstream connection
	LocalTxn         client.TxnOperator
	UpstreamExecutor SQLExecutor
	LocalExecutor    SQLExecutor

	// Execution state
	IterationLSN      uint64
	SubscriptionState int8 // subscription state: 0=running, 1=error, 2=pause, 3=dropped

	// Context information
	PrevSnapshotName    string
	PrevSnapshotTS      types.TS
	CurrentSnapshotName string
	CurrentSnapshotTS   types.TS
	// AObjectMap stores the mapping from upstream aobj to downstream object stats
	// This map is used to track appendable object transformations during CCPR sync
	AObjectMap         *AObjectMap
	TableIDs           map[TableKey]uint64
	IndexTableMappings map[string]string // Maps upstream_index_table_name to downstream_index_table_name
	ErrorMetadata      *ErrorMetadata    // Error metadata parsed from error_message
	IsStale            bool              // Whether the iteration is stale
}

type IterationContextJSON struct {
	// Task identification
	TaskID           string  `json:"task_id"`
	SubscriptionName string  `json:"subscription_name"`
	SrcInfo          SrcInfo `json:"src_info"`

	// Context information
	AObjectMap         map[string]AObjectMappingJSON `json:"aobject_map"` // AObjectMap as serializable map (key is upstream ObjectId as string)
	TableIDs           map[string]uint64             `json:"table_ids"`
	IndexTableMappings map[string]string             `json:"index_table_mappings"` // IndexTableMappings as serializable map (key is upstream_index_table_name, value is downstream_index_table_name)
}

type Job interface {
	Execute()
	WaitDone() any
	GetType() int8
}

type JobStats struct {
	FilterObjectPending   atomic.Int64
	FilterObjectRunning   atomic.Int64
	FilterObjectCompleted atomic.Int64

	GetChunkPending   atomic.Int64
	GetChunkRunning   atomic.Int64
	GetChunkCompleted atomic.Int64

	GetMetaPending   atomic.Int64
	GetMetaRunning   atomic.Int64
	GetMetaCompleted atomic.Int64

	WriteObjectPending   atomic.Int64
	WriteObjectRunning   atomic.Int64
	WriteObjectCompleted atomic.Int64
	// contains filtered or unexported fields
}

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

type MemoryStats struct {
	ObjectContentBytes    int64
	DecompressBufferBytes int64
	SortIndexBytes        int64
	RowOffsetMapBytes     int64
	TotalBytes            int64
	MaxBytes              int64
}

type MemoryType int

type MoCtlResponse struct {
	Method string             `json:"method"`
	Result []MoCtlResultEntry `json:"result"`
}

type MoCtlResultEntry struct {
	ReturnStr string `json:"ReturnStr"`
}

type MockResultScanner interface {
	Close() error
	Next() bool
	Scan(dest ...interface{}) error
	Err() error
}

type MultiClassifier []ErrorClassifier

type MySQLErrorClassifier struct{}

type ObjectStats struct {
	Stats       string   // Object stats information
	CreateAt    types.TS // Creation timestamp
	DeleteAt    types.TS // Deletion timestamp (0 if not deleted)
	IsTombstone bool     // Whether this is a tombstone (deleted object)
}

type ObjectWithTableInfo struct {
	Stats       objectio.ObjectStats
	DBName      string
	TableName   string
	IsTombstone bool
	Delete      bool
	FilterJob   *FilterObjectJob
}

type Operation func() error

type Policy struct {
	// MaxAttempts defines how many times the operation should be attempted in total.
	// Must be >= 1.
	MaxAttempts int

	// Backoff decides how long to wait between attempts. Optional; zero value means no backoff.
	Backoff BackoffStrategy

	// Classifier decides whether an error warrants another attempt. Optional; defaults to never retry.
	Classifier ErrorClassifier
}

type PublicationExecutorOption struct {
	GCInterval          time.Duration
	GCTTL               time.Duration
	SyncTaskInterval    time.Duration
	RetryOption         *ExecutorRetryOption    // Retry configuration for executor operations
	SQLExecutorRetryOpt *SQLExecutorRetryOption // Retry configuration for SQL executor operations
}

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

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

type SQLExecutor interface {
	Close() error
	Connect() error
	EndTxn(ctx context.Context, commit bool) error
	// ExecSQL executes a SQL statement with options
	// accountID: the account ID to use for tenant context; use InvalidAccountID if not applicable (e.g., for UpstreamExecutor)
	// useTxn: if true, execute within a transaction (must have active transaction for InternalSQLExecutor, not allowed for UpstreamExecutor)
	// if false, execute as autocommit (will create and commit transaction automatically for InternalSQLExecutor)
	// timeout: if > 0, each retry attempt will use a new context with this timeout; if 0, use the provided ctx directly
	// Returns: result, cancel function (may be nil if no timeout context was created), error
	// IMPORTANT: caller MUST call the cancel function after finishing with the result (if cancel is not nil)
	ExecSQL(ctx context.Context, ar *ActiveRoutine, accountID uint32, query string, useTxn bool, needRetry bool, timeout time.Duration) (*Result, context.CancelFunc, error)
	// ExecSQLInDatabase executes a SQL statement with a specific default database context
	// This is useful for DDL statements that may require a default database to be set
	ExecSQLInDatabase(ctx context.Context, ar *ActiveRoutine, accountID uint32, query string, database string, useTxn bool, needRetry bool, timeout time.Duration) (*Result, context.CancelFunc, error)
}

type SQLExecutorRetryOption struct {
	MaxRetries    int             // Maximum number of retries for retryable errors
	RetryInterval time.Duration   // Interval between retries
	Classifier    ErrorClassifier // Error classifier for retry logic
}

type SrcInfo struct {
	SyncLevel string // 'account', 'database', or 'table'
	DBName    string // Required for database/table level
	TableName string // Required for table level
	AccountID uint32 // Account ID for downstream operations
}

type StaleReadClassifier struct{}

type SyncProtectionResponse struct {
	Status  string `json:"status"`
	Code    string `json:"code,omitempty"`
	Message string `json:"message,omitempty"`
}

type SyncProtectionRetryOption struct {
	// InitialInterval is the initial retry interval (default: 1s)
	InitialInterval time.Duration
	// MaxTotalTime is the maximum total time to retry (default: 5min)
	// If 0 or negative, no retry is performed (fail immediately on first error)
	MaxTotalTime time.Duration
}

type SyncProtectionTTLClassifier struct{}

type TTLChecker func() bool

type TableDDLInfo struct {
	TableID         uint64
	TableCreateSQL  string
	Operation       int8     // DDLOperationCreate (1), DDLOperationAlter (2), DDLOperationDrop (3), DDLOperationAlterInplace (4), 0 means no operation
	AlterStatements []string // ALTER TABLE statements for inplace alter operations
}

type TableKey struct {
	DBName    string // Database name
	TableName string // Table name
}

type TableMetadata struct {
	RelID         uint64
	RelName       string
	RelDatabaseID uint64
	RelDatabase   string
	RelCreateSQL  sql.NullString
	AccountID     uint32
}

type TaskContext struct {
	TaskID string
	LSN    uint64
}

type TaskEntry struct {
	TaskID            string
	LSN               uint64
	State             int8       // iteration_state from mo_ccpr_log
	SubscriptionState int8       // subscription state: 0=running, 1=error, 2=pause, 3=dropped
	DropAt            *time.Time // drop timestamp from mo_ccpr_log
}

type UTHelper interface {
	// OnSnapshotCreated is called after a snapshot is created in the upstream
	OnSnapshotCreated(ctx context.Context, snapshotName string, snapshotTS types.TS) error
	// OnSQLExecFailed is called before each SQL execution attempt
	// errorCount: current error count for this SQL query (resets to 0 for each new query)
	// Returns: error to inject (nil means no error injection)
	OnSQLExecFailed(ctx context.Context, query string, errorCount int) error
}

type UTInjectionClassifier struct{}

type UpstreamConnConfig struct {
	Account  string // Account name (may be empty)
	User     string
	Password string
	Host     string
	Port     int
	Timeout  string
}

type UpstreamConnectionClassifier struct {
	MultiClassifier
}

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

type UpstreamSQLHelper interface {
	// HandleSpecialSQL checks if the SQL is a special statement and handles it directly
	// Returns (handled, result, error) where handled indicates if the statement was handled
	HandleSpecialSQL(ctx context.Context, query string) (bool, *Result, error)
}

type UpstreamSQLHelperFactory func(
	txnOp client.TxnOperator,
	engine engine.Engine,
	accountID uint32,
	executor executor.SQLExecutor,
	txnClient client.TxnClient,
) UpstreamSQLHelper

type Worker interface {
	Submit(taskID string, lsn uint64, state int8) error
	Stop()
	// RegisterSyncProtection registers a sync protection job for keepalive
	RegisterSyncProtection(jobID string, ttlExpireTS int64)
	// UnregisterSyncProtection unregisters a sync protection job
	UnregisterSyncProtection(jobID string)
	// GetSyncProtectionTTL returns the current TTL expiration timestamp for a job
	GetSyncProtectionTTL(jobID string) int64
}

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

type WriteObjectJobDuration struct {
	ObjectName string
	Size       int64
	Duration   time.Duration
}

type WriteObjectJobInfo interface {
	GetObjectName() string
	GetObjectSize() int64
}

type WriteObjectJobResult struct {
	Err error
}

type WriteObjectWorker interface {
	SubmitWriteObject(job Job) error
	Stop()
}