dmsengine

type Change struct {
	Time     time.Time
	Schema   string
	Table    string
	Op       Op
	Position string
	Before   Row
	After    Row
	Commit   bool
}

type Column struct {
	Name     string
	Native   string
	Kind     Kind
	Nullable bool
}

type DynamoKey struct {
	Name   string
	Number bool
}

type DynamoStore interface {
	CreateTable(ctx context.Context, name string, keys []DynamoKey) error
	DropTable(ctx context.Context, name string) error
	PutItem(ctx context.Context, table string, item map[string]DynamoValue) error
	DeleteItem(ctx context.Context, table string, key map[string]DynamoValue) error
}

type DynamoValue struct {
	Type string
	S    string
	B    []byte
}

type EndpointConfig struct {
	Engine   string
	Server   string
	Database string
	User     string
	Password string
	SSLMode  string
	// CertificatePEM is the endpoint's CertificateArn contents, trusted when SslMode asks for TLS.
	CertificatePEM string
	// SlotName names the PostgreSQL logical replication slot for source endpoints.
	SlotName string
	// S3Settings, KinesisSettings and DynamoDBSettings carry the raw settings JSON objects.
	S3Settings       string
	KinesisSettings  string
	DynamoDBSettings string
	Port             int
	// PollInterval spaces PostgreSQL change polls; zero picks a default.
	PollInterval time.Duration
}

type Family int

type Kind int

type Mappings struct {
	Rules []Rule `json:"rules"`
}

type ObjectStore interface {
	PutObject(ctx context.Context, bucket, key string, body []byte) error
	CheckBucket(ctx context.Context, bucket string) error
}

type Op string

type Opener interface {
	OpenSource(ctx context.Context, ep EndpointConfig) (Source, error)
	OpenTarget(ctx context.Context, ep EndpointConfig, src Family) (Target, error)
	// ReleaseSlot drops the replication slot a source endpoint created, if any.
	ReleaseSlot(ctx context.Context, ep EndpointConfig) error
}

type PrepMode string

type Progress struct {
	FullLoadStart  time.Time
	FullLoadFinish time.Time
	Checkpoint     string
	Tables         []TableStat
	Running        bool
}

type Result struct {
	Err    error
	Reason string
}

type Row []any

type Rule struct {
	Locator struct {
		Schema string `json:"schema-name"`
		Table  string `json:"table-name"`
		Column string `json:"column-name"`
	} `json:"object-locator"`
	ID     flexString `json:"rule-id"`
	Name   flexString `json:"rule-name"`
	Type   string     `json:"rule-type"`
	Action string     `json:"rule-action"`
	Target string     `json:"rule-target"`
	Value  string     `json:"value"`
	Old    string     `json:"old-value"`
}

type Source interface {
	Ping(ctx context.Context) error
	Tables(ctx context.Context) ([]Table, error)
	Scan(ctx context.Context, t Table, fn func(Row) error) error
	// Position captures the current change position; for PostgreSQL it also creates the replication slot.
	Position(ctx context.Context) (string, error)
	// Stream delivers changes from start until ctx is done or fn fails. Rows are in the table's column order.
	Stream(
		ctx context.Context,
		start StreamStart,
		tables func(schema, table string) (Table, bool),
		fn func(Change) error,
	) error
	// Release drops source-side resources such as replication slots.
	Release(ctx context.Context) error
	Close() error
}

type Stores struct {
	Objects ObjectStore
	Streams StreamStore
	Dynamo  DynamoStore
}

type StreamStart struct {
	Time     time.Time
	Position string
}

type StreamStore interface {
	PutRecord(ctx context.Context, stream, partitionKey string, data []byte) error
	CheckStream(ctx context.Context, stream string) error
}

type Table struct {
	Schema  string
	Name    string
	Columns []Column
	PK      []string
}

type TableRef struct{ Schema, Name string }

type TableStat struct {
	FullLoadStart     time.Time
	FullLoadEnd       time.Time
	LastUpdate        time.Time
	Schema            string
	Name              string
	State             string
	Error             string
	FullLoadRows      int64
	FullLoadErrorRows int64
	Inserts           int64
	Updates           int64
	Deletes           int64
	FullLoadReloaded  bool
}

type Target interface {
	Ping(ctx context.Context) error
	Prepare(ctx context.Context, t TargetTable, mode PrepMode) error
	Load(ctx context.Context, t TargetTable, rows []Row) error
	EndLoad(ctx context.Context, t TargetTable) error
	Apply(ctx context.Context, t TargetTable, c Change) error
	Commit(ctx context.Context) error
	Close() error
}

type TargetColumn struct {
	Column
	Source int
}

type TargetTable struct {
	Schema       string
	Name         string
	SourceSchema string
	SourceName   string
	Columns      []TargetColumn
	PK           []string
	SourceFamily Family
}

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

type TaskConfig struct {
	Source        Source
	Target        Target
	Now           func() time.Time
	OnProgress    func(Progress)
	MigrationType string
	CDCStart      string
	CDCStop       string
	Resume        string
	Settings      TaskSettings
	Mappings      Mappings
	PingBackoff   time.Duration
	SourceFamily  Family
}

type TaskSettings struct {
	PrepMode     PrepMode
	TargetSchema string
	CommitRate   int
}