adapter

package
v0.0.0-...-ba04163 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 25, 2025 License: AGPL-3.0 Imports: 7 Imported by: 0

Documentation

Overview

Package adapter provides the unified interface for all database adapters.

This package defines the contracts that database-specific implementations must follow, enabling a consistent way to interact with any database technology while respecting their unique characteristics.

Architecture

The adapter package follows an interface-driven design with several key components:

  • DatabaseAdapter: The main interface that all database adapters implement
  • Connection: Represents an active database connection with operation interfaces
  • InstanceConnection: Represents an instance-level (server) connection
  • Operation Interfaces: SchemaOperator, DataOperator, ReplicationOperator, MetadataOperator
  • Registry: Manages adapter registration and retrieval

Usage

To use an adapter, first register it with the global registry: 

import (
    "github.com/redbco/redb-open/pkg/anchor/adapter"
    "github.com/redbco/redb-open/services/anchor/internal/database/postgres"
)

func init() {
    adapter.Register(postgres.NewAdapter())
}

Then connect to a database: 

config := adapter.ConnectionConfig{
    DatabaseID:     "my-db",
    ConnectionType: "postgres",
    Host:           "localhost",
    Port:           5432,
    DatabaseName:   "myapp",
    Username:       "user",
    Password:       "pass",
}

conn, err := adapter.GlobalRegistry().Connect(ctx, config)
if err != nil {
    log.Fatal(err)
}
defer conn.Close()

Perform operations through the connection: 

// Schema discovery
schema, err := conn.SchemaOperations().DiscoverSchema(ctx)

// Data operations
data, err := conn.DataOperations().Fetch(ctx, "users", 100)

// Replication (if supported)
if repOps := conn.ReplicationOperations(); repOps != nil {
    if repOps.IsSupported() {
        status, err := repOps.GetStatus(ctx)
    }
}

Capability-Based Design

The adapter system is designed around database capabilities. Not all databases support all operations. Use nil checks and the IsSupported() method to check for capability support: 

conn, _ := adapter.GlobalRegistry().Connect(ctx, config)

// Check if schema operations are supported
if schemaOps := conn.SchemaOperations(); schemaOps != nil {
    schema, err := schemaOps.DiscoverSchema(ctx)
}

// Check if replication is supported
if repOps := conn.ReplicationOperations(); repOps != nil && repOps.IsSupported() {
    // Use replication operations
}

Error Handling

The adapter package provides standardized error types:

  • DatabaseError: Wraps database-specific errors with context
  • UnsupportedOperationError: Indicates an unsupported operation
  • ConnectionError: Indicates a connection failure
  • ConfigurationError: Indicates invalid configuration

Use the Is() and As() functions from the errors package to check error types: 

if adapter.IsUnsupported(err) {
    // Handle unsupported operation
}

if adapter.IsConnectionError(err) {
    // Handle connection error
}

Implementing a New Adapter

To implement a new database adapter:

1. Create a new package under services/anchor/internal/database/{dbname}

2. Implement the DatabaseAdapter interface: 

type Adapter struct{}

func NewAdapter() adapter.DatabaseAdapter {
    return &Adapter{}
}

func (a *Adapter) Type() dbcapabilities.DatabaseType {
    return dbcapabilities.YourDB
}

func (a *Adapter) Capabilities() dbcapabilities.Capability {
    return dbcapabilities.MustGet(dbcapabilities.YourDB)
}

3. Implement the Connection interface and operation interfaces

4. Register the adapter: 

func init() {
    adapter.Register(NewAdapter())
}

See the documentation in docs/DATABASE_ADAPTER_IMPLEMENTATION_GUIDE.md for detailed instructions on implementing a new adapter.

Thread Safety

All types in this package are designed to be thread-safe. The Registry uses mutex locks to protect concurrent access. Connection implementations should also be thread-safe.

Package adapter provides the unified interface for all database adapters. This package defines the contracts that database-specific implementations must follow.

Index

Constants

View Source 
const (
	// TransformDirect - direct field mapping with no transformation
	TransformDirect = "direct"
	// TransformCast - type casting (e.g., string to int)
	TransformCast = "cast"
	// TransformUppercase - convert to uppercase
	TransformUppercase = "uppercase"
	// TransformLowercase - convert to lowercase
	TransformLowercase = "lowercase"
	// TransformFunction - apply a custom function
	TransformFunction = "function"
	// TransformExpression - evaluate an expression
	TransformExpression = "expression"
	// TransformDefault - use default value if source is null
	TransformDefault = "default"
)

TransformationType constants

Variables

