Inventory Reservation Backend for Ecommerce Checkout

-- inventory/migrations/1_inventory.up.sql
CREATE TABLE stock (
    sku      TEXT PRIMARY KEY,
    on_hand  INT NOT NULL CHECK (on_hand >= 0),
    reserved INT NOT NULL DEFAULT 0 CHECK (reserved >= 0),
    CHECK (reserved <= on_hand)
);

CREATE TABLE reservations (
    id         BIGSERIAL PRIMARY KEY,
    cart_id    TEXT NOT NULL UNIQUE,
    items_fingerprint TEXT NOT NULL,
    status     TEXT NOT NULL DEFAULT 'held'
               CHECK (status IN ('held', 'committed', 'released')),
    expires_at TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE reservation_items (
    reservation_id BIGINT NOT NULL REFERENCES reservations (id),
    sku            TEXT NOT NULL REFERENCES stock (sku),
    qty            INT NOT NULL CHECK (qty > 0),
    PRIMARY KEY (reservation_id, sku)
);

CREATE INDEX reservations_expiring ON reservations (expires_at)
    WHERE status = 'held';

// inventory/inventory.go
package inventory

import (
	"context"
	"encoding/json"
	"errors"
	"sort"
	"time"

	"encore.dev/beta/errs"
	"encore.dev/storage/sqldb"
)

var db = sqldb.NewDatabase("inventory", sqldb.DatabaseConfig{
	Migrations: "./migrations",
})

const holdFor = 15 * time.Minute

type Item struct {
	SKU string `json:"sku"`
	Qty int    `json:"qty"`
}

type ReserveParams struct {
	CartID string `json:"cart_id"`
	Items  []Item `json:"items"`
}

type Reservation struct {
	ID        int64     `json:"id"`
	Status    string    `json:"status"`
	ExpiresAt time.Time `json:"expires_at"`
}

// Reserve holds stock for every item in the cart, or for none of them.
// Calling it again for the same cart and the same items returns the
// existing reservation. A different item list for that cart is rejected.
//
//encore:api private method=POST path=/reservations
func Reserve(ctx context.Context, p *ReserveParams) (*Reservation, error) {
	if p.CartID == "" || len(p.Items) == 0 {
		return nil, &errs.Error{Code: errs.InvalidArgument, Message: "cart_id and items are required"}
	}
	for _, it := range p.Items {
		if it.Qty <= 0 {
			return nil, &errs.Error{Code: errs.InvalidArgument, Message: "qty must be positive"}
		}
	}
	// Lock rows in a fixed order so two carts with the same SKUs
	// cannot deadlock each other.
	items := append([]Item(nil), p.Items...)
	sort.Slice(items, func(i, j int) bool { return items[i].SKU < items[j].SKU })
	fp := itemsFingerprint(items)

	tx, err := db.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	var r Reservation
	err = tx.QueryRow(ctx, `
		INSERT INTO reservations (cart_id, items_fingerprint, expires_at)
		VALUES ($1, $2, now() + $3 * interval '1 second')
		ON CONFLICT (cart_id) DO NOTHING
		RETURNING id, status, expires_at
	`, p.CartID, fp, int(holdFor.Seconds())).Scan(&r.ID, &r.Status, &r.ExpiresAt)
	if errors.Is(err, sqldb.ErrNoRows) {
		// Retry of a cart that already has a reservation.
		var existing string
		err = tx.QueryRow(ctx, `
			SELECT id, status, expires_at, items_fingerprint
			FROM reservations WHERE cart_id = $1
			FOR SHARE
		`, p.CartID).Scan(&r.ID, &r.Status, &r.ExpiresAt, &existing)
		if err != nil {
			return nil, err
		}
		if existing != fp {
			return nil, &errs.Error{Code: errs.FailedPrecondition, Message: "cart already reserved with different items"}
		}
		return &r, nil
	} else if err != nil {
		return nil, err
	}

	for _, it := range items {
		res, err := tx.Exec(ctx, `
			UPDATE stock SET reserved = reserved + $2
			WHERE sku = $1 AND on_hand - reserved >= $2
		`, it.SKU, it.Qty)
		if err != nil {
			return nil, err
		}
		if res.RowsAffected() == 0 {
			return nil, &errs.Error{Code: errs.FailedPrecondition, Message: "not enough stock for " + it.SKU}
		}
		if _, err := tx.Exec(ctx, `
			INSERT INTO reservation_items (reservation_id, sku, qty) VALUES ($1, $2, $3)
		`, r.ID, it.SKU, it.Qty); err != nil {
			return nil, err
		}
	}
	return &r, tx.Commit()
}

func itemsFingerprint(items []Item) string {
	b, err := json.Marshal(items)
	if err != nil {
		return ""
	}
	return string(b)
}

// Commit turns a held reservation into a sale after payment succeeds.
// It fails if the hold already expired or was released.
//
//encore:api private method=POST path=/reservations/:id/commit
func Commit(ctx context.Context, id int64) error {
	tx, err := db.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	var status string
	err = tx.QueryRow(ctx, `
		UPDATE reservations SET status = 'committed'
		WHERE id = $1 AND status = 'held' AND expires_at > now()
		RETURNING status
	`, id).Scan(&status)
	if errors.Is(err, sqldb.ErrNoRows) {
		return &errs.Error{Code: errs.FailedPrecondition, Message: "reservation is no longer held"}
	} else if err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `
		UPDATE stock s SET on_hand = s.on_hand - i.qty, reserved = s.reserved - i.qty
		FROM reservation_items i
		WHERE i.reservation_id = $1 AND s.sku = i.sku
	`, id); err != nil {
		return err
	}
	return tx.Commit()
}

