From e6f69f500aed2ac8a70e76ee47c8f6b160bb90e7 Mon Sep 17 00:00:00 2001 From: matst80 Date: Mon, 20 Jul 2026 00:13:24 +0200 Subject: [PATCH] uniform --- cmd/cart/main.go | 102 +----------- cmd/cart/pool-server.go | 47 +----- cmd/cart/projection.go | 61 +++++++ cmd/checkout/checkout_builder.go | 3 +- cmd/checkout/klarna-client.go | 14 +- cmd/checkout/main.go | 2 +- cmd/checkout/utils.go | 10 +- cmd/order/handlers.go | 6 + cmd/order/main.go | 272 +------------------------------ cmd/order/provider.go | 126 ++++++++++++++ cmd/order/stats.go | 201 +++++++++++++++++++++++ cmd/profile/main.go | 2 +- pkg/telemetry/logs.go | 76 ++++----- pkg/telemetry/otel.go | 6 +- 14 files changed, 457 insertions(+), 471 deletions(-) create mode 100644 cmd/cart/projection.go create mode 100644 cmd/order/provider.go create mode 100644 cmd/order/stats.go diff --git a/cmd/cart/main.go b/cmd/cart/main.go index 0c91874..d51042f 100644 --- a/cmd/cart/main.go +++ b/cmd/cart/main.go @@ -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) } diff --git a/cmd/cart/pool-server.go b/cmd/cart/pool-server.go index 28db29c..f9c8baf 100644 --- a/cmd/cart/pool-server.go +++ b/cmd/cart/pool-server.go @@ -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))) diff --git a/cmd/cart/projection.go b/cmd/cart/projection.go new file mode 100644 index 0000000..0e5f892 --- /dev/null +++ b/cmd/cart/projection.go @@ -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 +} diff --git a/cmd/checkout/checkout_builder.go b/cmd/checkout/checkout_builder.go index b7bf923..dbad763 100644 --- a/cmd/checkout/checkout_builder.go +++ b/cmd/checkout/checkout_builder.go @@ -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 diff --git a/cmd/checkout/klarna-client.go b/cmd/checkout/klarna-client.go index 410121a..1ab6e8d 100644 --- a/cmd/checkout/klarna-client.go +++ b/cmd/checkout/klarna-client.go @@ -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) { diff --git a/cmd/checkout/main.go b/cmd/checkout/main.go index 8cf417e..64323ce 100644 --- a/cmd/checkout/main.go +++ b/cmd/checkout/main.go @@ -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) } diff --git a/cmd/checkout/utils.go b/cmd/checkout/utils.go index 29421b9..59078ac 100644 --- a/cmd/checkout/utils.go +++ b/cmd/checkout/utils.go @@ -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 { diff --git a/cmd/order/handlers.go b/cmd/order/handlers.go index 09cc476..869d5bd 100644 --- a/cmd/order/handlers.go +++ b/cmd/order/handlers.go @@ -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) diff --git a/cmd/order/main.go b/cmd/order/main.go index acffc1b..fcc8731 100644 --- a/cmd/order/main.go +++ b/cmd/order/main.go @@ -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) diff --git a/cmd/order/provider.go b/cmd/order/provider.go new file mode 100644 index 0000000..9d84d62 --- /dev/null +++ b/cmd/order/provider.go @@ -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 +} diff --git a/cmd/order/stats.go b/cmd/order/stats.go new file mode 100644 index 0000000..e59ee6a --- /dev/null +++ b/cmd/order/stats.go @@ -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()) +} diff --git a/cmd/profile/main.go b/cmd/profile/main.go index ebdf5c7..77df527 100644 --- a/cmd/profile/main.go +++ b/cmd/profile/main.go @@ -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) } diff --git a/pkg/telemetry/logs.go b/pkg/telemetry/logs.go index d17b215..e10295b 100644 --- a/pkg/telemetry/logs.go +++ b/pkg/telemetry/logs.go @@ -12,22 +12,42 @@ import ( "go.opentelemetry.io/otel/sdk/log" ) -// NewLoggerProvider builds the OTLP log provider from the environment, or returns -// (nil, nil) when logs are disabled — in which case the caller MUST skip +// LoggerConfig controls the optional OTLP log pipeline. +// The zero value disables logging (Enabled = false, SampleRatio = 0). +type LoggerConfig struct { + Enabled bool + SampleRatio float64 // 0.0–1.0, clamped by NewLoggerProvider +} + +// LoggerConfigFromEnv reads OTEL_LOGS_ENABLED and OTEL_LOGS_SAMPLE_RATIO from +// the environment. Call this from main(); library consumers that want explicit +// control should construct LoggerConfig directly. +func LoggerConfigFromEnv() LoggerConfig { + var enabled bool + switch os.Getenv("OTEL_LOGS_ENABLED") { + case "1", "true", "TRUE", "True", "yes", "on": + enabled = true + } + ratio := 1.0 + if v := os.Getenv("OTEL_LOGS_SAMPLE_RATIO"); v != "" { + if r, err := strconv.ParseFloat(v, 64); err == nil && r >= 0 && r <= 1 { + ratio = r + } + } + return LoggerConfig{Enabled: enabled, SampleRatio: ratio} +} + +// NewLoggerProvider builds the OTLP log provider from cfg, or returns +// (nil, nil) when cfg.Enabled is false — in which case the caller MUST skip // global.SetLoggerProvider and the shutdown registration. // -// Env: -// -// OTEL_LOGS_ENABLED "1"/"true"/"yes"/"on" to enable the OTLP log pipeline. -// Default OFF: the batch processor clones its record ring -// on every export, which profiling showed to be the -// dominant heap allocator in these services. Traces and -// metrics are always on regardless of this flag. -// OTEL_LOGS_SAMPLE_RATIO fraction of records to export, 0.0–1.0 (default 1.0). -// e.g. 0.1 keeps ~10% of logs. Dropped records never reach -// the batch ring, so allocation churn falls ~proportionally. -func NewLoggerProvider(ctx context.Context) (*log.LoggerProvider, error) { - if !logsEnabled() { +// Use LoggerConfigFromEnv() to read the conventional OTEL_LOGS_ENABLED / +// OTEL_LOGS_SAMPLE_RATIO env vars, or construct LoggerConfig directly for +// explicit control (e.g. tests). Default OFF: the batch processor clones its +// record ring on every export, which profiling showed to be the dominant heap +// allocator in these services. Traces and metrics are always on regardless. +func NewLoggerProvider(ctx context.Context, cfg LoggerConfig) (*log.LoggerProvider, error) { + if !cfg.Enabled { return nil, nil } exporter, err := otlploggrpc.New(ctx) @@ -35,36 +55,12 @@ func NewLoggerProvider(ctx context.Context) (*log.LoggerProvider, error) { return nil, err } var proc log.Processor = log.NewBatchProcessor(exporter) - if ratio := sampleRatio(); ratio < 1.0 { - proc = &samplingProcessor{next: proc, ratio: ratio} + if cfg.SampleRatio < 1.0 { + proc = &samplingProcessor{next: proc, ratio: cfg.SampleRatio} } return log.NewLoggerProvider(log.WithProcessor(proc)), nil } -func logsEnabled() bool { - switch os.Getenv("OTEL_LOGS_ENABLED") { - case "1", "true", "TRUE", "True", "yes", "on": - return true - default: - return false - } -} - -func sampleRatio() float64 { - v := os.Getenv("OTEL_LOGS_SAMPLE_RATIO") - if v == "" { - return 1.0 - } - r, err := strconv.ParseFloat(v, 64) - if err != nil || r < 0 { - return 1.0 - } - if r > 1 { - return 1.0 - } - return r -} - // samplingProcessor forwards a random `ratio` fraction of records to the wrapped // processor and drops the rest before they reach the (allocation-heavy) batch // ring. This is head sampling: cheap, stateless, and per-record independent. diff --git a/pkg/telemetry/otel.go b/pkg/telemetry/otel.go index 7b3e954..f2b4c1a 100644 --- a/pkg/telemetry/otel.go +++ b/pkg/telemetry/otel.go @@ -15,11 +15,11 @@ import ( ) // SetupOTelSDK bootstraps the OpenTelemetry pipeline (propagator + traces + -// metrics, and an opt-in logger — see NewLoggerProvider). It is shared by every +// metrics, and an opt-in logger controlled by logCfg). It is shared by every // service main package; the service name comes from OTEL_RESOURCE_ATTRIBUTES, so // the same setup works everywhere. If it returns no error, call the returned // shutdown for proper cleanup. -func SetupOTelSDK(ctx context.Context) (func(context.Context) error, error) { +func SetupOTelSDK(ctx context.Context, logCfg LoggerConfig) (func(context.Context) error, error) { var shutdownFuncs []func(context.Context) error var err error @@ -58,7 +58,7 @@ func SetupOTelSDK(ctx context.Context) (func(context.Context) error, error) { // Logger provider is opt-in via OTEL_LOGS_ENABLED; nil means logs are off. // Traces + metrics above are always on. - loggerProvider, err := NewLoggerProvider(ctx) + loggerProvider, err := NewLoggerProvider(ctx, logCfg) if err != nil { handleErr(err) return shutdown, err