View Source 
var (
	// ErrOperationNotSupported is returned when an operation is not supported by the database
	ErrOperationNotSupported = errors.New("operation not supported by this database")

	// ErrConnectionClosed is returned when attempting to use a closed connection
	ErrConnectionClosed = errors.New("connection is closed")

	// ErrConnectionFailed is returned when a connection attempt fails
	ErrConnectionFailed = errors.New("connection failed")

	// ErrInvalidConfiguration is returned when the configuration is invalid
	ErrInvalidConfiguration = errors.New("invalid configuration")

	// ErrCapabilityNotFound is returned when a capability is not found
	ErrCapabilityNotFound = errors.New("capability not found")

	// ErrTableNotFound is returned when a table/collection is not found
	ErrTableNotFound = errors.New("table not found")

	// ErrDatabaseNotFound is returned when a database is not found
	ErrDatabaseNotFound = errors.New("database not found")

	// ErrAdapterNotFound is returned when an adapter is not registered
	ErrAdapterNotFound = errors.New("adapter not found")

	// ErrInvalidQuery is returned when a query is malformed
	ErrInvalidQuery = errors.New("invalid query")

	// ErrTransactionFailed is returned when a transaction fails
	ErrTransactionFailed = errors.New("transaction failed")

	// ErrAuthenticationFailed is returned when authentication fails
	ErrAuthenticationFailed = errors.New("authentication failed")

	// ErrPermissionDenied is returned when a permission is denied
	ErrPermissionDenied = errors.New("permission denied")

	// ErrInvalidData is returned when data is invalid or malformed
	ErrInvalidData = errors.New("invalid data")

	// ErrConfigurationError is returned when there's a configuration error
	ErrConfigurationError = errors.New("configuration error")
)

Standard adapter errors

Functions

func GetBool 

func GetBool(b *bool) bool

GetBool returns the bool value from a pointer, or false if nil. Helper function for optional bool fields.

func GetBoolPtr 

func GetBoolPtr(b bool) *bool

GetBoolPtr returns a pointer to a bool value. Helper function for optional bool fields.

func GetString 

func GetString(s *string) string

GetString returns the string value from a pointer, or empty string if nil. Helper function for optional string fields.

func GetStringPtr 

func GetStringPtr(s string) *string

GetStringPtr returns a pointer to a string value, or nil if the string is empty. Helper function for optional string fields.

func IsAuthenticationError 

func IsAuthenticationError(err error) bool

IsAuthenticationError checks if an error is an authentication error.

func IsConfigurationError 

func IsConfigurationError(err error) bool

IsConfigurationError checks if an error is a configuration error.

func IsConnectionError 

func IsConnectionError(err error) bool

IsConnectionError checks if an error is a connection error.

func IsRegistered 

func IsRegistered(dbType dbcapabilities.DatabaseType) bool

IsRegistered checks if an adapter is registered in the global registry.

func IsUnsupported 

func IsUnsupported(err error) bool

IsUnsupported checks if an error indicates an unsupported operation.

func IsUnsupportedOperator 

func IsUnsupportedOperator(op interface{}) bool

IsUnsupportedOperator checks if an operator is an unsupported operator. This can be used to detect when an operation is not available.

func ListRegistered 

func ListRegistered() []dbcapabilities.DatabaseType

ListRegistered returns all registered database types from the global registry.

func Register 

func Register(adapter DatabaseAdapter)

Register registers an adapter in the global registry.

func WrapError 

func WrapError(dbType dbcapabilities.DatabaseType, operation string, err error) error

WrapError wraps an error with database context. If the error is already a DatabaseError, it returns it as-is.

Types

type CDCEvent 

type CDCEvent struct {
	// Operation type (INSERT, UPDATE, DELETE, TRUNCATE)
	Operation CDCOperation `json:"operation"`

	// Target identification
	SchemaName string `json:"schema_name,omitempty"` // Schema/database name (optional)
	TableName  string `json:"table_name"`            // Table/collection name

	// Event data
	Data    map[string]interface{} `json:"data,omitempty"`     // New data for INSERT/UPDATE
	OldData map[string]interface{} `json:"old_data,omitempty"` // Old data for UPDATE/DELETE

	// Event metadata
	Timestamp     time.Time              `json:"timestamp"`                // Event timestamp
	LSN           string                 `json:"lsn,omitempty"`            // Log sequence number (database-specific)
	TransactionID string                 `json:"transaction_id,omitempty"` // Transaction identifier
	Metadata      map[string]interface{} `json:"metadata,omitempty"`       // Additional database-specific metadata
	SourceNode    string                 `json:"source_node,omitempty"`    // Source node ID (for mesh routing)
	TargetNode    string                 `json:"target_node,omitempty"`    // Target node ID (for mesh routing)
}

CDCEvent represents a standardized CDC event across all database types. This is the universal format that all database adapters must produce and consume.

func (*CDCEvent) Validate 

func (e *CDCEvent) Validate() error

Validate checks if the CDC event is valid.

type CDCEventFilter 

