Implement Broadcast Feature

realtime/events.go

@@ -1,20 +1,131 @@
-package realtime;
+package realtime
 
-// Channel Events
-const JOIN_EVENT = "phx_join"
-const REPLY_EVENT = "phx_reply"
+import (
+	"fmt"
+	"reflect"
+	"strings"
+)
 
-// DB Subscription Events
-const POSTGRES_CHANGE_EVENT = "postgres_changes"
+// Events that are used to communicate with the server
+const (
+	joinEvent  string = "phx_join"
+	replyEvent string = "phx_reply"
+	leaveEvent string = "phx_leave"
+	closeEvent string = "phx_close"
 
-// Broadcast Events
-const BROADCAST_EVENT = "broadcast"
+	// DB Subscription Events
+	postgresChangesEvent string = "postgres_changes"
 
-// Presence Events
-const PRESENCE_STATE_EVENT = "presence_state"
-const PRESENCE_DIFF_EVENT ="presence_diff"
+	// Broadcast Events
+	broadcastEvent string = "broadcast"
 
-// Other Events
-const SYS_EVENT = "system"
-const HEARTBEAT_EVENT = "heartbeat"
-const ACCESS_TOKEN_EVENT = "access_token"
+	// Presence Events
+	presenceStateEvent string = "presence_state"
+	presenceDiffEvent  string = "presence_diff"
+
+	// Other Events
+	systemEvent       string = "system"
+	heartbeatEvent    string = "heartbeat"
+	accessTokennEvent string = "access_token"
+)
+
+// Event "type" that the user can specify for channel to listen to
+const (
+	presenceEventType        string = "presence"
+	broadcastEventType       string = "broadcast"
+	postgresChangesEventType string = "postgres_changes"
+)
+
+// type eventFilter struct {}
+type eventFilter interface {}
+
+type postgresFilter struct {
+   Event  string `supabase:"required" json:"event"`
+	Schema string `supabase:"required" json:"schema"`
+	Table  string `supabase:"optional" json:"table,omitempty"`
+	Filter string `supabase:"optional" json:"filter,omitempty"`
+}
+
+type broadcastFilter struct {
+	Event string `supabase:"required"`
+}
+
+type presenceFilter struct {
+	Event string `supabase:"required"`
+}
+
+// Custom event that can be sent to the server
+type CustomEvent struct {
+   Event    string   `json:"event"`
+   Payload  any      `json:"payload"`
+   Type     string   `json:"type"`
+}
+
+// Verify if the given event type is supported
+func verifyEventType(eventType string) bool {
+	switch eventType {
+      case presenceEventType:
+         fallthrough
+      case broadcastEventType:
+         fallthrough
+      case postgresChangesEventType:
+         return true
+	}
+
+	return false
+}
+
+// Enforce client's filter object to follow a specific message
+// structure of certain events. Check messages.go for more
+// information on the struct of each event.
+// Only the following events are currently supported:
+//   - postgres_changes, broadcast, presence
+func createEventFilter(eventType string, filter map[string]string) (eventFilter, error) {
+	var filterType reflect.Type      // Type for filter
+	var filterConValue reflect.Value // Concrete value
+	var filterPtrValue reflect.Value // Pointer value to the concrete value
+	var missingFields []string
+
+	switch eventType {
+	case postgresChangesEvent:
+		filterPtrValue = reflect.ValueOf(&postgresFilter{})
+		break
+	case broadcastEvent:
+		filterPtrValue = reflect.ValueOf(&broadcastFilter{})
+		break
+	case presenceEventType:
+		filterPtrValue = reflect.ValueOf(&presenceFilter{})
+	default:
+		return nil, fmt.Errorf("Unsupported event type: %s", eventType)
+	}
+
+	// Get the underlying filter type to identify missing fields
+	filterConValue = filterPtrValue.Elem()
+	filterType = filterConValue.Type()
+	missingFields = make([]string, 0, filterType.NumField())
+
+	for i := 0; i < filterType.NumField(); i++ {
+		currField := filterType.Field(i)
+		currFieldName := strings.ToLower(currField.Name)
+		isRequired := currField.Tag.Get("supabase") == "required"
+
+		val, ok := filter[currFieldName]
+		if !ok && isRequired {
+			missingFields = append(missingFields, currFieldName)
+		}
+
+		// Set field to empty string when value for currFieldName is missing
+		filterConValue.Field(i).SetString(val)
+	}
+
+	if len(missingFields) != 0 {
+		return nil, fmt.Errorf("Criteria for %s is missing: %+v", eventType, missingFields)
+	}
+
+	filterFinal, ok := filterConValue.Interface().(eventFilter)
+	if !ok {
+		return nil, fmt.Errorf("Unexpected Error: cannot create event filter")
+	}
+
+	return filterFinal, nil
+}

realtime/messages.go

@@ -1,30 +1,160 @@
-package realtime;
+package realtime
 
