Repository: xataio/pgstream
Pull request #58 (2024-08-02 15:24:42)

Expose required internal libs
This PR moves some of the internal libraries to the pkg directory, since they need to be exposed when using pgstream as a library. This includes only small wrappers around backoff handling, tls and kafka, with their own config which is required for use with the stream, and for implementing new processors outside of the pgstream repository.


Patch:
diff --git a/cmd/config.go b/cmd/config.go
index b137b03..10f0c72 100644
--- a/cmd/config.go
+++ b/cmd/config.go
@@ -6,11 +6,11 @@ import (
 	"fmt"
 
 	"github.com/spf13/viper"
-	"github.com/xataio/pgstream/internal/backoff"
-	"github.com/xataio/pgstream/internal/kafka"
-	"github.com/xataio/pgstream/internal/tls"
+	"github.com/xataio/pgstream/pkg/backoff"
+	"github.com/xataio/pgstream/pkg/kafka"
 	pgschemalog "github.com/xataio/pgstream/pkg/schemalog/postgres"
 	"github.com/xataio/pgstream/pkg/stream"
+	"github.com/xataio/pgstream/pkg/tls"
 	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
 	kafkalistener "github.com/xataio/pgstream/pkg/wal/listener/kafka"
 	kafkaprocessor "github.com/xataio/pgstream/pkg/wal/processor/kafka"
@@ -245,8 +245,8 @@ func parseTranslatorConfig() *translator.Config {
 	}
 }
 