type CDCEventFilter struct {
	// Table filtering
	IncludeTables []string `json:"include_tables,omitempty"`
	ExcludeTables []string `json:"exclude_tables,omitempty"`

	// Operation filtering
	IncludeOperations []CDCOperation `json:"include_operations,omitempty"`
	ExcludeOperations []CDCOperation `json:"exclude_operations,omitempty"`

	// Data filtering (future: WHERE clause equivalent)
	Conditions map[string]interface{} `json:"conditions,omitempty"`
}

CDCEventFilter defines filtering criteria for CDC events. This will be used in future to filter which events are replicated.

type CDCOperation 

type CDCOperation string

CDCOperation represents the type of change in a CDC event. 

const (
	// CDCInsert represents an INSERT operation
	CDCInsert CDCOperation = "INSERT"
	// CDCUpdate represents an UPDATE operation
	CDCUpdate CDCOperation = "UPDATE"
	// CDCDelete represents a DELETE operation
	CDCDelete CDCOperation = "DELETE"
	// CDCTruncate represents a TRUNCATE operation
	CDCTruncate CDCOperation = "TRUNCATE"
)

type CDCStatistics 

type CDCStatistics struct {
	EventsProcessed    int64                  `json:"events_processed"`
	EventsFailed       int64                  `json:"events_failed"`
	LastEventTimestamp time.Time              `json:"last_event_timestamp"`
	LastEventLSN       string                 `json:"last_event_lsn,omitempty"`
	OperationCounts    map[CDCOperation]int64 `json:"operation_counts"`
	AverageLatency     time.Duration          `json:"average_latency"`
	CurrentLag         time.Duration          `json:"current_lag,omitempty"`
	BytesProcessed     int64                  `json:"bytes_processed"`
	AdditionalMetrics  map[string]interface{} `json:"additional_metrics,omitempty"`
}

CDCStatistics tracks CDC replication statistics.

func NewCDCStatistics 

func NewCDCStatistics() *CDCStatistics

NewCDCStatistics creates a new CDC statistics tracker.

func (*CDCStatistics) RecordEvent 

func (s *CDCStatistics) RecordEvent(event *CDCEvent, latency time.Duration)

RecordEvent records a successfully processed event.

func (*CDCStatistics) RecordFailure 

func (s *CDCStatistics) RecordFailure()

RecordFailure records a failed event.

type ConfigurationError 

type ConfigurationError struct {
	DatabaseType dbcapabilities.DatabaseType
	Field        string
	Reason       string
}

ConfigurationError is returned when a configuration error occurs.

func NewConfigurationError 

func NewConfigurationError(dbType dbcapabilities.DatabaseType, field string, reason string) *ConfigurationError

NewConfigurationError creates a new ConfigurationError.

func (*ConfigurationError) Error 

func (e *ConfigurationError) Error() string

Error implements the error interface.

func (*ConfigurationError) Is 

func (e *ConfigurationError) Is(target error) bool

Is checks if the error is ErrInvalidConfiguration.

type Connection 

type Connection interface {
	// Identity and status
	ID() string
	Type() dbcapabilities.DatabaseType
	IsConnected() bool

	// Lifecycle management
	Ping(ctx context.Context) error
	Close() error

	// Operation interfaces
	// Returns nil if the operation category is not supported by this database
	SchemaOperations() SchemaOperator
	DataOperations() DataOperator
	ReplicationOperations() ReplicationOperator
	MetadataOperations() MetadataOperator

	// Raw returns the underlying database-specific connection object.
	// Use this only when you need to perform operations not covered by the standard interfaces.
	// Type assertion is required when using Raw().
	Raw() interface{}

	// Configuration
	Config() ConnectionConfig
	Adapter() DatabaseAdapter
}

Connection represents an active connection to a specific database. This is the main interface for interacting with a database.

type ConnectionConfig 

type ConnectionConfig struct {
	// Core identifiers
	DatabaseID    string  `json:"databaseId"`
	TenantID      string  `json:"tenantId"`
	WorkspaceID   string  `json:"workspaceId"`
	EnvironmentID *string `json:"environmentId,omitempty"`
	InstanceID    string  `json:"instanceId,omitempty"`

	// Connection metadata
	Name        string `json:"name,omitempty"`
	Description string `json:"description,omitempty"`

	// Database type and vendor
	ConnectionType string `json:"connectionType"` // e.g., "postgres", "mysql"
	DatabaseVendor string `json:"databaseVendor"` // Cloud provider or "custom"

	// Connection details
	Host         string `json:"host"`
	Port         int    `json:"port"`
	Username     string `json:"username,omitempty"`
	Password     string `json:"password,omitempty"`
	DatabaseName string `json:"databaseName"`

	// SSL/TLS configuration
	SSL                   bool    `json:"ssl,omitempty"`
	SSLMode               string  `json:"sslMode,omitempty"` // verify-full, require, etc.
	SSLRejectUnauthorized *bool   `json:"sslRejectUnauthorized,omitempty"`
	SSLCert               *string `json:"sslCert,omitempty"`
	SSLKey                *string `json:"sslKey,omitempty"`
	SSLRootCert           *string `json:"sslRootCert,omitempty"`

	// Additional options
	Role              string `json:"role,omitempty"`
	ConnectedToNodeID string `json:"connectedToNodeId,omitempty"`
	OwnerID           string `json:"ownerId,omitempty"`
	Enabled           *bool  `json:"enabled,omitempty"`

	// Cloud/Object Storage credentials (S3, GCS, Azure Blob)
	AccessKeyID     string `json:"accessKeyId,omitempty"`
	SecretAccessKey string `json:"secretAccessKey,omitempty"`
	SessionToken    string `json:"sessionToken,omitempty"`
	Region          string `json:"region,omitempty"`
	PathStyle       bool   `json:"pathStyle,omitempty"`

	// BigQuery specific
	ProjectID       string `json:"projectId,omitempty"`
	CredentialsFile string `json:"credentialsFile,omitempty"`
	CredentialsJSON string `json:"credentialsJson,omitempty"`
	Location        string `json:"location,omitempty"`

	// InfluxDB specific
	Token        string `json:"token,omitempty"`
	Organization string `json:"organization,omitempty"`

	// Azure specific
	ConnectionString string `json:"connectionString,omitempty"`

	// Database-specific options (use sparingly)
	Options map[string]interface{} `json:"options,omitempty"`
}

