This example demonstrates real-time synchronization from a Postgres JSONB column to Antfly using:

1. Linear Merge API for efficient batch syncing
2. LISTEN/NOTIFY for real-time change detection
3. Periodic full syncs to catch any missed changes

Perfect for keeping Antfly in sync with your existing Postgres database!

Features#

✅ Real-time Updates - Changes in Postgres instantly sync to Antfly via LISTEN/NOTIFY ✅ Efficient Batching - Rapid changes are batched together (1-second window) ✅ Content Hash Optimization - Unchanged documents are skipped (no unnecessary writes) ✅ Automatic Deletion - Documents deleted from Postgres are removed from Antfly ✅ Periodic Full Sync - Configurable full sync to ensure consistency ✅ Graceful Shutdown - Clean exit with statistics on Ctrl+C ✅ Production Ready - Connection pooling, error handling, metrics

Architecture#

┌─────────────────┐
│   Postgres DB   │
│  (JSONB table)  │
└────┬───────┬────┘
     │       │
     │       │ LISTEN/NOTIFY real-time
     │       │ 
     │       │
     │   ┌───▼────────────┐          ┌──────────────┐
     │   │  postgres-sync │─────────▶│    Antfly    │
     │   │    (daemon)    │  Linear  │  (search &   │
     │   └───▲────────────┘  Merge   │   storage)   │
     │       │                       └──────────────┘
     │       │
     └───────┘ Periodic "insurance" sync every 5 min

How It Works#

1. Initial Full Sync:
   • Queries all rows from Postgres table
   • Uses Linear Merge API to sync to Antfly
   • Content hashing skips unchanged documents

// FullSync performs a complete sync of all data from Postgres to Antfly
func (ps *PostgresSync) FullSync(ctx context.Context) error {
	startTime := time.Now()

	// Query all records from Postgres
	query := fmt.Sprintf(`
		SELECT %s, %s
		FROM %s
		ORDER BY %s
	`, ps.config.IDColumn, ps.config.DataColumn, ps.config.TableName, ps.config.IDColumn)

	rows, err := ps.pgPool.Query(ctx, query)
	if err != nil {
		return fmt.Errorf("failed to query Postgres: %w", err)
	}
	defer rows.Close()

	// Collect records in batches
	var records []struct {
		ID   string
		Data map[string]any
	}

	for rows.Next() {
		var id string
		var data []byte

		if err := rows.Scan(&id, &data); err != nil {
			return fmt.Errorf("failed to scan row: %w", err)
		}

		var jsonData map[string]any
		if err := json.Unmarshal(data, &jsonData); err != nil {
			log.Printf("Warning: Failed to parse JSON for ID %s: %v", id, err)
			continue
		}

		records = append(records, struct {
			ID   string
			Data map[string]any
		}{ID: id, Data: jsonData})
	}

	if err := rows.Err(); err != nil {
		return fmt.Errorf("error iterating rows: %w", err)
	}

	fmt.Printf("Full sync: Found %d records in Postgres\n", len(records))

	if len(records) == 0 {
		fmt.Println("No records to sync")
		ps.stats.mu.Lock()
		ps.stats.LastFullSync = time.Now()
		ps.stats.mu.Unlock()
		return nil
	}

	// Sync in batches using Linear Merge
	totalUpserted := 0
	totalSkipped := 0
	totalDeleted := 0
	cursor := ""

	for i := 0; i < len(records); i += ps.config.BatchSize {
		end := min(i+ps.config.BatchSize, len(records))
		batch := records[i:end]

		// Convert to Antfly records map
		antflyRecords := make(map[string]any)
		for _, rec := range batch {
			// Add metadata
			doc := make(map[string]any)
			doc["id"] = rec.ID
			doc["data"] = rec.Data
			doc["source"] = "postgres"
			doc["synced_at"] = time.Now().Format(time.RFC3339)

			antflyRecords[rec.ID] = doc
		}

		// Perform linear merge
		result, err := ps.antflyClient.LinearMerge(
			ctx,
			ps.config.AntflyTable,
			antfly.LinearMergeRequest{
				Records:      antflyRecords,
				LastMergedId: cursor,
				DryRun:       false,
			},
		)
		if err != nil {
			ps.stats.mu.Lock()
			ps.stats.TotalErrors++
			ps.stats.mu.Unlock()
			return fmt.Errorf("linear merge failed: %w", err)
		}

		totalUpserted += result.Upserted
		totalSkipped += result.Skipped
		totalDeleted += result.Deleted

		// Update cursor for next batch
		if result.NextCursor != "" {
			cursor = result.NextCursor
		} else {
			// Find max ID in batch
			maxID := ""
			for id := range antflyRecords {
				if id > maxID {
					maxID = id
				}
			}
			cursor = maxID
		}

		fmt.Printf("  Batch %d-%d: %d upserted, %d skipped, %d deleted\n",
			i+1, end, result.Upserted, result.Skipped, result.Deleted)
	}

	duration := time.Since(startTime)

	fmt.Printf("✓ Full sync complete: %d upserted, %d skipped, %d deleted in %v\n",
		totalUpserted, totalSkipped, totalDeleted, duration)

	// Update stats
	ps.stats.mu.Lock()
	ps.stats.TotalSynced += int64(totalUpserted)
	ps.stats.TotalSkipped += int64(totalSkipped)
	ps.stats.TotalDeleted += int64(totalDeleted)
	ps.stats.LastFullSync = time.Now()
	ps.stats.mu.Unlock()

	return nil
}

