ha

Jan 30, 2026

Changes in this version

+ func ConnectHandler(opts ...connect.HandlerOption) (path string, handler http.Handler)

type Connector

+ func (c *Connector) Backup(ctx context.Context, writer io.Writer) error

Jan 23, 2026

Changes in this version

type ConnHooksConfig

+ GrpcToken string

type Option

+ func WithGrpcToken(token string) Option

Jan 19, 2026

Changes in this version

type Statement

+ func (s *Statement) AggregateFunctions() map[int]string

+ func (s *Statement) RewriteQueryToAggregate() (query string, aggregateFunctions map[int]string, newColumns map[int]int, ...)

Jan 16, 2026

Changes in this version

+ const TypePragma

type Statement

+ func (s *Statement) Visit(n sql.Node) (w sql.Visitor, node sql.Node, err error)

+ func (s *Statement) VisitEnd(n sql.Node) (sql.Node, error)

Jan 8, 2026

Changes in this version

type Statement

+ func (s *Statement) Limit() int

+ func (s *Statement) ProjectionFunctions() map[int]string

Jan 8, 2026

Changes in this version

+ func CrossShardQuery(ctx context.Context, stmt *Statement, args []driver.NamedValue, ...) (driver.Rows, error)

Jan 6, 2026

Changes in this version

type Statement

+ func (s *Statement) OrderBy() []string

Jan 6, 2026

Changes in this version

type ConnHooksConfig

+ QueryRouter *regexp.Regexp

type Option

+ func WithQueryRouter(re *regexp.Regexp) Option

Jan 3, 2026

Changes in this version

type ConnHooksConfig

+ TxSeqTrackerProvider TxSeqTrackerProvider

+ type TxSeqTracker interface

+ LatestSeq func() uint64

+ type TxSeqTrackerProvider func() TxSeqTracker

Jan 3, 2026

Changes in this version

type ConnHooksConfig

+ GrpcTimeout time.Duration

type Option

+ func WithGrpcTimeout(timeout time.Duration) Option

Jan 3, 2026

Changes in this version

type ConnHooksConfig

+ Subscriber Subscriber

Jan 3, 2026

Changes in this version

+ func ListReplicationIDs() []string

+ type ConnHooksConfig struct

+ CDC CDCPublisher

+ DisableDDLSync bool

+ Leader LeaderProvider

+ NodeName string

+ Publisher Publisher

+ ReplicationID string

type Connector

+ func LookupConnectorByReplicationID(id string) (*Connector, bool)

+ func (c *Connector) DB() *sql.DB

+ func (c *Connector) PubSeq() uint64

type Option

+ func WithGrpcPort(port int) Option

Dec 29, 2025

Changes in this version

type Option

+ func WithMySQLPass(pass string) Option

+ func WithMySQLUser(user string) Option

Dec 27, 2025

Changes in this version

type Option

+ func WithMySQLPort(port int) Option

Dec 20, 2025

Changes in this version

type Change

+ TsNs int64

type ChangeSet

+ func (cs *ChangeSet) DebeziumData() []DebeziumData

type Connector

+ func (c *Connector) CDCPublisher() CDCPublisher

+ type DebeziumData struct

+ Payload DebeziumPayload

+ Schema any

+ Transaction DebeziumTransaction

+ type DebeziumPayload struct

+ After map[string]any

+ Before map[string]any

+ Op string

+ Source DebeziumSource

+ TsNs int64

+ type DebeziumSource struct

+ Connector string

+ DB string

+ Name string

+ ServerID int64

+ Table string

+ TsNs int64

+ Version string

+ type DebeziumTransaction struct

+ ID string

type Option

+ func WithReplicationID(id string) Option

+ func WithReplicationPublisher(pub Publisher) Option

+ func WithReplicationSubscriber(sub Subscriber) Option

+ type Publisher interface

+ Publish func(cs *ChangeSet) error

+ Sequence func() uint64

+ type Subscriber interface

+ DeliveredInfo func(ctx context.Context, name string) (any, error)

+ LatestSeq func() uint64