ConnectionConfig contains the configuration for a database connection. This is a unified configuration that works across all database types.

type ConnectionError 

type ConnectionError struct {
	DatabaseType dbcapabilities.DatabaseType
	Host         string
	Port         int
	Cause        error
}

ConnectionError is returned when a connection error occurs.

func NewConnectionError 

func NewConnectionError(dbType dbcapabilities.DatabaseType, host string, port int, cause error) *ConnectionError

NewConnectionError creates a new ConnectionError.

func (*ConnectionError) Error 

func (e *ConnectionError) Error() string

Error implements the error interface.

func (*ConnectionError) Is 

func (e *ConnectionError) Is(target error) bool

Is checks if the error is ErrConnectionFailed.

func (*ConnectionError) Unwrap 

func (e *ConnectionError) Unwrap() error

Unwrap returns the underlying error.

type DataOperator 

type DataOperator interface {
	// Read operations
	Fetch(ctx context.Context, table string, limit int) ([]map[string]interface{}, error)
	FetchWithColumns(ctx context.Context, table string, columns []string, limit int) ([]map[string]interface{}, error)

	// Write operations
	Insert(ctx context.Context, table string, data []map[string]interface{}) (int64, error)
	Update(ctx context.Context, table string, data []map[string]interface{}, whereColumns []string) (int64, error)
	Upsert(ctx context.Context, table string, data []map[string]interface{}, uniqueColumns []string) (int64, error)
	Delete(ctx context.Context, table string, conditions map[string]interface{}) (int64, error)

	// Streaming for large datasets
	Stream(ctx context.Context, params StreamParams) (StreamResult, error)

	// Query execution (for databases supporting query languages)
	ExecuteQuery(ctx context.Context, query string, args ...interface{}) ([]interface{}, error)
	ExecuteCountQuery(ctx context.Context, query string) (int64, error)

	// Utility operations
	GetRowCount(ctx context.Context, table string, whereClause string) (int64, bool, error)
	Wipe(ctx context.Context) error
}

DataOperator handles data CRUD operations. All databases should support basic data operations.

type DatabaseAdapter 

type DatabaseAdapter interface {
	// Type returns the canonical database type identifier
	Type() dbcapabilities.DatabaseType

	// Capabilities returns the capability metadata for this database type
	Capabilities() dbcapabilities.Capability

	// Connect establishes a connection to a specific database
	Connect(ctx context.Context, config ConnectionConfig) (Connection, error)

	// ConnectInstance establishes a connection to a database instance (server-level)
	ConnectInstance(ctx context.Context, config InstanceConfig) (InstanceConnection, error)
}

DatabaseAdapter represents a database technology adapter. Each database type (PostgreSQL, MySQL, MongoDB, etc.) must implement this interface.

func Get 

func Get(dbType dbcapabilities.DatabaseType) (DatabaseAdapter, error)

Get retrieves an adapter from the global registry.

func GetByName 

func GetByName(name string) (DatabaseAdapter, error)

GetByName retrieves an adapter from the global registry by name.

type DatabaseError 

type DatabaseError struct {
	DatabaseType dbcapabilities.DatabaseType
	Operation    string
	Cause        error
	Context      map[string]interface{}
}

DatabaseError wraps database-specific errors with additional context. This provides a consistent error structure across all database types.

func NewDatabaseError 

func NewDatabaseError(dbType dbcapabilities.DatabaseType, operation string, cause error) *DatabaseError

NewDatabaseError creates a new DatabaseError.

func (*DatabaseError) Error 

func (e *DatabaseError) Error() string

Error implements the error interface.

func (*DatabaseError) Is 

func (e *DatabaseError) Is(target error) bool

