Refactor realtime to realtime

main.go

@@ -5,27 +5,11 @@ import (
 	"io/ioutil"
 	"log"
 	"net/http"
+	"time"
 )
 
-type Response struct {
-	Page       int    `json:"page"`
-	PerPage    int    `json:"perPage"`
-	TotalItems int    `json:"totalItems"`
-	TotalPages int    `json:"totalPages"`
-	Items      []Item `json:"items"`
-}
-
-type Item struct {
-	CollectionId   string `json:"collectionId"`
-	CollectionName string `json:"collectionName"`
-	Created        string `json:"created"`
-	Downloaded     bool   `json:"downloaded"`
-	Id             string `json:"id"`
-	Link           string `json:"link"`
-	Updated        string `json:"updated"`
-}
-
 const POCKETBASE_URL = `https://pocketbase-scratch.site.quack-lab.dev/api/collections`
+const POCKETBASE_REALTIME = `https://pocketbase-scratch.site.quack-lab.dev/api/realtime`
 const COLLECTION_NAME = "youtubedownload"
 const FULL_URL = POCKETBASE_URL + "/" + COLLECTION_NAME + "/records"
 
@@ -40,7 +24,7 @@ func main() {
 	defer res.Body.Close()
 	body, err := ioutil.ReadAll(res.Body)
 	if err != nil {
-		log.Println("Error reading response body:", err)
+		log.Printf("Error reading response body: %+v\n", err)
 		return
 	}
 	if res.StatusCode != http.StatusOK {
@@ -48,11 +32,19 @@ func main() {
 		return
 	}
 
-	var data Response
+	var data APIResponse
 	err = json.Unmarshal(body, &data)
 	if err != nil {
-		log.Println("Error unmarshaling JSON:", err)
+		log.Printf("Error unmarshaling JSON: %+v\n", err)
 		return
 	}
-	log.Printf("Data: %+v\n", data)
+	// log.Printf("Data: %+v\n", data)
+
+	listener := new(RealtimeListener)
+	listener.Url = POCKETBASE_REALTIME
+	listener.Collections = []string{COLLECTION_NAME}
+	listener.initialize()
+
+	log.Printf("zzz...\n")
+	time.Sleep(1 * time.Hour)
 }

realtime.go

@@ -0,0 +1,64 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"log"
+	"net/http"
+
+	"github.com/r3labs/sse"
+)
+
+type RealtimeListener struct {
+	Url         string
+	Collections []string
+	Chan        chan string
+	client      *sse.Client
+}
+
+type Subscription struct {
+	ClientId      string   `json:"clientId"`
+	Subscriptions []string `json:"subscriptions"`
+}
+
+func (listener RealtimeListener) handlePbConnect(msg *sse.Event) {
+	var connectEvent ConnectEvent
+	err := json.Unmarshal(msg.Data, &connectEvent)
+	if err != nil {
+		log.Printf("Error unmarshalling PB_CONNECT event: %v\n", err)
+		return
+	}
+	log.Printf("Client ID: %v\n", connectEvent.ClientId)
+
+	subscription := Subscription{
+		ClientId:      connectEvent.ClientId,
+		Subscriptions: []string{"workshop_link"},
+	}
+
+	body, err := json.Marshal(subscription)
+	if err != nil {
+		log.Printf("Error marshalling subscription body: %v\n", err)
+		return
+	}
+
+	resp, err := http.Post(POCKETBASE_REALTIME, "application/json", bytes.NewBuffer(body))
+	if err != nil {
+		log.Printf("Error posting subscription: %v\n", err)
+		return
+	}
+	defer resp.Body.Close()
+}
+
+func (listener RealtimeListener) handlePbEvent(msg *sse.Event) {
+	log.Println(msg)
+	// listener.Chan <- msg.Data
+}
+
+func (listener RealtimeListener) initialize() {
+	listener.client = sse.NewClient(listener.Url)
+	listener.client.Subscribe("PB_CONNECT", listener.handlePbConnect)
+
+	for _, collection := range listener.Collections {
+		listener.client.Subscribe(collection, listener.handlePbEvent)
+	}
+}