mcpField
S
supabase-realtime

supabase-realtime

github.com/nice-wolf-studio开发19321 次安装53 周前更新
代码仓库

关于 supabase-realtime

面向开发者的 Supabase Realtime 命令行操作指南，适用于编写监听器和简单监控脚本。内容涵盖数据库变更、广播与在线状态的 WebSocket 模式，并提供 REST 轮询作为替代方案。

主要功能

  1. 01WebSocket 连接示例
  2. 02表变更订阅与筛选
  3. 03广播消息模式
  4. 04在线状态跟踪示例
  5. 05REST 轮询替代方案

适用场景

  1. 01开发期间监控表变更
  2. 02在脚本中串流日志或事件
  3. 03测试 Supabase Realtime 订阅

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/nice-wolf-studio/claude-code-supabase-skills --skill supabase-realtime

安装后在 Claude Code 中输入/supabase-realtime确认已加载。

详细介绍

Supabase Realtime

概述

本技能介绍如何使用 Supabase Realtime 功能。Realtime 支持通过 WebSocket 连接监听数据库变更、广播消息和跟踪在线状态。

注意： Realtime 操作需要 WebSocket 支持，而在 bash 中实现较为复杂。本技能重点介绍使用现有工具的实用模式和示例。

前置条件

必需的环境变量：

export SUPABASE_URL="https://your-project.supabase.co"
export SUPABASE_KEY="your-anon-or-service-role-key"

其他工具：

  • 用于 WebSocket 连接的 websocat 或 wscat
  • 用于处理 JSON 的 jq

安装 websocat：

# macOS
brew install websocat

# Linux
wget https://github.com/vi/websocat/releases/download/v1.12.0/websocat.x86_64-unknown-linux-musl
chmod +x websocat.x86_64-unknown-linux-musl
sudo mv websocat.x86_64-unknown-linux-musl /usr/local/bin/websocat

WebSocket 连接

连接到 Supabase Realtime：

SUPABASE_URL="https://your-project.supabase.co"
SUPABASE_KEY="your-anon-key"

# Extract WebSocket URL (replace https:// with wss://)
WS_URL=$(echo "$SUPABASE_URL" | sed 's/https:/wss:/')

# Connect to realtime
websocat "${WS_URL}/realtime/v1/websocket?apikey=${SUPABASE_KEY}&vsn=1.0.0"

数据库变更订阅

订阅表变更

监听表中的所有变更：

#!/bin/bash

SUPABASE_URL="https://your-project.supabase.co"
SUPABASE_KEY="your-anon-key"
WS_URL=$(echo "$SUPABASE_URL" | sed 's/https:/wss:/')

# Create subscription message
SUB_MESSAGE='{
  "topic": "realtime:public:users",
  "event": "phx_join",
  "payload": {},
  "ref": "1"
}'

# Connect and subscribe
echo "$SUB_MESSAGE" | websocat "${WS_URL}/realtime/v1/websocket?apikey=${SUPABASE_KEY}&vsn=1.0.0"

订阅特定事件：

# Listen for INSERT events only
SUB_MESSAGE='{
  "topic": "realtime:public:users",
  "event": "phx_join",
  "payload": {
    "config": {
      "postgres_changes": [
        {
          "event": "INSERT",
          "schema": "public",
          "table": "users"
        }
      ]
    }
  },
  "ref": "1"
}'

echo "$SUB_MESSAGE" | websocat "${WS_URL}/realtime/v1/websocket?apikey=${SUPABASE_KEY}&vsn=1.0.0"

订阅 UPDATE 事件：

SUB_MESSAGE='{
  "topic": "realtime:public:products",
  "event": "phx_join",
  "payload": {
    "config": {
      "postgres_changes": [
        {
          "event": "UPDATE",
          "schema": "public",
          "table": "products"
        }
      ]
    }
  },
  "ref": "1"
}'

订阅 DELETE 事件：

SUB_MESSAGE='{
  "topic": "realtime:public:posts",
  "event": "phx_join",
  "payload": {
    "config": {
      "postgres_changes": [
        {
          "event": "DELETE",
          "schema": "public",
          "table": "posts"
        }
      ]
    }
  },
  "ref": "1"
}'

订阅所有事件（*、INSERT、UPDATE、DELETE）：

SUB_MESSAGE='{
  "topic": "realtime:public:orders",
  "event": "phx_join",
  "payload": {
    "config": {
      "postgres_changes": [
        {
          "event": "*",
          "schema": "public",
          "table": "orders"
        }
      ]
    }
  },
  "ref": "1"
}'
筛选订阅

监听符合筛选条件的变更：