Is checks if the error matches the target error.

func (*DatabaseError) Unwrap 

func (e *DatabaseError) Unwrap() error

Unwrap returns the underlying error.

func (*DatabaseError) WithContext 

func (e *DatabaseError) WithContext(key string, value interface{}) *DatabaseError

WithContext adds context to a DatabaseError.

type InstanceConfig 

type InstanceConfig struct {
	// Core identifiers
	InstanceID    string  `json:"instanceId"`
	TenantID      string  `json:"tenantId"`
	WorkspaceID   string  `json:"workspaceId"`
	EnvironmentID *string `json:"environmentId,omitempty"`

	// Connection metadata
	Name        string `json:"name,omitempty"`
	Description string `json:"description,omitempty"`

	// Database type and vendor
	ConnectionType string `json:"connectionType"`
	DatabaseVendor string `json:"databaseVendor"`

	// Instance information
	Version          string `json:"version,omitempty"`
	UniqueIdentifier string `json:"uniqueIdentifier,omitempty"`

	// Connection details
	Host         string `json:"host"`
	Port         int    `json:"port"`
	Username     string `json:"username,omitempty"`
	Password     string `json:"password,omitempty"`
	DatabaseName string `json:"databaseName"` // System database for connection

	// SSL/TLS configuration
	SSL                   bool    `json:"ssl,omitempty"`
	SSLMode               string  `json:"sslMode,omitempty"`
	SSLRejectUnauthorized *bool   `json:"sslRejectUnauthorized,omitempty"`
	SSLCert               *string `json:"sslCert,omitempty"`
	SSLKey                *string `json:"sslKey,omitempty"`
	SSLRootCert           *string `json:"sslRootCert,omitempty"`

	// Additional options
	Role              string `json:"role,omitempty"`
	ConnectedToNodeID string `json:"connectedToNodeId,omitempty"`
	OwnerID           string `json:"ownerId,omitempty"`
	Enabled           *bool  `json:"enabled,omitempty"`

	// Cloud/Object Storage credentials
	AccessKeyID     string `json:"accessKeyId,omitempty"`
	SecretAccessKey string `json:"secretAccessKey,omitempty"`
	SessionToken    string `json:"sessionToken,omitempty"`
	Region          string `json:"region,omitempty"`
	PathStyle       bool   `json:"pathStyle,omitempty"`

	// BigQuery specific
	ProjectID       string `json:"projectId,omitempty"`
	CredentialsFile string `json:"credentialsFile,omitempty"`
	CredentialsJSON string `json:"credentialsJson,omitempty"`
	Location        string `json:"location,omitempty"`

	// InfluxDB specific
	Token        string `json:"token,omitempty"`
	Organization string `json:"organization,omitempty"`

	// Azure specific
	ConnectionString string `json:"connectionString,omitempty"`

	// Database-specific options
	Options map[string]interface{} `json:"options,omitempty"`
}

InstanceConfig contains the configuration for a database instance connection.

type InstanceConnection 

type InstanceConnection interface {
	// Identity and status
	ID() string
	Type() dbcapabilities.DatabaseType
	IsConnected() bool

	// Lifecycle management
	Ping(ctx context.Context) error
	Close() error

	// Instance-level database management
	ListDatabases(ctx context.Context) ([]string, error)
	CreateDatabase(ctx context.Context, name string, options map[string]interface{}) error
	DropDatabase(ctx context.Context, name string, options map[string]interface{}) error

	// Metadata operations
	MetadataOperations() MetadataOperator

	// Raw returns the underlying database-specific connection object
	Raw() interface{}

	// Configuration
	Config() InstanceConfig
	Adapter() DatabaseAdapter
}

InstanceConnection represents an active connection to a database instance. This is used for instance-level operations like creating/dropping databases.

type LogicalDatabaseInfo 

type LogicalDatabaseInfo struct {
	Name      string
	SizeBytes int64
	Owner     string
	Encoding  string
	Collation string
	CreatedAt *time.Time
}

LogicalDatabaseInfo represents metadata about a logical database within an instance

type MetadataOperator 

type MetadataOperator interface {
	// Metadata collection
	CollectDatabaseMetadata(ctx context.Context) (map[string]interface{}, error)
	CollectInstanceMetadata(ctx context.Context) (map[string]interface{}, error)

	// Version and identification
	GetVersion(ctx context.Context) (string, error)
	GetUniqueIdentifier(ctx context.Context) (string, error)

	// Statistics
	GetDatabaseSize(ctx context.Context) (int64, error)
	GetTableCount(ctx context.Context) (int, error)

	// Command execution (for databases supporting admin commands)
	ExecuteCommand(ctx context.Context, command string) ([]byte, error)

	// Extended metrics and metadata collection (optional - may return nil for databases that don't support it)
	CollectInstanceMetrics(ctx context.Context) (map[string]interface{}, error)
	ListLogicalDatabases(ctx context.Context) ([]LogicalDatabaseInfo, error)
}

