Postgres Changes and CDC

---

Change Data Capture (CDC)

Change Data Capture is a pattern for tracking changes to database records. Instead of polling the database repeatedly, CDC captures changes as they occur and streams them to interested systems.

---

How PostgreSQL CDC Works

PostgreSQL has built-in support for CDC through its logical replication system.

Write-Ahead Logging (WAL)

Before PostgreSQL applies any change to disk, it writes to the WAL:

Database Operation:
INSERT INTO posts (title) VALUES ('Hello');

WAL Record Created:
┌─────────────────────────────────────────────────────────────┐
│ LSN: 0/1234567                                              │
│ Transaction ID: 5678                                        │
│ Table: public.posts                                         │
│ Operation: INSERT                                           │
│ New Tuple: (uuid, 'Hello', ...)                            │
│ Timestamp: 2024-01-15 10:30:00.123                         │
└─────────────────────────────────────────────────────────────┘

Logical Decoding

PostgreSQL can decode WAL into logical events:

WAL (Binary) ──→ Logical Decoder ──→ Change Events (Readable)

Change Event:
{
  "table": "posts",
  "schema": "public",
  "operation": "INSERT",
  "new_record": {...},
  "old_record": null
}

Replication Slots

A replication slot ensures change events aren't lost:

┌─────────────────────────────────────────────────────────────┐
│                    Replication Slot                          │
│                                                             │
│  - Tracks which WAL position has been consumed             │
│  - Prevents WAL from being deleted before reading          │
│  - Survives server restarts                                │
│  - Named identifier for the consumer                       │
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

Supabase Realtime and CDC

Supabase uses a publication/subscription model:

Database Setup

-- Supabase automatically creates a publication
-- You can check it with:
SELECT * FROM pg_publication;

-- Tables are added to the publication for realtime
-- This is managed in Supabase Dashboard under Database > Publications

Enabling Realtime for Tables

In the Supabase Dashboard:

1. Go to Database → Publications
2. Enable tables you want for realtime
3. Or use SQL:

-- Add a table to realtime publication
ALTER PUBLICATION supabase_realtime ADD TABLE posts;

-- Check which tables are in the publication
SELECT * FROM pg_publication_tables WHERE pubname = 'supabase_realtime';

---

Subscription Filters

You don't want every change—just the ones you care about.

Event Types

// All events
.on('postgres_changes', {
  event: '*',  // INSERT, UPDATE, DELETE
  schema: 'public',
  table: 'posts'
}, handler)

// Only inserts
.on('postgres_changes', {
  event: 'INSERT',
  schema: 'public',
  table: 'posts'
}, handler)

// Only updates
.on('postgres_changes', {
  event: 'UPDATE',
  schema: 'public',
  table: 'posts'
}, handler)

// Only deletes
.on('postgres_changes', {
  event: 'DELETE',
  schema: 'public',
  table: 'posts'
}, handler)

Row Filters

Filter by column values:

// Only posts in a specific channel
.on('postgres_changes', {
  event: 'INSERT',
  schema: 'public',
  table: 'posts',
  filter: 'channel_id=eq.123'
}, handler)

// Only high-priority items
.on('postgres_changes', {
  event: '*',
  schema: 'public',
  table: 'tasks',
  filter: 'priority=gt.5'
}, handler)

Filter Operators

Operator	Meaning	Example
eq	Equals	status=eq.active
neq	Not equals	status=neq.deleted
gt	Greater than	priority=gt.5
gte	Greater or equal	priority=gte.5
lt	Less than	priority=lt.5
lte	Less or equal	priority=lte.5
in	In list	status=in.(active,pending)

---

Change Payload Structure

When a change occurs, you receive a payload:

INSERT Payload

{
  schema: 'public',
  table: 'posts',
  commit_timestamp: '2024-01-15T10:30:00.000Z',
  eventType: 'INSERT',
  new: {
    id: 'uuid-123',
    title: 'Hello World',
    content: 'This is my first post',
    user_id: 'user-uuid',
    created_at: '2024-01-15T10:30:00.000Z'
  },
  old: {}  // Empty for INSERT
}

UPDATE Payload

{
  schema: 'public',
  table: 'posts',
  commit_timestamp: '2024-01-15T10:35:00.000Z',
  eventType: 'UPDATE',
  new: {
    id: 'uuid-123',
    title: 'Hello World - Updated',  // Changed
    content: 'This is my updated post',
    user_id: 'user-uuid',
    created_at: '2024-01-15T10:30:00.000Z'
  },
  old: {
    id: 'uuid-123'  // Primary key only by default
  }
}

DELETE Payload