# Only listen to changes where status = 'active'
SUB_MESSAGE='{
  "topic": "realtime:public:users",
  "event": "phx_join",
  "payload": {
    "config": {
      "postgres_changes": [
        {
          "event": "*",
          "schema": "public",
          "table": "users",
          "filter": "status=eq.active"
        }
      ]
    }
  },
  "ref": "1"
}'

广播消息

发送广播消息

向频道广播消息：

#!/bin/bash

SUPABASE_URL="https://your-project.supabase.co"
SUPABASE_KEY="your-anon-key"
WS_URL=$(echo "$SUPABASE_URL" | sed 's/https:/wss:/')

# Join channel first
JOIN_MESSAGE='{
  "topic": "realtime:chat-room-1",
  "event": "phx_join",
  "payload": {
    "config": {
      "broadcast": {
        "self": true
      }
    }
  },
  "ref": "1"
}'

# Broadcast message
BROADCAST_MESSAGE='{
  "topic": "realtime:chat-room-1",
  "event": "broadcast",
  "payload": {
    "type": "message",
    "event": "new_message",
    "payload": {
      "user": "Alice",
      "message": "Hello, World!"
    }
  },
  "ref": "2"
}'

# Send messages
{
  echo "$JOIN_MESSAGE"
  sleep 1
  echo "$BROADCAST_MESSAGE"
} | websocat "${WS_URL}/realtime/v1/websocket?apikey=${SUPABASE_KEY}&vsn=1.0.0"
监听广播消息

接收广播消息：

# Join channel and listen
JOIN_MESSAGE='{
  "topic": "realtime:chat-room-1",
  "event": "phx_join",
  "payload": {
    "config": {
      "broadcast": {
        "self": false
      }
    }
  },
  "ref": "1"
}'

echo "$JOIN_MESSAGE" | websocat "${WS_URL}/realtime/v1/websocket?apikey=${SUPABASE_KEY}&vsn=1.0.0"

在线状态跟踪

跟踪在线状态

加入带有在线状态的频道：

PRESENCE_MESSAGE='{
  "topic": "realtime:lobby",
  "event": "phx_join",
  "payload": {
    "config": {
      "presence": {
        "key": "user-123"
      }
    }
  },
  "ref": "1"
}'

# Track presence state
TRACK_MESSAGE='{
  "topic": "realtime:lobby",
  "event": "presence",
  "payload": {
    "type": "presence",
    "event": "track",
    "payload": {
      "user_id": "123",
      "username": "Alice",
      "status": "online"
    }
  },
  "ref": "2"
}'
取消跟踪在线状态

退出在线状态跟踪：

UNTRACK_MESSAGE='{
  "topic": "realtime:lobby",
  "event": "presence",
  "payload": {
    "type": "presence",
    "event": "untrack"
  },
  "ref": "3"
}'

实用模式

持续监听脚本
#!/bin/bash
# listen-to-changes.sh

SUPABASE_URL="https://your-project.supabase.co"
SUPABASE_KEY="your-anon-key"
WS_URL=$(echo "$SUPABASE_URL" | sed 's/https:/wss:/')
TABLE="users"

echo "Listening for changes on $TABLE table..."

# Subscribe to changes
SUB_MESSAGE='{
  "topic": "realtime:public:'"$TABLE"'",
  "event": "phx_join",
  "payload": {
    "config": {
      "postgres_changes": [
        {
          "event": "*",
          "schema": "public",
          "table": "'"$TABLE"'"
        }
      ]
    }
  },
  "ref": "1"
}'

# Listen continuously
echo "$SUB_MESSAGE" | websocat "${WS_URL}/realtime/v1/websocket?apikey=${SUPABASE_KEY}&vsn=1.0.0" | \
while IFS= read -r line; do
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $line" | jq '.'
done
使用处理器处理变更
#!/bin/bash
# process-changes.sh

handle_insert() {
    local record="$1"
    echo "New record inserted:"
    echo "$record" | jq '.payload.record'

    # Your custom logic here
    # Example: Send notification, update cache, etc.
}

handle_update() {
    local old_record="$1"
    local new_record="$2"
    echo "Record updated:"
    echo "Old: $(echo "$old_record" | jq -c '.')"
    echo "New: $(echo "$new_record" | jq -c '.')"
}

handle_delete() {
    local record="$1"
    echo "Record deleted:"
    echo "$record" | jq '.payload.old_record'
}

