core: Add nats handler for spaces

Author: OliverSchlueter

integrations/spaces-go-sdk/spaces/spaces.go

@@ -28,7 +28,7 @@ func (s *Service) GetSpace(id string) (*InternalSpace, error) {
 		return spaceFromCache, nil
 	}
 
-	resp, err := s.broker.Request("fancyspaces.space.get", []byte(id))
+	resp, err := s.broker.Request("fancyspaces.core.spaces.get", []byte(id))
 	if err != nil {
 		return nil, err
 	}

services/core/cmd/local/main.go

@@ -10,6 +10,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/OliverSchlueter/goutils/broker"
 	"github.com/OliverSchlueter/goutils/containers"
 	"github.com/OliverSchlueter/goutils/middleware"
 	"github.com/OliverSchlueter/goutils/sloki"
@@ -21,6 +22,9 @@ import (
 )
 
 func main() {
+	nc := containers.ConnectToNatsE2E()
+	b := broker.NewNatsBroker(&broker.NatsConfiguration{Nats: nc})
+
 	// Feature flags
 	fflags.DisableIssueSyncer.Enable()
 
@@ -64,6 +68,7 @@ func main() {
 	app.Start(app.Configuration{
 		Mux:        mux,
 		MavenMux:   mavenMux,
+		Broker:     b,
 		Mongo:      mc,
 		ClickHouse: ch,
 		MinIO:      mio,
@@ -101,6 +106,7 @@ func main() {
 	case os.Interrupt:
 		slog.Info("Received interrupt signal, shutting down...")
 
+		containers.DisconnectNats(nc)
 		containers.DisconnectMongo(mc)
 		containers.DisconnectClickhouse(ch)
 		containers.DisconnectMinIO(mio)

services/core/cmd/prod/main.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 	"syscall"
 
+	"github.com/OliverSchlueter/goutils/broker"
 	"github.com/OliverSchlueter/goutils/containers"
 	"github.com/OliverSchlueter/goutils/env"
 	"github.com/OliverSchlueter/goutils/middleware"
@@ -21,6 +22,9 @@ import (
 )
 
 const (
+	natsUrlEnv       = "NATS_URL"
+	natsAuthTokenEnv = "NATS_AUTH_TOKEN"
+
 	usersPathEnv = "USERS_PATH"
 
 	mongodbUrlEnv = "MONGODB_URL"
@@ -34,6 +38,9 @@ const (
 )
 
 func main() {
+	nc := containers.ConnectToNats(env.MustGetStr(natsUrlEnv), env.MustGetStr(natsAuthTokenEnv))
+	b := broker.NewNatsBroker(&broker.NatsConfiguration{Nats: nc})
+
 	// Setup logging
 	logService := sloki.NewService(sloki.Configuration{
 		URL:          "http://localhost:3100/loki/api/v1/push",
@@ -72,6 +79,7 @@ func main() {
 	app.Start(app.Configuration{
 		Mux:        mux,
 		MavenMux:   mavenMux,
+		Broker:     b,
 		Mongo:      mc,
 		ClickHouse: ch,
 		MinIO:      mio,
@@ -116,6 +124,7 @@ func main() {
 	case os.Interrupt:
 		slog.Info("Received interrupt signal, shutting down...")
 
+		containers.DisconnectNats(nc)
 		containers.DisconnectMongo(mc)
 		containers.DisconnectClickhouse(ch)
 		containers.DisconnectMinIO(mio)

services/core/internal/app/app.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
+	"github.com/OliverSchlueter/goutils/broker"
 	"github.com/OliverSchlueter/goutils/ratelimit"
 	"github.com/OliverSchlueter/goutils/sitemapgen"
 	"github.com/fancyinnovations/fancyspaces/core/internal/analytics"
@@ -42,6 +43,7 @@ const apiPrefix = "/api/v1"
 type Configuration struct {
 	Mux        *http.ServeMux
 	MavenMux   *http.ServeMux
+	Broker     broker.Broker
 	Mongo      *mongo.Database
 	ClickHouse driver.Conn
 	MinIO      *minio.Client
@@ -72,6 +74,13 @@ func Start(cfg Configuration) {
 		Analytics:   as,
 	})
 	sh.Register(apiPrefix, cfg.Mux)
+	snh := spacesHandler.NewNatsHandler(spacesHandler.NatsConfiguration{
+		Broker: cfg.Broker,
+		Store:  spacesStore,
+	})
+	if err := snh.Register(); err != nil {
+		panic(fmt.Errorf("could not register spaces nats handler: %w", err))
+	}
 
 	// Versions
 	versionsDB := mongoVersionsDB.NewDB(&mongoVersionsDB.Configuration{

services/core/internal/spaces/handler/nats_handler.go

@@ -0,0 +1,75 @@
+package handler
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+
+	"github.com/OliverSchlueter/goutils/broker"
+	"github.com/OliverSchlueter/goutils/middleware"
+	"github.com/OliverSchlueter/goutils/problems"
+	"github.com/OliverSchlueter/goutils/sloki"
+	"github.com/fancyinnovations/fancyspaces/core/internal/spaces"
+	spaces2 "github.com/fancyinnovations/fancyspaces/integrations/spaces-go-sdk/spaces"
+	"github.com/nats-io/nats.go"
+)
+
+type NatsHandler struct {
+	broker broker.Broker
+	store  *spaces.Store
+}
+
+type NatsConfiguration struct {
+	Broker broker.Broker
+	Store  *spaces.Store
+}
+
+func NewNatsHandler(cfg NatsConfiguration) *NatsHandler {
+	return &NatsHandler{
+		broker: cfg.Broker,
+		store:  cfg.Store,
+	}
+}
+
+func (h *NatsHandler) Register() error {
+	if err := h.broker.SubscribeQueue("fancyspaces.core.spaces.get", "fancyspaces.core.spaces.get", middleware.NatsLogging(h.handleGet)); err != nil {
+		return fmt.Errorf("could not subscribe to nats subject: %w", err)
+	}
+
+	return nil
+}
+
+func (h *NatsHandler) handleGet(msg *nats.Msg) {
+	id := string(msg.Data)
+
+	if id == "" {
+		problems.ValidationError("space_id", "Space ID is required").WriteToBroker(h.broker, msg.Reply)
+		return
+	}
+
+	space, err := h.store.GetByID(id)
+	if err != nil {
+		if errors.Is(err, spaces.ErrSpaceNotFound) {
+			problems.NotFound("Space", id).WriteToBroker(h.broker, msg.Reply)
+			return
+		}
+		slog.Error("Could not get space", sloki.WrapError(err))
+		problems.InternalServerError("").WriteToBroker(h.broker, msg.Reply)
+		return
+	}
+
+	response, err := json.Marshal(spaces2.InternalSpace{
+		Space:             *space,
+		AnalyticsWriteKey: space.AnalyticsSettings.WriteKey,
+	})
+	if err != nil {
+		slog.Error("failed to marshal space", sloki.WrapError(err))
+		problems.InternalServerError("Could not marshal space").WriteToBroker(h.broker, msg.Reply)
+		return
+	}
+
+	if err := h.broker.Publish(msg.Reply, response); err != nil {
+		slog.Error("failed to publish space response", sloki.WrapError(err))
+	}
+}