-type TemplateMsg struct {
-   Event string `json:"event"`
-   Topic string `json:"topic"`
-   Ref string `json:"ref"`
+import (
+	"encoding/json"
+	"strconv"
+	"time"
+)
+
+// This is a general message strucutre. It follows the message protocol
+// of the phoenix server:
+/*
+   {
+      event: string,
+      topic: string,
+      payload: [key: string]: boolean | int | string | any,
+      ref: string
+   }
+*/
+type Msg struct {
+   Metadata
+   Payload any `json:"payload"`
+}
+
+// Generic message that contains raw payload. It can be used
+// as a tagged union, where the event field can be used to
+// determine the structure of the payload.
+type RawMsg struct {
+	Metadata
+	Payload json.RawMessage `json:"payload"`
 }
 
-type ConnectionMsg struct {
-   TemplateMsg
+// The other fields besides the payload that make up a message.
+// It describes other information about a message such as type of event,
+// the topic the message belongs to, and its reference.
+type Metadata struct {
+	Event string `json:"event"`
+	Topic string `json:"topic"`
+	Ref   string `json:"ref"`
+}
+
+// Payload for the conection message for when client first joins the channel. 
+// More info: https://supabase.com/docs/guides/realtime/protocol#connection
+type ConnectionPayload struct {
+   Config struct {
+      Broadcast struct {
+         Self bool `json:"self"`
+      } `json:"broadcast,omitempty"`
+
+      Presence struct {
+         Key string `json:"key"`
+      } `json:"presence,omitempty"`
 
-   Payload struct {
-      Data struct {
-         Schema string `json:"schema"`
-         Table string `json:"table"`
-         CommitTime string `json:"commit_timestamp"`
-         EventType string  `json:"eventType"`
-         New map[string]string `json:"new"`
-         Old map[string]string `json:"old"`
-         Errors string `json:"errors"`
-      } `json:"data"`
-   } `json:"payload"`
+      PostgresChanges []postgresFilter `json:"postgres_changes,omitempty"`
+   } `json:"config"`
 }
 
-type HearbeatMsg struct {
-   TemplateMsg
+// Payload of the server's first response of three upon joining channel. 
+// It contains details about subscribed postgres events.
+// More info: https://supabase.com/docs/guides/realtime/protocol#connection
+type ReplyPayload struct {
+   Response struct {
+      PostgresChanges []struct{
+         ID int `json:"id"`
+         postgresFilter
+      } `json:"postgres_changes"`
+   } `json:"response"`
+   Status string `json:"status"`
+}
+
+// Payload of the server's second response of three upon joining channel. 
+// It contains details about the status of subscribing to PostgresSQL.
+// More info: https://supabase.com/docs/guides/realtime/protocol#system-messages
+type SystemPayload struct {
+   Channel     string `json:"channel"`
+   Extension   string `json:"extension"`
+   Message     string `json:"message"`
+   Status      string `json:"status"`
+}
+
+// Payload of the server's third response of three upon joining channel. 
+// It contains details about the Presence feature of Supabase.
+// More info: https://supabase.com/docs/guides/realtime/protocol#state-update
+type PresenceStatePayload map[string]struct{
+   Metas []struct{
+      Ref   string   `json:"phx_ref"` 
+      Name  string   `json:"name"` 
+      T     float64  `json:"t"` 
+   } `json:"metas,omitempty"`
+}
+
+// Payload of the server's response when there is a postgres_changes event. 
+// More info: https://supabase.com/docs/guides/realtime/protocol#system-messages
+type PostgresCDCPayload struct {
+   Data struct {
+      Schema     string          `json:"schema"`
+      Table      string          `json:"table"`
+      CommitTime string          `json:"commit_timestamp"`
+      Record     map[string]any  `json:"record"`
+      Columns    []struct{
+         Name string `json:"name"`
+         Type string `json:"type"`
+      } `json:"columns"`
+      ActionType  string          `json:"type"`
+      Old        map[string]any  `json:"old_record"`
+      Errors     string          `json:"errors"`
+   } `json:"data"`
+   IDs []int `json:"ids"`
+}
+
+// Payload of the server's response when there is a broadcast event. 
+// More info: https://supabase.com/docs/guides/realtime/protocol#broadcast-message
+type BroadcastPayload struct {
+   Event    string `json:"event"`
+   Payload  any    `json:"payload"`
+   Type     string `json:"type"`
+}
+
+// create a template message
+func createMsgMetadata(event string, topic string) *Metadata {
+	return &Metadata{
+		Event: event,
+		Topic: topic,
+		Ref:   "",
+	}
+}
+
+// create a connection message depending on event type
+func createConnectionMessage(topic string, bindings []*binding) *Msg {
+	msg := &Msg{}
+
+   // Fill out the message template
+   msg.Metadata = *createMsgMetadata(joinEvent, topic)
+   msg.Metadata.Ref = strconv.FormatInt(time.Now().Unix(), 10)
+
+   // Fill out the payload
+   payload := &ConnectionPayload{}
+   for _, bind := range bindings {
+      filter := bind.filter
+      switch filter.(type) {
+         case postgresFilter:
+            if payload.Config.PostgresChanges == nil {
+               payload.Config.PostgresChanges = make([]postgresFilter, 0, 1)
+            }
+            payload.Config.PostgresChanges = append(payload.Config.PostgresChanges, filter.(postgresFilter))
+            break
+         case broadcastFilter:
+            payload.Config.Broadcast.Self = true
+            break
+         case presenceFilter:
+            payload.Config.Presence.Key = ""
+            break
+         default:
+            panic("TYPE ASSERTION FAILED: expecting one of postgresFilter, broadcastFilter, or presenceFilter")
+      }
+   }
+
+   msg.Payload = payload
 
-   Payload struct {
-   } `json:"payload"`
+	return msg
 }