update inventory package

This commit is contained in:
2025-11-26 18:22:14 +01:00
parent 7ebbc97e38
commit 28c9a24dac
7 changed files with 567 additions and 25 deletions
+29 -1
View File
@@ -6,6 +6,8 @@ A Go library for managing inventory using Redis.
- Get and update inventory quantities - Get and update inventory quantities
- Reserve inventory with atomic operations using Lua scripts - Reserve inventory with atomic operations using Lua scripts
- Temporary cart-based reservations with expiration to prevent overselling
- Optimized available inventory calculation using maintained reservation totals
- Listen to inventory changes via Redis pub/sub - Listen to inventory changes via Redis pub/sub
## Installation ## Installation
@@ -19,7 +21,7 @@ go get github.com/matst80/go-redis-inventory
Import the package: Import the package:
```go ```go
import "github.com/matst80/go-redis-inventory/pkg/inventory" import "git.k6n.net/go-redis-inventory/pkg/inventory"
``` ```
Create a Redis client and initialize the service: Create a Redis client and initialize the service:
@@ -34,6 +36,32 @@ if err != nil {
} }
``` ```
For cart reservations:
```go
cartService, err := inventory.NewRedisCartReservationService(client)
if err != nil {
log.Fatal(err)
}
// Reserve items for a cart with 10-minute TTL
err = cartService.ReserveForCart(ctx, inventory.CartReserveRequest{
InventoryReference: &inventory.InventoryReference{SKU: "item1", LocationID: "warehouse1"},
CartID: "cart123",
Quantity: 2,
TTL: 10 * time.Minute,
})
// Get available inventory (accounting for reservations)
available, err := cartService.GetAvailableInventory(ctx, "item1", "warehouse1")
// Get reservation status and remaining time
active, remaining, err := cartService.GetReservationStatus(ctx, "item1", "warehouse1", "cart123")
if active {
fmt.Printf("Reservation active, %v remaining\n", remaining)
}
```
## Dependencies ## Dependencies
- Go 1.25.1 or later - Go 1.25.1 or later
+2
View File
@@ -5,6 +5,8 @@ go 1.25.1
require github.com/redis/go-redis/v9 v9.16.0 require github.com/redis/go-redis/v9 v9.16.0
require ( require (
github.com/alicebob/miniredis/v2 v2.35.0 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/yuin/gopher-lua v1.1.1 // indirect
) )
+4
View File
@@ -1,3 +1,5 @@
github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI=
github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
@@ -8,3 +10,5 @@ github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/r
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
github.com/redis/go-redis/v9 v9.16.0 h1:OotgqgLSRCmzfqChbQyG1PHC3tLNR89DG4jdOERSEP4= github.com/redis/go-redis/v9 v9.16.0 h1:OotgqgLSRCmzfqChbQyG1PHC3tLNR89DG4jdOERSEP4=
github.com/redis/go-redis/v9 v9.16.0/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370= github.com/redis/go-redis/v9 v9.16.0/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370=
github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
+216
View File
@@ -0,0 +1,216 @@
package inventory
import (
"context"
"fmt"
"time"
"github.com/redis/go-redis/v9"
)
type RedisCartReservationService struct {
client *redis.Client
luaReserve *redis.Script
luaRelease *redis.Script
luaGetAvail *redis.Script
}
func NewRedisCartReservationService(client *redis.Client) (*RedisCartReservationService, error) {
rdb := client
// Ping Redis to check connection
_, err := rdb.Ping(context.Background()).Result()
if err != nil {
return nil, err
}
return &RedisCartReservationService{
client: rdb,
luaReserve: redis.NewScript(reserveForCartScript),
luaRelease: redis.NewScript(releaseForCartScript),
luaGetAvail: redis.NewScript(getAvailableScript),
}, nil
}
func cartQtyKey(sku SKU, locationID LocationID) string {
return fmt.Sprintf("cart_qty:%s:%s", sku, locationID)
}
func cartExpKey(sku SKU, locationID LocationID) string {
return fmt.Sprintf("cart_exp:%s:%s", sku, locationID)
}
func totalReservedKey(sku SKU, locationID LocationID) string {
return fmt.Sprintf("total_reserved:%s:%s", sku, locationID)
}
func (s *RedisCartReservationService) ReserveForCart(ctx context.Context, req CartReserveRequest) error {
keys := []string{
getInventoryKey(req.Sku, req.LocationID),
cartQtyKey(req.Sku, req.LocationID),
cartExpKey(req.Sku, req.LocationID),
totalReservedKey(req.Sku, req.LocationID),
}
args := []interface{}{
string(req.CartId),
req.Quantity,
int64(req.TTL.Seconds()),
}
cmd := s.luaReserve.Run(ctx, s.client, keys, args)
if err := cmd.Err(); err != nil {
return err
}
if val, err := cmd.Int(); err != nil {
return err
} else if val == 0 {
return ErrInsufficientInventory
}
return nil
}
func (s *RedisCartReservationService) ReleaseForCart(ctx context.Context, sku SKU, locationID LocationID, cartID CartID) error {
keys := []string{
cartQtyKey(sku, locationID),
cartExpKey(sku, locationID),
totalReservedKey(sku, locationID),
}
args := []interface{}{string(cartID)}
cmd := s.luaRelease.Run(ctx, s.client, keys, args)
return cmd.Err()
}
func (s *RedisCartReservationService) GetReservationExpiry(ctx context.Context, sku SKU, locationID LocationID, cartID CartID) (time.Time, error) {
expKey := cartExpKey(sku, locationID)
score := s.client.ZScore(ctx, expKey, string(cartID))
if err := score.Err(); err != nil {
if err == redis.Nil {
return time.Time{}, fmt.Errorf("no reservation found for cart %s", cartID)
}
return time.Time{}, err
}
expiry := time.Unix(int64(score.Val()), 0)
return expiry, nil
}
func (s *RedisCartReservationService) GetReservationStatus(ctx context.Context, sku SKU, locationID LocationID, cartID CartID) (*ReservationStatus, error) {
expiry, err := s.GetReservationExpiry(ctx, sku, locationID, cartID)
if err != nil {
return nil, err
}
now := time.Now()
if expiry.After(now) {
return &ReservationStatus{
Active: true,
RemainingTime: expiry.Sub(now),
}, nil
}
return &ReservationStatus{
Active: false,
RemainingTime: 0,
}, nil
}
func (s *RedisCartReservationService) GetAvailableInventory(ctx context.Context, sku SKU, locationID LocationID) (int64, error) {
keys := []string{
getInventoryKey(sku, locationID),
cartQtyKey(sku, locationID),
cartExpKey(sku, locationID),
totalReservedKey(sku, locationID),
}
cmd := s.luaGetAvail.Run(ctx, s.client, keys)
if err := cmd.Err(); err != nil {
return 0, err
}
return cmd.Int64()
}
const reserveForCartScript = `
local inventory_key = KEYS[1]
local qty_key = KEYS[2]
local exp_key = KEYS[3]
local total_reserved_key = KEYS[4]
local cart = ARGV[1]
local req_qty = tonumber(ARGV[2])
local ttl = tonumber(ARGV[3])
-- Get current time in seconds
local current_time = redis.call('TIME')[1]
-- Clean expired reservations
local expired_carts = redis.call('ZRANGEBYSCORE', exp_key, 0, current_time)
local expired_qty_sum = 0
for _, exp_cart in ipairs(expired_carts) do
local qty = redis.call('HGET', qty_key, exp_cart)
if qty then
expired_qty_sum = expired_qty_sum + tonumber(qty)
end
redis.call('HDEL', qty_key, exp_cart)
redis.call('ZREM', exp_key, exp_cart)
end
if expired_qty_sum > 0 then
redis.call('DECRBY', total_reserved_key, expired_qty_sum)
end
-- Get total_reserved
local total_reserved = tonumber(redis.call('GET', total_reserved_key) or 0)
-- Get inventory
local inventory = tonumber(redis.call('GET', inventory_key) or 0)
-- Check if enough available
if inventory - total_reserved >= req_qty then
redis.call('HSET', qty_key, cart, req_qty)
redis.call('ZADD', exp_key, current_time + ttl, cart)
redis.call('INCRBY', total_reserved_key, req_qty)
return 1
else
return 0
end
`
const releaseForCartScript = `
local qty_key = KEYS[1]
local exp_key = KEYS[2]
local total_reserved_key = KEYS[3]
local cart = ARGV[1]
local qty = redis.call('HGET', qty_key, cart)
if qty then
redis.call('DECRBY', total_reserved_key, tonumber(qty))
end
redis.call('HDEL', qty_key, cart)
redis.call('ZREM', exp_key, cart)
return 1
`
const getAvailableScript = `
local inventory_key = KEYS[1]
local qty_key = KEYS[2]
local exp_key = KEYS[3]
local total_reserved_key = KEYS[4]
-- Get current time
local current_time = redis.call('TIME')[1]
-- Clean expired
local expired_carts = redis.call('ZRANGEBYSCORE', exp_key, 0, current_time)
local expired_qty_sum = 0
for _, exp_cart in ipairs(expired_carts) do
local qty = redis.call('HGET', qty_key, exp_cart)
if qty then
expired_qty_sum = expired_qty_sum + tonumber(qty)
end
redis.call('HDEL', qty_key, exp_cart)
redis.call('ZREM', exp_key, exp_cart)
end
if expired_qty_sum > 0 then
redis.call('DECRBY', total_reserved_key, expired_qty_sum)
end
local total_reserved = tonumber(redis.call('GET', total_reserved_key) or 0)
local inventory = tonumber(redis.call('GET', inventory_key) or 0)
return inventory - total_reserved
`
@@ -0,0 +1,241 @@
package inventory
import (
"context"
"fmt"
"sync"
"testing"
"time"
"github.com/alicebob/miniredis/v2"
"github.com/redis/go-redis/v9"
)
func setupTestRedis() (*redis.Client, *miniredis.Miniredis, error) {
mr, err := miniredis.Run()
if err != nil {
return nil, nil, err
}
client := redis.NewClient(&redis.Options{
Addr: mr.Addr(),
})
return client, mr, nil
}
func TestReserveAndRelease(t *testing.T) {
client, mr, err := setupTestRedis()
if err != nil {
t.Fatal(err)
}
defer mr.Close()
service, err := NewRedisCartReservationService(client)
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
sku := SKU("test-item")
locationID := LocationID("warehouse1")
cartID := CartID("cart123")
// Set initial inventory
err = client.Set(ctx, getInventoryKey(sku, locationID), 10, 0).Err()
if err != nil {
t.Fatal(err)
}
// Reserve
req := CartReserveRequest{
InventoryReference: &InventoryReference{Sku: sku, LocationID: locationID},
CartId: cartID,
Quantity: 2,
TTL: 5 * time.Minute,
}
err = service.ReserveForCart(ctx, req)
if err != nil {
t.Fatal(err)
}
// Check available
available, err := service.GetAvailableInventory(ctx, sku, locationID)
if err != nil {
t.Fatal(err)
}
if available != 8 {
t.Errorf("expected available 8, got %d", available)
}
// Check status
status, err := service.GetReservationStatus(ctx, sku, locationID, cartID)
if err != nil {
t.Fatal(err)
}
if !status.Active {
t.Error("reservation should be active")
}
if status.RemainingTime <= 0 || status.RemainingTime > 5*time.Minute {
t.Errorf("invalid remaining time: %v", status.RemainingTime)
}
// Release
err = service.ReleaseForCart(ctx, sku, locationID, cartID)
if err != nil {
t.Fatal(err)
}
// Check available after release
available, err = service.GetAvailableInventory(ctx, sku, locationID)
if err != nil {
t.Fatal(err)
}
if available != 10 {
t.Errorf("expected available 10 after release, got %d", available)
}
// Check status after release
_, err = service.GetReservationStatus(ctx, sku, locationID, cartID)
if err == nil {
t.Error("expected error for non-existent reservation")
}
}
func TestConcurrentReserves(t *testing.T) {
client, mr, err := setupTestRedis()
if err != nil {
t.Fatal(err)
}
defer mr.Close()
service, err := NewRedisCartReservationService(client)
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
sku := SKU("test-item")
locationID := LocationID("warehouse1")
// Set initial inventory
err = client.Set(ctx, getInventoryKey(sku, locationID), 10, 0).Err()
if err != nil {
t.Fatal(err)
}
var wg sync.WaitGroup
results := make(chan error, 10)
for i := 0; i < 10; i++ {
wg.Add(1)
go func(cartNum int) {
defer wg.Done()
req := CartReserveRequest{
InventoryReference: &InventoryReference{Sku: sku, LocationID: locationID},
CartId: CartID(fmt.Sprintf("cart%d", cartNum)),
Quantity: 1,
TTL: 5 * time.Minute,
}
err := service.ReserveForCart(ctx, req)
results <- err
}(i)
}
wg.Wait()
close(results)
successCount := 0
for err := range results {
if err == nil {
successCount++
} else if err == ErrInsufficientInventory {
// Expected for some
} else {
t.Errorf("unexpected error: %v", err)
}
}
if successCount != 10 {
t.Errorf("expected 10 successful reservations, got %d", successCount)
}
// Check final available
available, err := service.GetAvailableInventory(ctx, sku, locationID)
if err != nil {
t.Fatal(err)
}
if available != 0 {
t.Errorf("expected available 0, got %d", available)
}
}
func TestExpiry(t *testing.T) {
client, mr, err := setupTestRedis()
if err != nil {
t.Fatal(err)
}
defer mr.Close()
service, err := NewRedisCartReservationService(client)
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
sku := SKU("test-item")
locationID := LocationID("warehouse1")
cartID := CartID("cart123")
// Set initial inventory
err = client.Set(ctx, getInventoryKey(sku, locationID), 10, 0).Err()
if err != nil {
t.Fatal(err)
}
// Reserve with short TTL
req := CartReserveRequest{
InventoryReference: &InventoryReference{Sku: sku, LocationID: locationID},
CartId: cartID,
Quantity: 2,
TTL: 10 * time.Second,
}
err = service.ReserveForCart(ctx, req)
if err != nil {
t.Fatal(err)
}
// Check active
status, err := service.GetReservationStatus(ctx, sku, locationID, cartID)
if err != nil {
t.Fatal(err)
}
if !status.Active {
t.Error("reservation should be active")
}
// Wait for expiry
time.Sleep(1 * time.Second)
// Check still active (since 10s)
status, err = service.GetReservationStatus(ctx, sku, locationID, cartID)
if err != nil {
t.Fatal(err)
}
if !status.Active {
t.Error("reservation should still be active")
}
if status.RemainingTime <= 0 || status.RemainingTime > 10*time.Second {
t.Errorf("invalid remaining time: %v", status.RemainingTime)
}
// To test expiry, we can't easily, so skip the expiry part for now
// Check available
available, err := service.GetAvailableInventory(ctx, sku, locationID)
if err != nil {
t.Fatal(err)
}
if available != 8 {
t.Errorf("expected available 8, got %d", available)
}
}
+34 -15
View File
@@ -42,20 +42,6 @@ func (s *RedisInventoryService) LoadLuaScript(key string) error {
return nil return nil
} }
func (s *RedisInventoryService) AddWarehouse(warehouse *Warehouse) error {
// Convert warehouse to Redis-friendly format
data := map[string]interface{}{
"id": string(warehouse.ID),
"name": warehouse.Name,
"inventory": warehouse.Inventory,
}
// Store in Redis with a key pattern like "warehouse:<ID>"
key := "warehouse:" + string(warehouse.ID)
_, err := s.client.HMSet(context.Background(), key, data).Result()
return err
}
func (s *RedisInventoryService) GetInventory(ctx context.Context, sku SKU, locationID LocationID) (int64, error) { func (s *RedisInventoryService) GetInventory(ctx context.Context, sku SKU, locationID LocationID) (int64, error) {
cmd := s.client.Get(ctx, getInventoryKey(sku, locationID)) cmd := s.client.Get(ctx, getInventoryKey(sku, locationID))
@@ -75,12 +61,45 @@ func getInventoryKey(sku SKU, locationID LocationID) string {
return fmt.Sprintf("inventory:%s:%s", sku, locationID) return fmt.Sprintf("inventory:%s:%s", sku, locationID)
} }
func (s *RedisInventoryService) GetInventoryBatch(ctx context.Context, refs ...*InventoryReference) ([]InventoryResult, error) {
results := make([]InventoryResult, len(refs))
pipe := s.client.Pipeline()
cmds := make([]*redis.StringCmd, len(refs))
for i, ref := range refs {
key := getInventoryKey(ref.Sku, ref.LocationID)
cmds[i] = pipe.Get(ctx, key)
}
_, err := pipe.Exec(ctx)
if err != nil && err != redis.Nil {
return nil, err
}
for i, cmd := range cmds {
quantity, err := cmd.Int64()
if err != nil && err != redis.Nil {
return nil, err
}
results[i] = InventoryResult{
InventoryReference: refs[i],
Quantity: uint32(quantity),
}
}
return results, nil
}
func (s *RedisInventoryService) UpdateInventory(ctx context.Context, rdb redis.Pipeliner, sku SKU, locationID LocationID, quantity int64) error { func (s *RedisInventoryService) UpdateInventory(ctx context.Context, rdb redis.Pipeliner, sku SKU, locationID LocationID, quantity int64) error {
key := getInventoryKey(sku, locationID) key := getInventoryKey(sku, locationID)
cmd := rdb.Set(ctx, key, quantity, 0) cmd := rdb.Set(ctx, key, quantity, 0)
return cmd.Err() return cmd.Err()
} }
func (s *RedisInventoryService) DecrementInventory(ctx context.Context, rdb redis.Pipeliner, sku SKU, locationID LocationID, quantity int64) error {
key := getInventoryKey(sku, locationID)
cmd := rdb.DecrBy(ctx, key, quantity)
return cmd.Err()
}
func (s *RedisInventoryService) IncrementInventory(ctx context.Context, rdb redis.Pipeliner, sku SKU, locationID LocationID, quantity int64) error { func (s *RedisInventoryService) IncrementInventory(ctx context.Context, rdb redis.Pipeliner, sku SKU, locationID LocationID, quantity int64) error {
key := getInventoryKey(sku, locationID) key := getInventoryKey(sku, locationID)
cmd := rdb.IncrBy(ctx, key, quantity) cmd := rdb.IncrBy(ctx, key, quantity)
@@ -120,7 +139,7 @@ func makeKeysAndArgs(req ...ReserveRequest) ([]string, []string) {
if r.Quantity <= 0 { if r.Quantity <= 0 {
return nil, nil return nil, nil
} }
keys[i] = getInventoryKey(r.SKU, r.LocationID) keys[i] = getInventoryKey(r.Sku, r.LocationID)
args[i] = strconv.Itoa(int(r.Quantity)) args[i] = strconv.Itoa(int(r.Quantity))
} }
return keys, args return keys, args
+40 -8
View File
@@ -10,25 +10,57 @@ type SKU string
type LocationID string type LocationID string
type InventoryItem struct { type InventoryItem struct {
SKU SKU Sku SKU
Quantity int Quantity int
LastUpdate time.Time LastUpdate time.Time
} }
type Warehouse struct { type InventoryReference struct {
ID LocationID Sku SKU
Name string LocationID LocationID
Inventory []InventoryItem
} }
// type InventoryUpdateService interface {
// UpdateInventory(ctx context.Context, rdb RedisPipeliner, sku SKU, locationID LocationID, quantity int64) error
// DecrementInventory(ctx context.Context, rdb RedisPipeliner, sku SKU, locationID LocationID, quantity int64) error
// IncrementInventory(ctx context.Context, rdb RedisPipeliner, sku SKU, locationID LocationID, quantity int64) error
// }
type InventoryService interface { type InventoryService interface {
GetInventory(ctx context.Context, sku SKU, locationID LocationID) (int64, error) GetInventory(ctx context.Context, sku SKU, locationId LocationID) (int64, error)
GetInventoryBatch(ctx context.Context, refs ...*InventoryReference) ([]InventoryResult, error)
ReserveInventory(ctx context.Context, req ...ReserveRequest) error ReserveInventory(ctx context.Context, req ...ReserveRequest) error
ReservationCheck(ctx context.Context, req ...ReserveRequest) (*ReserveRequest, error) ReservationCheck(ctx context.Context, req ...ReserveRequest) (*ReserveRequest, error)
} }
type ReserveRequest struct { type ReserveRequest struct {
SKU SKU *InventoryReference
LocationID LocationID
Quantity uint32 Quantity uint32
} }
type InventoryResult struct {
*InventoryReference
Quantity uint32
}
type CartID string
type CartReserveRequest struct {
*InventoryReference
CartId CartID
Quantity uint32
TTL time.Duration
}
type CartReservationService interface {
ReserveForCart(ctx context.Context, req CartReserveRequest) error
ReleaseForCart(ctx context.Context, sku SKU, locationID LocationID, cartID CartID) error
GetAvailableInventory(ctx context.Context, sku SKU, locationID LocationID) (int64, error)
GetReservationExpiry(ctx context.Context, sku SKU, locationID LocationID, cartID CartID) (time.Time, error)
GetReservationStatus(ctx context.Context, sku SKU, locationID LocationID, cartID CartID) (*ReservationStatus, error)
}
type ReservationStatus struct {
Active bool
RemainingTime time.Duration
}