feat: add Supabase Realtime integration (#20)

README.md

@@ -2,8 +2,8 @@ An isomorphic Go client for Supabase.
 
 ## Features
 
-- [ ] Integration with [Supabase.Realtime](https://github.com/supabase-community/realtime-go)
-  - Realtime listeners for database changes
+- [x] Integration with [Supabase.Realtime](https://github.com/supabase-community/realtime-go)
+  - Realtime listeners for database changes (Broadcast, Presence, Postgres CDC)
 - [x] Integration with [Postgrest](https://github.com/supabase-community/postgrest-go)
   - Access your database using a REST API generated from your schema & database functions
 - [x] Integration with [Gotrue](https://github.com/supabase-community/gotrue-go)
@@ -122,3 +122,66 @@ client.EnableTokenAutoRefresh(session)
 // - Retry failed refreshes with exponential backoff
 // - Update all service clients with new tokens
 ```
+
+### Realtime
+
+The client includes support for Supabase Realtime features including Broadcast, Presence, and Postgres Change Data Capture (CDC).
+
+**Important:** The Realtime client uses lazy connection. You must explicitly call `Connect()` before subscribing to channels.
+
+#### Basic Usage
+
+```go
+ctx := context.Background()
+
+// Connect to the Realtime server
+err = client.Realtime.Connect(ctx)
+if err != nil {
+    log.Fatal("Failed to connect:", err)
+}
+defer client.Realtime.Disconnect()
+
+// Create and subscribe to a channel
+channel := client.Realtime.Channel("room:123", nil)
+err = channel.Subscribe(ctx, func(state realtime.SubscribeState, err error) {
+    if err != nil {
+        log.Printf("Subscription error: %v", err)
+        return
+    }
+    log.Printf("Subscription state: %v", state)
+})
+```
+
+#### Listening for Postgres Changes
+
+```go
+channel.OnPostgresChange("INSERT", func(event realtime.PostgresChangeEvent) {
+    log.Printf("New row inserted in %s: %s", event.Table, event.Payload)
+})
+```
+
+#### Broadcast Messages
+
+```go
+// Listen for broadcast events
+channel.OnBroadcast("cursor-move", func(payload json.RawMessage) {
+    log.Printf("Cursor moved: %s", payload)
+})
+
+// Send a broadcast message
+channel.SendBroadcast("cursor-move", map[string]interface{}{
+    "x": 100,
+    "y": 200,
+})
+```
+
+#### Non-Standard URLs
+
+For self-hosted instances or custom URL formats where the project reference cannot be automatically extracted, provide it explicitly:
+
+```go
+options := &supabase.ClientOptions{
+    ProjectRef: "your-project-ref",
+}
+client, err := supabase.NewClient("https://supabase.yourcompany.com", key, options)
+```

client.go

@@ -3,12 +3,16 @@ package supabase
 import (
 	"errors"
 	"log"
+	"net"
+	"net/url"
+	"strings"
 	"time"
 
 	"github.com/supabase-community/auth-go"
 	"github.com/supabase-community/auth-go/types"
 	"github.com/supabase-community/functions-go"
 	"github.com/supabase-community/postgrest-go"
+	"github.com/supabase-community/realtime-go/realtime"
 	storage_go "github.com/supabase-community/storage-go"
 )
 
@@ -26,25 +30,43 @@ type Client struct {
 	// Auth is an interface. We don't need a pointer to an interface.
 	Auth      auth.Client
 	Functions *functions.Client
+	Realtime  realtime.IRealtimeClient
 	options   clientOptions
 }
 
 type clientOptions struct {
-	url     string
-	headers map[string]string
+	url        string
+	headers    map[string]string
+	projectRef string
 }
 
 type ClientOptions struct {
-	Headers map[string]string
-	Schema  string
+	Headers    map[string]string
+	Schema     string
+	ProjectRef string
+}
+
+func extractProjectRef(rawURL string) string {
+	parsedURL, err := url.Parse(rawURL)
+	if err != nil {
+		return ""
+	}
+	host := parsedURL.Hostname()
+	if host == "localhost" || net.ParseIP(host) != nil {
+		return ""
+	}
+	parts := strings.Split(host, ".")
+	if len(parts) >= 2 {
+		return parts[0]
+	}
+	return ""
 }
 
 // NewClient creates a new Supabase client.
 // url is the Supabase URL.
 // key is the Supabase API key.
 // options is the Supabase client options.
 func NewClient(url, key string, options *ClientOptions) (*Client, error) {
-
 	if url == "" || key == "" {
 		return nil, errors.New("url and key are required")
 	}
@@ -65,6 +87,14 @@ func NewClient(url, key string, options *ClientOptions) (*Client, error) {
 	// map is pass by reference, so this gets updated by rest of function
 	client.options.headers = headers
 
+	var projectRef string
+	if options != nil && options.ProjectRef != "" {
+		projectRef = options.ProjectRef
+	} else {
+		projectRef = extractProjectRef(url)
+	}
+	client.options.projectRef = projectRef
+
 	var schema string
 	if options != nil && options.Schema != "" {
 		schema = options.Schema
@@ -77,6 +107,10 @@ func NewClient(url, key string, options *ClientOptions) (*Client, error) {
 	client.Auth = auth.New(url, key).WithCustomAuthURL(url + AUTH_URL)
 	client.Functions = functions.NewClient(url+FUNCTIONS_URL, key, headers)
 
+	if projectRef != "" {
+		client.Realtime = realtime.NewRealtimeClient(projectRef, key)
+	}
+
 	return client, nil
 }
 
@@ -158,7 +192,11 @@ func (c *Client) UpdateAuthSession(session types.Session) {
 	c.Auth = c.Auth.WithToken(session.AccessToken)
 	c.rest.SetAuthToken(session.AccessToken)
 	c.options.headers["Authorization"] = "Bearer " + session.AccessToken
+
+	if c.Realtime != nil {
+		_ = c.Realtime.SetAuth(session.AccessToken)
+	}
+
 	c.Storage = storage_go.NewClient(c.options.url+STORAGE_URL, session.AccessToken, c.options.headers)
 	c.Functions = functions.NewClient(c.options.url+FUNCTIONS_URL, session.AccessToken, c.options.headers)
-
 }

client_test.go

@@ -63,3 +63,36 @@ func TestFunctions(t *testing.T) {
 	}
 	t.Logf("function invokation result: %v", result)
 }
+
+func TestNewClientWithProjectRef(t *testing.T) {
+	options := &supabase.ClientOptions{
+		ProjectRef: "testproject",
+	}
+	client, err := supabase.NewClient("https://custom.domain.com", API_KEY, options)
+	if err != nil {
+		t.Fatalf("cannot initialize client: %v", err)
+	}
+	if client.Realtime == nil {
+		t.Error("Realtime client should be initialized when ProjectRef is provided")
+	}
+}
+
+func TestNewClientRealtimeNil(t *testing.T) {
+	client, err := supabase.NewClient("http://localhost:54321", API_KEY, nil)
+	if err != nil {
+		t.Fatalf("cannot initialize client: %v", err)
+	}
+	if client.Realtime != nil {
+		t.Error("Realtime client should be nil for localhost URLs without ProjectRef")
+	}
+}
+
+func TestNewClientRealtimeInitialized(t *testing.T) {
+	client, err := supabase.NewClient(API_URL, API_KEY, nil)
+	if err != nil {
+		t.Fatalf("cannot initialize client: %v", err)
+	}
+	if client.Realtime == nil {
+		t.Error("Realtime client should be initialized for standard Supabase URLs")
+	}
+}

go.mod

@@ -1,6 +1,6 @@
 module github.com/supabase-community/supabase-go
 
-go 1.21.1
+go 1.22
 
 toolchain go1.22.1
 
@@ -9,12 +9,14 @@ require (
 	github.com/supabase-community/auth-go v1.4.0
 	github.com/supabase-community/functions-go v0.0.0-20220927045802-22373e6cb51d
 	github.com/supabase-community/postgrest-go v0.0.11
+	github.com/supabase-community/realtime-go v0.1.1
 	github.com/supabase-community/storage-go v0.7.0
 )
 
 require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/tomnomnom/linkheader v0.0.0-20180905144013-02ca5825eb80 // indirect
+	nhooyr.io/websocket v1.8.11 // indirect
 )
 
 replace github.com/supabase-community/postgrest-go => github.com/roja/postgrest-go v0.0.11

go.sum

@@ -12,13 +12,17 @@ github.com/roja/functions-go v0.0.2 h1:fh+1XHtZ7HdJJAwHoJCyT6IHIHXIBpQ6fcS4dWDCK
 github.com/roja/functions-go v0.0.2/go.mod h1:nnIju6x3+OZSojtGQCQzu0h3kv4HdIZk+UWCnNxtSak=
 github.com/roja/postgrest-go v0.0.11 h1:vGgsfL4+Tvkpbukneo4fPy/43gbeWSSlKmum9+pwDCI=
 github.com/roja/postgrest-go v0.0.11/go.mod h1:cw6LfzMyK42AOSBA1bQ/HZ381trIJyuui2GWhraW7Cc=
-github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
-github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/supabase-community/auth-go v1.4.0 h1:5dS0NB9TKdT3zwaQqHJhrTFXvAcIC2/EDVC9dEV0wnY=
 github.com/supabase-community/auth-go v1.4.0/go.mod h1:Ww/iTwJNJZh//HwEl0i/vgC2JUgRwU+s3B8HfBqClas=
+github.com/supabase-community/realtime-go v0.1.1 h1:2S2Q2TgPcvnFMjutsNRRTCXnBdNd6hLlZ6GqE3JX1UY=
+github.com/supabase-community/realtime-go v0.1.1/go.mod h1:vp4ArMwXO9iMOxpR+h7sSP2g7Ei3JRzpkttXLXP4i+U=
 github.com/supabase-community/storage-go v0.7.0 h1:cJ8HLbbnL54H5rHPtHfiwtpRwcbDfA3in9HL/ucHnqA=
 github.com/supabase-community/storage-go v0.7.0/go.mod h1:oBKcJf5rcUXy3Uj9eS5wR6mvpwbmvkjOtAA+4tGcdvQ=
 github.com/tomnomnom/linkheader v0.0.0-20180905144013-02ca5825eb80 h1:nrZ3ySNYwJbSpD6ce9duiP+QkD3JuLCcWkdaehUS/3Y=
 github.com/tomnomnom/linkheader v0.0.0-20180905144013-02ca5825eb80/go.mod h1:iFyPdL66DjUD96XmzVL3ZntbzcflLnznH0fr99w5VqE=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+nhooyr.io/websocket v1.8.11 h1:f/qXNc2/3DpoSZkHt1DQu6rj4zGC8JmkkLkWss0MgN0=
+nhooyr.io/websocket v1.8.11/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c=

internal_test.go

@@ -0,0 +1,76 @@
+package supabase
+
+import "testing"
+
+func TestExtractProjectRef(t *testing.T) {
+	tests := []struct {
+		name     string
+		url      string
+		expected string
+	}{
+		{
+			name:     "standard supabase.co URL",
+			url:      "https://abc.supabase.co",
+			expected: "abc",
+		},
+		{
+			name:     "supabase.co URL with path",
+			url:      "https://myproject.supabase.co/rest/v1",
+			expected: "myproject",
+		},
+		{
+			name:     "supabase.in regional URL",
+			url:      "https://abc.supabase.in",
+			expected: "abc",
+		},
+		{
+			name:     "custom domain",
+			url:      "https://abc.example.com",
+			expected: "abc",
+		},
+		{
+			name:     "localhost URL",
+			url:      "http://localhost:54321",
+			expected: "",
+		},
+		{
+			name:     "localhost without port",
+			url:      "http://localhost",
+			expected: "",
+		},
+		{
+			name:     "IPv4 address",
+			url:      "http://127.0.0.1:54321",
+			expected: "",
+		},
+		{
+			name:     "IPv6 address",
+			url:      "http://[::1]:54321",
+			expected: "",
+		},
+		{
+			name:     "invalid URL",
+			url:      "not-a-url",
+			expected: "",
+		},
+		{
+			name:     "empty URL",
+			url:      "",
+			expected: "",
+		},
+		{
+			name:     "single label hostname",
+			url:      "http://supabase",
+			expected: "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := extractProjectRef(tt.url)
+			if result != tt.expected {
+				t.Errorf("extractProjectRef(%q) = %q, want %q", tt.url, result, tt.expected)
+			}
+		})
+	}
+}