inventory deployment

cmd/inventory/main.go

@@ -2,20 +2,81 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"log"
-	"time"
+	"net/http"
+	"os"
+	"strings"
+	"sync"
 
-	"git.tornberg.me/go-cart-actor/pkg/inventory"
+	"git.tornberg.me/mats/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 {
+	service *inventory.RedisInventoryService
+}
+
+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) {
+	// Parse path: /inventory/{sku}/{location}
+	path := r.URL.Path
+	parts := strings.Split(strings.Trim(path, "/"), "/")
+	if len(parts) != 3 || parts[0] != "inventory" {
+		http.Error(w, "Invalid path", http.StatusBadRequest)
+		return
+	}
+	sku := inventory.SKU(parts[1])
+	locationID := inventory.LocationID(parts[2])
+
+	quantity, err := srv.service.GetInventory(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)
+}
+
+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:     "10.10.3.18:6379",
-		Password: "slaskredis", // no password set
-		DB:       0,            // use default DB
+		Addr:     redisAddress,
+		Password: redisPassword, // no password set
+		DB:       0,             // use default DB
 		MaintNotificationsConfig: &maintnotifications.Config{
 			Mode: maintnotifications.ModeDisabled,
 		},
@@ -26,51 +87,56 @@ func main() {
 		return
 	}
 
-	// Start inventory change listener
-	listener := inventory.NewInventoryChangeListener(rdb, ctx, func(changes []inventory.InventoryChange) {
-		for _, change := range changes {
-			log.Printf("Inventory change detected: SKU %s at location %s now has %d", change.SKU, change.StockLocationID, change.Value)
+	server := &Server{service: s}
+
+	// Set up HTTP routes
+	http.HandleFunc("/livez", server.livezHandler)
+	http.HandleFunc("/readyz", server.readyzHandler)
+	http.HandleFunc("/inventory/", server.getInventoryHandler)
+
+	stockhandler := &StockHandler{
+		MainStockLocationID: inventory.LocationID(country),
+		rdb:                 rdb,
+		ctx:                 ctx,
+		svc:                 *s,
+	}
+
+	amqpUrl, ok := os.LookupEnv("RABBIT_HOST")
+	if ok {
+		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)
 		}
-	})
-	log.Println("Starting inventory change listener...")
-	go listener.Start()
-	listener.WaitReady()
-	log.Println("Listener is ready, proceeding with inventory operations...")
-
-	rdb.Pipelined(ctx, func(p redis.Pipeliner) error {
-		s.UpdateInventory(p, "1", "1", 10)
-		s.UpdateInventory(p, "2", "2", 20)
-		s.UpdateInventory(p, "3", "3", 30)
-		s.UpdateInventory(p, "4", "4", 40)
-		return nil
-	})
-	err = s.ReserveInventory(
-		inventory.ReserveRequest{
-			SKU:        "1",
-			LocationID: "1",
-			Quantity:   3,
-		},
-		inventory.ReserveRequest{
-			SKU:        "2",
-			LocationID: "2",
-			Quantity:   15,
-		},
-		inventory.ReserveRequest{
-			SKU:        "3",
-			LocationID: "3",
-			Quantity:   25,
-		},
-	)
-	if err != nil {
-		log.Printf("Unable to reserve inventory: %v", err)
 	}
-	v, err := s.GetInventory("1", "1")
-	if err != nil {
-		log.Printf("Unable to get inventory: %v", err)
-		return
-	}
-	log.Printf("Inventory after reservation: %v", v)
 
-	// Wait a bit for listener to process messages
-	time.Sleep(2 * time.Second)
+	// Start HTTP server
+	log.Println("Starting HTTP server on :8080")
+	log.Fatal(http.ListenAndServe(":8080", nil))
 }

cmd/inventory/stockhandler.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"context"
+	"log"
+	"strconv"
+	"strings"
+	"sync"
+
+	"git.tornberg.me/mats/go-redis-inventory/pkg/inventory"
+	"github.com/matst80/slask-finder/pkg/types"
+	"github.com/redis/go-redis/v9"
+)
+
+type StockHandler struct {
+	rdb                 *redis.Client
+	ctx                 context.Context
+	svc                 inventory.RedisInventoryService
+	MainStockLocationID inventory.LocationID
+}
+
+func (s *StockHandler) HandleItem(item types.Item, wg *sync.WaitGroup) {
+	wg.Go(func() {
+		pipe := s.rdb.Pipeline()
+		centralStockString, ok := item.GetStringFieldValue(3)
+		if !ok {
+			centralStockString = "0"
+		}
+		centralStockString = strings.Replace(centralStockString, "+", "", -1)
+		centralStockString = strings.Replace(centralStockString, "<", "", -1)
+		centralStockString = strings.Replace(centralStockString, ">", "", -1)
+		centralStock, err := strconv.ParseInt(centralStockString, 10, 64)
+
+		if err != nil {
+			log.Printf("unable to parse central stock for item %s: %v", item.GetSku(), err)
+			centralStock = 0
+		} else {
+			s.svc.UpdateInventory(pipe, inventory.SKU(item.GetSku()), s.MainStockLocationID, int64(centralStock))
+		}
+		for id, value := range item.GetStock() {
+			s.svc.UpdateInventory(pipe, inventory.SKU(item.GetSku()), inventory.LocationID(id), int64(value))
+		}
+		_, err = pipe.Exec(s.ctx)
+		if err != nil {
+			log.Printf("unable to update stock: %v", err)
+		}
+	})
+}