75 lines
2.6 KiB
Go
75 lines
2.6 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"time"
|
|
|
|
"git.k6n.net/mats/go-redis-inventory/pkg/inventory"
|
|
contract "git.k6n.net/mats/platform/order"
|
|
"git.k6n.net/mats/platform/rabbit"
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
// commitOrder applies an order.created event to inventory. The sale already
|
|
// happened, so this is an unconditional permanent decrement (it may drive stock
|
|
// negative — visible oversell, reconciled downstream as backorder), NOT a
|
|
// check-and-reserve that could refuse. It then releases the cart's TTL hold and
|
|
// emits the resulting level crossing. Idempotent per order via a Redis marker,
|
|
// since the bus is at-least-once.
|
|
func commitOrder(ctx context.Context, rdb *redis.Client, svc *inventory.RedisInventoryService, resv *inventory.RedisCartReservationService, conn *rabbit.Conn, country string, low int64, c contract.Created) {
|
|
if c.OrderID == "" {
|
|
return
|
|
}
|
|
// Idempotency: first writer wins; redelivery is a no-op.
|
|
fresh, err := rdb.SetNX(ctx, "invcommit:"+c.OrderID, "1", 30*24*time.Hour).Result()
|
|
if err != nil {
|
|
log.Printf("inventory: commit idempotency order=%s: %v", c.OrderID, err)
|
|
return
|
|
}
|
|
if !fresh {
|
|
return // already committed
|
|
}
|
|
// Default location for lines that don't carry their own: the order country
|
|
// (central stock), falling back to the service's configured country.
|
|
defaultLoc := c.Country
|
|
if defaultLoc == "" {
|
|
defaultLoc = country
|
|
}
|
|
for _, line := range c.Lines {
|
|
if line.SKU == "" || line.Quantity <= 0 {
|
|
continue
|
|
}
|
|
// Drop-ship items are fulfilled by the supplier — never decrement
|
|
// local stock for them.
|
|
if line.DropShip {
|
|
continue
|
|
}
|
|
loc := inventory.LocationID(defaultLoc)
|
|
if line.Location != "" {
|
|
loc = inventory.LocationID(line.Location)
|
|
}
|
|
sku := inventory.SKU(line.SKU)
|
|
pipe := rdb.Pipeline()
|
|
if err := svc.DecrementInventory(ctx, pipe, sku, loc, int64(line.Quantity)); err != nil {
|
|
log.Printf("inventory: commit decrement sku=%s order=%s: %v", sku, c.OrderID, err)
|
|
continue
|
|
}
|
|
if _, err := pipe.Exec(ctx); err != nil {
|
|
log.Printf("inventory: commit exec sku=%s order=%s: %v", sku, c.OrderID, err)
|
|
continue
|
|
}
|
|
// Release the cart's TTL hold now the sale is committed (best-effort —
|
|
// it would expire on its own anyway).
|
|
if c.CartID != "" && resv != nil {
|
|
if err := resv.ReleaseForCart(ctx, sku, loc, inventory.CartID(c.CartID)); err != nil {
|
|
log.Printf("inventory: commit release sku=%s cart=%s: %v", sku, c.CartID, err)
|
|
}
|
|
}
|
|
// Emit the level crossing from the new on-hand.
|
|
if newQty, err := svc.GetInventory(ctx, sku, loc); err == nil {
|
|
emitLevelIfChanged(ctx, rdb, conn, country, low, string(sku), string(loc), newQty)
|
|
}
|
|
}
|
|
}
|