+ RemoveConsumer func(ctx context.Context, name string) error

+ SetDB func(*sql.DB)

+ Start func() error

Dec 3, 2025

Changes in this version

type Change

+ func (c Change) PKNewValues() []any

Dec 3, 2025

Changes in this version

type Change

+ PKColumns []string

+ func (c Change) PKColumnsNames() []string

+ func (c Change) PKOldValues() []any

type RowIdentify

+ const PK

Nov 13, 2025

Changes in this version

+ const TXCookieName

+ func ListDSN() []string

type Connector

+ func (c *Connector) ResponseWriter(w http.ResponseWriter) http.ResponseWriter

Nov 13, 2025

Changes in this version

type Connector

+ func (c *Connector) LeaderProvider() LeaderProvider

+ func (c *Connector) Start(db *sql.DB) error

+ type DynamicLeader struct

+ func (d *DynamicLeader) IsLeader() bool

+ func (d *DynamicLeader) Ready() chan struct{}

+ func (d *DynamicLeader) RedirectTarget() string

type LeaderProvider

+ Ready func() chan struct{}

type Option

+ func WithClusterSize(size int) Option

+ func WithLeaderElectionLocalTarget(localEndpoint string) Option

+ func WithLeaderProvider(p LeaderProvider) Option

type StaticLeader

+ func (s *StaticLeader) Ready() chan struct{}

Nov 12, 2025

Changes in this version

type Connector

+ func LookupConnector(dsn string) *Connector

+ func (c *Connector) ConsistentReader(timeout time.Duration, methods ...string) func(http.Handler) http.Handler

+ func (c *Connector) ConsistentReaderFunc(h http.HandlerFunc, timeout time.Duration, methods ...string) http.HandlerFunc

+ func (c *Connector) ForwardToLeader(timeout time.Duration, methods ...string) func(http.Handler) http.Handler

+ func (c *Connector) ForwardToLeaderFunc(h http.HandlerFunc, timeout time.Duration, methods ...string) http.HandlerFunc

+ type LeaderProvider interface

+ IsLeader func() bool

+ RedirectTarget func() string

type Option

+ func WithAutoStart(enabled bool) Option

+ func WithLeader(p LeaderProvider) Option

+ type StaticLeader struct

+ Target string

+ func (s *StaticLeader) IsLeader() bool

+ func (s *StaticLeader) RedirectTarget() string

Nov 10, 2025

Changes in this version

type ChangeSet

+ func (cs *ChangeSet) SetApplyStrategy(fn applyStrategyFn)

+ type NATSSubscriberConfig struct

+ ConnProvider ConnHooksProvider

+ DB *sql.DB

+ Durable string

+ Interceptor ChangeSetInterceptor

+ NatsConn *nats.Conn

+ Node string

+ Policy string

+ RowIdentify RowIdentify

+ Stream string

+ Subject string

type Option

+ func WithRowIdentify(i RowIdentify) Option

+ type RowIdentify string

+ const Full

+ const Rowid

Nov 10, 2025

Changes in this version

type AsyncNATSPublisher

+ func (p *AsyncNATSPublisher) Sequence() uint64

+ func (p *AsyncNATSPublisher) SetSequence(s uint64)

type CDCPublisher

+ Sequence func() uint64

+ SetSequence func(uint64)

type ChangeSet

+ LocalProdSeq uint64

+ Subject string

type JSONPublisher

+ func (p *JSONPublisher) Sequence() uint64

+ func (p *JSONPublisher) SetSequence(s uint64)

type NATSPublisher

+ func (p *NATSPublisher) Sequence() uint64

+ func (p *NATSPublisher) SetSequence(s uint64)

type NoopPublisher

+ func (p *NoopPublisher) Sequence() uint64

+ func (p *NoopPublisher) SetSequence(s uint64)

+ type PubSequenceProvider interface

+ Sequence func() uint64

+ SetSequence func(uint64)

type WriterPublisher

+ func (p *WriterPublisher) Sequence() uint64

+ func (p *WriterPublisher) SetSequence(s uint64)

Nov 9, 2025

Changes in this version

