go-order-manager/main.go

package main

import (
	"compress/gzip"
	"context"
	"encoding/gob"
	"encoding/json"
	"log"
	"net/http"
	"os"
	"runtime"
	"sync"
	"time"

	"git.k6n.net/mats/platform/httpserver"
)

var amqpUrl = os.Getenv("AMQP_URL")

func init() {
	gob.Register([]Order{})
	if amqpUrl == "" {
		log.Fatal("AMQP_URL environment variable is not set")
	}
}

type PersistingOrderHandler struct {
	mu       sync.RWMutex
	Orders   map[string]Order `json:"orders"`
	fileName string
}

func (h *PersistingOrderHandler) Load() error {
	file, err := os.Open(h.fileName)
	if err != nil {
		return err
	}
	defer runtime.GC()
	defer file.Close()

	zipReader, err := gzip.NewReader(file)
	if err != nil {
		return err
	}

	enc := json.NewDecoder(zipReader)
	defer zipReader.Close()

	h.mu.Lock()
	defer h.mu.Unlock()
	tmp := Order{}
	for err == nil {
		if err = enc.Decode(&tmp); err == nil {
			h.Orders[tmp.ID] = tmp
		}
	}
	enc = nil

	if err.Error() == "EOF" {
		return nil
	}

	return err
}

func (h *PersistingOrderHandler) GetById(id string) (*Order, bool) {
	h.mu.RLock()
	defer h.mu.RUnlock()
	order, ok := h.Orders[id]
	return &order, ok
}

func (h *PersistingOrderHandler) Save() error {
	file, err := os.Create(h.fileName + ".tmp")
	if err != nil {
		return err
	}

	defer runtime.GC()
	defer file.Close()
	zipWriter := gzip.NewWriter(file)
	enc := json.NewEncoder(zipWriter)
	defer zipWriter.Close()
	h.mu.Lock()
	defer h.mu.Unlock()

	for _, order := range h.Orders {
		err = enc.Encode(order)
		if err != nil {
			return err
		}
	}

	enc = nil
	return os.Rename(h.fileName+".tmp", h.fileName)
}

func (h *PersistingOrderHandler) GetLatest() []Order {
	h.mu.RLock()
	defer h.mu.RUnlock()
	ret := make([]Order, 0, len(h.Orders))
	for _, order := range h.Orders {
		ret = append(ret, order)
	}
	return ret
}

func (h *PersistingOrderHandler) OrderPlaced(order Order) {
	// Here you would implement the logic to persist the order
	log.Printf("Order placed: %s", order.ID)
	h.mu.Lock()
	h.Orders[order.ID] = order
	h.mu.Unlock()
	go func() {
		err := h.Save()
		if err != nil {
			log.Printf("error saving: %v", err)
		}
	}()

}

func main() {
	handler := &PersistingOrderHandler{
		Orders:   make(map[string]Order),
		mu:       sync.RWMutex{},
		fileName: "data/order.dbz",
	}
	err := handler.Load()
	if err != nil {
		//log.Fatal(err)
	}
	client := &RabbitTransportClient{
		Url:        amqpUrl,
		ClientName: "OrderClient",
	}

	err = client.Connect(handler)
	if err != nil {
		log.Fatalf("Failed to connect to RabbitMQ: %v", err)
	}
	// Note: client.Close() is called via a shutdown hook in
	// httpserver.RunServerWithShutdown below — NOT via defer, so it runs
	// BEFORE the HTTP server shuts down (stopping new AMQP deliveries
	// before we stop accepting HTTP requests).

	// Routes are served at their CANONICAL public paths under /api/orders, so
	// ingress routes straight to this service with no rewrite/strip — you can
	// curl the pod or the ingress and the path is identical (deploy/API-ROUTING.md).
	// /healthz stays pod-local for probes (never exposed at the edge).
	mux := http.NewServeMux()
	httpserver.RegisterProbes(mux, nil)
	mux.HandleFunc("GET /api/orders", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		json.NewEncoder(w).Encode(handler.GetLatest())
	})
	mux.HandleFunc("GET /api/orders/{id}", func(w http.ResponseWriter, r *http.Request) {
		order_id := r.PathValue("id")

		order, ok := handler.GetById(order_id)
		if !ok {
			w.WriteHeader(http.StatusNotFound)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		json.NewEncoder(w).Encode(order)
	})
	mux.HandleFunc("POST /api/orders/{id}/capture", func(w http.ResponseWriter, r *http.Request) {
		order_id := r.PathValue("id")
		capture := &CaptureData{}
		json.NewDecoder(r.Body).Decode(capture)
		w.WriteHeader(http.StatusNotImplemented)
		w.Write([]byte(order_id))
	})

	// Graceful shutdown: flush pending order saves + close the AMQP connection
	// before the server stops. The previous code used a bare http.ListenAndServe
	// with no signal handling — a SIGTERM killed the process mid-save, risking
	// a corrupt .dbz file or a dropped AMQP ack. RunServerWithShutdown handles
	// SIGINT/SIGTERM, runs the save hook, then shuts down the HTTP server.
	srv := &http.Server{
		Addr:              ":8080",
		Handler:           mux,
		ReadHeaderTimeout: 10 * time.Second,
	}
	if err := httpserver.RunServerWithShutdown(srv, "order-manager", 10*time.Second, 5*time.Second,
		func(context.Context) error {
			// Flush the in-memory order map to disk so a SIGTERM doesn't lose
			// orders that were placed since the last async Save().
			return handler.Save()
		},
		func(context.Context) error {
			// Close the AMQP connection cleanly so in-flight deliveries are
			// requeued, not lost.
			client.Close()
			return nil
		},
	); err != nil {
		log.Fatalf("server: %v", err)
	}
}