// release returns held stock. The status check makes it safe to call
// more than once for the same reservation.
func release(ctx context.Context, id int64) error {
	tx, err := db.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	res, err := tx.Exec(ctx, `
		UPDATE reservations SET status = 'released'
		WHERE id = $1 AND status = 'held'
	`, id)
	if err != nil {
		return err
	}
	if res.RowsAffected() == 0 {
		return nil
	}
	if _, err := tx.Exec(ctx, `
		UPDATE stock s SET reserved = s.reserved - i.qty
		FROM reservation_items i
		WHERE i.reservation_id = $1 AND s.sku = i.sku
	`, id); err != nil {
		return err
	}
	return tx.Commit()
}

// payments/payments.go
package payments

import "encore.dev/pubsub"

type PaymentFailedEvent struct {
	ReservationID int64  `json:"reservation_id"`
	PaymentID     string `json:"payment_id"`
	Reason        string `json:"reason"`
}

// Published by the payment provider's webhook handler in this service.
var PaymentFailed = pubsub.NewTopic[*PaymentFailedEvent]("payment-failed", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

// inventory/release.go
package inventory

import (
	"context"

	"encore.app/payments"
	"encore.dev/cron"
	"encore.dev/pubsub"
)

// Release the hold as soon as the payment fails.
var _ = pubsub.NewSubscription(payments.PaymentFailed, "release-hold",
	pubsub.SubscriptionConfig[*payments.PaymentFailedEvent]{
		Handler: func(ctx context.Context, e *payments.PaymentFailedEvent) error {
			return release(ctx, e.ReservationID)
		},
	},
)

// Sweep holds whose checkout was abandoned and never sent an event.
var _ = cron.NewJob("release-expired-holds", cron.JobConfig{
	Title:    "Release expired stock holds",
	Every:    5 * cron.Minute,
	Endpoint: ReleaseExpired,
})

//encore:api private
func ReleaseExpired(ctx context.Context) error {
	rows, err := db.Query(ctx, `
		SELECT id FROM reservations
		WHERE status = 'held' AND expires_at < now()
		ORDER BY expires_at
		LIMIT 500
	`)
	if err != nil {
		return err
	}
	var ids []int64
	for rows.Next() {
		var id int64
		if err := rows.Scan(&id); err != nil {
			rows.Close()
			return err
		}
		ids = append(ids, id)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return err
	}
	for _, id := range ids {
		if err := release(ctx, id); err != nil {
			return err
		}
	}
	return nil
}