feature/backoffice (#6)

Co-authored-by: matst80 <mats.tornberg@gmail.com>
Reviewed-on: https://git.tornberg.me/mats/go-cart-actor/pulls/6
Co-authored-by: Mats Törnberg <mats@tornberg.me>
Co-committed-by: Mats Törnberg <mats@tornberg.me>

cmd/backoffice/fileserver.go

@@ -0,0 +1,187 @@
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"os"
+	"path/filepath"
+	"regexp"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"git.tornberg.me/go-cart-actor/pkg/actor"
+	"git.tornberg.me/go-cart-actor/pkg/cart"
+)
+
+type FileServer struct {
+	// Define fields here
+	dataDir string
+	storage actor.LogStorage[cart.CartGrain]
+}
+
+func NewFileServer(dataDir string, storage actor.LogStorage[cart.CartGrain]) *FileServer {
+	return &FileServer{
+		dataDir: dataDir,
+		storage: storage,
+	}
+}
+
+func isValidId(id string) (uint64, bool) {
+	if nr, err := strconv.ParseUint(id, 10, 64); err == nil {
+		return nr, true
+	}
+	if nr, ok := cart.ParseCartId(id); ok {
+		return uint64(nr), true
+	}
+	return 0, false
+}
+
+func isValidFileId(name string) (uint64, bool) {
+
+	parts := strings.Split(name, ".")
+	if len(parts) > 1 && parts[1] == "events" {
+		idStr := parts[0]
+
+		return isValidId(idStr)
+	}
+	return 0, false
+}
+
+var cartFileRe = regexp.MustCompile(`^(\d+)\.events\.log$`)
+
+func listCartFiles(dir string) ([]CartFileInfo, error) {
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return []CartFileInfo{}, nil
+		}
+		return nil, err
+	}
+	out := make([]CartFileInfo, 0)
+	for _, e := range entries {
+		if e.IsDir() {
+			continue
+		}
+		id, valid := isValidFileId(e.Name())
+		if !valid {
+			continue
+		}
+
+		info, err := e.Info()
+		if err != nil {
+			continue
+		}
+		info.Sys()
+		out = append(out, CartFileInfo{
+			ID:       fmt.Sprintf("%d", id),
+			CartId:   cart.CartId(id),
+			Size:     info.Size(),
+			Modified: info.ModTime(),
+			System:   info.Sys(),
+		})
+	}
+	return out, nil
+}
+
+func readRawLogLines(path string) ([]json.RawMessage, error) {
+	fh, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer fh.Close()
+	lines := make([]json.RawMessage, 0, 64)
+	s := bufio.NewScanner(fh)
+	// increase buffer to handle larger JSON lines
+	buf := make([]byte, 0, 1024*1024)
+	s.Buffer(buf, 1024*1024)
+	for s.Scan() {
+		line := s.Bytes()
+		if line == nil {
+			continue
+		}
+		lines = append(lines, line)
+	}
+	if err := s.Err(); err != nil {
+		return nil, err
+	}
+	return lines, nil
+}
+
+func writeJSON(w http.ResponseWriter, status int, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(v)
+}
+
+func (fs *FileServer) CartsHandler(w http.ResponseWriter, r *http.Request) {
+	list, err := listCartFiles(fs.dataDir)
+	if err != nil {
+		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
+		return
+	}
+	// sort by modified desc
+	sort.Slice(list, func(i, j int) bool { return list[i].Modified.After(list[j].Modified) })
+
+	writeJSON(w, http.StatusOK, map[string]any{
+		"count": len(list),
+		"carts": list,
+	})
+}
+
+type JsonError struct {
+	Error string `json:"error"`
+}
+
+func (fs *FileServer) CartHandler(w http.ResponseWriter, r *http.Request) {
+	idStr := r.PathValue("id")
+	if idStr == "" {
+		writeJSON(w, http.StatusBadRequest, JsonError{Error: "missing id"})
+		return
+	}
+	id, ok := isValidId(idStr)
+	if !ok {
+		writeJSON(w, http.StatusBadRequest, JsonError{Error: "invalid id"})
+		return
+	}
+	// reconstruct state from event log if present
+	grain := cart.NewCartGrain(id, time.Now())
+
+	err := fs.storage.LoadEvents(id, grain)
+	if err != nil {
+		writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()})
+		return
+	}
+
+	path := filepath.Join(fs.dataDir, fmt.Sprintf("%d.events.log", id))
+	info, err := os.Stat(path)
+	if err != nil && errors.Is(err, os.ErrNotExist) {
+		writeJSON(w, http.StatusNotFound, JsonError{Error: "cart not found"})
+		return
+	} else if err != nil {
+		writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()})
+		return
+	}
+	lines, err := readRawLogLines(path)
+	if err != nil {
+		writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()})
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{
+		"id":        id,
+		"cartId":    cart.CartId(id).String(),
+		"state":     grain,
+		"mutations": lines,
+		"meta": map[string]any{
+			"size":     info.Size(),
+			"modified": info.ModTime(),
+			"path":     path,
+			"system":   info.Sys(),
+		},
+	})
+}

