listener

pkg/inventory/doc.go

@@ -14,9 +14,6 @@
 // adapter.go bridges to platform/inventory.ReservationPolicy so other
 // services depend on the platform interface, not on this package.
 //
-// listener.go subscribes to the inventory_changed Redis pub/sub channel
-// and mirrors upstream events into the local cache.
-//
 // Request and reservation DTOs (CartReserveRequest, ReserveRequest,
 // CartID, ReservationStatus, ReservationSummary) now live in
 // request_types.go. Storage types (InventoryItem, InventoryReference,

pkg/inventory/listener.go

@@ -1,85 +0,0 @@
-package inventory
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"log"
-	"strings"
-
-	"github.com/redis/go-redis/v9"
-)
-
-type InventoryChange struct {
-	SKU             string `json:"sku"`
-	StockLocationID string `json:"stock_location_id"`
-	Value           int64  `json:"value"`
-}
-
-type InventoryChangeHandler func(changes []InventoryChange)
-
-type InventoryChangeListener struct {
-	client  *redis.Client
-	ctx     context.Context
-	handler InventoryChangeHandler
-	ready   chan struct{}
-}
-
-func NewInventoryChangeListener(client *redis.Client, ctx context.Context, handler InventoryChangeHandler) *InventoryChangeListener {
-	return &InventoryChangeListener{
-		client:  client,
-		ctx:     ctx,
-		handler: handler,
-		ready:   make(chan struct{}),
-	}
-}
-
-func (l *InventoryChangeListener) WaitReady() {
-	<-l.ready
-}
-
-func (l *InventoryChangeListener) Start() error {
-	pubsub := l.client.Subscribe(l.ctx, "inventory_changed")
-	defer pubsub.Close()
-
-	// Signal that we're ready to receive messages
-	close(l.ready)
-
-	ch := pubsub.Channel()
-
-	for msg := range ch {
-		var payload map[string]int64
-		if err := json.Unmarshal([]byte(msg.Payload), &payload); err != nil {
-			log.Printf("Failed to unmarshal inventory change message: %v", err)
-			continue
-		}
-
-		changes := make([]InventoryChange, 0, len(payload))
-		for key, newValue := range payload {
-			// Parse key: "inventory:sku:location"
-			parts := strings.Split(key, ":")
-			if len(parts) != 3 || parts[0] != "inventory" {
-				log.Printf("Invalid inventory key format: %s", key)
-				continue
-			}
-			sku := parts[1]
-			locationID := parts[2]
-			changes = append(changes, InventoryChange{
-				SKU:             sku,
-				StockLocationID: locationID,
-				Value:           newValue,
-			})
-		}
-
-		if l.handler != nil {
-			l.handler(changes)
-		} else {
-			// Default logging
-			for _, change := range changes {
-				fmt.Printf("Inventory changed: SKU %s at location %s -> %d\n", change.SKU, change.StockLocationID, change.Value)
-			}
-		}
-	}
-
-	return nil
-}

pkg/inventory/redis_service.go

@@ -2,7 +2,6 @@ package inventory
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"strconv"
@@ -106,26 +105,6 @@ func (s *RedisInventoryService) IncrementInventory(ctx context.Context, rdb redi
 	return cmd.Err()
 }
 
-func (s *RedisInventoryService) SendInventoryChanged(ctx context.Context, sku SKU, locationID LocationID) error {
-	key := getInventoryKey(sku, locationID)
-	currentValCmd := s.client.Get(ctx, key)
-	if err := currentValCmd.Err(); err != nil {
-		return err
-	}
-	currentVal, err := currentValCmd.Int64()
-	if err != nil {
-		return err
-	}
-	payloadMap := map[string]int64{key: currentVal}
-	payloadBytes, err := json.Marshal(payloadMap)
-	if err != nil {
-		return err
-	}
-	payload := string(payloadBytes)
-	cmd := s.client.Publish(ctx, "inventory_changed", payload)
-	return cmd.Err()
-}
-
 var (
 	ErrInsufficientInventory = errors.New("insufficient inventory")
 	ErrInvalidQuantity       = errors.New("invalid quantity")
@@ -282,12 +261,7 @@ for i = 1, num_keys do
   local new_val = new_values[i]
 
   -- Set the key to its new calculated value
-  redis.call('SET', key, new_val)
-end
-local message_payload = cjson.encode(payload)
-
--- Publish the JSON-encoded message to the specified channel
-redis.call('PUBLISH', "inventory_changed", message_payload)
+  redis.call('SET', key, new_val)	end
 -- Return 1 to indicate the operation was successful
 return 1
 `)