2. Real-time Updates (LISTEN/NOTIFY):

   • Postgres trigger fires on INSERT/UPDATE/DELETE
   • Sends notification with change details
   • Go daemon receives notification
   • Batches rapid changes (1-second window)
   • Syncs batch to Antfly via Linear Merge

3. Periodic Full Sync:

   • Runs every N minutes (configurable)
   • Ensures consistency if any notifications were missed
   • Catches documents modified outside triggers

Prerequisites#

1. Postgres Database#

# Using Docker
docker run --name postgres-antfly-demo \
  -e POSTGRES_PASSWORD=secret \
  -e POSTGRES_DB=antfly_demo \
  -p 5432:5432 \
  -d postgres:16

# Or use existing Postgres instance

2. Antfly Running#

cd /path/to/antfly
go run ./cmd/antfly swarm

3. Build the sync tool#

cd /path/to/antfly
go build -o postgres-sync ./examples/postgres-sync

Quick Start#

Step 1: Set up Postgres schema#

# Connect to your Postgres database
psql postgresql://postgres:secret@localhost:5432/antfly_demo

# Run the schema setup
\i examples/postgres-sync/schema.sql

This creates:

• documents table with JSONB data column
• Triggers for LISTEN/NOTIFY on changes
• Sample data (5 documents)

Step 2: Start the sync daemon#

export POSTGRES_URL="postgresql://postgres:secret@localhost:5432/antfly_demo"

./postgres-sync \
  --postgres "$POSTGRES_URL" \
  --antfly http://localhost:8080/api/v1 \
  --pg-table documents \
  --antfly-table postgres_docs \
  --create-table \
  --full-sync-interval 5m

You should see:

=== Postgres to Antfly Real-time Sync ===
Postgres: postgresql://postgres:***@localhost:5432/antfly_demo
Antfly: http://localhost:8080/api/v1
Table: documents.data -> postgres_docs
Full sync interval: 5m0s

✓ Created Antfly table 'postgres_docs'
Performing initial full sync...
Full sync: Found 5 records in Postgres
  Batch 1-5: 5 upserted, 0 skipped, 0 deleted
✓ Full sync complete: 5 upserted, 0 skipped, 0 deleted in 123ms

Starting real-time sync (LISTEN/NOTIFY)...
✓ Listening on channel 'documents_changes'
✓ Real-time sync active

Sync is running. Press Ctrl+C to stop.

Step 3: Test real-time sync#