# Listen and process
websocat "${WS_URL}/realtime/v1/websocket?apikey=${SUPABASE_KEY}&vsn=1.0.0" | \
while IFS= read -r line; do
    event_type=$(echo "$line" | jq -r '.payload.data.type // empty')

    case "$event_type" in
        "INSERT")
            handle_insert "$(echo "$line" | jq '.payload.data')"
            ;;
        "UPDATE")
            handle_update \
                "$(echo "$line" | jq '.payload.data.old_record')" \
                "$(echo "$line" | jq '.payload.data.record')"
            ;;
        "DELETE")
            handle_delete "$(echo "$line" | jq '.payload.data')"
            ;;
    esac
done
多表监听器
#!/bin/bash
# listen-multiple-tables.sh

TABLES=("users" "posts" "comments")

for table in "${TABLES[@]}"; do
    (
        echo "Starting listener for $table"
        SUB_MESSAGE='{
          "topic": "realtime:public:'"$table"'",
          "event": "phx_join",
          "payload": {
            "config": {
              "postgres_changes": [{"event": "*", "schema": "public", "table": "'"$table"'"}]
            }
          },
          "ref": "1"
        }'

        echo "$SUB_MESSAGE" | websocat "${WS_URL}/realtime/v1/websocket?apikey=${SUPABASE_KEY}&vsn=1.0.0" | \
        while IFS= read -r line; do
            echo "[$table] $line"
        done
    ) &
done

wait

消息格式

订阅确认
{
  "event": "phx_reply",
  "payload": {
    "response": {
      "postgres_changes": [
        {
          "id": "12345",
          "event": "*",
          "schema": "public",
          "table": "users"
        }
      ]
    },
    "status": "ok"
  },
  "ref": "1",
  "topic": "realtime:public:users"
}
INSERT 事件
{
  "event": "postgres_changes",
  "payload": {
    "data": {
      "commit_timestamp": "2023-01-01T12:00:00Z",
      "record": {
        "id": 123,
        "name": "John Doe",
        "email": "[email protected]"
      },
      "schema": "public",
      "table": "users",
      "type": "INSERT"
    },
    "ids": [12345]
  },
  "topic": "realtime:public:users"
}
UPDATE 事件
{
  "event": "postgres_changes",
  "payload": {
    "data": {
      "commit_timestamp": "2023-01-01T12:00:00Z",
      "old_record": {
        "id": 123,
        "name": "John Doe"
      },
      "record": {
        "id": 123,
        "name": "Jane Doe"
      },
      "schema": "public",
      "table": "users",
      "type": "UPDATE"
    }
  }
}
DELETE 事件
{
  "event": "postgres_changes",
  "payload": {
    "data": {
      "commit_timestamp": "2023-01-01T12:00:00Z",
      "old_record": {
        "id": 123,
        "name": "John Doe"
      },
      "schema": "public",
      "table": "users",
      "type": "DELETE"
    }
  }
}

替代方案：REST 轮询

对于 WebSocket 不实用的简单场景，可以考虑使用轮询：

#!/bin/bash
# poll-changes.sh

source "$(dirname "${BASH_SOURCE[0]}")/../../scripts/supabase-api.sh"

LAST_TIMESTAMP=$(date -u +%Y-%m-%dT%H:%M:%SZ)

while true; do
    # Get records created/updated since last check
    new_records=$(supabase_get "/rest/v1/users?updated_at=gt.${LAST_TIMESTAMP}&order=updated_at.asc")

    if [[ "$new_records" != "[]" ]]; then
        echo "New changes detected:"
        echo "$new_records" | jq '.'

        # Update timestamp
        LAST_TIMESTAMP=$(echo "$new_records" | jq -r '.[-1].updated_at')
    fi

    # Poll every 5 seconds
    sleep 5
done

Realtime 配置

在 Supabase Dashboard 中启用 Realtime：

  1. 前往 Database > Replication
  2. 为要监听的表启用复制
  3. 选择要发布的事件（INSERT、UPDATE、DELETE）

行级安全： Realtime 遵循 RLS 策略。用户只能接收自己有权访问的行的变更。

限制

  • WebSocket 连接需要持续管理连接
  • bash 并不适合处理 WebSocket（生产环境可考虑使用 Node.js/Python）
  • 连接断开时需要重连逻辑
  • Realtime 受 Supabase 套餐的连接数限制

使用场景

适合在 bash 中使用 Realtime 的场景：

  • 开发/调试工具
  • 简单的监控脚本
  • 日志流传输
  • 测试实时功能

更适合使用其他语言的场景：

  • 生产环境聊天应用
  • 复杂的在线状态跟踪
  • 多频道协调
  • 需要自动重连

API 文档

完整的 Supabase Realtime 文档：https://supabase.com/docs/guides/realtime

评论

支持 Markdown