{
  schema: 'public',
  table: 'posts',
  commit_timestamp: '2024-01-15T10:40:00.000Z',
  eventType: 'DELETE',
  new: {},  // Empty for DELETE
  old: {
    id: 'uuid-123'  // Primary key only
  }
}

---

Replica Identity

By default, the old record only contains the primary key. To get full old record:

Setting Replica Identity

-- Full old record in change events
ALTER TABLE posts REPLICA IDENTITY FULL;

-- Default: only primary key
ALTER TABLE posts REPLICA IDENTITY DEFAULT;

-- Using a specific index
ALTER TABLE posts REPLICA IDENTITY USING INDEX posts_unique_idx;

With FULL Replica Identity

// UPDATE now includes complete old record
{
  eventType: 'UPDATE',
  new: {
    id: 'uuid-123',
    title: 'New Title',
    content: 'New content'
  },
  old: {
    id: 'uuid-123',
    title: 'Old Title',      // Now included!
    content: 'Old content'    // Now included!
  }
}

Trade-offs

Setting	Pros	Cons
DEFAULT	Less WAL size	Limited old data
FULL	Complete change info	Larger WAL, more I/O
USING INDEX	Efficient for large rows	Must have unique index

---

RLS and Postgres Changes

Row Level Security filters what changes you receive:

How It Works

1. Database change occurs (INSERT/UPDATE/DELETE)
2. Realtime server receives change
3. For each subscriber:
   a. Get subscriber's JWT (auth context)
   b. Check RLS policy against change data
   c. If policy passes, send change
   d. If policy fails, don't send

Example

-- RLS Policy
CREATE POLICY "Users see own posts"
  ON posts FOR SELECT
  USING (user_id = auth.uid());

// User A subscribes
supabase.channel('posts').on('postgres_changes', {
  event: '*',
  schema: 'public',
  table: 'posts'
}, handler).subscribe()

// User B creates a post
await supabase.from('posts').insert({
  title: 'User B Post',
  user_id: userBId
})

// User A does NOT receive this event
// Because: userBId != userAId (fails RLS)

Important Implications

• Subscriptions automatically respect your security model
• No separate authorization logic needed for realtime
• BUT: More complex policies = more evaluation overhead

---

Handling Changes in Your Application

Pattern: Optimistic Updates with Realtime Sync

// Component state
const [posts, setPosts] = useState([])

// Initial load
useEffect(() => {
  supabase.from('posts').select('*').then(({ data }) => {
    setPosts(data)
  })
}, [])

// Subscribe to changes
useEffect(() => {
  const channel = supabase
    .channel('posts-changes')
    .on('postgres_changes', {
      event: '*',
      schema: 'public',
      table: 'posts'
    }, (payload) => {
      switch (payload.eventType) {
        case 'INSERT':
          setPosts(prev => [...prev, payload.new])
          break
        case 'UPDATE':
          setPosts(prev =>
            prev.map(p => p.id === payload.new.id ? payload.new : p)
          )
          break
        case 'DELETE':
          setPosts(prev =>
            prev.filter(p => p.id !== payload.old.id)
          )
          break
      }
    })
    .subscribe()

  return () => {
    supabase.removeChannel(channel)
  }
}, [])

Pattern: Deduplicate Local Changes

// Track pending local changes
const pendingChanges = new Set()

async function createPost(post) {
  const tempId = crypto.randomUUID()
  pendingChanges.add(tempId)

  // Optimistic update
  setPosts(prev => [...prev, { ...post, id: tempId }])

  // Actual insert
  const { data } = await supabase
    .from('posts')
    .insert(post)
    .select()
    .single()

  // Remove pending marker
  pendingChanges.delete(tempId)

  // Replace temp with real
  setPosts(prev =>
    prev.map(p => p.id === tempId ? data : p)
  )
}

// In realtime handler
if (payload.eventType === 'INSERT') {
  // Only add if not our own pending change
  if (!pendingChanges.has(payload.new.id)) {
    setPosts(prev => [...prev, payload.new])
  }
}

---

Key Takeaways

1. CDC captures changes as they happen: No polling needed
2. PostgreSQL WAL is the source: Reliable, transactional
3. Filters reduce noise: Only receive what you need
4. Replica Identity controls old data: FULL gives complete history
5. RLS applies automatically: Security is consistent
6. Update UI reactively: Handle INSERT, UPDATE, DELETE events

---

Next Steps

Beyond database changes, Supabase Realtime offers Broadcast and Presence channels for building collaborative features.

---

Change Data Capture turns your database into an event source. Every write becomes a message, enabling reactive architectures that respond to data as it changes.