type Option

+ func WithCDCID(id string) Option

Nov 7, 2025

Changes in this version

+ func Shutdown()

Nov 7, 2025

Changes in this version

+ type AsyncNATSPublisher struct

+ func NewAsyncNATSPublisher(nc *nats.Conn, subject string, timeout time.Duration, ...) (*AsyncNATSPublisher, error)

+ func (p *AsyncNATSPublisher) Close() error

+ func (p *AsyncNATSPublisher) Publish(cs *ChangeSet) error

type Option

+ func WithAsyncPublisher() Option

+ func WithAsyncPublisherOutboxDir(dir string) Option

Oct 21, 2025

Changes in this version

type CDCSubscriber

+ SetDB func(*sql.DB)

type Connector

+ func (c *Connector) Snapshotter() DBSnapshotter

+ func (c *Connector) Subscriber() CDCSubscriber

type DBSnapshotter

+ SetDB func(*sql.DB)

+ Start func() error

type NATSSnapshotter

+ func (s *NATSSnapshotter) SetDB(db *sql.DB)

+ func (s *NATSSnapshotter) Start() error

type NATSSubscriber

+ func (s *NATSSubscriber) SetDB(db *sql.DB)

type NoopSnapshotter

+ func (s *NoopSnapshotter) SetDB(_ *sql.DB)

+ func (s *NoopSnapshotter) Start() error

type NoopSubscriber

+ func (*NoopSubscriber) SetDB(db *sql.DB)

Oct 18, 2025

Changes in this version

+ var ErrSnapshotterNotConfigured = errors.New("snapshotter not configured")

+ var ErrSubscriberNotConfigured = errors.New("subscriber not configured")

type Connector

+ func GetConnector(dsn string) *Connector

type Statement

+ func UnverifiedStatement(source string, hasDistinct bool, hasReturning bool, typ string, ...) *Statement

+ func (s *Statement) ModifiesDatabase() bool

Oct 16, 2025

Changes in this version

+ type BackupFn func(context.Context, *sql.DB, io.Writer) error

type ChangeSet

+ func (cs *ChangeSet) SetConnProvider(connProvider ConnHooksProvider)

+ type ConnHooksFactory func(nodeName string, filename string, disableDDSSync bool, publisher CDCPublisher) ConnHooksProvider

+ type ConnHooksProvider interface

+ DisableHooks func(*sql.Conn) error

+ EnableHooks func(*sql.Conn) error

+ RegisterHooks func(driver.Conn) (driver.Conn, error)

type Connector

+ func (c *Connector) NodeName() string

+ func (c *Connector) Publisher() CDCPublisher

+ type DriverProvider interface

+ ConnWithoutHooks func() (*sql.Conn, error)

+ EnableHooks func(conn *sql.Conn)

+ OnConnect func(c driver.Conn) (driver.Conn, error)

type Option

+ func NameToOptions(name string) (string, []Option, error)

type Statement

+ func (s *Statement) SourceWithIfExists() string

Oct 12, 2025

Changes in this version

type Option

+ func WithReplicationStream(stream string) Option

Oct 11, 2025

Changes in this version

type Change

+ Args []any

+ Command string

Oct 10, 2025

Changes in this version

+ type CDCSubscriber interface

+ DeliveredInfo func(ctx context.Context, name string) (any, error)

+ LatestSeq func() uint64

+ RemoveConsumer func(ctx context.Context, name string) error

+ Start func() error

type ChangeSet

+ func (cs *ChangeSet) SetInterceptor(interceptor ChangeSetInterceptor)

+ type ChangeSetInterceptor interface

+ AfterApply func(*ChangeSet, *sql.DB, error) error

+ BeforeApply func(*ChangeSet, *sql.DB) (skip bool, err error)

+ type ChangeSetSerializer func(*ChangeSet) ([]byte, error)

+ type DBSnapshotter interface

+ LatestSnapshot func(ctx context.Context) (sequence uint64, reader io.ReadCloser, err error)

+ TakeSnapshot func(ctx context.Context, db *sql.DB) (sequence uint64, err error)

