metadata

type CaptureObservation interface {
	Elector

	// Run runs
	// the `eclector.RunElection` and other background tasks.
	// controllerCallback will be called when the capture campaign as the controller.
	Run(
		ctx context.Context,
		controllerCallback func(context.Context, ControllerObservation) error,
	) error

	// Advance advances some changefeed progresses that are collected from processors.
	Advance(cp CaptureProgress) error

	// OwnerChanges fetch owner modifications.
	OwnerChanges() <-chan ScheduledChangefeed

	// OnOwnerLaunched create an owner observation for a changefeed owner.
	OnOwnerLaunched(cf ChangefeedUUID) OwnerObservation

	// PostOwnerRemoved inform the metadata storage when an owner exits.
	PostOwnerRemoved(cf ChangefeedUUID, taskPosition ChangefeedProgress) error
}

type CaptureProgress map[ChangefeedUUID]ChangefeedProgress

type ChangefeedIdent struct {
	// UUID is generated internally by TiCDC to distinguish between changefeeds with the same ID.
	// Note that it can't be specified by the user.
	UUID ChangefeedUUID `gorm:"column:uuid;type:bigint(20) unsigned;primaryKey" json:"uuid"`

	// Namespace and ID pair is unique in one ticdc cluster. And in the current implementation,
	// Namespace can only be set to `default`.
	Namespace string `gorm:"column:namespace;type:varchar(128);not null;uniqueIndex:namespace,priority:1" json:"namespace"`
	ID        string `gorm:"column:id;type:varchar(128);not null;uniqueIndex:namespace,priority:2" json:"id"`
}

type ChangefeedInfo struct {
	ChangefeedIdent

	UpstreamID uint64 `gorm:"column:upstream_id;type:bigint(20) unsigned;not null;index:upstream_id,priority:1" json:"upstream_id"`
	SinkURI    string `gorm:"column:sink_uri;type:text;not null" json:"sink_uri"`
	StartTs    uint64 `gorm:"column:start_ts;type:bigint(20) unsigned;not null" json:"start_ts"`
	TargetTs   uint64 `gorm:"column:target_ts;type:bigint(20) unsigned;not null" json:"target_ts"`
	// Note that pointer type is used here for compatibility with the old model, and config should never be nil in practice.
	Config *config.ReplicaConfig `gorm:"column:config;type:longtext;not null" json:"config"`
}

type ChangefeedProgress model.ChangeFeedStatus

type ChangefeedState struct {
	ChangefeedUUID ChangefeedUUID      `gorm:"column:changefeed_uuid;type:bigint(20) unsigned;primaryKey" json:"changefeed_uuid"`
	State          model.FeedState     `gorm:"column:state;type:text;not null" json:"state"`
	Warning        *model.RunningError `gorm:"column:warning;type:text" json:"warning"`
	Error          *model.RunningError `gorm:"column:error;type:text" json:"error"`
}

type ChangefeedUUID = uint64

type ControllerObservation interface {
	// CreateChangefeed creates a changefeed, UUID will be filled into the input ChangefeedInfo.
	CreateChangefeed(cf *ChangefeedInfo, up *model.UpstreamInfo) (ChangefeedIdent, error)

	// RemoveChangefeed removes a changefeed, will mark it as removed and stop the owner and processors asynchronizely.
	RemoveChangefeed(cf ChangefeedUUID) error

	// CleanupChangefeed cleans up a changefeed, will delete info, schdule and state metadata.
	CleanupChangefeed(cf ChangefeedUUID) error

	// RefreshCaptures Fetch the latest capture list in the TiCDC cluster.
	RefreshCaptures() (captures []*model.CaptureInfo, changed bool)

	// SetOwner Schedule a changefeed owner to a given target.
	// Notes:
	//   * the target capture can fetch the event by `OwnerChanges`.
	//   * target state can only be `SchedLaunched` or `SchedRemoving`.
	SetOwner(target ScheduledChangefeed) error

	// GetChangefeedSchedule Get current schedule of the given changefeed.
	GetChangefeedSchedule(cf ChangefeedUUID) (ScheduledChangefeed, error)

	// ScheduleSnapshot Get a snapshot of all changefeeds current schedule.
	ScheduleSnapshot() ([]ScheduledChangefeed, []*model.CaptureInfo, error)
}

type Elector interface {
	// Self tells the caller who am I.
	Self() *model.CaptureInfo

	// RunElection runs the elector to continuously campaign for leadership
	// until the context is canceled.
	// onTakeControl will be called when the capture campaign as the controller.
	RunElection(ctx context.Context, onTakeControl func(ctx context.Context) error) error

	// GetController returns the last observed controller whose lease is still valid.
	GetController() (*model.CaptureInfo, error)

	// GetCaptures queries some or all captures.
	GetCaptures(...model.CaptureID) ([]*model.CaptureInfo, error)
}

type OwnerObservation interface {
	// Self returns the changefeed info of the owner.
	Self() ChangefeedUUID

	// UpdateChangefeed updates changefeed metadata, must be called on a paused one.
	UpdateChangefeed(*ChangefeedInfo) error

	// PauseChangefeed pauses a changefeed.
	PauseChangefeed() error

	// ResumeChangefeed resumes a changefeed.
	ResumeChangefeed() error

	// SetChangefeedFinished set the changefeed to state finished.
	SetChangefeedFinished() error

	// SetChangefeedRemoved set the changefeed to state removed.
	SetChangefeedRemoved() error

	// SetChangefeedFailed set the changefeed to state failed.
	SetChangefeedFailed(err *model.RunningError) error

	// SetChangefeedWarning set the changefeed to state warning.
	SetChangefeedWarning(warn *model.RunningError) error

	// SetChangefeedPending sets the changefeed to state pending.
	SetChangefeedPending(err *model.RunningError) error
}

type Querier interface {
	// GetChangefeeds queries some or all changefeeds.
	GetChangefeed(...ChangefeedUUID) ([]*ChangefeedInfo, error)
	// GetChangefeedState queries some or all changefeed states.
	GetChangefeedState(...ChangefeedUUID) ([]*ChangefeedState, error)
	// GetChangefeedProgress queries some or all changefeed progresses.
	GetChangefeedProgress(...ChangefeedUUID) (map[ChangefeedUUID]ChangefeedProgress, error)
}

type SchedState int

type ScheduleError struct {
	Msg string
}

type ScheduledChangefeed struct {
	ChangefeedUUID ChangefeedUUID   `gorm:"column:changefeed_uuid;type:bigint(20) unsigned;primaryKey" json:"changefeed_uuid"`
	Owner          *model.CaptureID `gorm:"column:owner;type:varchar(128)" json:"owner"`
	OwnerState     SchedState       `gorm:"column:owner_state;type:text;not null" json:"owner_state"`
	// Processors is always equal to the owner in the current implementation.
	Processors *model.CaptureID `gorm:"column:processors;type:text" json:"processors"`
	// TaskPosition is used to initialize changefeed owner on the capture.
	TaskPosition ChangefeedProgress `gorm:"column:task_position;type:text;not null" json:"task_position"`
}