bluesky-nats-proxy/main.go

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"strings"
	"time"

	"github.com/gorilla/websocket"
	"github.com/nats-io/nats.go"
)

type BlueskyPost struct {
	Op   string `json:"op"`
	Repo string `json:"repo"`
	Record struct {
		Text      string    `json:"text"`
		CreatedAt time.Time `json:"createdAt"`
	} `json:"record"`
}

func main() {
	// Connect to NATS
	nc, err := nats.Connect(nats.DefaultURL)
	if err != nil {
		log.Fatal("Error connecting to NATS:", err)
	}
	defer nc.Close()

	// Connect to WebSocket
	c, _, err := websocket.DefaultDialer.Dial("wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post", nil)
	if err != nil {
		log.Fatal("Error connecting to WebSocket:", err)
	}
	defer c.Close()

	// Handle WebSocket messages
	go func() {
		for {
			_, message, err := c.ReadMessage()
			if err != nil {
				log.Println("Error reading WebSocket message:", err)
				return
			}

			var post BlueskyPost
			if err := json.Unmarshal(message, &post); err != nil {
				log.Println("Error unmarshaling JSON:", err)
				continue
			}

			// Extract user ID from repo field
			userID := strings.Split(post.Repo, ".")[0]

			// Determine post type (simplified for this example)
			postType := "text"
			if strings.Contains(post.Record.Text, "http") {
				postType = "link"
			}

			// Create a more detailed NATS subject
			subject := fmt.Sprintf("bsky.feed.post.user.%s.type.%s.time.%s", 
				userID, 
				postType, 
				post.Record.CreatedAt.Format("20060102"))

			// Publish the JSON message to NATS with the detailed subject
			if err := nc.Publish(subject, message); err != nil {
				log.Println("Error publishing to NATS:", err)
			} else {
				log.Printf("Published message to subject: %s", subject)
			}
		}
	}()

	// Wait for interrupt signal to gracefully shut down
	interrupt := make(chan os.Signal, 1)
	signal.Notify(interrupt, os.Interrupt)
	<-interrupt

	log.Println("Shutting down...")
}