feat: make configurable hyper table check for timescale db (#40)

* feat: make configurable hyper table check for timescale db

* feat: make configurable hyper table check for timescale db

---------

Co-authored-by: Serhat Karabulut <serhat.karabulut@trendyolgo.com>

Author: 3n0ugh

README.md

@@ -154,6 +154,7 @@ This setup ensures continuous data synchronization and minimal downtime in captu
 | `slot.createIfNotExists`                |   bool   |    no    |    -    | Create replication slot if not exists. Otherwise, return `replication slot is not exists` error.      |                                                                                                                                                    |
 | `slot.name`                             |  string  |   yes    |    -    | Set the logical replication slot name                                                                 | Should be unique and descriptive.                                                                                                                  |
 | `slot.slotActivityCheckerInterval`      |   int    |    no    |  1000   | Set the slot activity check interval time in milliseconds                                             | Specify as an integer value in milliseconds (e.g., `1000` for 1 second).                                                                           |
+| `extensionSupport.enableTimescaleDB`    |   bool   |    no    |  false  | Enable support for TimescaleDB hypertables. Ensures proper handling of compressed chunks during replication. |                                                                                                                                                    |
 
 ### API
 

config/config.go

@@ -14,16 +14,17 @@ import (
 )
 
 type Config struct {
-	Logger      LoggerConfig       `json:"logger" yaml:"logger"`
-	Host        string             `json:"host" yaml:"host"`
-	Username    string             `json:"username" yaml:"username"`
-	Password    string             `json:"password" yaml:"password"`
-	Database    string             `json:"database" yaml:"database"`
-	Publication publication.Config `json:"publication" yaml:"publication"`
-	Slot        slot.Config        `json:"slot" yaml:"slot"`
-	Port        int                `json:"port" yaml:"port"`
-	Metric      MetricConfig       `json:"metric" yaml:"metric"`
-	DebugMode   bool               `json:"debugMode" yaml:"debugMode"`
+	Logger           LoggerConfig       `json:"logger" yaml:"logger"`
+	Host             string             `json:"host" yaml:"host"`
+	Username         string             `json:"username" yaml:"username"`
+	Password         string             `json:"password" yaml:"password"`
+	Database         string             `json:"database" yaml:"database"`
+	Publication      publication.Config `json:"publication" yaml:"publication"`
+	Slot             slot.Config        `json:"slot" yaml:"slot"`
+	Port             int                `json:"port" yaml:"port"`
+	Metric           MetricConfig       `json:"metric" yaml:"metric"`
+	DebugMode        bool               `json:"debugMode" yaml:"debugMode"`
+	ExtensionSupport ExtensionSupport   `json:"extensionSupport" yaml:"extensionSupport"`
 }
 
 type MetricConfig struct {
@@ -35,6 +36,10 @@ type LoggerConfig struct {
 	LogLevel slog.Level    `json:"level" yaml:"level"` // if custom logger is nil, set the slog log level
 }
 
+type ExtensionSupport struct {
+	EnableTimeScaleDB bool `json:"enableTimeScaleDB" yaml:"EnableTimeScaleDB"`
+}
+
 func (c *Config) DSN() string {
 	// URL-encode username and password to handle special characters
 	encodedUsername := url.QueryEscape(c.Username)

connector.go

@@ -112,14 +112,16 @@ func NewConnector(ctx context.Context, cfg config.Config, listenerFunc replicati
 
 	prometheusRegistry := metric.NewRegistry(m)
 
-	tdb, err := timescaledb.NewTimescaleDB(ctx, cfg.DSN())
-	if err != nil {
-		return nil, err
-	}
-
-	_, err = tdb.FindHyperTables(ctx)
-	if err != nil {
-		return nil, err
+	var tdb *timescaledb.TimescaleDB
+	if cfg.ExtensionSupport.EnableTimeScaleDB {
+		tdb, err = timescaledb.NewTimescaleDB(ctx, cfg.DSN())
+		if err != nil {
+			return nil, err
+		}
+		_, err = tdb.FindHyperTables(ctx)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	return &connector{
@@ -156,7 +158,10 @@ func (c *connector) Start(ctx context.Context) {
 
 	logger.Info("slot captured")
 	go c.slot.Metrics(ctx)
-	go c.timescaleDB.SyncHyperTables(ctx)
+
+	if c.timescaleDB != nil {
+		go c.timescaleDB.SyncHyperTables(ctx)
+	}
 
 	signal.Notify(c.cancelCh, syscall.SIGTERM, syscall.SIGINT, syscall.SIGABRT, syscall.SIGQUIT)