MetadataOperator handles metadata collection and introspection. All databases should support basic metadata operations.

type NotFoundError 

type NotFoundError struct {
	DatabaseType dbcapabilities.DatabaseType
	ResourceType string
	ResourceName string
}

NotFoundError is returned when a resource is not found.

func NewNotFoundError 

func NewNotFoundError(dbType dbcapabilities.DatabaseType, resourceType string, resourceName string) *NotFoundError

NewNotFoundError creates a new NotFoundError.

func (*NotFoundError) Error 

func (e *NotFoundError) Error() string

Error implements the error interface.

func (*NotFoundError) Is 

func (e *NotFoundError) Is(target error) bool

Is checks if the error is ErrTableNotFound or ErrDatabaseNotFound.

type Registry 

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

Registry manages the registration and retrieval of database adapters.

func GlobalRegistry 

func GlobalRegistry() *Registry

GlobalRegistry returns the global adapter registry.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry creates a new adapter registry.

func (*Registry) Clear 

func (r *Registry) Clear()

Clear removes all adapters from the registry.

func (*Registry) Connect 

func (r *Registry) Connect(ctx context.Context, config ConnectionConfig) (Connection, error)

Connect creates a new database connection using the registered adapter.

func (*Registry) ConnectInstance 

func (r *Registry) ConnectInstance(ctx context.Context, config InstanceConfig) (InstanceConnection, error)

ConnectInstance creates a new instance connection using the registered adapter.

func (*Registry) Get 

func (r *Registry) Get(dbType dbcapabilities.DatabaseType) (DatabaseAdapter, error)

Get retrieves a registered adapter by database type. Returns ErrAdapterNotFound if the adapter is not registered.

func (*Registry) GetByName 

func (r *Registry) GetByName(name string) (DatabaseAdapter, error)

GetByName retrieves a registered adapter by database name or alias. Returns ErrAdapterNotFound if the adapter is not registered.

func (*Registry) GetCapabilities 

func (r *Registry) GetCapabilities(dbType dbcapabilities.DatabaseType) (dbcapabilities.Capability, error)

GetCapabilities returns the capabilities for a database type.

func (*Registry) GetCapabilitiesByName 

func (r *Registry) GetCapabilitiesByName(name string) (dbcapabilities.Capability, error)

GetCapabilitiesByName returns the capabilities for a database by name or alias.

func (*Registry) IsRegistered 

func (r *Registry) IsRegistered(dbType dbcapabilities.DatabaseType) bool

IsRegistered checks if an adapter is registered for the given database type.

func (*Registry) ListRegistered 

func (r *Registry) ListRegistered() []dbcapabilities.DatabaseType

ListRegistered returns a list of all registered database types.

func (*Registry) Register 

func (r *Registry) Register(adapter DatabaseAdapter)

Register registers a database adapter. If an adapter for the same database type is already registered, it will be replaced.

func (*Registry) Unregister 

func (r *Registry) Unregister(dbType dbcapabilities.DatabaseType)

Unregister removes an adapter from the registry.

type ReplicationConfig 

type ReplicationConfig struct {
	// Core identifiers
	ReplicationID string  `json:"replicationId"`
	DatabaseID    string  `json:"databaseId"`
	TenantID      string  `json:"tenantId"`
	WorkspaceID   string  `json:"workspaceId"`
	EnvironmentID *string `json:"environmentId,omitempty"`

	// Replication metadata
	ReplicationName string `json:"replicationName"`

	// Database type
	ConnectionType string `json:"connectionType"`
	DatabaseVendor string `json:"databaseVendor"`

	// Connection details (may reuse database connection)
	Host         string `json:"host"`
	Port         int    `json:"port"`
	Username     string `json:"username,omitempty"`
	Password     string `json:"password,omitempty"`
	DatabaseName string `json:"databaseName"`

	// SSL/TLS configuration
	SSL         bool   `json:"ssl,omitempty"`
	SSLMode     string `json:"sslMode,omitempty"`
	SSLCert     string `json:"sslCert,omitempty"`
	SSLKey      string `json:"sslKey,omitempty"`
	SSLRootCert string `json:"sslRootCert,omitempty"`

	// Replication scope
	TableNames      []string `json:"tableNames,omitempty"`      // Tables to replicate
	CollectionNames []string `json:"collectionNames,omitempty"` // Collections to replicate
	IndexNames      []string `json:"indexNames,omitempty"`      // Indexes to replicate
	KeyPatterns     []string `json:"keyPatterns,omitempty"`     // Key patterns to replicate

	// Database-specific replication options
	SlotName        string   `json:"slotName,omitempty"`        // PostgreSQL replication slot
	PublicationName string   `json:"publicationName,omitempty"` // PostgreSQL publication
	StreamNames     []string `json:"streamNames,omitempty"`     // Snowflake streams

	// Resume/checkpoint support
	StartPosition string `json:"startPosition,omitempty"` // Starting position for resume (LSN, binlog position, etc.)

	// Event handling
	EventHandler func(map[string]interface{}) `json:"-"` // Event callback function

	// Additional options
	Role              string                 `json:"role,omitempty"`
	ConnectedToNodeID string                 `json:"connectedToNodeId,omitempty"`
	OwnerID           string                 `json:"ownerId,omitempty"`
	Enabled           *bool                  `json:"enabled,omitempty"`
	Options           map[string]interface{} `json:"options,omitempty"`
}