In another terminal, connect to Postgres and make changes:

psql $POSTGRES_URL

-- Insert a new document
INSERT INTO documents (id, data) VALUES
  ('test_001', '{"title": "Real-time Test", "content": "This syncs instantly!"}');

-- Update a document
UPDATE documents
SET data = data || '{"updated": true}'
WHERE id = 'doc_001';

-- Delete a document
DELETE FROM documents WHERE id = 'test_001';

Watch the sync daemon output:

← Change detected: INSERT test_001
→ Real-time sync: 1 upserted, 0 skipped

← Change detected: UPDATE doc_001
→ Real-time sync: 1 upserted, 0 skipped

← Change detected: DELETE test_001
→ Real-time sync: 1 deleted

Configuration Options#

Usage Examples#

Basic Usage#

# Minimal config (uses environment variable)
export POSTGRES_URL="postgresql://user:pass@localhost/db"
./postgres-sync --create-table

Custom Table Names#

./postgres-sync \
  --postgres "$POSTGRES_URL" \
  --pg-table products \
  --id-column product_id \
  --data-column metadata \
  --antfly-table product_catalog

Disable Periodic Sync (Real-time Only)#

./postgres-sync \
  --postgres "$POSTGRES_URL" \
  --full-sync-interval 0

High-Volume Sync#

./postgres-sync \
  --postgres "$POSTGRES_URL" \
  --batch-size 5000 \
  --full-sync-interval 30m

Document Schema#

Each Postgres row is stored in Antfly as:

{
  "id": "doc_001",
  "data": {
    "title": "Getting Started",
    "content": "Welcome to Antfly",
    "category": "tutorial"
  },
  "source": "postgres",
  "synced_at": "2024-01-15T10:30:00Z"
}

• id: From Postgres ID column
• data: The JSONB column content
• source: Always "postgres"
• synced_at: Timestamp of last sync

Testing Real-time Sync#

We provide a demo SQL script with various test scenarios:

psql $POSTGRES_URL -f examples/postgres-sync/demo-changes.sql

This demonstrates:

• Single inserts
• Bulk inserts (batching)
• Updates
• Deletes
• Transactional changes
• Random update generator

Interactive Testing#

-- 1. Insert new documents
INSERT INTO documents (id, data) VALUES
  ('demo_001', '{"title": "Demo", "content": "Testing sync"}');

-- Watch sync daemon: "← Change detected: INSERT demo_001"

-- 2. Bulk insert (tests batching)
INSERT INTO documents (id, data)
SELECT
  'bulk_' || i,
  jsonb_build_object('title', 'Bulk Doc ' || i, 'index', i)
FROM generate_series(1, 100) AS i;

-- Watch sync daemon batch them together!

-- 3. Update multiple records
UPDATE documents
SET data = data || '{"updated": true}'
WHERE id LIKE 'bulk_%';

-- 4. Delete them
DELETE FROM documents WHERE id LIKE 'bulk_%';

Monitoring#

The sync daemon prints statistics every 30 seconds:

--- Sync Statistics ---
Total synced: 1,234
Total skipped: 5,678
Total deleted: 42
Real-time updates: 89
Errors: 0
Last full sync: 2m30s ago
Last real-time sync: 5s ago
----------------------

How LISTEN/NOTIFY Works#

Postgres Side#

When you INSERT/UPDATE/DELETE a row, a trigger fires:

CREATE TRIGGER documents_change_trigger
  AFTER INSERT OR UPDATE OR DELETE ON documents
  FOR EACH ROW
  EXECUTE FUNCTION notify_document_change();

The function sends a notification:

PERFORM pg_notify('documents_changes', json_build_object(
  'operation', 'INSERT',
  'id', NEW.id,
  'data', NEW.data,
  'timestamp', NOW()
)::text);

Go Side#

The sync daemon listens for notifications:

// StartRealtimeSync starts listening for Postgres notifications
func (ps *PostgresSync) StartRealtimeSync(ctx context.Context) error {
	// Use a dedicated connection for LISTEN
	conn, err := pgx.Connect(ctx, ps.config.PostgresURL)
	if err != nil {
		return fmt.Errorf("failed to create listen connection: %w", err)
	}
	defer func() {
		if err := conn.Close(ctx); err != nil {
			log.Printf("Warning: Failed to close connection: %v", err)
		}
	}()

	// Start listening for notifications
	channelName := ps.config.TableName + "_changes"
	_, err = conn.Exec(ctx, "LISTEN "+pgx.Identifier{channelName}.Sanitize())
	if err != nil {
		return fmt.Errorf("failed to LISTEN: %w", err)
	}

	fmt.Printf("✓ Listening on channel '%s'\n", channelName)

	// Buffer for batching rapid changes
	changeBatch := make(map[string]ChangeEvent)
	var batchMu sync.Mutex
	batchTicker := time.NewTicker(1 * time.Second)
	defer batchTicker.Stop()

	// Process batched changes
	processBatch := func() {
		batchMu.Lock()
		if len(changeBatch) == 0 {
			batchMu.Unlock()
			return
		}

		// Copy and clear batch
		toProcess := make(map[string]ChangeEvent, len(changeBatch))
		maps.Copy(toProcess, changeBatch)
		changeBatch = make(map[string]ChangeEvent)
		batchMu.Unlock()

		// Process the batch
		if err := ps.processBatchedChanges(ctx, toProcess); err != nil {
			log.Printf("Error processing batch: %v", err)
			ps.stats.mu.Lock()
			ps.stats.TotalErrors++
			ps.stats.mu.Unlock()
		}
	}

	// Goroutine to process batches periodically
	go func() {
		for {
			select {
			case <-batchTicker.C:
				processBatch()
			case <-ctx.Done():
				return
			}
		}
	}()

	// Listen for notifications
	for {
		notification, err := conn.WaitForNotification(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return nil // Context cancelled, clean exit
			}
			return fmt.Errorf("notification error: %w", err)
		}

		// Parse the notification payload
		var event ChangeEvent
		if err := json.Unmarshal([]byte(notification.Payload), &event); err != nil {
			log.Printf("Warning: Failed to parse notification: %v", err)
			continue
		}

		// Add to batch
		batchMu.Lock()
		changeBatch[event.ID] = event
		batchMu.Unlock()

		fmt.Printf("← Change detected: %s %s\n", event.Operation, event.ID)
	}
}

Batching Strategy#

Rapid changes are batched to avoid overwhelming Antfly:

• Changes accumulate in a 1-second window
• Batch is processed every second
• Multiple changes to the same document are de-duplicated
• Linear Merge API handles the batch efficiently

Example:

0.0s: INSERT doc_001
0.1s: UPDATE doc_001
0.3s: INSERT doc_002
0.5s: UPDATE doc_001
1.0s: → Process batch {doc_001, doc_002} (2 records)

Performance Characteristics#

Optimization Tips#

1. Batch Size: Increase for large tables (up to 10,000)
2. Full Sync Interval: Reduce if notifications are unreliable
3. Connection Pool: Increase for high-volume tables
4. Indexes: Add GIN index on JSONB for faster queries

Troubleshooting#

Connection Issues#

Error: failed to connect to Postgres

Solution: Check Postgres URL and network:

psql "$POSTGRES_URL" -c "SELECT 1"

No Notifications Received#

# Test notifications manually
psql $POSTGRES_URL

-- Terminal 1:
LISTEN documents_changes;

-- Terminal 2:
INSERT INTO documents (id, data) VALUES ('test', '{}');

-- Terminal 1 should show: Asynchronous notification received

Common issues:

• Trigger not created → Run schema.sql again
• Wrong channel name → Check --pg-table matches table name
• Connection dropped → Daemon will reconnect automatically

Documents Not Syncing#

Check the daemon logs for errors:

Error processing batch: linear merge failed: ...

Common issues:

• Antfly not running → Start Antfly first
• Table doesn't exist → Use --create-table flag
• Invalid JSON in JSONB column → Check Postgres data

