Supabase数据库变更监听 #

Dashboard > Database > Replication

步骤
├── 1. 找到目标表
├── 2. 开启Supabase Realtime
└── 3. 保存

-- 启用单个表
ALTER PUBLICATION supabase_realtime ADD TABLE messages;

-- 启用多个表
ALTER PUBLICATION supabase_realtime ADD TABLE messages, users, posts;

-- 查看已启用的表
SELECT schemaname, tablename 
FROM pg_publication_tables 
WHERE pubname = 'supabase_realtime';

const channel = supabase
  .channel('all-changes')
  .on(
    'postgres_changes',
    {
      event: '*',
      schema: 'public',
      table: 'messages',
    },
    (payload) => {
      const { eventType, new: newRecord, old: oldRecord } = payload
      
      switch (eventType) {
        case 'INSERT':
          console.log('New record:', newRecord)
          break
        case 'UPDATE':
          console.log('Updated from:', oldRecord)
          console.log('Updated to:', newRecord)
          break
        case 'DELETE':
          console.log('Deleted:', oldRecord)
          break
      }
    }
  )
  .subscribe()

const channel = supabase
  .channel('inserts')
  .on(
    'postgres_changes',
    {
      event: 'INSERT',
      schema: 'public',
      table: 'messages',
    },
    (payload) => {
      console.log('New message:', payload.new)
    }
  )
  .subscribe()

import { useState, useEffect } from 'react'
import { supabase } from '../lib/supabase'

interface Message {
  id: number
  content: string
  user_id: string
  created_at: string
}

export function ChatRoom({ roomId }: { roomId: string }) {
  const [messages, setMessages] = useState<Message[]>([])

  useEffect(() => {
    // 加载历史消息
    loadMessages()

    // 监听新消息
    const channel = supabase
      .channel(`room-${roomId}`)
      .on(
        'postgres_changes',
        {
          event: 'INSERT',
          schema: 'public',
          table: 'messages',
          filter: `room_id=eq.${roomId}`,
        },
        (payload) => {
          setMessages((prev) => [...prev, payload.new as Message])
        }
      )
      .subscribe()

    return () => {
      channel.unsubscribe()
    }
  }, [roomId])

  async function loadMessages() {
    const { data } = await supabase
      .from('messages')
      .select('*')
      .eq('room_id', roomId)
      .order('created_at', { ascending: true })

    setMessages(data || [])
  }

  async function sendMessage(content: string) {
    await supabase.from('messages').insert({
      room_id: roomId,
      content,
      user_id: (await supabase.auth.getUser()).data.user?.id,
    })
  }

  return (
    <div>
      <div className="messages">
        {messages.map((msg) => (
          <div key={msg.id}>{msg.content}</div>
        ))}
      </div>
      <input
        onKeyDown={(e) => {
          if (e.key === 'Enter') {
            sendMessage(e.currentTarget.value)
            e.currentTarget.value = ''
          }
        }}
      />
    </div>
  )
}

const channel = supabase
  .channel('updates')
  .on(
    'postgres_changes',
    {
      event: 'UPDATE',
      schema: 'public',
      table: 'users',
    },
    (payload) => {
      console.log('Before:', payload.old)
      console.log('After:', payload.new)
    }
  )
  .subscribe()

// 使用触发器记录字段变化
// 在数据库中创建
CREATE OR REPLACE FUNCTION notify_status_change()
RETURNS TRIGGER AS $$
BEGIN
    IF OLD.status IS DISTINCT FROM NEW.status THEN
        PERFORM pg_notify(
            'status_changed',
            json_build_object(
                'id', NEW.id,
                'old_status', OLD.status,
                'new_status', NEW.status
            )::text
        );
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

const channel = supabase
  .channel('deletes')
  .on(
    'postgres_changes',
    {
      event: 'DELETE',
      schema: 'public',
      table: 'messages',
    },
    (payload) => {
      console.log('Deleted message:', payload.old)
      // payload.old 包含被删除记录的数据
    }
  )
  .subscribe()