+ type JSONPublisher struct

+ func NewJSONPublisher(w io.Writer) *JSONPublisher

+ func (p *JSONPublisher) Publish(cs *ChangeSet) error

+ type NATSPublisher struct

+ func NewNATSPublisher(nc *nats.Conn, subject string, timeout time.Duration, ...) (*NATSPublisher, error)

+ func (p *NATSPublisher) Publish(cs *ChangeSet) error

+ type NATSSnapshotter struct

+ func NewNATSSnapshotter(ctx context.Context, nc *nats.Conn, replicas int, stream string, db *sql.DB, ...) (*NATSSnapshotter, error)

+ func (s *NATSSnapshotter) LatestSnapshot(ctx context.Context) (uint64, io.ReadCloser, error)

+ func (s *NATSSnapshotter) LatestSnapshotSequence(ctx context.Context) (uint64, error)

+ func (s *NATSSnapshotter) TakeSnapshot(ctx context.Context, db *sql.DB) (sequence uint64, err error)

+ type NATSSubscriber struct

+ func NewNATSSubscriber(node string, durable string, nc *nats.Conn, subject string, policy string, ...) (*NATSSubscriber, error)

+ func (s *NATSSubscriber) DeliveredInfo(ctx context.Context, name string) (any, error)

+ func (s *NATSSubscriber) LatestSeq() uint64

+ func (s *NATSSubscriber) RemoveConsumer(ctx context.Context, name string) error

+ func (s *NATSSubscriber) Start() error

+ type NoopPublisher struct

+ func NewNoopPublisher() *NoopPublisher

+ func (p *NoopPublisher) Publish(cs *ChangeSet) error

+ type NoopSnapshotter struct

+ func NewNoopSnapshotter() *NoopSnapshotter

+ func (s *NoopSnapshotter) LatestSnapshot(ctx context.Context) (uint64, io.ReadCloser, error)

+ func (s *NoopSnapshotter) TakeSnapshot(ctx context.Context, db *sql.DB) (sequence uint64, err error)

+ type NoopSubscriber struct

+ func NewNoopSubscriber() *NoopSubscriber

+ func (*NoopSubscriber) DeliveredInfo(ctx context.Context, name string) (any, error)

+ func (*NoopSubscriber) LatestSeq() uint64

+ func (*NoopSubscriber) RemoveConsumer(ctx context.Context, name string) error

+ func (*NoopSubscriber) Start() error

type Option

+ func WithCDCSubscriber(sub CDCSubscriber) Option

+ func WithChangeSetInterceptor(interceptor ChangeSetInterceptor) Option

+ func WithDBSnapshotter(snap DBSnapshotter) Option

+ type WriterPublisher struct

+ func NewWriterPublisher(w io.Writer, serializer ChangeSetSerializer) *WriterPublisher

+ func (p *WriterPublisher) Publish(cs *ChangeSet) error

Oct 10, 2025

Changes in this version

+ func LatestSnapshot(ctx context.Context, options ...Option) (sequence uint64, reader io.ReadCloser, err error)

type ChangeSet

+ Filename string

type Option

+ func WithDisableDDLSync() Option

+ func WithWaitFor(ch chan struct{}) Option

type Statement

+ func ParseStatement(ctx context.Context, sql string) (*Statement, error)

Oct 9, 2025

Changes in this version

type Driver

+ Options []Option

Oct 9, 2025

Changes in this version

+ type ConnectHookFn func(conn *sqlite3.SQLiteConn) error

+ type Driver struct

+ ConnectHook ConnectHookFn

+ Extensions []string

+ func (d *Driver) Open(name string) (driver.Conn, error)

+ func (d *Driver) OpenConnector(name string) (driver.Connector, error)

type Option

+ func WithConnectHook(fn ConnectHookFn) Option

Oct 8, 2025

Changes in this version

+ const DefaultStream

+ const TypeAlterTable

+ const TypeAnalyze

+ const TypeBegin

+ const TypeCommit

+ const TypeCreateIndex

+ const TypeCreateTable

+ const TypeCreateTrigger