ReplicationConfig contains the configuration for a replication connection.

type ReplicationOperator 

type ReplicationOperator interface {
	// Capability checking
	IsSupported() bool
	GetSupportedMechanisms() []string

	// Setup and management
	CheckPrerequisites(ctx context.Context) error
	Connect(ctx context.Context, config ReplicationConfig) (ReplicationSource, error)

	// Status and monitoring
	GetStatus(ctx context.Context) (map[string]interface{}, error)
	GetLag(ctx context.Context) (map[string]interface{}, error)

	// Slot/publication management (where applicable)
	ListSlots(ctx context.Context) ([]map[string]interface{}, error)
	DropSlot(ctx context.Context, slotName string) error
	ListPublications(ctx context.Context) ([]map[string]interface{}, error)
	DropPublication(ctx context.Context, publicationName string) error

	// CDC Event handling - these methods enable database-agnostic CDC
	// ParseEvent converts a raw database-specific event to a standardized CDCEvent
	ParseEvent(ctx context.Context, rawEvent map[string]interface{}) (*CDCEvent, error)

	// ApplyCDCEvent applies a standardized CDC event to this database
	// This method handles INSERT, UPDATE, DELETE operations in a database-specific way
	ApplyCDCEvent(ctx context.Context, event *CDCEvent) error

	// TransformData applies transformation rules to event data
	// Returns the transformed data ready for application to target database
	// transformationServiceEndpoint is the gRPC endpoint for the transformation service (optional, for custom transformations)
	TransformData(ctx context.Context, data map[string]interface{}, rules []TransformationRule, transformationServiceEndpoint string) (map[string]interface{}, error)
}

ReplicationOperator handles Change Data Capture (CDC) and replication operations. Only databases with CDC support will implement this fully.

func NewUnsupportedReplicationOperator 

func NewUnsupportedReplicationOperator(dbType dbcapabilities.DatabaseType) ReplicationOperator

NewUnsupportedReplicationOperator creates a new unsupported replication operator.

type ReplicationSource 

type ReplicationSource interface {
	// Identity
	GetSourceID() string
	GetDatabaseID() string

	// Status
	GetStatus() map[string]interface{}
	GetMetadata() map[string]interface{}
	IsActive() bool

	// Lifecycle
	Start() error
	Stop() error
	Close() error

	// Position tracking for graceful shutdown and resume
	// GetPosition returns the current replication position (LSN, binlog position, etc.)
	// The format is database-specific: PostgreSQL uses LSN, MySQL uses binlog file:position
	GetPosition() (string, error)

	// SetPosition sets the starting replication position for resume operations
	// This should be called before Start() to resume from a specific position
	SetPosition(position string) error

	// SaveCheckpoint persists the current position for crash recovery
	// This is typically called periodically or after processing batches of events
	SaveCheckpoint(ctx context.Context, position string) error
}

ReplicationSource represents an active replication connection.

type SchemaOperator 

type SchemaOperator interface {
	// DiscoverSchema retrieves the complete schema of the database as a UnifiedModel
	DiscoverSchema(ctx context.Context) (*unifiedmodel.UnifiedModel, error)

	// CreateStructure creates database objects from a UnifiedModel
	CreateStructure(ctx context.Context, model *unifiedmodel.UnifiedModel) error

	// ListTables returns the names of all tables/collections in the database
	ListTables(ctx context.Context) ([]string, error)

	// GetTableSchema retrieves the schema for a specific table/collection
	GetTableSchema(ctx context.Context, tableName string) (*unifiedmodel.Table, error)
}

SchemaOperator handles schema discovery and modification operations. Not all databases support all schema operations.

func NewUnsupportedSchemaOperator 

func NewUnsupportedSchemaOperator(dbType dbcapabilities.DatabaseType) SchemaOperator

NewUnsupportedSchemaOperator creates a new unsupported schema operator.

type StreamParams 

type StreamParams struct {
	Table     string   // Table/collection name
	Columns   []string // Specific columns to fetch (empty = all)
	BatchSize int32    // Number of rows per batch
	Offset    int64    // Starting offset
	OrderBy   string   // Column to order by (optional)
}

StreamParams configures streaming operations for large datasets.

type StreamResult 

type StreamResult struct {
	Data       []map[string]interface{} // The batch of data
	HasMore    bool                     // Whether more data is available
	NextCursor string                   // Cursor for pagination
}