-func parseTLSConfig(prefix string) *tls.Config {
-	return &tls.Config{
+func parseTLSConfig(prefix string) tls.Config {
+	return tls.Config{
 		Enabled:        viper.GetBool(fmt.Sprintf("%s_TLS_ENABLED", prefix)),
 		CaCertFile:     viper.GetString(fmt.Sprintf("%s_TLS_CA_CERT_FILE", prefix)),
 		ClientCertFile: viper.GetString(fmt.Sprintf("%s_TLS_CLIENT_CERT_FILE", prefix)),
diff --git a/internal/backoff/backoff.go b/pkg/backoff/backoff.go
similarity index 100%
rename from internal/backoff/backoff.go
rename to pkg/backoff/backoff.go
diff --git a/internal/backoff/mocks/mock_backoff.go b/pkg/backoff/mocks/mock_backoff.go
similarity index 87%
rename from internal/backoff/mocks/mock_backoff.go
rename to pkg/backoff/mocks/mock_backoff.go
index aef8da2..6a3be66 100644
--- a/internal/backoff/mocks/mock_backoff.go
+++ b/pkg/backoff/mocks/mock_backoff.go
@@ -2,7 +2,7 @@
 
 package mocks
 
-import "github.com/xataio/pgstream/internal/backoff"
+import "github.com/xataio/pgstream/pkg/backoff"
 
 type Backoff struct {
 	RetryNotifyFn func(backoff.Operation, backoff.Notify) error
diff --git a/internal/kafka/config.go b/pkg/kafka/config.go
similarity index 90%
rename from internal/kafka/config.go
rename to pkg/kafka/config.go
index 37473de..55fc7fc 100644
--- a/internal/kafka/config.go
+++ b/pkg/kafka/config.go
@@ -2,12 +2,12 @@
 
 package kafka
 
-import tlslib "github.com/xataio/pgstream/internal/tls"
+import tlslib "github.com/xataio/pgstream/pkg/tls"
 
 type ConnConfig struct {
 	Servers []string
 	Topic   TopicConfig
-	TLS     *tlslib.Config
+	TLS     tlslib.Config
 }
 
 type TopicConfig struct {
diff --git a/internal/kafka/conn.go b/pkg/kafka/conn.go
similarity index 94%
rename from internal/kafka/conn.go
rename to pkg/kafka/conn.go
index 36256df..df9ef78 100644
--- a/internal/kafka/conn.go
+++ b/pkg/kafka/conn.go
@@ -9,7 +9,7 @@ import (
 	"strconv"
 	"time"
 
-	tlslib "github.com/xataio/pgstream/internal/tls"
+	tlslib "github.com/xataio/pgstream/pkg/tls"
 
 	"github.com/segmentio/kafka-go"
 )
@@ -17,7 +17,7 @@ import (
 // withConnection creates a connection that can be used by the kafka operation
 // passed in the parameters. This ensures the cleanup of all connection resources.
 func withConnection(config *ConnConfig, kafkaOperation func(conn *kafka.Conn) error) error {
-	dialer, err := buildDialer(config.TLS)
+	dialer, err := buildDialer(&config.TLS)
 	if err != nil {
 		return err
 	}
diff --git a/internal/kafka/instrumentation/instrumented_kafka_writer.go b/pkg/kafka/instrumentation/instrumented_kafka_writer.go
similarity index 97%
rename from internal/kafka/instrumentation/instrumented_kafka_writer.go
rename to pkg/kafka/instrumentation/instrumented_kafka_writer.go
index 5258db6..7467a1d 100644
--- a/internal/kafka/instrumentation/instrumented_kafka_writer.go
+++ b/pkg/kafka/instrumentation/instrumented_kafka_writer.go
@@ -7,7 +7,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/xataio/pgstream/internal/kafka"
+	"github.com/xataio/pgstream/pkg/kafka"
 
 	"go.opentelemetry.io/otel/metric"
 )
diff --git a/internal/kafka/kafka_offset_parser.go b/pkg/kafka/kafka_offset_parser.go
similarity index 100%
rename from internal/kafka/kafka_offset_parser.go
rename to pkg/kafka/kafka_offset_parser.go
diff --git a/internal/kafka/kafka_offset_parser_test.go b/pkg/kafka/kafka_offset_parser_test.go
similarity index 100%
rename from internal/kafka/kafka_offset_parser_test.go
rename to pkg/kafka/kafka_offset_parser_test.go
diff --git a/internal/kafka/kafka_reader.go b/pkg/kafka/kafka_reader.go
similarity index 98%
rename from internal/kafka/kafka_reader.go
rename to pkg/kafka/kafka_reader.go
index 0916cb3..609337e 100644
--- a/internal/kafka/kafka_reader.go
+++ b/pkg/kafka/kafka_reader.go
@@ -44,7 +44,7 @@ func NewReader(config ReaderConfig, logger loglib.Logger) (*Reader, error) {
 		return nil, fmt.Errorf("unsupported start offset [%s], must be one of [%s, %s]", config.ConsumerGroupStartOffset, earliestOffset, latestOffset)
 	}
 
-	dialer, err := buildDialer(config.Conn.TLS)
+	dialer, err := buildDialer(&config.Conn.TLS)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/kafka/kafka_writer.go b/pkg/kafka/kafka_writer.go
similarity index 96%
rename from internal/kafka/kafka_writer.go
rename to pkg/kafka/kafka_writer.go
index 66513af..04ba0f4 100644
--- a/internal/kafka/kafka_writer.go
+++ b/pkg/kafka/kafka_writer.go
@@ -8,8 +8,8 @@ import (
 	"time"
 
 	"github.com/segmentio/kafka-go"
-	tlslib "github.com/xataio/pgstream/internal/tls"
 	loglib "github.com/xataio/pgstream/pkg/log"
+	tlslib "github.com/xataio/pgstream/pkg/tls"
 )
 
 type MessageWriter interface {
@@ -56,7 +56,7 @@ func NewWriter(config WriterConfig, logger loglib.Logger) (*Writer, error) {
 		}
 	}
 
-	transport, err := buildTransport(config.Conn.TLS)
+	transport, err := buildTransport(&config.Conn.TLS)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/kafka/log.go b/pkg/kafka/log.go
similarity index 100%
rename from internal/kafka/log.go
rename to pkg/kafka/log.go
diff --git a/internal/kafka/mocks/mock_kafka_parser.go b/pkg/kafka/mocks/mock_kafka_parser.go
similarity index 87%
rename from internal/kafka/mocks/mock_kafka_parser.go
rename to pkg/kafka/mocks/mock_kafka_parser.go
index bb5fa38..d1d314e 100644
--- a/internal/kafka/mocks/mock_kafka_parser.go
+++ b/pkg/kafka/mocks/mock_kafka_parser.go
@@ -2,7 +2,7 @@
 
 package mocks
 
-import "github.com/xataio/pgstream/internal/kafka"
+import "github.com/xataio/pgstream/pkg/kafka"
 
 type OffsetParser struct {
 	ToStringFn   func(o *kafka.Offset) string
diff --git a/internal/kafka/mocks/mock_kafka_reader.go b/pkg/kafka/mocks/mock_kafka_reader.go
similarity index 92%
rename from internal/kafka/mocks/mock_kafka_reader.go
rename to pkg/kafka/mocks/mock_kafka_reader.go
index 62bc746..63f8149 100644
--- a/internal/kafka/mocks/mock_kafka_reader.go
+++ b/pkg/kafka/mocks/mock_kafka_reader.go
@@ -5,7 +5,7 @@ package mocks
 import (
 	"context"
 
-	"github.com/xataio/pgstream/internal/kafka"
+	"github.com/xataio/pgstream/pkg/kafka"
 )
 
 type Reader struct {
diff --git a/internal/kafka/mocks/mock_kafka_writer.go b/pkg/kafka/mocks/mock_kafka_writer.go
similarity index 92%
rename from internal/kafka/mocks/mock_kafka_writer.go
rename to pkg/kafka/mocks/mock_kafka_writer.go
index f366988..4e0c233 100644
--- a/internal/kafka/mocks/mock_kafka_writer.go
+++ b/pkg/kafka/mocks/mock_kafka_writer.go
@@ -6,7 +6,7 @@ import (
 	"context"
 	"sync/atomic"
 
-	"github.com/xataio/pgstream/internal/kafka"
+	"github.com/xataio/pgstream/pkg/kafka"
 )
 
 type Writer struct {
diff --git a/internal/tls/tls.go b/pkg/tls/tls.go
similarity index 100%
rename from internal/tls/tls.go
rename to pkg/tls/tls.go
diff --git a/pkg/wal/checkpointer/kafka/wal_kafka_checkpointer.go b/pkg/wal/checkpointer/kafka/wal_kafka_checkpointer.go
index 1ee144c..1a9e5e8 100644
--- a/pkg/wal/checkpointer/kafka/wal_kafka_checkpointer.go
+++ b/pkg/wal/checkpointer/kafka/wal_kafka_checkpointer.go
@@ -7,8 +7,8 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/xataio/pgstream/internal/backoff"
-	"github.com/xataio/pgstream/internal/kafka"
+	"github.com/xataio/pgstream/pkg/backoff"
+	"github.com/xataio/pgstream/pkg/kafka"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/wal"
 )
diff --git a/pkg/wal/listener/kafka/wal_kafka_reader.go b/pkg/wal/listener/kafka/wal_kafka_reader.go
index a5aa0d7..1c8ff90 100644
--- a/pkg/wal/listener/kafka/wal_kafka_reader.go
+++ b/pkg/wal/listener/kafka/wal_kafka_reader.go
@@ -9,7 +9,7 @@ import (
 	"fmt"
 	"runtime/debug"
 
-	"github.com/xataio/pgstream/internal/kafka"
+	"github.com/xataio/pgstream/pkg/kafka"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/wal"
 )
diff --git a/pkg/wal/processor/kafka/config.go b/pkg/wal/processor/kafka/config.go
index 3639fd7..b57f27e 100644
--- a/pkg/wal/processor/kafka/config.go
+++ b/pkg/wal/processor/kafka/config.go
@@ -6,7 +6,7 @@ import (
 	"errors"
 	"time"
 
-	"github.com/xataio/pgstream/internal/kafka"
+	"github.com/xataio/pgstream/pkg/kafka"
 )
 
 type Config struct {
diff --git a/pkg/wal/processor/kafka/wal_kafka_batch_writer.go b/pkg/wal/processor/kafka/wal_kafka_batch_writer.go
index 898c563..5d2d8a7 100644
--- a/pkg/wal/processor/kafka/wal_kafka_batch_writer.go
+++ b/pkg/wal/processor/kafka/wal_kafka_batch_writer.go
@@ -10,9 +10,9 @@ import (
 	"runtime/debug"
 	"time"
 
-	"github.com/xataio/pgstream/internal/kafka"
-	kafkainstrumentation "github.com/xataio/pgstream/internal/kafka/instrumentation"
 	synclib "github.com/xataio/pgstream/internal/sync"
+	"github.com/xataio/pgstream/pkg/kafka"
+	kafkainstrumentation "github.com/xataio/pgstream/pkg/kafka/instrumentation"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/wal"
 	"github.com/xataio/pgstream/pkg/wal/checkpointer"
diff --git a/pkg/wal/processor/kafka/wal_kafka_msg_batch.go b/pkg/wal/processor/kafka/wal_kafka_msg_batch.go
index ad35dea..492a1bd 100644
--- a/pkg/wal/processor/kafka/wal_kafka_msg_batch.go
+++ b/pkg/wal/processor/kafka/wal_kafka_msg_batch.go
@@ -3,7 +3,7 @@
 package kafka
 
 import (
-	"github.com/xataio/pgstream/internal/kafka"
+	"github.com/xataio/pgstream/pkg/kafka"
 	"github.com/xataio/pgstream/pkg/wal"
 )
 
diff --git a/pkg/wal/processor/search/config.go b/pkg/wal/processor/search/config.go
index 4b0a6d0..16db131 100644
--- a/pkg/wal/processor/search/config.go
+++ b/pkg/wal/processor/search/config.go
@@ -5,7 +5,7 @@ package search
 import (
 	"time"
 
-	"github.com/xataio/pgstream/internal/backoff"
+	"github.com/xataio/pgstream/pkg/backoff"
 )
 
 type IndexerConfig struct {
diff --git a/pkg/wal/processor/search/search_schema_cleaner.go b/pkg/wal/processor/search/search_schema_cleaner.go
index 4b200b1..e318800 100644
--- a/pkg/wal/processor/search/search_schema_cleaner.go
+++ b/pkg/wal/processor/search/search_schema_cleaner.go
@@ -8,7 +8,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/xataio/pgstream/internal/backoff"
+	"github.com/xataio/pgstream/pkg/backoff"
 	loglib "github.com/xataio/pgstream/pkg/log"
 )
 
diff --git a/pkg/wal/processor/search/search_store_retrier.go b/pkg/wal/processor/search/search_store_retrier.go
index 1372b56..c08ae7c 100644
--- a/pkg/wal/processor/search/search_store_retrier.go
+++ b/pkg/wal/processor/search/search_store_retrier.go
@@ -9,7 +9,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/xataio/pgstream/internal/backoff"
+	"github.com/xataio/pgstream/pkg/backoff"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/schemalog"
 )


Test patch:
diff --git a/pkg/stream/integration/helper_test.go b/pkg/stream/integration/helper_test.go
index 1419268..586ce4c 100644
--- a/pkg/stream/integration/helper_test.go
+++ b/pkg/stream/integration/helper_test.go
@@ -10,13 +10,13 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
-	kafkalib "github.com/xataio/pgstream/internal/kafka"
 	"github.com/xataio/pgstream/internal/log/zerolog"
 	pglib "github.com/xataio/pgstream/internal/postgres"
-	"github.com/xataio/pgstream/internal/tls"
+	kafkalib "github.com/xataio/pgstream/pkg/kafka"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	schemalogpg "github.com/xataio/pgstream/pkg/schemalog/postgres"
 	"github.com/xataio/pgstream/pkg/stream"
+	"github.com/xataio/pgstream/pkg/tls"
 	"github.com/xataio/pgstream/pkg/wal"
 	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
 	kafkalistener "github.com/xataio/pgstream/pkg/wal/listener/kafka"
@@ -174,7 +174,7 @@ func testKafkaCfg() kafkalib.ConnConfig {
 			Name:       "integration-tests",
 			AutoCreate: true,
 		},
-		TLS: &tls.Config{
+		TLS: tls.Config{
 			Enabled: false,
 		},
 	}
diff --git a/pkg/wal/checkpointer/kafka/wal_kafka_checkpointer_test.go b/pkg/wal/checkpointer/kafka/wal_kafka_checkpointer_test.go
index f18f688..f216349 100644
--- a/pkg/wal/checkpointer/kafka/wal_kafka_checkpointer_test.go
+++ b/pkg/wal/checkpointer/kafka/wal_kafka_checkpointer_test.go
@@ -10,10 +10,10 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"github.com/xataio/pgstream/internal/backoff"
-	backoffmocks "github.com/xataio/pgstream/internal/backoff/mocks"
-	"github.com/xataio/pgstream/internal/kafka"
-	kafkamocks "github.com/xataio/pgstream/internal/kafka/mocks"
+	"github.com/xataio/pgstream/pkg/backoff"
+	backoffmocks "github.com/xataio/pgstream/pkg/backoff/mocks"
+	"github.com/xataio/pgstream/pkg/kafka"
+	kafkamocks "github.com/xataio/pgstream/pkg/kafka/mocks"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/wal"
 )
diff --git a/pkg/wal/listener/kafka/wal_kafka_reader_test.go b/pkg/wal/listener/kafka/wal_kafka_reader_test.go
index 86e8f50..954d602 100644
--- a/pkg/wal/listener/kafka/wal_kafka_reader_test.go
+++ b/pkg/wal/listener/kafka/wal_kafka_reader_test.go
@@ -11,8 +11,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
-	"github.com/xataio/pgstream/internal/kafka"
-	kafkamocks "github.com/xataio/pgstream/internal/kafka/mocks"
+	"github.com/xataio/pgstream/pkg/kafka"
+	kafkamocks "github.com/xataio/pgstream/pkg/kafka/mocks"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/wal"
 )
diff --git a/pkg/wal/processor/kafka/wal_kafka_batch_writer_test.go b/pkg/wal/processor/kafka/wal_kafka_batch_writer_test.go
index 32eeb9f..dc9a3cc 100644
--- a/pkg/wal/processor/kafka/wal_kafka_batch_writer_test.go
+++ b/pkg/wal/processor/kafka/wal_kafka_batch_writer_test.go
@@ -12,10 +12,10 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
-	"github.com/xataio/pgstream/internal/kafka"
-	kafkamocks "github.com/xataio/pgstream/internal/kafka/mocks"
 	synclib "github.com/xataio/pgstream/internal/sync"
 	syncmocks "github.com/xataio/pgstream/internal/sync/mocks"
+	"github.com/xataio/pgstream/pkg/kafka"
+	kafkamocks "github.com/xataio/pgstream/pkg/kafka/mocks"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/schemalog"
 	"github.com/xataio/pgstream/pkg/wal"
diff --git a/pkg/wal/processor/search/search_schema_cleaner_test.go b/pkg/wal/processor/search/search_schema_cleaner_test.go
index 8a01bd2..7809e05 100644
--- a/pkg/wal/processor/search/search_schema_cleaner_test.go
+++ b/pkg/wal/processor/search/search_schema_cleaner_test.go
@@ -10,8 +10,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
-	"github.com/xataio/pgstream/internal/backoff"
-	"github.com/xataio/pgstream/internal/backoff/mocks"
+	"github.com/xataio/pgstream/pkg/backoff"
+	"github.com/xataio/pgstream/pkg/backoff/mocks"
 	loglib "github.com/xataio/pgstream/pkg/log"
 )
 
diff --git a/pkg/wal/processor/search/search_store_retrier_test.go b/pkg/wal/processor/search/search_store_retrier_test.go
index 9134ed9..7921053 100644
--- a/pkg/wal/processor/search/search_store_retrier_test.go
+++ b/pkg/wal/processor/search/search_store_retrier_test.go
@@ -8,7 +8,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
-	"github.com/xataio/pgstream/internal/backoff"
+	"github.com/xataio/pgstream/pkg/backoff"
 	loglib "github.com/xataio/pgstream/pkg/log"
 )