mats/go-cart-actor
1
0
Fork 0
Code Issues Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

uniform
Build and Publish / BuildAndDeployArm64 (push) Failing after 7s
Details
Build and Publish / BuildAndDeployAmd64 (push) Failing after 7s
Details

Browse Source
This commit is contained in:
mats
2026-07-20 00:13:24 +02:00
parent 83ed477aa4
commit e6f69f500a
14 changed files with 457 additions and 471 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/cart/main.go
+2 -100
View File
@@ -84,15 +84,7 @@ func loadUCPCartSigner() *ucp.SigningConfig {
return cfg
}
func getCountryFromHost(host string) string {
if strings.Contains(strings.ToLower(host), "-no") {
return "no"
}
if strings.Contains(strings.ToLower(host), "-se") {
return "se"
}
return "se"
}
type MutationContext struct {
VoucherService voucher.Service
@@ -103,62 +95,6 @@ type CartChangeEvent struct {
Mutations []actor.ApplyResult `json:"mutations"`
}
// catalogProjectionCache is the cart's per-pod, cold-start-ready catalog
// projection cache, backed by platform/catalog.ProjectionStore. It consumes the
// catalog.projection_published wire — a full snapshot (begin/chunk/end, framed by
// epoch) plus incremental deltas — into its OWN local .bin and serves catalog
// facts by SKU (price, tax_class, inventory_tracked, image, …) at add-to-cart /
// checkout-reserve decisions. Stock is NOT cached — queried synchronously from
// inventory per docs/inventory-shape.md.
//
// Clustering: the cart runs N replicas. Each pod has its OWN exclusive bus queue
// (rabbit.ListenToPattern declares an exclusive server-named queue, so every pod
// receives the full stream) and its OWN pod-local .bin. It's a replicated read
// cache — no leader, no shared volume; N pods writing one file would corrupt it.
// See docs/catalog-projection.md.
//
// The cart stores the full catalog.Projection (identity transform): its
// add-to-cart overlay (projection_overlay.go) consumes nearly every field, so
// there's nothing to subset. Deletes ride as tombstones in the store's overlay
// (IsDeleted) so a removed SKU isn't re-fetched before the next snapshot folds
// the delete into the base. Cold-grace: an SKU not yet in the base/overlay
// resolves via the existing PRODUCT_BASE_URL HTTP fetch + overlay.
type catalogProjectionCache struct {
store *catalog.ProjectionStore[catalog.Projection]
}
// identityProjection is the cart's mandatory transform — it stores the full
// Projection because the overlay uses almost all of it.
func identityProjection(p catalog.Projection) catalog.Projection { return p }
// newCatalogProjectionCache opens the per-pod store under dir. dir MUST be
// pod-local (container fs / an emptyDir in k8s) — never a shared volume.
func newCatalogProjectionCache(dir string) (*catalogProjectionCache, error) {
s, err := catalog.NewProjectionStore(dir, "cart-projection.bin", identityProjection, nil)
if err != nil {
return nil, err
}
return &catalogProjectionCache{store: s}, nil
}
// HandleFrame feeds one catalog.projection_published event (snapshot frame or
// delta) into the store; framing/epoch ordering live in platform/catalog.
func (c *catalogProjectionCache) HandleFrame(meta map[string]string, payload []byte) error {
return c.store.HandleFrame(meta, payload)
}
func (c *catalogProjectionCache) Get(sku string) (catalog.Projection, bool) { return c.store.Get(sku) }
// IsDeleted reports whether the bus has explicitly deleted sku (overlay
// tombstone) — call sites skip the HTTP fallback for it.
func (c *catalogProjectionCache) IsDeleted(sku string) bool { return c.store.IsDeleted(sku) }
// Len reports cached entries (base + overlay) — debug/log gauge only.
func (c *catalogProjectionCache) Len() int {
_, base, overlay := c.store.Stats()
return base + overlay
}
func main() {
// cartPort is the bare HTTP port. It drives both the local listener and the
@@ -184,8 +120,6 @@ func main() {
promotionService := promotions.NewPromotionService(nil)
//inventoryPubSub := actor.NewPubSub[inventory.InventoryChange]()
rdb := redis.NewClient(&redis.Options{
Addr: redisAddress,
Password: redisPassword,
@@ -228,44 +162,12 @@ func main() {
_, span := tracer.Start(ctx, fmt.Sprintf("Spawn cart id %d", id))
defer span.End()
ret := cart.NewCartGrain(id, time.Now())
// Set baseline lastChange at spawn; replay may update it to last event timestamp.
// inventoryPubSub.Subscribe(ret.HandleInventoryChange)
err := diskStorage.LoadEvents(ctx, id, ret)
// if err == nil && inventoryService != nil {
// refs := make([]*inventory.InventoryReference, 0)
// for _, item := range ret.Items {
// refs = append(refs, &inventory.InventoryReference{
// SKU: inventory.SKU(item.Sku),
// LocationID: getLocationId(item),
// })
// }
// _, span := tracer.Start(ctx, "update inventory")
// defer span.End()
// res, err := inventoryService.GetInventoryBatch(ctx, refs...)
// if err != nil {
// log.Printf("unable to update inventory %v", err)
// } else {
// for _, update := range res {
// for _, item := range ret.Items {
// if matchesSkuAndLocation(update, *item) && update.Quantity != uint32(item.Stock) {
// // maybe apply an update to give visibility to the cart
// item.Stock = uint16(update.Quantity)
// }
// }
// }
// }
// }
return ret, err
},
Destroy: func(grain actor.Grain[cart.CartGrain]) error {
// cart, err := grain.GetCurrentState()
// if err != nil {
// return err
// }
//inventoryPubSub.Unsubscribe(cart.HandleInventoryChange)
return nil
},
SpawnHost: func(host string) (actor.Host[cart.CartGrain], error) {
@@ -411,7 +313,7 @@ func main() {
log.Print("cart: abandonment recovery DISABLED (CART_ABANDONED_DELAY not set or <= 0)")
}
otelShutdown, err := telemetry.SetupOTelSDK(ctx)
otelShutdown, err := telemetry.SetupOTelSDK(ctx, telemetry.LoggerConfigFromEnv())
if err != nil {
log.Fatalf("Unable to start otel %v", err)
}
cmd/cart/pool-server.go
+2 -45
View File
@@ -16,6 +16,7 @@ import (
messages "git.k6n.net/mats/go-cart-actor/proto/cart"
"git.k6n.net/mats/go-cart-actor/pkg/voucher"
"git.k6n.net/mats/platform/conv"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/anypb"
@@ -70,7 +71,7 @@ func (s *PoolServer) GetCartHandler(w http.ResponseWriter, r *http.Request, id c
func (s *PoolServer) AddSkuToCartHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
sku := r.PathValue("sku")
country := getCountryFromHost(r.Host)
country := conv.CountryFromHost(r.Host)
if s.idx == nil {
log.Printf("No index present")
}
@@ -440,40 +441,6 @@ func (s *PoolServer) AddSkuRequestHandler(w http.ResponseWriter, r *http.Request
return s.WriteResult(w, reply)
}
// func (s *PoolServer) HandleConfirmation(w http.ResponseWriter, r *http.Request, id CartId) error {
// orderId := r.PathValue("orderId")
// if orderId == "" {
// return fmt.Errorf("orderId is empty")
// }
// order, err := KlarnaInstance.GetOrder(orderId)
// if err != nil {
// return err
// }
// w.Header().Set("Content-Type", "application/json")
// w.Header().Set("X-Pod-Name", s.pod_name)
// w.Header().Set("Cache-Control", "no-cache")
// w.Header().Set("Access-Control-Allow-Origin", "*")
// w.WriteHeader(http.StatusOK)
// return json.NewEncoder(w).Encode(order)
// }
// func (s *PoolServer) HandleCheckout(w http.ResponseWriter, r *http.Request, id CartId) error {
// klarnaOrder, err := s.CreateOrUpdateCheckout(r.Host, id)
// if err != nil {
// return err
// }
// s.ApplyCheckoutStarted(klarnaOrder, id)
// w.Header().Set("Content-Type", "application/json")
// return json.NewEncoder(w).Encode(klarnaOrder)
// }
//
// Removed leftover legacy block after CookieCartIdHandler (obsolete code referencing cid/legacy)
func (s *PoolServer) RemoveCartCookie(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error {
// Clear cart cookie (breaking change: do not issue a new legacy id here)
http.SetCookie(w, &http.Cookie{
@@ -758,13 +725,6 @@ func (s *PoolServer) ApplyAnywhere(ctx context.Context, cartId cart.CartId, msgs
func (s *PoolServer) Serve(mux *http.ServeMux) {
// mux.HandleFunc("OPTIONS /cart", func(w http.ResponseWriter, r *http.Request) {
// w.Header().Set("Access-Control-Allow-Origin", "*")
// w.Header().Set("Access-Control-Allow-Methods", "GET, PUT, POST, DELETE")
// w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
// w.WriteHeader(http.StatusOK)
// })
handleFunc := func(pattern string, handlerFunc func(http.ResponseWriter, *http.Request)) {
attr := attribute.String("http.route", pattern)
mux.HandleFunc(pattern, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -800,9 +760,6 @@ func (s *PoolServer) Serve(mux *http.ServeMux) {
handleFunc("DELETE /cart/item/{itemId}/marking", CookieCartIdHandler(s.ProxyHandler(s.RemoveLineItemMarkingHandler)))
handleFunc("PUT /cart/item/{itemId}/custom-fields", CookieCartIdHandler(s.ProxyHandler(s.SetCustomFieldsHandler)))
//mux.HandleFunc("GET /cart/checkout", CookieCartIdHandler(s.ProxyHandler(s.HandleCheckout)))
//mux.HandleFunc("GET /cart/confirmation/{orderId}", CookieCartIdHandler(s.ProxyHandler(s.HandleConfirmation)))
handleFunc("GET /cart/byid/{id}", CartIdHandler(s.ProxyHandler(s.GetCartHandler)))
handleFunc("POST /cart/byid/{id}", CartIdHandler(s.ProxyHandler(s.AddSkuRequestHandler)))
cmd/cart/projection.go
+61
View File
@@ -0,0 +1,61 @@
package main
import (
"git.k6n.net/mats/platform/catalog"
)
// catalogProjectionCache is the cart's per-pod, cold-start-ready catalog
// projection cache, backed by platform/catalog.ProjectionStore. It consumes the
// catalog.projection_published wire — a full snapshot (begin/chunk/end, framed by
// epoch) plus incremental deltas — into its OWN local .bin and serves catalog
// facts by SKU (price, tax_class, inventory_tracked, image, …) at add-to-cart /
// checkout-reserve decisions. Stock is NOT cached — queried synchronously from
// inventory per docs/inventory-shape.md.
//
// Clustering: the cart runs N replicas. Each pod has its OWN exclusive bus queue
// (rabbit.ListenToPattern declares an exclusive server-named queue, so every pod
// receives the full stream) and its OWN pod-local .bin. It's a replicated read
// cache — no leader, no shared volume; N pods writing one file would corrupt it.
// See docs/catalog-projection.md.
//
// The cart stores the full catalog.Projection (identity transform): its
// add-to-cart overlay (projection_overlay.go) consumes nearly every field, so
// there's nothing to subset. Deletes ride as tombstones in the store's overlay
// (IsDeleted) so a removed SKU isn't re-fetched before the next snapshot folds
// the delete into the base. Cold-grace: an SKU not yet in the base/overlay
// resolves via the existing PRODUCT_BASE_URL HTTP fetch + overlay.
type catalogProjectionCache struct {
store *catalog.ProjectionStore[catalog.Projection]
}
// identityProjection is the cart's mandatory transform — it stores the full
// Projection because the overlay uses almost all of it.
func identityProjection(p catalog.Projection) catalog.Projection { return p }
// newCatalogProjectionCache opens the per-pod store under dir. dir MUST be
// pod-local (container fs / an emptyDir in k8s) — never a shared volume.
func newCatalogProjectionCache(dir string) (*catalogProjectionCache, error) {
s, err := catalog.NewProjectionStore(dir, "cart-projection.bin", identityProjection, nil)
if err != nil {
return nil, err
}
return &catalogProjectionCache{store: s}, nil
}
// HandleFrame feeds one catalog.projection_published event (snapshot frame or
// delta) into the store; framing/epoch ordering live in platform/catalog.
func (c *catalogProjectionCache) HandleFrame(meta map[string]string, payload []byte) error {
return c.store.HandleFrame(meta, payload)
}
func (c *catalogProjectionCache) Get(sku string) (catalog.Projection, bool) { return c.store.Get(sku) }
// IsDeleted reports whether the bus has explicitly deleted sku (overlay
// tombstone) — call sites skip the HTTP fallback for it.
func (c *catalogProjectionCache) IsDeleted(sku string) bool { return c.store.IsDeleted(sku) }
// Len reports cached entries (base + overlay) — debug/log gauge only.
func (c *catalogProjectionCache) Len() int {
_, base, overlay := c.store.Stats()
return base + overlay
}
cmd/checkout/checkout_builder.go
+2 -1
View File
@@ -9,6 +9,7 @@ import (
"git.k6n.net/mats/go-cart-actor/pkg/cart"
"git.k6n.net/mats/go-cart-actor/pkg/checkout"
"git.k6n.net/mats/platform/conv"
"git.k6n.net/mats/platform/httpclient"
"git.k6n.net/mats/platform/tax"
adyenCheckout "github.com/adyen/adyen-go-api-library/v21/src/checkout"
@@ -228,7 +229,7 @@ func defaultAdyenTaxRate(tp tax.Provider, country string) int64 {
func GetCheckoutMetaFromRequest(r *http.Request) *CheckoutMeta {
host := getOriginalHost(r)
country := getCountryFromHost(host)
country := conv.CountryFromHost(host)
siteUrl := fmt.Sprintf("%s://%s", getScheme(r), host)
if checkoutPublicURL != "" {
siteUrl = checkoutPublicURL
cmd/checkout/klarna-client.go
+9 -5
View File
@@ -1,6 +1,7 @@
package main
import (
"bytes"
"context"
"encoding/json"
"fmt"
@@ -60,12 +61,15 @@ func (k *KlarnaClient) getOrderResponse(res *http.Response) (*CheckoutOrder, err
}
return &klarnaOrderResponse, nil
}
body, err := io.ReadAll(res.Body)
if err == nil {
log.Println(string(body))
body, readErr := io.ReadAll(res.Body)
if len(body) > 0 {
log.Printf("klarna error body: %s", string(body))
return nil, fmt.Errorf("klarna: %s: %s", res.Status, string(bytes.TrimSpace(body)))
}
return nil, fmt.Errorf("%s", res.Status)
if readErr != nil {
return nil, fmt.Errorf("klarna: %s (body unreadable: %v)", res.Status, readErr)
}
return nil, fmt.Errorf("klarna: %s", res.Status)
}
func (k *KlarnaClient) CreateOrder(ctx context.Context, reader io.Reader) (*CheckoutOrder, error) {
cmd/checkout/main.go
+1 -1
View File
@@ -237,7 +237,7 @@ func main() {
log.Print("checkout: data retention / GDPR purge disabled (CHECKOUT_RETENTION_DAYS not set or <= 0)")
}
otelShutdown, err := telemetry.SetupOTelSDK(ctx)
otelShutdown, err := telemetry.SetupOTelSDK(ctx, telemetry.LoggerConfigFromEnv())
if err != nil {
log.Fatalf("Unable to start otel %v", err)
}
cmd/checkout/utils.go
+1 -9
View File
@@ -60,15 +60,7 @@ func getCountryFromCurrency(currency string) string {
return "se"
}
func getCountryFromHost(host string) string {
if strings.Contains(strings.ToLower(host), "-no") {
return "no"
}
if strings.Contains(strings.ToLower(host), "-se") {
return "se"
}
return ""
}
func (a *CheckoutPoolServer) reserveInventory(ctx context.Context, grain *checkout.CheckoutGrain) error {
if a.reservationPolicy == nil {
cmd/order/handlers.go
+6
View File
@@ -109,6 +109,12 @@ func (s *server) applyAndRespond(w http.ResponseWriter, r *http.Request, id orde
s.logger.Error("idempotency record failed", "key", idemKey, "orderId", id.String(), "err", err)
}
}
// Invalidate the stats cache asynchronously so the dashboard picks up
// the new state without blocking the HTTP response. Nil-safe: tests and
// minimal configs can omit the cache.
if s.statsCache != nil {
go s.statsCache.Rebuild()
}
grain, err := s.applier.Get(r.Context(), uint64(id))
if err != nil {
writeErr(w, http.StatusInternalServerError, err)
cmd/order/main.go
+6 -266
View File
@@ -12,12 +12,10 @@ import (
"log"
"log/slog"
"net/http"
"net/mail"
"os"
"path/filepath"
"strconv"
"strings"
"sort"
"sync"
"time"
@@ -56,6 +54,7 @@ type server struct {
idem *idempotency.Store
logger *slog.Logger
inventoryReservations order.InventoryReservationService
statsCache *statsCache
}
func main() {
@@ -229,8 +228,13 @@ func main() {
taxProvider: taxProvider,
defaultFlow: def.Name, dataDir: dataDir, idem: idem, logger: logger,
inventoryReservations: inventoryReservations,
statsCache: newStatsCache(dataDir, pool),
}
// Warm the stats cache before accepting requests so the first dashboard
// hit is instant. Rebuilds asynchronously after every mutation thereafter.
s.statsCache.Rebuild()
mux := http.NewServeMux()
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })
@@ -498,160 +502,6 @@ func (s *server) buildPlaceOrder(id order.OrderId, req *checkoutReq) *messages.P
return po
}
// --- dashboard stats ------------------------------------------------------
type orderAlert struct {
Severity string `json:"severity"` // warn | error
Message string `json:"message"`
OrderId string `json:"orderId,omitempty"`
}
type statusBucket struct {
Status order.Status `json:"status"`
Count int `json:"count"`
Total int64 `json:"total"`
}
type revenueSnapshot struct {
Total int64 `json:"total"`
Captured int64 `json:"captured"`
Refunded int64 `json:"refunded"`
Pending int64 `json:"pending"`
}
type dailyRevenue struct {
Date string `json:"date"` // YYYY-MM-DD
Total int64 `json:"total"`
}
type orderStatsResponse struct {
OrdersTotal int `json:"ordersTotal"`
Revenue revenueSnapshot `json:"revenue"`
ByStatus []statusBucket `json:"byStatus"`
RecentOrders []orderSummary `json:"recentOrders"`
DailyRevenue30 []dailyRevenue `json:"dailyRevenue30,omitempty"`
Alerts []orderAlert `json:"alerts"`
}
// handleStats scans all order logs and returns aggregated dashboard stats.
func (s *server) handleStats(w http.ResponseWriter, r *http.Request) {
matches, _ := filepath.Glob(filepath.Join(s.dataDir, "*.events.log"))
now := time.Now()
var totalOrders int
var totalRevenue, capturedRevenue, refundedRevenue int64
byStatus := map[order.Status]int{}
statusTotal := map[order.Status]int64{} // total amount per status
// Daily revenue for the last 30 days (keys are "2026-06-01" sortable strings).
dailyByDay := map[string]int64{}
for i := 0; i < 30; i++ {
day := now.AddDate(0, 0, -i).Format("2006-01-02")
dailyByDay[day] = 0
}
// Collect all order summaries, then sort by placedAt and take top 10.
var allOrders []orderSummary
var alerts []orderAlert
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
}
totalOrders++
byStatus[g.Status]++
statusTotal[g.Status] += g.TotalAmount.Int64()
totalRevenue += g.TotalAmount.Int64()
capturedRevenue += g.CapturedAmount.Int64()
refundedRevenue += g.RefundedAmount.Int64()
// Daily revenue — parse placedAt to extract date.
if g.PlacedAt != "" {
if placed, parseErr := time.Parse(time.RFC3339, g.PlacedAt); parseErr == nil {
dayKey := placed.Format("2006-01-02")
if _, ok := dailyByDay[dayKey]; ok {
dailyByDay[dayKey] += g.TotalAmount.Int64()
}
}
}
allOrders = append(allOrders, orderSummary{
OrderId: order.OrderId(raw).String(),
Reference: g.OrderReference,
Status: g.Status,
TotalAmount: g.TotalAmount.Int64(),
Currency: g.Currency,
PlacedAt: g.PlacedAt,
})
// Alerts.
if g.Status == order.StatusPending {
alerts = append(alerts, orderAlert{
Severity: "warn",
Message: "Payment still pending",
OrderId: order.OrderId(raw).String(),
})
}
if g.Status == order.StatusCancelled {
alerts = append(alerts, orderAlert{
Severity: "warn",
Message: "Order was cancelled",
OrderId: order.OrderId(raw).String(),
})
}
}
// Sort ALL orders by placedAt descending, take top 10 as "recent".
sort.Slice(allOrders, func(i, j int) bool {
return allOrders[i].PlacedAt > allOrders[j].PlacedAt
})
recent := allOrders
if len(recent) > 10 {
recent = recent[:10]
}
// Cap alerts
if len(alerts) > 50 {
alerts = alerts[:50]
}
// Build status buckets array with totals.
buckets := make([]statusBucket, 0, len(byStatus))
for st, cnt := range byStatus {
buckets = append(buckets, statusBucket{Status: st, Count: cnt, Total: statusTotal[st]})
}
sort.Slice(buckets, func(i, j int) bool {
return buckets[i].Count > buckets[j].Count
})
// Daily revenue as an ordered slice.
var dailyRev []dailyRevenue
for i := 29; i >= 0; i-- {
day := now.AddDate(0, 0, -i).Format("2006-01-02")
dailyRev = append(dailyRev, dailyRevenue{Date: day, Total: dailyByDay[day]})
}
writeJSON(w, http.StatusOK, orderStatsResponse{
OrdersTotal: totalOrders,
Revenue: revenueSnapshot{
Total: totalRevenue,
Captured: capturedRevenue,
Refunded: refundedRevenue,
Pending: totalRevenue - capturedRevenue - refundedRevenue,
},
ByStatus: buckets,
RecentOrders: recent,
DailyRevenue30: dailyRev,
Alerts: alerts,
})
}
// --- reads ----------------------------------------------------------------
func (s *server) handleGet(w http.ResponseWriter, r *http.Request) {
@@ -743,116 +593,6 @@ func (s *server) handleList(w http.ResponseWriter, r *http.Request) {
// --- helpers --------------------------------------------------------------
// selectTaxProvider picks the tax provider from the environment. Default is
// NordicTaxProvider with SE as the default country (matching the current
// Nordic-only footprint). Set TAX_PROVIDER=static for dev and tests
// (no country awareness).
func selectTaxProvider() tax.Provider {
provider := os.Getenv("TAX_PROVIDER")
switch provider {
case "static":
return tax.NewStatic()
case "nordic":
return tax.NewNordic(config.EnvString("TAX_DEFAULT_COUNTRY", "SE"))
default:
return tax.NewNordic(config.EnvString("TAX_DEFAULT_COUNTRY", "SE"))
}
}
// 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()
}
// selectEmailSender picks the flow-email transport. Supports SMTP and
// MailerSend. When no sender is configured the hook still exists in
// capabilities but errors at run time if a flow tries to use it.
func selectEmailSender(logger *slog.Logger) (order.EmailSender, error) {
kind := config.EnvString("ORDER_EMAIL_SENDER", "")
if kind == "" && os.Getenv("ORDER_EMAIL_SMTP_HOST") == "" {
return nil, nil
}
if kind == "" {
kind = "smtp"
}
fromAddr := strings.TrimSpace(os.Getenv("ORDER_EMAIL_FROM_ADDRESS"))
if fromAddr == "" {
return nil, fmt.Errorf("ORDER_EMAIL_FROM_ADDRESS is required when email sender is configured")
}
from := mail.Address{
Name: strings.TrimSpace(os.Getenv("ORDER_EMAIL_FROM_NAME")),
Address: fromAddr,
}
switch kind {
case "smtp":
sender, err := order.NewSMTPEmailSender(order.SMTPEmailSenderConfig{
Host: config.EnvString("ORDER_EMAIL_SMTP_HOST", ""),
Port: config.EnvInt("ORDER_EMAIL_SMTP_PORT", 587, func(n int) bool { return n > 0 }),
Username: os.Getenv("ORDER_EMAIL_SMTP_USERNAME"),
Password: os.Getenv("ORDER_EMAIL_SMTP_PASSWORD"),
DefaultFrom: from,
Timeout: config.EnvDuration("ORDER_EMAIL_SMTP_TIMEOUT", 10*time.Second),
})
if err != nil {
return nil, err
}
logger.Info("email sender enabled", "kind", "smtp", "host", config.EnvString("ORDER_EMAIL_SMTP_HOST", ""), "from", from.Address)
return sender, nil
case "mailersend":
apiKey := os.Getenv("MAILERSEND_API_KEY")
if apiKey == "" {
return nil, fmt.Errorf("MAILERSEND_API_KEY is required for mailersend sender")
}
sender, err := order.NewMailerSendEmailSender(apiKey, from)
if err != nil {
return nil, err
}
logger.Info("email sender enabled", "kind", "mailersend", "from", from.Address)
return sender, nil
default:
return nil, fmt.Errorf("unsupported ORDER_EMAIL_SENDER %q (supported: smtp, mailersend)", kind)
}
}
func selectInventoryReservationService() (order.InventoryReservationService, error) {
return order.NewHTTPInventoryReservationService(config.EnvString("INVENTORY_URL", "http://localhost:8080"), nil)
}
// loadUCPOrderSigner loads the UCP ECDSA signing key from the path specified in
// the UCP_SIGNING_KEY_PATH environment variable. Returns nil if unset or unreadable.
func loadUCPOrderSigner() *ucp.SigningConfig {
path := os.Getenv("UCP_SIGNING_KEY_PATH")
if path == "" {
return nil
}
pemData, err := os.ReadFile(path)
if err != nil {
log.Printf("ucp signing: cannot read key %s: %v", path, err)
return nil
}
cfg, err := ucp.NewSigningConfigFromPEM(pemData, "k6n-ecdsa-2026")
if err != nil {
log.Printf("ucp signing: invalid key at %s: %v", path, err)
return nil
}
return cfg
}
func writeJSON(w http.ResponseWriter, status int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
cmd/order/provider.go
+126
View File
@@ -0,0 +1,126 @@
package main
import (
"fmt"
"log"
"log/slog"
"net/mail"
"os"
"strings"
"time"
"git.k6n.net/mats/go-cart-actor/internal/ucp"
"git.k6n.net/mats/go-cart-actor/pkg/order"
"git.k6n.net/mats/platform/config"
"git.k6n.net/mats/platform/tax"
)
// selectTaxProvider picks the tax provider from the environment. Default is
// NordicTaxProvider with SE as the default country (matching the current
// Nordic-only footprint). Set TAX_PROVIDER=static for dev and tests
// (no country awareness).
func selectTaxProvider() tax.Provider {
provider := os.Getenv("TAX_PROVIDER")
switch provider {
case "static":
return tax.NewStatic()
case "nordic":
return tax.NewNordic(config.EnvString("TAX_DEFAULT_COUNTRY", "SE"))
default:
return tax.NewNordic(config.EnvString("TAX_DEFAULT_COUNTRY", "SE"))
}
}
// 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()
}
// selectEmailSender picks the flow-email transport. Supports SMTP and
// MailerSend. When no sender is configured the hook still exists in
// capabilities but errors at run time if a flow tries to use it.
func selectEmailSender(logger *slog.Logger) (order.EmailSender, error) {
kind := config.EnvString("ORDER_EMAIL_SENDER", "")
if kind == "" && os.Getenv("ORDER_EMAIL_SMTP_HOST") == "" {
return nil, nil
}
if kind == "" {
kind = "smtp"
}
fromAddr := strings.TrimSpace(os.Getenv("ORDER_EMAIL_FROM_ADDRESS"))
if fromAddr == "" {
return nil, fmt.Errorf("ORDER_EMAIL_FROM_ADDRESS is required when email sender is configured")
}
from := mail.Address{
Name: strings.TrimSpace(os.Getenv("ORDER_EMAIL_FROM_NAME")),
Address: fromAddr,
}
switch kind {
case "smtp":
sender, err := order.NewSMTPEmailSender(order.SMTPEmailSenderConfig{
Host: config.EnvString("ORDER_EMAIL_SMTP_HOST", ""),
Port: config.EnvInt("ORDER_EMAIL_SMTP_PORT", 587, func(n int) bool { return n > 0 }),
Username: os.Getenv("ORDER_EMAIL_SMTP_USERNAME"),
Password: os.Getenv("ORDER_EMAIL_SMTP_PASSWORD"),
DefaultFrom: from,
Timeout: config.EnvDuration("ORDER_EMAIL_SMTP_TIMEOUT", 10*time.Second),
})
if err != nil {
return nil, err
}
logger.Info("email sender enabled", "kind", "smtp", "host", config.EnvString("ORDER_EMAIL_SMTP_HOST", ""), "from", from.Address)
return sender, nil
case "mailersend":
apiKey := os.Getenv("MAILERSEND_API_KEY")
if apiKey == "" {
return nil, fmt.Errorf("MAILERSEND_API_KEY is required for mailersend sender")
}
sender, err := order.NewMailerSendEmailSender(apiKey, from)
if err != nil {
return nil, err
}
logger.Info("email sender enabled", "kind", "mailersend", "from", from.Address)
return sender, nil
default:
return nil, fmt.Errorf("unsupported ORDER_EMAIL_SENDER %q (supported: smtp, mailersend)", kind)
}
}
func selectInventoryReservationService() (order.InventoryReservationService, error) {
return order.NewHTTPInventoryReservationService(config.EnvString("INVENTORY_URL", "http://localhost:8080"), nil)
}
// loadUCPOrderSigner loads the UCP ECDSA signing key from the path specified in
// the UCP_SIGNING_KEY_PATH environment variable. Returns nil if unset or unreadable.
func loadUCPOrderSigner() *ucp.SigningConfig {
path := os.Getenv("UCP_SIGNING_KEY_PATH")
if path == "" {
return nil
}
pemData, err := os.ReadFile(path)
if err != nil {
log.Printf("ucp signing: cannot read key %s: %v", path, err)
return nil
}
cfg, err := ucp.NewSigningConfigFromPEM(pemData, "k6n-ecdsa-2026")
if err != nil {
log.Printf("ucp signing: invalid key at %s: %v", path, err)
return nil
}
return cfg
}
cmd/order/stats.go
+201
View File
@@ -0,0 +1,201 @@
package main
import (
"context"
"net/http"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"time"
"git.k6n.net/mats/go-cart-actor/pkg/actor"
"git.k6n.net/mats/go-cart-actor/pkg/order"
)
type orderAlert struct {
Severity string `json:"severity"` // warn | error
Message string `json:"message"`
OrderId string `json:"orderId,omitempty"`
}
type statusBucket struct {
Status order.Status `json:"status"`
Count int `json:"count"`
Total int64 `json:"total"`
}
type revenueSnapshot struct {
Total int64 `json:"total"`
Captured int64 `json:"captured"`
Refunded int64 `json:"refunded"`
Pending int64 `json:"pending"`
}
type dailyRevenue struct {
Date string `json:"date"` // YYYY-MM-DD
Total int64 `json:"total"`
}
type orderStatsResponse struct {
OrdersTotal int `json:"ordersTotal"`
Revenue revenueSnapshot `json:"revenue"`
ByStatus []statusBucket `json:"byStatus"`
RecentOrders []orderSummary `json:"recentOrders"`
DailyRevenue30 []dailyRevenue `json:"dailyRevenue30,omitempty"`
Alerts []orderAlert `json:"alerts"`
}
// statsCache holds an in-memory aggregate of all orders, rebuilt on mutation
// instead of scanning *.events.log on every request.
type statsCache struct {
mu sync.RWMutex
data orderStatsResponse
pool *actor.SimpleGrainPool[order.OrderGrain]
dataDir string
}
func newStatsCache(dataDir string, pool *actor.SimpleGrainPool[order.OrderGrain]) *statsCache {
return &statsCache{dataDir: dataDir, pool: pool}
}
// Get returns the cached stats. Call Rebuild() on startup to warm the cache;
// if never built, returns an empty response.
func (c *statsCache) Get() orderStatsResponse {
c.mu.RLock()
defer c.mu.RUnlock()
return c.data
}
// Rebuild rebuilds the stats cache from disk. Safe to call concurrently;
// intended as a goroutine after each mutation so it never blocks an HTTP
// response.
func (c *statsCache) Rebuild() {
c.rebuild()
}
func (c *statsCache) rebuild() {
if c.pool == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
matches, _ := filepath.Glob(filepath.Join(c.dataDir, "*.events.log"))
now := time.Now()
var totalOrders int
var totalRevenue, capturedRevenue, refundedRevenue int64
byStatus := map[order.Status]int{}
statusTotal := map[order.Status]int64{}
dailyByDay := map[string]int64{}
for i := 0; i < 30; i++ {
day := now.AddDate(0, 0, -i).Format("2006-01-02")
dailyByDay[day] = 0
}
var allOrders []orderSummary
var alerts []orderAlert
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 := c.pool.Get(ctx, raw)
if err != nil || g.Status == order.StatusNew {
continue
}
totalOrders++
byStatus[g.Status]++
statusTotal[g.Status] += g.TotalAmount.Int64()
totalRevenue += g.TotalAmount.Int64()
capturedRevenue += g.CapturedAmount.Int64()
refundedRevenue += g.RefundedAmount.Int64()
if g.PlacedAt != "" {
if placed, parseErr := time.Parse(time.RFC3339, g.PlacedAt); parseErr == nil {
dayKey := placed.Format("2006-01-02")
if _, ok := dailyByDay[dayKey]; ok {
dailyByDay[dayKey] += g.TotalAmount.Int64()
}
}
}
allOrders = append(allOrders, orderSummary{
OrderId: order.OrderId(raw).String(),
Reference: g.OrderReference,
Status: g.Status,
TotalAmount: g.TotalAmount.Int64(),
Currency: g.Currency,
PlacedAt: g.PlacedAt,
})
if g.Status == order.StatusPending {
alerts = append(alerts, orderAlert{
Severity: "warn",
Message: "Payment still pending",
OrderId: order.OrderId(raw).String(),
})
}
if g.Status == order.StatusCancelled {
alerts = append(alerts, orderAlert{
Severity: "warn",
Message: "Order was cancelled",
OrderId: order.OrderId(raw).String(),
})
}
}
sort.Slice(allOrders, func(i, j int) bool {
return allOrders[i].PlacedAt > allOrders[j].PlacedAt
})
recent := allOrders
if len(recent) > 10 {
recent = recent[:10]
}
if len(alerts) > 50 {
alerts = alerts[:50]
}
buckets := make([]statusBucket, 0, len(byStatus))
for st, cnt := range byStatus {
buckets = append(buckets, statusBucket{Status: st, Count: cnt, Total: statusTotal[st]})
}
sort.Slice(buckets, func(i, j int) bool {
return buckets[i].Count > buckets[j].Count
})
var dailyRev []dailyRevenue
for i := 29; i >= 0; i-- {
day := now.AddDate(0, 0, -i).Format("2006-01-02")
dailyRev = append(dailyRev, dailyRevenue{Date: day, Total: dailyByDay[day]})
}
c.mu.Lock()
c.data = orderStatsResponse{
OrdersTotal: totalOrders,
Revenue: revenueSnapshot{
Total: totalRevenue,
Captured: capturedRevenue,
Refunded: refundedRevenue,
Pending: totalRevenue - capturedRevenue - refundedRevenue,
},
ByStatus: buckets,
RecentOrders: recent,
DailyRevenue30: dailyRev,
Alerts: alerts,
}
c.mu.Unlock()
}
// handleStats returns dashboard stats from the in-memory cache (O(1) read,
// rebuilt after each mutation) instead of scanning all *.events.log on every
// request.
func (s *server) handleStats(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.statsCache.Get())
}
cmd/profile/main.go
+1 -1
View File
@@ -220,7 +220,7 @@ func main() {
log.Print("profile: data retention / GDPR purge disabled (PROFILE_RETENTION_DAYS not set or <= 0)")
}
otelShutdown, err := telemetry.SetupOTelSDK(ctx)
otelShutdown, err := telemetry.SetupOTelSDK(ctx, telemetry.LoggerConfigFromEnv())
if err != nil {
log.Fatalf("Unable to start otel %v", err)
}