StreamResult contains the result of a streaming operation.

type TransformationRule 

type TransformationRule struct {
	// Source field identification
	SourceColumn string `json:"source_column"`
	SourceTable  string `json:"source_table,omitempty"`

	// Target field identification
	TargetColumn string `json:"target_column"`
	TargetTable  string `json:"target_table,omitempty"`

	// Transformation configuration
	TransformationType string                 `json:"transformation_type"`           // direct, cast, function, expression
	TransformationName string                 `json:"transformation_name,omitempty"` // Name of transformation function (e.g., "reverse", "uppercase")
	Parameters         map[string]interface{} `json:"parameters,omitempty"`

	// Metadata
	Description string `json:"description,omitempty"`
}

TransformationRule defines how to transform data from source to target. This is used for field mapping, type conversion, and value transformation.

type UnsupportedOperationError 

type UnsupportedOperationError struct {
	DatabaseType dbcapabilities.DatabaseType
	Operation    string
	Reason       string
}

UnsupportedOperationError is returned when an operation is not supported.

func NewUnsupportedOperationError 

func NewUnsupportedOperationError(dbType dbcapabilities.DatabaseType, operation string, reason string) *UnsupportedOperationError

NewUnsupportedOperationError creates a new UnsupportedOperationError.

func (*UnsupportedOperationError) Error 

func (e *UnsupportedOperationError) Error() string

Error implements the error interface.

func (*UnsupportedOperationError) Is 

func (e *UnsupportedOperationError) Is(target error) bool

Is checks if the error is ErrOperationNotSupported.

type UnsupportedReplicationOperator 

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

UnsupportedReplicationOperator is a nil object pattern for databases that don't support replication.

func (*UnsupportedReplicationOperator) ApplyCDCEvent 

func (u *UnsupportedReplicationOperator) ApplyCDCEvent(ctx context.Context, event *CDCEvent) error

func (*UnsupportedReplicationOperator) CheckPrerequisites 

func (u *UnsupportedReplicationOperator) CheckPrerequisites(ctx context.Context) error

func (*UnsupportedReplicationOperator) Connect 

func (u *UnsupportedReplicationOperator) Connect(ctx context.Context, config ReplicationConfig) (ReplicationSource, error)

func (*UnsupportedReplicationOperator) DropPublication 

func (u *UnsupportedReplicationOperator) DropPublication(ctx context.Context, publicationName string) error

func (*UnsupportedReplicationOperator) DropSlot 

func (u *UnsupportedReplicationOperator) DropSlot(ctx context.Context, slotName string) error

func (*UnsupportedReplicationOperator) GetLag 

func (u *UnsupportedReplicationOperator) GetLag(ctx context.Context) (map[string]interface{}, error)

func (*UnsupportedReplicationOperator) GetStatus 

func (u *UnsupportedReplicationOperator) GetStatus(ctx context.Context) (map[string]interface{}, error)

func (*UnsupportedReplicationOperator) GetSupportedMechanisms 

func (u *UnsupportedReplicationOperator) GetSupportedMechanisms() []string

func (*UnsupportedReplicationOperator) IsSupported 

func (u *UnsupportedReplicationOperator) IsSupported() bool

func (*UnsupportedReplicationOperator) ListPublications 

func (u *UnsupportedReplicationOperator) ListPublications(ctx context.Context) ([]map[string]interface{}, error)

func (*UnsupportedReplicationOperator) ListSlots 

func (u *UnsupportedReplicationOperator) ListSlots(ctx context.Context) ([]map[string]interface{}, error)

func (*UnsupportedReplicationOperator) ParseEvent 

func (u *UnsupportedReplicationOperator) ParseEvent(ctx context.Context, rawEvent map[string]interface{}) (*CDCEvent, error)

func (*UnsupportedReplicationOperator) TransformData 

func (u *UnsupportedReplicationOperator) TransformData(ctx context.Context, data map[string]interface{}, rules []TransformationRule, transformationServiceEndpoint string) (map[string]interface{}, error)

type UnsupportedSchemaOperator 

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

UnsupportedSchemaOperator is a nil object pattern for databases that don't support schema operations.

func (*UnsupportedSchemaOperator) CreateStructure 

func (u *UnsupportedSchemaOperator) CreateStructure(ctx context.Context, model *unifiedmodel.UnifiedModel) error

func (*UnsupportedSchemaOperator) DiscoverSchema 

func (u *UnsupportedSchemaOperator) DiscoverSchema(ctx context.Context) (*unifiedmodel.UnifiedModel, error)

func (*UnsupportedSchemaOperator) GetTableSchema 

func (u *UnsupportedSchemaOperator) GetTableSchema(ctx context.Context, tableName string) (*unifiedmodel.Table, error)

func (*UnsupportedSchemaOperator) ListTables 

func (u *UnsupportedSchemaOperator) ListTables(ctx context.Context) ([]string, error)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.