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

package main

import (
	"context"
	"encoding/json"
	"log"
	"net/http"
	"os"
	"sync"

	"github.com/matst80/go-redis-inventory/pkg/inventory"
	"github.com/matst80/slask-finder/pkg/index"
	"github.com/matst80/slask-finder/pkg/messaging"
	"github.com/redis/go-redis/v9"
	"github.com/redis/go-redis/v9/maintnotifications"

	amqp "github.com/rabbitmq/amqp091-go"
)

type Server struct {
	inventoryService   *inventory.RedisInventoryService
	reservationService *inventory.RedisCartReservationService
}

func (srv *Server) livezHandler(w http.ResponseWriter, r *http.Request) {
	w.WriteHeader(http.StatusOK)
	w.Write([]byte("OK"))
}

func (srv *Server) readyzHandler(w http.ResponseWriter, r *http.Request) {
	w.WriteHeader(http.StatusOK)
	w.Write([]byte("OK"))
}

func (srv *Server) getInventoryHandler(w http.ResponseWriter, r *http.Request) {

	sku := inventory.SKU(r.PathValue("sku"))
	locationID := inventory.LocationID(r.PathValue("locationId"))

	quantity, err := srv.inventoryService.GetInventory(r.Context(), sku, locationID)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	response := map[string]int64{"quantity": quantity}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

func (srv *Server) getReservationHandler(w http.ResponseWriter, r *http.Request) {
	sku := inventory.SKU(r.PathValue("sku"))
	locationID := inventory.LocationID(r.PathValue("locationId"))

	summary, err := srv.reservationService.GetReservationSummary(r.Context(), sku, locationID)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(summary)
}

var country = "se"
var redisAddress = "10.10.3.18:6379"
var redisPassword = "slaskredis"

func init() {
	// Override redis config from environment variables if set
	if addr, ok := os.LookupEnv("REDIS_ADDRESS"); ok {
		redisAddress = addr
	}
	if password, ok := os.LookupEnv("REDIS_PASSWORD"); ok {
		redisPassword = password
	}
	if ctry, ok := os.LookupEnv("COUNTRY"); ok {
		country = ctry
	}
}

func main() {
	var ctx = context.Background()
	rdb := redis.NewClient(&redis.Options{
		Addr:     redisAddress,
		Password: redisPassword, // no password set
		DB:       0,             // use default DB
		MaintNotificationsConfig: &maintnotifications.Config{
			Mode: maintnotifications.ModeDisabled,
		},
	})
	s, err := inventory.NewRedisInventoryService(rdb)
	if err != nil {
		log.Fatalf("Unable to connect to inventory redis: %v", err)
		return
	}

	r, err := inventory.NewRedisCartReservationService(rdb)
	if err != nil {
		log.Fatalf("Unable to connect to reservation redis: %v", err)
		return
	}

	server := &Server{inventoryService: s, reservationService: r}

	// Set up HTTP routes
	http.HandleFunc("/livez", server.livezHandler)
	http.HandleFunc("/readyz", server.readyzHandler)
	http.HandleFunc("/inventory/{sku}/{locationId}", server.getInventoryHandler)
	http.HandleFunc("/reservations/{sku}/{locationId}", server.getReservationHandler)

	stockhandler := &StockHandler{
		MainStockLocationID: inventory.LocationID(country),
		rdb:                 rdb,
		ctx:                 ctx,
		svc:                 *s,
	}

	amqpUrl, ok := os.LookupEnv("RABBIT_HOST")
	if ok {
		log.Printf("Connecting to rabbitmq")
		conn, err := amqp.DialConfig(amqpUrl, amqp.Config{
			Properties: amqp.NewConnectionProperties(),
		})
		//a.conn = conn
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %v", err)
		}
		ch, err := conn.Channel()
		if err != nil {
			log.Fatalf("Failed to open a channel: %v", err)
		}
		// items listener
		err = messaging.ListenToTopic(ch, country, "item_added", func(d amqp.Delivery) error {
			var items []*index.DataItem
			err := json.Unmarshal(d.Body, &items)
			if err == nil {
				log.Printf("Got upserts %d, message count %d", len(items), d.MessageCount)
				wg := &sync.WaitGroup{}
				for _, item := range items {
					stockhandler.HandleItem(item, wg)
				}
				wg.Wait()
				log.Print("Batch done...")
			} else {
				log.Printf("Failed to unmarshal upsert message %v", err)
			}
			return err
		})
		if err != nil {
			log.Fatalf("Failed to listen to item_added topic: %v", err)
		}
	}

	// Start HTTP server
	log.Println("Starting HTTP server on :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}