cmd/backoffice/hub.go

@@ -0,0 +1,248 @@
+package main
+
+import (
+	"bufio"
+	"crypto/sha1"
+	"encoding/base64"
+	"encoding/binary"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"time"
+)
+
+// Hub manages websocket clients and broadcasts messages to them.
+type Hub struct {
+	register   chan *Client
+	unregister chan *Client
+	broadcast  chan []byte
+	clients    map[*Client]bool
+}
+
+// Client represents a single websocket client connection.
+type Client struct {
+	hub  *Hub
+	conn net.Conn
+	send chan []byte
+}
+
+// NewHub constructs a new Hub instance.
+func NewHub() *Hub {
+	return &Hub{
+		register:   make(chan *Client),
+		unregister: make(chan *Client),
+		broadcast:  make(chan []byte, 1024),
+		clients:    make(map[*Client]bool),
+	}
+}
+
+// Run starts the hub event loop.
+func (h *Hub) Run() {
+	for {
+		select {
+		case c := <-h.register:
+			h.clients[c] = true
+		case c := <-h.unregister:
+			if _, ok := h.clients[c]; ok {
+				delete(h.clients, c)
+				close(c.send)
+				_ = c.conn.Close()
+			}
+		case msg := <-h.broadcast:
+			for c := range h.clients {
+				select {
+				case c.send <- msg:
+				default:
+					// Client is slow or dead; drop it.
+					delete(h.clients, c)
+					close(c.send)
+					_ = c.conn.Close()
+				}
+			}
+		}
+	}
+}
+
+// computeAccept computes the Sec-WebSocket-Accept header value.
+func computeAccept(key string) string {
+	const magic = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
+	h := sha1.New()
+	h.Write([]byte(key + magic))
+	return base64.StdEncoding.EncodeToString(h.Sum(nil))
+}
+
+// ServeWS upgrades the HTTP request to a WebSocket connection and registers a client.
+func (h *Hub) ServeWS(w http.ResponseWriter, r *http.Request) {
+	if !strings.Contains(strings.ToLower(r.Header.Get("Connection")), "upgrade") || strings.ToLower(r.Header.Get("Upgrade")) != "websocket" {
+		http.Error(w, "upgrade required", http.StatusBadRequest)
+		return
+	}
+	key := r.Header.Get("Sec-WebSocket-Key")
+	if key == "" {
+		http.Error(w, "missing Sec-WebSocket-Key", http.StatusBadRequest)
+		return
+	}
+	accept := computeAccept(key)
+
+	hj, ok := w.(http.Hijacker)
+	if !ok {
+		http.Error(w, "websocket not supported", http.StatusInternalServerError)
+		return
+	}
+	conn, buf, err := hj.Hijack()
+	if err != nil {
+		return
+	}
+
+	// Write the upgrade response
+	response := "HTTP/1.1 101 Switching Protocols\r\n" +
+		"Upgrade: websocket\r\n" +
+		"Connection: Upgrade\r\n" +
+		"Sec-WebSocket-Accept: " + accept + "\r\n" +
+		"\r\n"
+	if _, err := buf.WriteString(response); err != nil {
+		_ = conn.Close()
+		return
+	}
+	if err := buf.Flush(); err != nil {
+		_ = conn.Close()
+		return
+	}
+
+	client := &Client{
+		hub:  h,
+		conn: conn,
+		send: make(chan []byte, 256),
+	}
+	h.register <- client
+	go client.writePump()
+	go client.readPump()
+}
+
+// writeWSFrame writes a single WebSocket frame to the writer.
+func writeWSFrame(w io.Writer, opcode byte, payload []byte) error {
+	// FIN set, opcode as provided
+	header := []byte{0x80 | (opcode & 0x0F)}
+	l := len(payload)
+	switch {
+	case l < 126:
+		header = append(header, byte(l))
+	case l <= 65535:
+		ext := make([]byte, 2)
+		binary.BigEndian.PutUint16(ext, uint16(l))
+		header = append(header, 126)
+		header = append(header, ext...)
+	default:
+		ext := make([]byte, 8)
+		binary.BigEndian.PutUint64(ext, uint64(l))
+		header = append(header, 127)
+		header = append(header, ext...)
+	}
+	if _, err := w.Write(header); err != nil {
+		return err
+	}
+	if l > 0 {
+		if _, err := w.Write(payload); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// readPump handles control frames from the client and discards other incoming frames.
+// This server is broadcast-only to clients.
+func (c *Client) readPump() {
+	defer func() {
+		c.hub.unregister <- c
+	}()
+	reader := bufio.NewReader(c.conn)
+	for {
+		// Read first two bytes
+		b1, err := reader.ReadByte()
+		if err != nil {
+			return
+		}
+		b2, err := reader.ReadByte()
+		if err != nil {
+			return
+		}
+		opcode := b1 & 0x0F
+		masked := (b2 & 0x80) != 0
+		length := int64(b2 & 0x7F)
+		if length == 126 {
+			ext := make([]byte, 2)
+			if _, err := io.ReadFull(reader, ext); err != nil {
+				return
+			}
+			length = int64(binary.BigEndian.Uint16(ext))
+		} else if length == 127 {
+			ext := make([]byte, 8)
+			if _, err := io.ReadFull(reader, ext); err != nil {
+				return
+			}
+			length = int64(binary.BigEndian.Uint64(ext))
+		}
+		var maskKey [4]byte
+		if masked {
+			if _, err := io.ReadFull(reader, maskKey[:]); err != nil {
+				return
+			}
+		}
+
+		// Handle Ping -> Pong
+		if opcode == 0x9 && length <= 125 {
+			payload := make([]byte, length)
+			if _, err := io.ReadFull(reader, payload); err != nil {
+				return
+			}
+			// Unmask if masked
+			if masked {
+				for i := int64(0); i < length; i++ {
+					payload[i] ^= maskKey[i%4]
+				}
+			}
+			_ = writeWSFrame(c.conn, 0xA, payload) // best-effort pong
+			continue
+		}
+
+		// Close frame
+		if opcode == 0x8 {
+			// Drain payload if any, then exit
+			if _, err := io.CopyN(io.Discard, reader, length); err != nil {
+				return
+			}
+			return
+		}
+
+		// For other frames, just discard payload
+		if _, err := io.CopyN(io.Discard, reader, length); err != nil {
+			return
+		}
+	}
+}
+
+// writePump sends queued messages to the client and pings periodically to keep the connection alive.
+func (c *Client) writePump() {
+	ticker := time.NewTicker(30 * time.Second)
+	defer func() {
+		ticker.Stop()
+		_ = c.conn.Close()
+	}()
+	for {
+		select {
+		case msg, ok := <-c.send:
+			if !ok {
+				// try to send close frame
+				_ = writeWSFrame(c.conn, 0x8, nil)
+				return
+			}
+			if err := writeWSFrame(c.conn, 0x1, msg); err != nil {
+				return
+			}
+		case <-ticker.C:
+			// Send a ping to keep connections alive behind proxies
+			_ = writeWSFrame(c.conn, 0x9, []byte("ping"))
+		}
+	}
+}

cmd/backoffice/main.go

@@ -1,5 +1,149 @@
 package main
 
-func main() {
-	// Your code here
+import (
+	"context"
+	"errors"
+	"log"
+	"net/http"
+	"os"
+	"time"
+
+	actor "git.tornberg.me/go-cart-actor/pkg/actor"
+	"git.tornberg.me/go-cart-actor/pkg/cart"
+	"github.com/matst80/slask-finder/pkg/messaging"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+type CartFileInfo struct {
+	ID       string      `json:"id"`
+	CartId   cart.CartId `json:"cartId"`
+	Size     int64       `json:"size"`
+	Modified time.Time   `json:"modified"`
+	System   any         `json:"system"`
+}
+
+func envOrDefault(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}
+
+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 := cart.NewCartMultationRegistry()
+	diskStorage := actor.NewDiskStorage[cart.CartGrain](dataDir, reg)
+
+	fs := NewFileServer(dataDir, diskStorage)
+
+	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
 }