224 lines
6.7 KiB
Go
224 lines
6.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"hash/fnv"
|
|
"log/slog"
|
|
"sync"
|
|
"time"
|
|
|
|
messages "git.k6n.net/mats/go-cart-actor/proto/order"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
// rabbitPublisher is the broker-delivery side used by the outbox relay. It dials
|
|
// lazily and reconnects on failure, so the relay can be started even when the
|
|
// broker is down at boot: messages accumulate durably in the outbox and drain
|
|
// once the broker is reachable.
|
|
type rabbitPublisher struct {
|
|
url string
|
|
mu sync.Mutex
|
|
conn *amqp.Connection
|
|
ch *amqp.Channel
|
|
}
|
|
|
|
func newRabbitPublisher(url string) *rabbitPublisher {
|
|
return &rabbitPublisher{url: url}
|
|
}
|
|
|
|
func (p *rabbitPublisher) connect() error {
|
|
conn, err := amqp.Dial(p.url)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ch, err := conn.Channel()
|
|
if err != nil {
|
|
_ = conn.Close()
|
|
return err
|
|
}
|
|
p.conn, p.ch = conn, ch
|
|
return nil
|
|
}
|
|
|
|
func (p *rabbitPublisher) reset() {
|
|
if p.ch != nil {
|
|
_ = p.ch.Close()
|
|
}
|
|
if p.conn != nil {
|
|
_ = p.conn.Close()
|
|
}
|
|
p.ch, p.conn = nil, nil
|
|
}
|
|
|
|
func (p *rabbitPublisher) Publish(exchange, routingKey string, body []byte) error {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
if p.ch == nil {
|
|
if err := p.connect(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
// Bound the publish so a half-open/black-holed connection can't wedge the
|
|
// outbox relay indefinitely; on timeout the error path resets and the relay
|
|
// retries on its next tick.
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
err := p.ch.PublishWithContext(ctx, exchange, routingKey, false, false,
|
|
amqp.Publishing{ContentType: "application/json", Body: body})
|
|
if err != nil {
|
|
p.reset() // force a reconnect on the next attempt
|
|
}
|
|
return err
|
|
}
|
|
|
|
// This ingester lets the order service supersede the legacy go-order-manager:
|
|
// it consumes the same "order-queue" the Klarna/Adyen checkout publishes to and
|
|
// records each completed order as an event-sourced grain. Those orders are
|
|
// already paid by the external processor, so we record payment with provider
|
|
// "legacy" (place -> authorize -> capture) rather than charging again.
|
|
|
|
// legacyOrder is the subset of go-order-manager's Order JSON we need.
|
|
type legacyOrder struct {
|
|
ID string `json:"order_id"`
|
|
PurchaseCountry string `json:"purchase_country"`
|
|
PurchaseCurrency string `json:"purchase_currency"`
|
|
Locale string `json:"locale"`
|
|
OrderAmount int64 `json:"order_amount"`
|
|
OrderTaxAmount int64 `json:"order_tax_amount"`
|
|
OrderLines []legacyLine `json:"order_lines"`
|
|
Customer *legacyPerson `json:"customer,omitempty"`
|
|
BillingAddress *legacyAddr `json:"billing_address,omitempty"`
|
|
}
|
|
|
|
type legacyLine struct {
|
|
Reference string `json:"reference"`
|
|
Name string `json:"name"`
|
|
Quantity int32 `json:"quantity"`
|
|
UnitPrice int64 `json:"unit_price"`
|
|
TaxRate int32 `json:"tax_rate"`
|
|
TotalAmount int64 `json:"total_amount"`
|
|
TotalTaxAmount int64 `json:"total_tax_amount"`
|
|
}
|
|
|
|
type legacyPerson struct {
|
|
Email string `json:"email,omitempty"`
|
|
}
|
|
type legacyAddr struct {
|
|
Email string `json:"email,omitempty"`
|
|
GivenName string `json:"given_name,omitempty"`
|
|
FamilyName string `json:"family_name,omitempty"`
|
|
}
|
|
|
|
// legacyOrderID maps a legacy string order id to a stable grain id, so a
|
|
// redelivered message resolves to the same grain (PlaceOrder then no-ops).
|
|
func legacyOrderID(ref string) uint64 {
|
|
h := fnv.New64a()
|
|
_, _ = h.Write([]byte(ref))
|
|
id := h.Sum64()
|
|
if id == 0 {
|
|
id = 1
|
|
}
|
|
return id
|
|
}
|
|
|
|
// ingestLegacyOrder records one legacy order as an event-sourced grain. It is
|
|
// idempotent: if the order is already placed, PlaceOrder is rejected and we
|
|
// stop (the order already exists). Testable without a broker.
|
|
func ingestLegacyOrder(ctx context.Context, app *orderedApplier, body []byte) error {
|
|
var lo legacyOrder
|
|
if err := json.Unmarshal(body, &lo); err != nil {
|
|
return err
|
|
}
|
|
if lo.ID == "" || len(lo.OrderLines) == 0 {
|
|
return nil // nothing actionable
|
|
}
|
|
id := legacyOrderID(lo.ID)
|
|
ms := time.Now().UnixMilli()
|
|
|
|
po := &messages.PlaceOrder{
|
|
OrderReference: lo.ID,
|
|
Currency: lo.PurchaseCurrency,
|
|
Locale: lo.Locale,
|
|
Country: lo.PurchaseCountry,
|
|
TotalAmount: lo.OrderAmount,
|
|
TotalTax: lo.OrderTaxAmount,
|
|
PlacedAtMs: ms,
|
|
}
|
|
if lo.Customer != nil {
|
|
po.CustomerEmail = lo.Customer.Email
|
|
}
|
|
if po.CustomerEmail == "" && lo.BillingAddress != nil {
|
|
po.CustomerEmail = lo.BillingAddress.Email
|
|
po.CustomerName = lo.BillingAddress.GivenName + " " + lo.BillingAddress.FamilyName
|
|
}
|
|
for _, l := range lo.OrderLines {
|
|
po.Lines = append(po.Lines, &messages.OrderLine{
|
|
Reference: l.Reference, Sku: l.Reference, Name: l.Name,
|
|
Quantity: l.Quantity, UnitPrice: l.UnitPrice, TaxRate: l.TaxRate,
|
|
TotalAmount: l.TotalAmount, TotalTax: l.TotalTaxAmount,
|
|
})
|
|
}
|
|
|
|
if err := applyOne(ctx, app, id, po); err != nil {
|
|
// Already placed (redelivery) or invalid — nothing more to do.
|
|
return nil
|
|
}
|
|
// Record the externally-settled payment so the grain reaches "captured".
|
|
ref := "legacy-" + lo.ID
|
|
_ = applyOne(ctx, app, id, &messages.AuthorizePayment{Provider: "legacy", Amount: lo.OrderAmount, Reference: ref, AtMs: ms})
|
|
_ = applyOne(ctx, app, id, &messages.CapturePayment{Provider: "legacy", Amount: lo.OrderAmount, Reference: ref, AtMs: ms})
|
|
return nil
|
|
}
|
|
|
|
// startLegacyIngest connects to AMQP and consumes "order-queue" until ctx is
|
|
// done. Best-effort: a connection failure is logged and ingestion is skipped
|
|
// (the HTTP checkout path keeps working regardless).
|
|
func startLegacyIngest(ctx context.Context, amqpURL string, app *orderedApplier, logger *slog.Logger) {
|
|
conn, err := amqp.Dial(amqpURL)
|
|
if err != nil {
|
|
logger.Warn("legacy order ingest disabled: amqp dial failed", "err", err)
|
|
return
|
|
}
|
|
ch, err := conn.Channel()
|
|
if err != nil {
|
|
logger.Warn("legacy order ingest disabled: channel failed", "err", err)
|
|
_ = conn.Close()
|
|
return
|
|
}
|
|
q, err := ch.QueueDeclare("order-queue", false, false, false, false, nil)
|
|
if err != nil {
|
|
logger.Warn("legacy order ingest disabled: queue declare failed", "err", err)
|
|
_ = ch.Close()
|
|
_ = conn.Close()
|
|
return
|
|
}
|
|
msgs, err := ch.Consume(q.Name, "order-service", true, false, false, false, nil)
|
|
if err != nil {
|
|
logger.Warn("legacy order ingest disabled: consume failed", "err", err)
|
|
_ = ch.Close()
|
|
_ = conn.Close()
|
|
return
|
|
}
|
|
logger.Info("ingesting legacy orders from order-queue")
|
|
go func() {
|
|
defer conn.Close()
|
|
defer ch.Close()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case d, ok := <-msgs:
|
|
if !ok {
|
|
logger.Warn("legacy order ingest: channel closed")
|
|
return
|
|
}
|
|
if err := ingestLegacyOrder(ctx, app, d.Body); err != nil {
|
|
logger.Error("legacy order ingest failed", "err", err)
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|