import { useState, useEffect } from 'react'
import { supabase } from '../lib/supabase'

export function TodoList() {
  const [todos, setTodos] = useState<any[]>([])

  useEffect(() => {
    loadTodos()

    const channel = supabase
      .channel('todos-changes')
      .on('postgres_changes', {
        event: 'INSERT',
        schema: 'public',
        table: 'todos',
      }, (payload) => {
        setTodos((prev) => [...prev, payload.new])
      })
      .on('postgres_changes', {
        event: 'DELETE',
        schema: 'public',
        table: 'todos',
      }, (payload) => {
        setTodos((prev) => prev.filter(t => t.id !== payload.old.id))
      })
      .on('postgres_changes', {
        event: 'UPDATE',
        schema: 'public',
        table: 'todos',
      }, (payload) => {
        setTodos((prev) => 
          prev.map(t => t.id === payload.new.id ? payload.new : t)
        )
      })
      .subscribe()

    return () => channel.unsubscribe()
  }, [])

  async function loadTodos() {
    const { data } = await supabase.from('todos').select('*')
    setTodos(data || [])
  }

  return (
    <ul>
      {todos.map((todo) => (
        <li key={todo.id}>{todo.title}</li>
      ))}
    </ul>
  )
}

// 只监听特定用户的变更
const channel = supabase
  .channel('user-todos')
  .on(
    'postgres_changes',
    {
      event: '*',
      schema: 'public',
      table: 'todos',
      filter: 'user_id=eq.user-123',
    },
    (payload) => {
      console.log('User todo changed:', payload)
    }
  )
  .subscribe()

// 组合过滤
const channel = supabase
  .channel('filtered')
  .on(
    'postgres_changes',
    {
      event: '*',
      schema: 'public',
      table: 'orders',
      filter: 'status=eq.pending AND total=gt.1000',
    },
    (payload) => {
      console.log('High value pending order:', payload)
    }
  )
  .subscribe()

{
  commit_timestamp: '2024-01-15T10:30:00Z',
  eventType: 'INSERT',
  schema: 'public',
  table: 'messages',
  new: {
    id: 1,
    content: 'Hello',
    user_id: 'user-123',
    created_at: '2024-01-15T10:30:00Z'
  },
  old: {},
  errors: null
}

{
  commit_timestamp: '2024-01-15T10:30:00Z',
  eventType: 'UPDATE',
  schema: 'public',
  table: 'messages',
  new: {
    id: 1,
    content: 'Hello Updated',
    user_id: 'user-123',
    created_at: '2024-01-15T10:30:00Z'
  },
  old: {
    id: 1,
    content: 'Hello',
    user_id: 'user-123',
    created_at: '2024-01-15T10:30:00Z'
  },
  errors: null
}

{
  commit_timestamp: '2024-01-15T10:30:00Z',
  eventType: 'DELETE',
  schema: 'public',
  table: 'messages',
  new: {},
  old: {
    id: 1,
    content: 'Hello',
    user_id: 'user-123',
    created_at: '2024-01-15T10:30:00Z'
  },
  errors: null
}

// 使用自定义Hook管理订阅
function useRealtimeSubscription(
  table: string,
  event: 'INSERT' | 'UPDATE' | 'DELETE' | '*',
  callback: (payload: any) => void,
  filter?: string
) {
  useEffect(() => {
    const channel = supabase
      .channel(`${table}-${event}`)
      .on(
        'postgres_changes',
        {
          event,
          schema: 'public',
          table,
          ...(filter && { filter }),
        },
        callback
      )
      .subscribe()

    return () => {
      channel.unsubscribe()
    }
  }, [table, event, callback, filter])
}

const channel = supabase
  .channel('error-handling')
  .on('postgres_changes', { ... }, (payload) => {
    if (payload.errors) {
      console.error('Subscription error:', payload.errors)
      return
    }
    // 处理数据
  })
  .subscribe((status, err) => {
    if (err) {
      console.error('Connection error:', err)
    }
  })