[!] 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=