path: root/main.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"log/slog"
	"os"
	"os/signal"
	"runtime"
	"time"

	_ "github.com/bluesky-social/indigo/api/bsky"
	"github.com/bluesky-social/jetstream/pkg/client"
	"github.com/bluesky-social/jetstream/pkg/client/schedulers/parallel"
	"github.com/bluesky-social/jetstream/pkg/models"
	"github.com/nats-io/nats.go"
)

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

	// Create Jetstream client with proper configuration
	config := &client.ClientConfig{
		WebsocketURL: "wss://jetstream1.us-east.bsky.network/subscribe",
	}
	logger := slog.Default()
	scheduler := parallel.NewScheduler(runtime.NumCPU(), "nats-proxy", logger, func(ctx context.Context, e *models.Event) error {
		if e != nil && e.Commit != nil && e.Commit.Collection != "" {
			// Get the collection from the commit
			collection := e.Commit.Collection

			// Determine the subject based on the collection
			subject := fmt.Sprintf("bluesky.%s", collection)

			// Convert the event to JSON
			jsonData, err := json.Marshal(e)
			if err != nil {
				log.Printf("Error marshaling event: %v", err)
				return err
			}

			// Publish the event to NATS
			err = nc.Publish(subject, jsonData)
			if err != nil {
				log.Printf("Error publishing to NATS: %v", err)
				return err
			}
		}

		return nil
	})
	log.Println("Created config, logger, scheduler")

	go func() {
		cursor := time.Now().UnixMicro()

		totalServers := 2 // See: https://github.com/bluesky-social/jetstream?tab=readme-ov-file#public-instances

		for i := 0; true; i++ {
			jc, err := client.NewClient(config, logger, scheduler)
			if err != nil {
				log.Fatalf("Error creating Jetstream client: %v", err)
			}

			if err := jc.ConnectAndRead(context.Background(), &cursor); err != nil {
				log.Printf("failed to connect: %v\n", err)
			}

			time.Sleep(1 * time.Second)

			// alternate between available jetstream servers
			config.WebsocketURL = fmt.Sprintf("wss://jetstream%d.us-east.bsky.network/subscribe", i%totalServers+1)
			log.Printf("connecting to %s instead\n", config.WebsocketURL)
		}
	}()

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

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