[!] rewrite with consolidated sync package (#7)

* Complete etcd and postgresql consolidation into internal/sync package - Consolidated all PostgreSQL operations from internal/db into internal/sync/postgresql.go - Consolidated all etcd operations from internal/etcd into internal/sync/etcd.go - Replaced COPY with INSERT ON CONFLICT using pgx.Batch for better upsert handling - Added SendBatch method to PgxIface interface for batch operations - Moved all types, connection functions, and retry logic - All build tests passing - Remove old packages - Consolidate some configs

Author: pashagolub

.github/copilot-instructions.md

@@ -0,0 +1,104 @@
+# GitHub Copilot Instructions: etcd_fdw Single Package Architecture
+
+## Project Context
+
+**Tool**: etcd_fdw - Bidirectional synchronization between etcd and PostgreSQL 
+**Language**: Go 1.25 
+**Architecture**: Single package consolidation (refactoring from separate internal/etcd and internal/db packages) 
+**Database**: PostgreSQL with single `etcd` table using revision status encoding 
+**Key Principle**: KISS (Keep It Simple, Stupid) - maximum simplicity, minimal complexity
+
+## Current Refactor Context
+
+**Objective**: Consolidate `internal/etcd` and `internal/db` packages into single `internal/sync` package 
+**Dependencies**: Only use existing packages from go.mod (pgx/v5, etcd client/v3, logrus, go-retry, testcontainers) 
+**Constraints**: No new dependencies, use INSERT ON CONFLICT with pgx.Batch instead of COPY, minimal test cases
+
+## Architecture Overview
+
+### Single Table Design
+```sql
+CREATE TABLE etcd (
+ ts timestamp with time zone NOT NULL DEFAULT now(),
+ key text NOT NULL,
+ value text,
+ revision bigint NOT NULL,
+ tombstone boolean NOT NULL DEFAULT false,
+ PRIMARY KEY(key, revision)
+);
+```
+
+**Revision Encoding**:
+- `revision = -1`: Pending sync to etcd (PostgreSQL → etcd)
+- `revision > 0`: Real etcd revision (etcd → PostgreSQL)
+
+### Package Structure (Target)
+```
+internal/sync/
+├── sync.go # Main service orchestration
+├── postgresql.go # PostgreSQL operations (consolidated from internal/db)
+├── etcd.go # etcd client operations (consolidated from internal/etcd) 
+├── config.go # Connection management
+└── sync_test.go # Consolidated minimal tests
+```
+
+## Key Functions to Consolidate
+
+### From internal/db/postgres.go:
+- `BulkInsert()` - Replace COPY with INSERT ON CONFLICT using pgx.Batch
+- `GetPendingRecords()` - Get records with revision = -1
+- `UpdateRevision()` - Update revision after etcd sync
+- `GetLatestRevision()` - For watch resume points
+
+### From internal/etcd/client.go:
+- `NewEtcdClient()` - etcd connection management
+- `GetAllKeys()` - Initial sync from etcd
+- `WatchPrefix()` - Continuous etcd monitoring
+- `Put()`, `Delete()` - etcd operations
+
+### From internal/sync/sync.go (existing):
+- `Service` - Main orchestration service
+- `Start()` - Bidirectional sync coordination
+
+## Code Style Preferences
+
+**Simplicity First**:
+- No nested function calls unless necessary
+- Direct error handling (no complex wrapping)
+- Minimal logging (only essential events)
+- Concrete types over interfaces within package
+- pgx.Batch for bulk operations instead of COPY
+
+**Testing**:
+- Minimal test cases covering maximum functionality
+- Use testcontainers for integration tests
+- No bloated test code
+- Focus on essential behavior validation
+
+**Error Handling**:
+- Use existing retry logic from internal/retry package
+- Return errors directly with context
+- Log errors at appropriate levels only
+
+## Recent Changes
+
+1. **Specification Created**: Single package architecture refactor defined
+2. **Research Completed**: Package consolidation strategy determined
+3. **Data Model Designed**: Unified structures for sync package
+4. **Contracts Defined**: Public API for consolidated package
+5. **Quickstart Created**: Step-by-step refactor process
+
+## Current Task Context
+
+**Phase**: Implementation planning complete, ready for /tasks command
+**Next**: Generate specific implementation tasks for package consolidation
+**Focus**: Maintain identical behavior while simplifying architecture
+
+## Key Principles for Implementation
+
+- Preserve all existing functionality exactly
+- Maintain test coverage without bloating
+- Use only existing dependencies from go.mod
+- Follow KISS principle throughout
+- No performance regression
+- Simplify imports and reduce cognitive complexity

cmd/etcd_fdw/integration_test.go

@@ -14,8 +14,7 @@ import (
 "github.com/testcontainers/testcontainers-go/modules/postgres"
 "github.com/testcontainers/testcontainers-go/wait"
 
-"github.com/cybertec-postgresql/etcd_fdw/internal/db"
-"github.com/cybertec-postgresql/etcd_fdw/internal/etcd"
+"github.com/cybertec-postgresql/etcd_fdw/internal/sync"
 )
 
 func setupPostgreSQLContainer(ctx context.Context, t *testing.T) (*pgxpool.Pool, testcontainers.Container) {
@@ -54,7 +53,7 @@ func setupPostgreSQLContainer(ctx context.Context, t *testing.T) (*pgxpool.Pool,
 return pool, pgContainer
 }
 
-func setupEtcdContainer(ctx context.Context, t *testing.T) (*etcd.EtcdClient, testcontainers.Container) {
+func setupEtcdContainer(ctx context.Context, t *testing.T) (*sync.EtcdClient, testcontainers.Container) {
 etcdContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
 ContainerRequest: testcontainers.ContainerRequest{
 Image: "quay.io/coreos/etcd:v3.5.9",
@@ -77,13 +76,13 @@ func setupEtcdContainer(ctx context.Context, t *testing.T) (*etcd.EtcdClient, te
 require.NoError(t, err)
 
 dsn := "etcd://" + endpoint + "/test"
-etcdClient, err := etcd.NewEtcdClient(dsn)
+etcdClient, err := sync.NewEtcdClient(dsn)
 require.NoError(t, err)
 
 return etcdClient, etcdContainer
 }
 
-func setupTestContainers(t *testing.T) (*pgxpool.Pool, *etcd.EtcdClient, func()) {
+func setupTestContainers(t *testing.T) (*pgxpool.Pool, *sync.EtcdClient, func()) {
 ctx := context.Background()
 
 pool, pgContainer := setupPostgreSQLContainer(ctx, t)
@@ -118,18 +117,18 @@ func TestPollingMechanism(t *testing.T) {
 require.NoError(t, err)
 
 // Test GetPendingRecords function
-pendingRecords, err := db.GetPendingRecords(ctx, pool)
+pendingRecords, err := sync.GetPendingRecords(ctx, pool)
 require.NoError(t, err)
 assert.Len(t, pendingRecords, 1)
 assert.Equal(t, "test/polling/key1", pendingRecords[0].Key)
 assert.Equal(t, "value1", pendingRecords[0].Value)
 
 // Test UpdateRevision function
-err = db.UpdateRevision(ctx, pool, "test/polling/key1", 123)
+err = sync.UpdateRevision(ctx, pool, "test/polling/key1", 123)
 require.NoError(t, err)
 
 // Verify record was updated
-pendingAfterUpdate, err := db.GetPendingRecords(ctx, pool)
+pendingAfterUpdate, err := sync.GetPendingRecords(ctx, pool)
 require.NoError(t, err)
 assert.Len(t, pendingAfterUpdate, 0, "No pending records should remain after update")
 
@@ -155,7 +154,7 @@ func TestBulkInsert(t *testing.T) {
 defer cancel()
 
 // Prepare test records
-records := []db.KeyValueRecord{
+records := []sync.KeyValueRecord{
 {
 Key: "test/bulk/key1",
 Value: ("value1"),
@@ -180,7 +179,7 @@ func TestBulkInsert(t *testing.T) {
 }
 
 // Test BulkInsert function
-err := db.BulkInsert(ctx, pool, records)
+err := sync.BulkInsert(ctx, pool, records)
 require.NoError(t, err)
 
 // Verify records were inserted correctly
@@ -228,7 +227,7 @@ func TestInsertPendingRecord(t *testing.T) {
 defer cancel()
 
 // Test inserting a new pending record
-err := db.InsertPendingRecord(ctx, pool, "test/pending/key1", ("value1"), false)
+err := sync.InsertPendingRecord(ctx, pool, "test/pending/key1", ("value1"), false)
 require.NoError(t, err)
 
 // Verify record was inserted with revision = -1
@@ -244,7 +243,7 @@ func TestInsertPendingRecord(t *testing.T) {
 assert.Equal(t, "value1", value.String)
 
 // Test inserting second record with same key (should create new record with different timestamp)
-err = db.InsertPendingRecord(ctx, pool, "test/pending/key1", ("updated_value"), false)
+err = sync.InsertPendingRecord(ctx, pool, "test/pending/key1", ("updated_value"), false)
 require.NoError(t, err)
 
 // Verify both records exist (different timestamps, both with revision = -1)
@@ -257,7 +256,7 @@ func TestInsertPendingRecord(t *testing.T) {
 assert.Equal(t, 1, count, "Should have 1 pending records for the same key with latest value")
 
 // Test inserting tombstone record
-err = db.InsertPendingRecord(ctx, pool, "test/pending/key2", "", true)
+err = sync.InsertPendingRecord(ctx, pool, "test/pending/key2", "", true)
 require.NoError(t, err)
 
 // Verify tombstone record
@@ -284,7 +283,7 @@ func TestGetLatestRevision(t *testing.T) {
 defer cancel()
 
 // Test with empty table
-latestRevision, err := db.GetLatestRevision(ctx, pool)
+latestRevision, err := sync.GetLatestRevision(ctx, pool)
 require.NoError(t, err)
 assert.Equal(t, int64(0), latestRevision)
 
@@ -299,7 +298,7 @@ func TestGetLatestRevision(t *testing.T) {
 require.NoError(t, err)
 
 // Test latest revision (should ignore -1 pending records)
-latestRevision, err = db.GetLatestRevision(ctx, pool)
+latestRevision, err = sync.GetLatestRevision(ctx, pool)
 require.NoError(t, err)
 assert.Equal(t, int64(150), latestRevision)
 }
@@ -327,7 +326,7 @@ func TestPendingRecordFiltering(t *testing.T) {
 require.NoError(t, err)
 
 // Test GetPendingRecords only returns revision = -1
-pendingRecords, err := db.GetPendingRecords(ctx, pool)
+pendingRecords, err := sync.GetPendingRecords(ctx, pool)
 require.NoError(t, err)
 assert.Len(t, pendingRecords, 3)
 
@@ -345,7 +344,7 @@ func TestPendingRecordFiltering(t *testing.T) {
 for _, record := range pendingRecords {
 if record.Key == "test/filter/pending2" {
 assert.True(t, record.Tombstone)
-assert.Nil(t, record.Value)
+assert.Equal(t, "", record.Value) // Empty string for tombstones
 }
 }
 }
@@ -362,22 +361,22 @@ func TestConflictResolution(t *testing.T) {
 defer cancel()
 
 // Insert a pending record
-err := db.InsertPendingRecord(ctx, pool, "test/conflict/key1", "pending_value", false)
+err := sync.InsertPendingRecord(ctx, pool, "test/conflict/key1", "pending_value", false)
 require.NoError(t, err)
 
 // Verify it's pending
-pendingRecords, err := db.GetPendingRecords(ctx, pool)
+pendingRecords, err := sync.GetPendingRecords(ctx, pool)
 require.NoError(t, err)
 assert.Len(t, pendingRecords, 1)
 assert.Equal(t, "test/conflict/key1", pendingRecords[0].Key)
 assert.Equal(t, int64(-1), pendingRecords[0].Revision)
 
 // Simulate etcd sync by updating revision
-err = db.UpdateRevision(ctx, pool, "test/conflict/key1", 300)
+err = sync.UpdateRevision(ctx, pool, "test/conflict/key1", 300)
 require.NoError(t, err)
 
 // Verify record is no longer pending
-pendingAfterUpdate, err := db.GetPendingRecords(ctx, pool)
+pendingAfterUpdate, err := sync.GetPendingRecords(ctx, pool)
 require.NoError(t, err)
 assert.Len(t, pendingAfterUpdate, 0)
 
@@ -391,7 +390,7 @@ func TestConflictResolution(t *testing.T) {
 assert.Equal(t, int64(300), revision)
 
 // Test updating non-existent pending record (should fail gracefully)
-err = db.UpdateRevision(ctx, pool, "test/conflict/nonexistent", 400)
+err = sync.UpdateRevision(ctx, pool, "test/conflict/nonexistent", 400)
 assert.Error(t, err)
 assert.Contains(t, err.Error(), "no pending record found")
 }
@@ -411,10 +410,10 @@ func TestPerformanceOpsPerSecond(t *testing.T) {
 recordCount := 1000
 start := time.Now()
 
-records := make([]db.KeyValueRecord, recordCount)
+records := make([]sync.KeyValueRecord, recordCount)
 for i := 0; i < recordCount; i++ {
 value := fmt.Sprintf("test_value_%d", i)
-records[i] = db.KeyValueRecord{
+records[i] = sync.KeyValueRecord{
 Key: fmt.Sprintf("test/perf/key%d", i),
 Value: value,
 Revision: int64(i + 1),
@@ -423,7 +422,7 @@ func TestPerformanceOpsPerSecond(t *testing.T) {
 }
 }
 
-err := db.BulkInsert(ctx, pool, records)
+err := sync.BulkInsert(ctx, pool, records)
 require.NoError(t, err)
 
 elapsed := time.Since(start)
@@ -454,11 +453,11 @@ func TestPerformanceSyncLatency(t *testing.T) {
 // Insert pending record
 key := fmt.Sprintf("test/latency/key%d", i)
 value := fmt.Sprintf("test_value_%d", i)
-err := db.InsertPendingRecord(ctx, pool, key, value, false)
+err := sync.InsertPendingRecord(ctx, pool, key, value, false)
 require.NoError(t, err)
 
 // Update revision (simulating sync completion)
-err = db.UpdateRevision(ctx, pool, key, int64(i+1))
+err = sync.UpdateRevision(ctx, pool, key, int64(i+1))
 require.NoError(t, err)
 
 latency := time.Since(start)

cmd/etcd_fdw/main.go

@@ -13,8 +13,6 @@ import (
 "github.com/jessevdk/go-flags"
 "github.com/sirupsen/logrus"
 
-"github.com/cybertec-postgresql/etcd_fdw/internal/db"
-"github.com/cybertec-postgresql/etcd_fdw/internal/etcd"
 "github.com/cybertec-postgresql/etcd_fdw/internal/log"
 "github.com/cybertec-postgresql/etcd_fdw/internal/sync"
 )
@@ -131,25 +129,25 @@ func main() {
 }()
 
 // Connect to PostgreSQL with retry logic
-var pgPool db.PgxPoolIface
-if pgPool, err = db.NewWithRetry(ctx, config.PostgresDSN); err != nil {
+pgPool, err := sync.NewWithRetry(ctx, config.PostgresDSN)
+if err != nil {
 logrus.WithError(err).Fatal("Failed to connect to PostgreSQL after retries")
 }
 defer pgPool.Close()
 
 // Connect to etcd with retry logic
-var etcdClient *etcd.EtcdClient
+var etcdClient *sync.EtcdClient
 if config.EtcdDSN != "" {
 var err error
-etcdClient, err = etcd.NewEtcdClientWithRetry(ctx, config.EtcdDSN)
+etcdClient, err = sync.NewEtcdClientWithRetry(ctx, config.EtcdDSN)
 if err != nil {
 logrus.WithError(err).Fatal("Failed to connect to etcd after retries")
 }
 defer etcdClient.Close()
 }
 
 // Get prefix from etcd DSN
-prefix := etcd.GetPrefix(config.EtcdDSN)
+prefix := sync.GetPrefix(config.EtcdDSN)
 
 // Parse polling interval
 pollingInterval, err := time.ParseDuration(config.PollingInterval)

go.mod

@@ -7,7 +7,6 @@ require (
 github.com/jackc/pgx/v5 v5.7.6
 github.com/jessevdk/go-flags v1.6.1
 github.com/pashagolub/pgxmock/v4 v4.8.0
-github.com/sethvargo/go-retry v0.3.0
 github.com/sirupsen/logrus v1.9.3
 github.com/stretchr/testify v1.11.1
 github.com/testcontainers/testcontainers-go v0.38.0

go.sum

@@ -119,8 +119,6 @@ github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF
 github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
 github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
 github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
-github.com/sethvargo/go-retry v0.3.0 h1:EEt31A35QhrcRZtrYFDTBg91cqZVnFL2navjDrah2SE=
-github.com/sethvargo/go-retry v0.3.0/go.mod h1:mNX17F0C/HguQMyMyJxcnU471gOZGxCLyYaFyAZraas=
 github.com/shirou/gopsutil/v4 v4.25.5 h1:rtd9piuSMGeU8g1RMXjZs9y9luK5BwtnG7dZaQUJAsc=
 github.com/shirou/gopsutil/v4 v4.25.5/go.mod h1:PfybzyydfZcN+JMMjkF6Zb8Mq1A/VcogFFg7hj50W9c=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=