463 lines
16 KiB
Go
463 lines
16 KiB
Go
// Command order is a single-node HTTP entrypoint for the order grain. It exposes
|
|
// a simple checkout that creates an order by running the place-and-pay flow
|
|
// (place -> mock authorize -> capture) on the actor framework, plus read APIs
|
|
// for an order and its event timeline. No clustering, no external payment — the
|
|
// simplest path to a real, event-sourced, captured order.
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"log/slog"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.k6n.net/mats/go-cart-actor/pkg/actor"
|
|
"git.k6n.net/mats/go-cart-actor/pkg/flow"
|
|
"git.k6n.net/mats/go-cart-actor/pkg/idempotency"
|
|
"git.k6n.net/mats/go-cart-actor/pkg/order"
|
|
"git.k6n.net/mats/go-cart-actor/pkg/outbox"
|
|
messages "git.k6n.net/mats/go-cart-actor/proto/order"
|
|
|
|
"google.golang.org/protobuf/encoding/protojson"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
type server struct {
|
|
pool *actor.SimpleGrainPool[order.OrderGrain]
|
|
applier *orderedApplier
|
|
storage *actor.DiskStorage[order.OrderGrain]
|
|
reg actor.MutationRegistry
|
|
engine *flow.Engine
|
|
freg *flow.Registry
|
|
flows *flowStore
|
|
provider order.PaymentProvider
|
|
defaultFlow string
|
|
dataDir string
|
|
idem *idempotency.Store
|
|
logger *slog.Logger
|
|
}
|
|
|
|
func main() {
|
|
addr := envOr("ORDER_ADDR", ":8092") // :8090 cart, :8091 redirector
|
|
dataDir := envOr("ORDER_DATA", "data/orders")
|
|
flowsDir := envOr("ORDER_FLOWS", "data/order-flows")
|
|
if err := os.MkdirAll(dataDir, 0o755); err != nil {
|
|
log.Fatalf("create data dir: %v", err)
|
|
}
|
|
logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
|
|
|
|
reg := actor.NewMutationRegistry()
|
|
order.RegisterMutations(reg)
|
|
storage := actor.NewDiskStorage[order.OrderGrain](dataDir, reg)
|
|
|
|
pool, err := actor.NewSimpleGrainPool(actor.GrainPoolConfig[order.OrderGrain]{
|
|
Hostname: "order-1",
|
|
Spawn: func(ctx context.Context, id uint64) (actor.Grain[order.OrderGrain], error) {
|
|
g := order.NewOrderGrain(id, time.Now())
|
|
// Replay any persisted history so the resident grain is current.
|
|
if err := storage.LoadEvents(ctx, id, g); err != nil {
|
|
return nil, err
|
|
}
|
|
return g, nil
|
|
},
|
|
SpawnHost: func(string) (actor.Host[order.OrderGrain], error) {
|
|
return nil, fmt.Errorf("order service is single-node")
|
|
},
|
|
Destroy: func(actor.Grain[order.OrderGrain]) error { return nil },
|
|
TTL: time.Hour,
|
|
PoolSize: 10000,
|
|
// Storage is intentionally nil here: the pool persists mutations in a
|
|
// fire-and-forget goroutine, so concurrent appends for one grain can land
|
|
// out of order on disk — fatal for an event log, where replay order is the
|
|
// state. We persist through orderedApplier instead (synchronous, ordered,
|
|
// and only for mutations that actually applied). Replay on spawn still uses
|
|
// `storage` directly in Spawn above.
|
|
MutationRegistry: reg,
|
|
Storage: nil,
|
|
})
|
|
if err != nil {
|
|
log.Fatalf("create pool: %v", err)
|
|
}
|
|
|
|
applier := &orderedApplier{pool: pool, storage: storage}
|
|
provider := selectProvider(logger)
|
|
|
|
// Flow registry: generic hooks + the order lifecycle actions + predicates,
|
|
// paying via the selected provider. The amqp_emit hook bridges saga steps to
|
|
// the broker (publisher is nil without AMQP — the hook then errors at run
|
|
// time, which is logged, not fatal).
|
|
amqpURL := os.Getenv("AMQP_URL")
|
|
var emitPub order.Publisher
|
|
if amqpURL != "" {
|
|
// The amqp_emit hook writes to a durable outbox rather than the broker
|
|
// directly; a relay goroutine drains it to RabbitMQ (reconnecting), so an
|
|
// event survives a broker outage or a crash after the state change.
|
|
box, err := outbox.Open(filepath.Join(dataDir, "outbox.log"))
|
|
if err != nil {
|
|
// Durability-critical (same as the idempotency store): fail fast rather
|
|
// than silently publish events nowhere for the whole process lifetime.
|
|
log.Fatalf("open outbox: %v", err)
|
|
}
|
|
go box.Run(context.Background(), newRabbitPublisher(amqpURL), 5*time.Second, logger)
|
|
emitPub = box
|
|
}
|
|
freg := flow.NewRegistry()
|
|
flow.RegisterBuiltinHooks(freg)
|
|
order.RegisterFlowActions(freg, applier, provider)
|
|
order.RegisterEmitHook(freg, emitPub)
|
|
engine := flow.NewEngine(freg, logger)
|
|
|
|
def, err := order.EmbeddedFlow("place-and-pay")
|
|
if err != nil {
|
|
log.Fatalf("load flow: %v", err)
|
|
}
|
|
flows, err := newFlowStore(flowsDir, engine, map[string]*flow.Definition{def.Name: def})
|
|
if err != nil {
|
|
log.Fatalf("load flows: %v", err)
|
|
}
|
|
|
|
// Durable idempotency store for order creation — survives a restart, so a
|
|
// client retry after a crash returns the existing order instead of creating
|
|
// a duplicate.
|
|
idem, err := idempotency.Open(filepath.Join(dataDir, "idempotency.log"))
|
|
if err != nil {
|
|
log.Fatalf("open idempotency store: %v", err)
|
|
}
|
|
|
|
s := &server{
|
|
pool: pool, applier: applier, storage: storage, reg: reg,
|
|
engine: engine, freg: freg, flows: flows, provider: provider,
|
|
defaultFlow: def.Name, dataDir: dataDir, idem: idem, logger: logger,
|
|
}
|
|
|
|
mux := http.NewServeMux()
|
|
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })
|
|
|
|
// Checkout + order reads.
|
|
mux.HandleFunc("POST /checkout", s.handleCheckout)
|
|
mux.HandleFunc("POST /api/orders/from-checkout", s.handleFromCheckout)
|
|
mux.HandleFunc("GET /api/orders", s.handleList)
|
|
mux.HandleFunc("GET /api/orders/{id}", s.handleGet)
|
|
mux.HandleFunc("GET /api/orders/{id}/events", s.handleEvents)
|
|
|
|
// Post-purchase lifecycle (the grain already supports these transitions).
|
|
mux.HandleFunc("POST /api/orders/{id}/fulfillments", s.handleFulfill)
|
|
mux.HandleFunc("POST /api/orders/{id}/complete", s.handleComplete)
|
|
mux.HandleFunc("POST /api/orders/{id}/cancel", s.handleCancel)
|
|
mux.HandleFunc("POST /api/orders/{id}/returns", s.handleReturn)
|
|
mux.HandleFunc("POST /api/orders/{id}/refunds", s.handleRefund)
|
|
|
|
// Saga / flow admin (backoffice editor).
|
|
mux.HandleFunc("GET /sagas/capabilities", s.handleCapabilities)
|
|
mux.HandleFunc("GET /sagas/flows", s.handleListFlows)
|
|
mux.HandleFunc("GET /sagas/flows/{name}", s.handleGetFlow)
|
|
mux.HandleFunc("PUT /sagas/flows/{name}", s.handleSaveFlow)
|
|
|
|
// Ingest legacy checkout orders (Klarna/Adyen → order-queue) so this service
|
|
// is the single source of orders, superseding go-order-manager.
|
|
if amqpURL != "" {
|
|
startLegacyIngest(context.Background(), amqpURL, applier, logger)
|
|
}
|
|
|
|
logger.Info("order service listening", "addr", addr, "data", dataDir, "flows", flowsDir)
|
|
if err := http.ListenAndServe(addr, mux); err != nil {
|
|
log.Fatalf("serve: %v", err)
|
|
}
|
|
}
|
|
|
|
// orderedApplier wraps the grain pool to persist the event log correctly: it
|
|
// appends each applied mutation synchronously and in call order (the pool's own
|
|
// storage path is async/unordered), and only persists mutations whose handler
|
|
// succeeded — a rejected transition must never enter the log. A single mutex
|
|
// serialises all appends, which is ample for a single-node checkout service.
|
|
type orderedApplier struct {
|
|
pool *actor.SimpleGrainPool[order.OrderGrain]
|
|
storage *actor.DiskStorage[order.OrderGrain]
|
|
mu sync.Mutex
|
|
}
|
|
|
|
func (a *orderedApplier) Get(ctx context.Context, id uint64) (*order.OrderGrain, error) {
|
|
return a.pool.Get(ctx, id)
|
|
}
|
|
|
|
func (a *orderedApplier) Apply(ctx context.Context, id uint64, mutation ...proto.Message) (*actor.MutationResult[order.OrderGrain], error) {
|
|
res, err := a.pool.Apply(ctx, id, mutation...)
|
|
if err != nil || res == nil {
|
|
return res, err
|
|
}
|
|
// Persist only the mutations that actually applied (result index aligns with
|
|
// the input order).
|
|
applied := make([]proto.Message, 0, len(mutation))
|
|
for i, m := range mutation {
|
|
if i < len(res.Mutations) && res.Mutations[i].Error != nil {
|
|
continue
|
|
}
|
|
applied = append(applied, m)
|
|
}
|
|
if len(applied) > 0 {
|
|
a.mu.Lock()
|
|
appendErr := a.storage.AppendMutations(id, applied...)
|
|
a.mu.Unlock()
|
|
if appendErr != nil {
|
|
return res, appendErr
|
|
}
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
// --- checkout -------------------------------------------------------------
|
|
|
|
type lineReq struct {
|
|
Reference string `json:"reference"`
|
|
Sku string `json:"sku"`
|
|
Name string `json:"name"`
|
|
Quantity int32 `json:"quantity"`
|
|
UnitPrice int64 `json:"unitPrice"` // inc-vat, minor units
|
|
TaxRate int32 `json:"taxRate"` // percent
|
|
}
|
|
|
|
type checkoutReq struct {
|
|
OrderReference string `json:"orderReference,omitempty"`
|
|
CartId string `json:"cartId,omitempty"`
|
|
Currency string `json:"currency,omitempty"`
|
|
Locale string `json:"locale,omitempty"`
|
|
Country string `json:"country,omitempty"`
|
|
CustomerEmail string `json:"customerEmail,omitempty"`
|
|
CustomerName string `json:"customerName,omitempty"`
|
|
BillingAddress json.RawMessage `json:"billingAddress,omitempty"`
|
|
ShippingAddress json.RawMessage `json:"shippingAddress,omitempty"`
|
|
Lines []lineReq `json:"lines"`
|
|
}
|
|
|
|
func (s *server) handleCheckout(w http.ResponseWriter, r *http.Request) {
|
|
var req checkoutReq
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeErr(w, http.StatusBadRequest, fmt.Errorf("bad body: %w", err))
|
|
return
|
|
}
|
|
if len(req.Lines) == 0 {
|
|
writeErr(w, http.StatusBadRequest, fmt.Errorf("checkout requires at least one line"))
|
|
return
|
|
}
|
|
|
|
id, err := order.NewOrderId()
|
|
if err != nil {
|
|
writeErr(w, http.StatusInternalServerError, err)
|
|
return
|
|
}
|
|
if req.OrderReference == "" {
|
|
req.OrderReference = "ord-" + id.String()
|
|
}
|
|
if req.Currency == "" {
|
|
req.Currency = "SEK"
|
|
}
|
|
|
|
flowName := r.URL.Query().Get("flow")
|
|
if flowName == "" {
|
|
flowName = s.defaultFlow
|
|
}
|
|
def, ok := s.flows.get(flowName)
|
|
if !ok {
|
|
writeErr(w, http.StatusBadRequest, fmt.Errorf("unknown flow %q", flowName))
|
|
return
|
|
}
|
|
|
|
po := buildPlaceOrder(id, &req)
|
|
|
|
st := flow.NewState(uint64(id), s.logger)
|
|
st.Vars[order.PlaceOrderVar] = po
|
|
|
|
res, runErr := s.engine.Run(r.Context(), def, st)
|
|
grain, getErr := s.pool.Get(r.Context(), uint64(id))
|
|
if getErr != nil {
|
|
writeErr(w, http.StatusInternalServerError, getErr)
|
|
return
|
|
}
|
|
|
|
status := http.StatusCreated
|
|
if runErr != nil {
|
|
// The flow compensated; the order exists but is not paid. Report 402 with
|
|
// the (now cancelled/failed) order and the flow trace so the caller knows.
|
|
status = http.StatusPaymentRequired
|
|
s.logger.Warn("checkout flow failed", "orderId", id.String(), "err", runErr)
|
|
}
|
|
writeJSON(w, status, map[string]any{
|
|
"orderId": id.String(),
|
|
"flow": res,
|
|
"order": grain,
|
|
})
|
|
}
|
|
|
|
// buildPlaceOrder converts a checkout request into the PlaceOrder event,
|
|
// computing per-line and order totals (inc-vat) from the lines.
|
|
func buildPlaceOrder(id order.OrderId, req *checkoutReq) *messages.PlaceOrder {
|
|
po := &messages.PlaceOrder{
|
|
OrderReference: req.OrderReference,
|
|
CartId: req.CartId,
|
|
Currency: req.Currency,
|
|
Locale: req.Locale,
|
|
Country: req.Country,
|
|
CustomerEmail: req.CustomerEmail,
|
|
CustomerName: req.CustomerName,
|
|
BillingAddress: req.BillingAddress,
|
|
ShippingAddress: req.ShippingAddress,
|
|
PlacedAtMs: time.Now().UnixMilli(),
|
|
}
|
|
var total, totalTax int64
|
|
for _, l := range req.Lines {
|
|
lineTotal := l.UnitPrice * int64(l.Quantity)
|
|
var lineTax int64
|
|
if l.TaxRate > 0 {
|
|
// inc-vat -> tax portion = total * rate / (100 + rate)
|
|
lineTax = lineTotal * int64(l.TaxRate) / int64(100+l.TaxRate)
|
|
}
|
|
total += lineTotal
|
|
totalTax += lineTax
|
|
po.Lines = append(po.Lines, &messages.OrderLine{
|
|
Reference: l.Reference,
|
|
Sku: l.Sku,
|
|
Name: l.Name,
|
|
Quantity: l.Quantity,
|
|
UnitPrice: l.UnitPrice,
|
|
TaxRate: l.TaxRate,
|
|
TotalAmount: lineTotal,
|
|
TotalTax: lineTax,
|
|
})
|
|
}
|
|
po.TotalAmount = total
|
|
po.TotalTax = totalTax
|
|
return po
|
|
}
|
|
|
|
// --- reads ----------------------------------------------------------------
|
|
|
|
func (s *server) handleGet(w http.ResponseWriter, r *http.Request) {
|
|
id, ok := order.ParseOrderId(r.PathValue("id"))
|
|
if !ok {
|
|
writeErr(w, http.StatusBadRequest, fmt.Errorf("invalid order id"))
|
|
return
|
|
}
|
|
grain, err := s.pool.Get(r.Context(), uint64(id))
|
|
if err != nil {
|
|
writeErr(w, http.StatusInternalServerError, err)
|
|
return
|
|
}
|
|
if grain.Status == order.StatusNew {
|
|
writeErr(w, http.StatusNotFound, fmt.Errorf("order not found"))
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"orderId": id.String(), "order": grain})
|
|
}
|
|
|
|
type eventView struct {
|
|
Type string `json:"type"`
|
|
Time string `json:"time,omitempty"`
|
|
Payload json.RawMessage `json:"payload,omitempty"`
|
|
}
|
|
|
|
// handleEvents returns the raw event timeline by reading the grain's append-only
|
|
// log (the source of truth) without re-applying it to live state.
|
|
func (s *server) handleEvents(w http.ResponseWriter, r *http.Request) {
|
|
id, ok := order.ParseOrderId(r.PathValue("id"))
|
|
if !ok {
|
|
writeErr(w, http.StatusBadRequest, fmt.Errorf("invalid order id"))
|
|
return
|
|
}
|
|
var events []eventView
|
|
throwaway := order.NewOrderGrain(uint64(id), time.Now())
|
|
err := s.storage.LoadEventsFunc(r.Context(), uint64(id), throwaway,
|
|
func(msg proto.Message, _ int, ts time.Time) bool {
|
|
name, _ := s.reg.GetTypeName(msg)
|
|
payload, _ := protojson.Marshal(msg)
|
|
events = append(events, eventView{Type: name, Time: ts.UTC().Format(time.RFC3339), Payload: payload})
|
|
return false // collect only; do not apply
|
|
})
|
|
if err != nil {
|
|
writeErr(w, http.StatusInternalServerError, err)
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"orderId": id.String(), "events": events})
|
|
}
|
|
|
|
type orderSummary struct {
|
|
OrderId string `json:"orderId"`
|
|
Reference string `json:"reference,omitempty"`
|
|
Status order.Status `json:"status"`
|
|
TotalAmount int64 `json:"totalAmount"`
|
|
CapturedAmount int64 `json:"capturedAmount"`
|
|
Currency string `json:"currency,omitempty"`
|
|
PlacedAt string `json:"placedAt,omitempty"`
|
|
}
|
|
|
|
// handleList scans the data dir for order logs and summarises each (replaying it
|
|
// through the pool). Fine for a single-node simple checkout; a production list
|
|
// would use an index/projection store.
|
|
func (s *server) handleList(w http.ResponseWriter, r *http.Request) {
|
|
matches, _ := filepath.Glob(filepath.Join(s.dataDir, "*.events.log"))
|
|
out := make([]orderSummary, 0, len(matches))
|
|
for _, m := range matches {
|
|
base := strings.TrimSuffix(filepath.Base(m), ".events.log")
|
|
raw, err := strconv.ParseUint(base, 10, 64)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
g, err := s.pool.Get(r.Context(), raw)
|
|
if err != nil || g.Status == order.StatusNew {
|
|
continue
|
|
}
|
|
out = append(out, orderSummary{
|
|
OrderId: order.OrderId(raw).String(),
|
|
Reference: g.OrderReference,
|
|
Status: g.Status,
|
|
TotalAmount: g.TotalAmount,
|
|
CapturedAmount: g.CapturedAmount,
|
|
Currency: g.Currency,
|
|
PlacedAt: g.PlacedAt,
|
|
})
|
|
}
|
|
writeJSON(w, http.StatusOK, out)
|
|
}
|
|
|
|
// --- helpers --------------------------------------------------------------
|
|
|
|
// selectProvider picks the payment provider from the environment. Default is
|
|
// the dependency-free mock; set PAYMENT_PROVIDER=stripe + STRIPE_SECRET_KEY for
|
|
// real Stripe (STRIPE_API_BASE / STRIPE_PAYMENT_METHOD optionally override).
|
|
func selectProvider(logger *slog.Logger) order.PaymentProvider {
|
|
if os.Getenv("PAYMENT_PROVIDER") == "stripe" {
|
|
key := os.Getenv("STRIPE_SECRET_KEY")
|
|
if key == "" {
|
|
logger.Warn("PAYMENT_PROVIDER=stripe but STRIPE_SECRET_KEY is empty; falling back to mock")
|
|
return order.NewMockProvider()
|
|
}
|
|
logger.Info("using stripe payment provider")
|
|
return order.NewStripeProvider(key, os.Getenv("STRIPE_API_BASE"), os.Getenv("STRIPE_PAYMENT_METHOD"))
|
|
}
|
|
return order.NewMockProvider()
|
|
}
|
|
|
|
func envOr(key, def string) string {
|
|
if v := os.Getenv(key); v != "" {
|
|
return v
|
|
}
|
|
return def
|
|
}
|
|
|
|
func writeJSON(w http.ResponseWriter, status int, v any) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(status)
|
|
_ = json.NewEncoder(w).Encode(v)
|
|
}
|
|
|
|
func writeErr(w http.ResponseWriter, status int, err error) {
|
|
writeJSON(w, status, map[string]string{"error": err.Error()})
|
|
}
|