+ const TypeCreateView

+ const TypeCreateVirtualTable

+ const TypeDelete

+ const TypeDrop

+ const TypeExplain

+ const TypeInsert

+ const TypeOther

+ const TypeRelease

+ const TypeRollback

+ const TypeSavepoint

+ const TypeSelect

+ const TypeUpdate

+ const TypeVacuum

+ var ErrInvalidSQL = fmt.Errorf("invalid SQL")

+ var ErrNatsNotConfigured = errors.New("NATS not configured")

+ func Backup(ctx context.Context, db *sql.DB, w io.Writer) error

+ type CDCPublisher interface

+ Publish func(cs *ChangeSet) error

+ type Change struct

+ Columns []string

+ Database string

+ NewRowID int64

+ NewValues []any

+ OldRowID int64

+ OldValues []any

+ Operation string

+ SQL string

+ SQLArgs []any

+ Table string

+ type ChangeSet struct

+ Changes []Change

+ Node string

+ ProcessID int64

+ StreamSeq uint64

+ Timestamp int64

+ func NewChangeSet(node string, publisher CDCPublisher) *ChangeSet

+ func (cs *ChangeSet) AddChange(change Change)

+ func (cs *ChangeSet) Apply(db *sql.DB) error

+ func (cs *ChangeSet) Clear()

+ func (cs *ChangeSet) Send(pub CDCPublisher) error

+ type Conn struct

+ func (c *Conn) Exec(query string, args []driver.Value) (driver.Result, error)

+ func (c *Conn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error)

+ type Connector struct

+ func NewConnector(dsn string, options ...Option) (*Connector, error)

+ func (c *Connector) Close()

+ func (c *Connector) Connect(ctx context.Context) (driver.Conn, error)

+ func (c *Connector) DeliveredInfo(ctx context.Context, name string) ([]*jetstream.ConsumerInfo, error)

+ func (c *Connector) Driver() driver.Driver

+ func (c *Connector) LatestSeq() uint64

+ func (c *Connector) LatestSnapshot(ctx context.Context) (uint64, io.ReadCloser, error)

+ func (c *Connector) RemoveConsumer(ctx context.Context, name string) error

+ func (c *Connector) TakeSnapshot(ctx context.Context, db *sql.DB) (sequence uint64, err error)

+ type EmbeddedNatsConfig struct

+ EnableLogs bool

+ File string

+ Name string

+ Pass string

+ Port int

+ StoreDir string

+ User string

+ type Option func(*Connector)

+ func WithCDCPublisher(pub CDCPublisher) Option

+ func WithDeliverPolicy(deliverPolicy string) Option

+ func WithEmbeddedNatsConfig(cfg *EmbeddedNatsConfig) Option

+ func WithExtensions(extensions ...string) Option

+ func WithName(name string) Option

+ func WithNatsOptions(options ...nats.Option) Option

+ func WithPublisherTimeout(timeout time.Duration) Option

+ func WithReplicas(replicas int) Option

+ func WithReplicationSubject(subject string) Option

+ func WithReplicationURL(url string) Option

+ func WithSnapshotInterval(interval time.Duration) Option

+ func WithStreamMaxAge(maxAge time.Duration) Option

+ type SequenceProvider interface

+ LatestSeq func() uint64

+ type Statement struct

+ func Parse(ctx context.Context, sql string) (*Statement, error)

+ func (s *Statement) Begin() bool

+ func (s *Statement) Columns() []string

+ func (s *Statement) Commit() bool

+ func (s *Statement) DDL() bool

+ func (s *Statement) HasDistinct() bool

+ func (s *Statement) HasReturning() bool

+ func (s *Statement) IsCreateTable() bool

+ func (s *Statement) IsDelete() bool

+ func (s *Statement) IsExplain() bool

+ func (s *Statement) IsInsert() bool

+ func (s *Statement) IsSelect() bool

+ func (s *Statement) IsUpdate() bool

+ func (s *Statement) Parameters() []string

+ func (s *Statement) Rollback() bool

+ func (s *Statement) Source() string

+ func (s *Statement) Type() string