go-cart-actor/cmd/backoffice/main.go

package main

import (
	"context"
	"errors"
	"log"
	"net/http"
	"os"
	"regexp"
	"time"

	actor "git.tornberg.me/go-cart-actor/pkg/actor"
	"git.tornberg.me/go-cart-actor/pkg/cart"
	messages "git.tornberg.me/go-cart-actor/pkg/messages"
	"github.com/matst80/slask-finder/pkg/messaging"
	amqp "github.com/rabbitmq/amqp091-go"
)

type CartFileInfo struct {
	ID       uint64      `json:"id"`
	CartId   cart.CartId `json:"cart_id"`
	Path     string      `json:"path"`
	Size     int64       `json:"size"`
	Modified time.Time   `json:"modified"`
}

func envOrDefault(key, def string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return def
}

var cartFileRe = regexp.MustCompile(`^(\d+)\.events\.log$`)

var globalDisk *actor.DiskStorage[cart.CartGrain]

func buildRegistry() actor.MutationRegistry {
	reg := actor.NewMutationRegistry()
	reg.RegisterMutations(
		actor.NewMutation(cart.AddItem, func() *messages.AddItem { return &messages.AddItem{} }),
		actor.NewMutation(cart.ChangeQuantity, func() *messages.ChangeQuantity { return &messages.ChangeQuantity{} }),
		actor.NewMutation(cart.RemoveItem, func() *messages.RemoveItem { return &messages.RemoveItem{} }),
		actor.NewMutation(cart.InitializeCheckout, func() *messages.InitializeCheckout { return &messages.InitializeCheckout{} }),
		actor.NewMutation(cart.OrderCreated, func() *messages.OrderCreated { return &messages.OrderCreated{} }),
		actor.NewMutation(cart.RemoveDelivery, func() *messages.RemoveDelivery { return &messages.RemoveDelivery{} }),
		actor.NewMutation(cart.SetDelivery, func() *messages.SetDelivery { return &messages.SetDelivery{} }),
		actor.NewMutation(cart.SetPickupPoint, func() *messages.SetPickupPoint { return &messages.SetPickupPoint{} }),
		actor.NewMutation(cart.ClearCart, func() *messages.ClearCartRequest { return &messages.ClearCartRequest{} }),
		actor.NewMutation(cart.AddVoucher, func() *messages.AddVoucher { return &messages.AddVoucher{} }),
		actor.NewMutation(cart.RemoveVoucher, func() *messages.RemoveVoucher { return &messages.RemoveVoucher{} }),
	)
	return reg
}

func startMutationConsumer(ctx context.Context, conn *amqp.Connection, hub *Hub) error {
	ch, err := conn.Channel()
	if err != nil {
		_ = conn.Close()
		return err
	}
	msgs, err := messaging.DeclareBindAndConsume(ch, "cart", "mutation")
	if err != nil {
		_ = ch.Close()
		return err
	}

	go func() {
		defer ch.Close()

		for {
			select {
			case <-ctx.Done():
				return
			case m, ok := <-msgs:
				if !ok {
					log.Fatalf("connection closed")
					continue
				}
				// Log and broadcast to all websocket clients
				log.Printf("mutation event: %s", string(m.Body))

				if hub != nil {
					select {
					case hub.broadcast <- m.Body:
					default:
						// if hub queue is full, drop to avoid blocking
					}
				}
				if err := m.Ack(false); err != nil {
					log.Printf("error acknowledging message: %v", err)
				}
			}
		}
	}()
	return nil
}

func main() {
	dataDir := envOrDefault("DATA_DIR", "data")
	addr := envOrDefault("ADDR", ":8080")
	amqpURL := os.Getenv("AMQP_URL")

	_ = os.MkdirAll(dataDir, 0755)

	reg := buildRegistry()
	globalDisk = actor.NewDiskStorage[cart.CartGrain](dataDir, reg)

	fs := NewFileServer(dataDir)

	hub := NewHub()
	go hub.Run()

	mux := http.NewServeMux()
	mux.HandleFunc("GET /carts", fs.CartsHandler)
	mux.HandleFunc("GET /cart/{id}", fs.CartHandler)
	mux.HandleFunc("/ws", hub.ServeWS)
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte("ok"))
	})
	mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte("ok"))
	})
	mux.HandleFunc("/livez", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte("ok"))
	})

	// Global CORS middleware allowing all origins and handling preflight
	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Access-Control-Allow-Origin", "*")
		w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS")
		w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Requested-With")
		w.Header().Set("Access-Control-Expose-Headers", "*")
		if r.Method == http.MethodOptions {
			w.WriteHeader(http.StatusNoContent)
			return
		}
		mux.ServeHTTP(w, r)
	})

	srv := &http.Server{
		Addr:         addr,
		Handler:      handler,
		ReadTimeout:  10 * time.Second,
		WriteTimeout: 30 * time.Second,
		IdleTimeout:  60 * time.Second,
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	if amqpURL != "" {
		conn, err := amqp.Dial(amqpURL)
		if err != nil {
			log.Fatalf("failed to connect to RabbitMQ: %w", err)
		}
		if err := startMutationConsumer(ctx, conn, hub); err != nil {
			log.Printf("AMQP listener disabled: %v", err)
		} else {
			log.Printf("AMQP listener connected")
		}
	}

	log.Printf("backoffice HTTP listening on %s (dataDir=%s)", addr, dataDir)
	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
		log.Fatalf("http server error: %v", err)
	}

	// server stopped
}