High Memory Usage#

If syncing a very large table:

1. Reduce --batch-size (try 500 or 1000)
2. Increase --full-sync-interval (try 30m or 1h)
3. Add connection pool limits

Production Deployment#

Docker Compose#

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: production
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - postgres-data:/var/lib/postgresql/data
      - ./schema.sql:/docker-entrypoint-initdb.d/01-schema.sql

  antfly:
    image: antfly:latest
    command: ["swarm"]
    ports:
      - "8080:8080"

  postgres-sync:
    build: ./examples/postgres-sync
    environment:
      POSTGRES_URL: postgresql://postgres:${DB_PASSWORD}@postgres/production
    command:
      - --antfly=http://antfly:8080/api/v1
      - --create-table
      - --full-sync-interval=10m
    depends_on:
      - postgres
      - antfly
    restart: unless-stopped

volumes:
  postgres-data:

Kubernetes#

apiVersion: apps/v1
kind: Deployment
metadata:
  name: postgres-sync
spec:
  replicas: 1  # Only run one instance (LISTEN/NOTIFY is single-consumer)
  template:
    spec:
      containers:
      - name: postgres-sync
        image: antfly/postgres-sync:latest
        env:
        - name: POSTGRES_URL
          valueFrom:
            secretKeyRef:
              name: postgres-credentials
              key: url
        args:
        - --antfly=http://antfly-service:8080/api/v1
        - --create-table
        - --full-sync-interval=10m

systemd Service#

[Unit]
Description=Antfly Postgres Sync
After=network.target postgresql.service

[Service]
Type=simple
User=antfly
Environment=POSTGRES_URL=postgresql://localhost/production
ExecStart=/usr/local/bin/postgres-sync \
  --antfly=http://localhost:8080/api/v1 \
  --create-table \
  --full-sync-interval=10m
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

Advanced Use Cases#

Multi-table Sync#

Run multiple sync daemons for different tables:

# Terminal 1: Sync products table
./postgres-sync --pg-table products --antfly-table products_search

# Terminal 2: Sync customers table
./postgres-sync --pg-table customers --antfly-table customers_search

Conditional Sync#

Modify notify_document_change() to filter:

CREATE OR REPLACE FUNCTION notify_document_change()
RETURNS TRIGGER AS $$
BEGIN
  -- Only sync published documents
  IF (NEW.data->>'status' = 'published') THEN
    PERFORM pg_notify(...);
  END IF;
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

Transform Data During Sync#

Modify the sync tool to transform data:

// In processBatchedChanges()
doc := make(map[string]interface{})
doc["id"] = id
doc["data"] = jsonData

// Add computed fields
if title, ok := jsonData["title"].(string); ok {
  doc["title_lowercase"] = strings.ToLower(title)
}

// Add embeddings (if configured)
if content, ok := jsonData["content"].(string); ok {
  embedding := generateEmbedding(content)
  doc["embedding"] = embedding
}

Comparison with Other Approaches#

LISTEN/NOTIFY is the sweet spot for most use cases!

Limitations#

⚠️ Single Consumer: Only one sync daemon should run per table (LISTEN is not load-balanced)

⚠️ Notification Loss: If daemon is down, notifications are lost (periodic full sync recovers)

⚠️ Payload Size: Postgres notification payload is limited to 8KB (we only send ID + operation, not full data)

⚠️ Transaction Delay: Notifications only fire on COMMIT (delayed for long transactions)

Extending the Example#

Ideas for customization:

1. Add Filtering: Only sync certain document types
2. Add Enrichment: Generate embeddings during sync
3. Add Validation: Validate JSON schema before syncing
4. Add Metrics: Export Prometheus metrics
5. Add Dead Letter Queue: Store failed syncs for retry
6. Multi-tenancy: Support multiple databases

Related Examples#

• docsaf - Sync documentation files to Antfly
• Linear Merge API docs - See work-log/006-create-linear-merge-api/

Project Files#