feature/backoffice #6

Merged
mats merged 24 commits from feature/backoffice into main 2025-10-16 09:46:07 +02:00
32 changed files with 906 additions and 163 deletions
+3
View File
@@ -47,6 +47,9 @@ jobs:
docker push registry.knatofs.se/go-cart-actor-amd64:${{ needs.Metadata.outputs.version }}
- name: Apply deployment manifests
run: kubectl apply -f deployment/deployment.yaml -n cart
- name: Rollout amd64 backoffice deployment
run: |
kubectl rollout restart deployment/cart-backoffice-x86 -n cart
- name: Rollout amd64 deployment (pin to version)
run: |
kubectl set image deployment/cart-actor-x86 -n cart cart-actor-amd64=registry.knatofs.se/go-cart-actor-amd64:${{ needs.Metadata.outputs.version }}
+8
View File
@@ -59,6 +59,13 @@ RUN --mount=type=cache,target=/go/build-cache \
-X main.BuildDate=${BUILD_DATE}" \
-o /out/go-cart-actor ./cmd/cart
RUN --mount=type=cache,target=/go/build-cache \
go build -trimpath -ldflags="-s -w \
-X main.Version=${VERSION} \
-X main.GitCommit=${GIT_COMMIT} \
-X main.BuildDate=${BUILD_DATE}" \
-o /out/go-cart-backoffice ./cmd/backoffice
############################
# Runtime Stage
############################
@@ -67,6 +74,7 @@ FROM gcr.io/distroless/static-debian12:nonroot AS runtime
WORKDIR /
COPY --from=build /out/go-cart-actor /go-cart-actor
COPY --from=build /out/go-cart-backoffice /go-cart-backoffice
# Document (not expose forcibly) typical ports: 8080 (HTTP), 1337 (gRPC)
EXPOSE 8080 1337
+187
View File
@@ -0,0 +1,187 @@
package main
import (
"bufio"
"encoding/json"
"errors"
"fmt"
"net/http"
"os"
"path/filepath"
"regexp"
"sort"
"strconv"
"strings"
"time"
"git.tornberg.me/go-cart-actor/pkg/actor"
"git.tornberg.me/go-cart-actor/pkg/cart"
)
type FileServer struct {
// Define fields here
dataDir string
storage actor.LogStorage[cart.CartGrain]
}
func NewFileServer(dataDir string, storage actor.LogStorage[cart.CartGrain]) *FileServer {
return &FileServer{
dataDir: dataDir,
storage: storage,
}
}
func isValidId(id string) (uint64, bool) {
if nr, err := strconv.ParseUint(id, 10, 64); err == nil {
return nr, true
}
if nr, ok := cart.ParseCartId(id); ok {
return uint64(nr), true
}
return 0, false
}
func isValidFileId(name string) (uint64, bool) {
parts := strings.Split(name, ".")
if len(parts) > 1 && parts[1] == "events" {
idStr := parts[0]
return isValidId(idStr)
}
return 0, false
}
var cartFileRe = regexp.MustCompile(`^(\d+)\.events\.log$`)
func listCartFiles(dir string) ([]CartFileInfo, error) {
entries, err := os.ReadDir(dir)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
return []CartFileInfo{}, nil
}
return nil, err
}
out := make([]CartFileInfo, 0)
for _, e := range entries {
if e.IsDir() {
continue
}
id, valid := isValidFileId(e.Name())
if !valid {
continue
}
info, err := e.Info()
if err != nil {
continue
}
info.Sys()
out = append(out, CartFileInfo{
ID: fmt.Sprintf("%d", id),
CartId: cart.CartId(id),
Size: info.Size(),
Modified: info.ModTime(),
System: info.Sys(),
})
}
return out, nil
}
func readRawLogLines(path string) ([]json.RawMessage, error) {
fh, err := os.Open(path)
if err != nil {
return nil, err
}
defer fh.Close()
lines := make([]json.RawMessage, 0, 64)
s := bufio.NewScanner(fh)
// increase buffer to handle larger JSON lines
buf := make([]byte, 0, 1024*1024)
s.Buffer(buf, 1024*1024)
for s.Scan() {
line := s.Bytes()
if line == nil {
continue
}
lines = append(lines, line)
}
if err := s.Err(); err != nil {
return nil, err
}
return lines, nil
}
func writeJSON(w http.ResponseWriter, status int, v any) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Cache-Control", "no-cache")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(v)
}
func (fs *FileServer) CartsHandler(w http.ResponseWriter, r *http.Request) {
list, err := listCartFiles(fs.dataDir)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
return
}
// sort by modified desc
sort.Slice(list, func(i, j int) bool { return list[i].Modified.After(list[j].Modified) })
writeJSON(w, http.StatusOK, map[string]any{
"count": len(list),
"carts": list,
})
}
type JsonError struct {
Error string `json:"error"`
}
func (fs *FileServer) CartHandler(w http.ResponseWriter, r *http.Request) {
idStr := r.PathValue("id")
if idStr == "" {
writeJSON(w, http.StatusBadRequest, JsonError{Error: "missing id"})
return
}
id, ok := isValidId(idStr)
if !ok {
writeJSON(w, http.StatusBadRequest, JsonError{Error: "invalid id"})
return
}
// reconstruct state from event log if present
grain := cart.NewCartGrain(id, time.Now())
err := fs.storage.LoadEvents(id, grain)
if err != nil {
writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()})
return
}
path := filepath.Join(fs.dataDir, fmt.Sprintf("%d.events.log", id))
info, err := os.Stat(path)
if err != nil && errors.Is(err, os.ErrNotExist) {
writeJSON(w, http.StatusNotFound, JsonError{Error: "cart not found"})
return
} else if err != nil {
writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()})
return
}
lines, err := readRawLogLines(path)
if err != nil {
writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()})
return
}
writeJSON(w, http.StatusOK, map[string]any{
"id": id,
"cartId": cart.CartId(id).String(),
"state": grain,
"mutations": lines,
"meta": map[string]any{
"size": info.Size(),
"modified": info.ModTime(),
"path": path,
"system": info.Sys(),
},
})
}
+248
View File
@@ -0,0 +1,248 @@
package main
import (
"bufio"
"crypto/sha1"
"encoding/base64"
"encoding/binary"
"io"
"net"
"net/http"
"strings"
"time"
)
// Hub manages websocket clients and broadcasts messages to them.
type Hub struct {
register chan *Client
unregister chan *Client
broadcast chan []byte
clients map[*Client]bool
}
// Client represents a single websocket client connection.
type Client struct {
hub *Hub
conn net.Conn
send chan []byte
}
// NewHub constructs a new Hub instance.
func NewHub() *Hub {
return &Hub{
register: make(chan *Client),
unregister: make(chan *Client),
broadcast: make(chan []byte, 1024),
clients: make(map[*Client]bool),
}
}
// Run starts the hub event loop.
func (h *Hub) Run() {
for {
select {
case c := <-h.register:
h.clients[c] = true
case c := <-h.unregister:
if _, ok := h.clients[c]; ok {
delete(h.clients, c)
close(c.send)
_ = c.conn.Close()
}
case msg := <-h.broadcast:
for c := range h.clients {
select {
case c.send <- msg:
default:
// Client is slow or dead; drop it.
delete(h.clients, c)
close(c.send)
_ = c.conn.Close()
}
}
}
}
}
// computeAccept computes the Sec-WebSocket-Accept header value.
func computeAccept(key string) string {
const magic = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
h := sha1.New()
h.Write([]byte(key + magic))
return base64.StdEncoding.EncodeToString(h.Sum(nil))
}
// ServeWS upgrades the HTTP request to a WebSocket connection and registers a client.
func (h *Hub) ServeWS(w http.ResponseWriter, r *http.Request) {
if !strings.Contains(strings.ToLower(r.Header.Get("Connection")), "upgrade") || strings.ToLower(r.Header.Get("Upgrade")) != "websocket" {
http.Error(w, "upgrade required", http.StatusBadRequest)
return
}
key := r.Header.Get("Sec-WebSocket-Key")
if key == "" {
http.Error(w, "missing Sec-WebSocket-Key", http.StatusBadRequest)
return
}
accept := computeAccept(key)
hj, ok := w.(http.Hijacker)
if !ok {
http.Error(w, "websocket not supported", http.StatusInternalServerError)
return
}
conn, buf, err := hj.Hijack()
if err != nil {
return
}
// Write the upgrade response
response := "HTTP/1.1 101 Switching Protocols\r\n" +
"Upgrade: websocket\r\n" +
"Connection: Upgrade\r\n" +
"Sec-WebSocket-Accept: " + accept + "\r\n" +
"\r\n"
if _, err := buf.WriteString(response); err != nil {
_ = conn.Close()
return
}
if err := buf.Flush(); err != nil {
_ = conn.Close()
return
}
client := &Client{
hub: h,
conn: conn,
send: make(chan []byte, 256),
}
h.register <- client
go client.writePump()
go client.readPump()
}
// writeWSFrame writes a single WebSocket frame to the writer.
func writeWSFrame(w io.Writer, opcode byte, payload []byte) error {
// FIN set, opcode as provided
header := []byte{0x80 | (opcode & 0x0F)}
l := len(payload)
switch {
case l < 126:
header = append(header, byte(l))
case l <= 65535:
ext := make([]byte, 2)
binary.BigEndian.PutUint16(ext, uint16(l))
header = append(header, 126)
header = append(header, ext...)
default:
ext := make([]byte, 8)
binary.BigEndian.PutUint64(ext, uint64(l))
header = append(header, 127)
header = append(header, ext...)
}
if _, err := w.Write(header); err != nil {
return err
}
if l > 0 {
if _, err := w.Write(payload); err != nil {
return err
}
}
return nil
}
// readPump handles control frames from the client and discards other incoming frames.
// This server is broadcast-only to clients.
func (c *Client) readPump() {
defer func() {
c.hub.unregister <- c
}()
reader := bufio.NewReader(c.conn)
for {
// Read first two bytes
b1, err := reader.ReadByte()
if err != nil {
return
}
b2, err := reader.ReadByte()
if err != nil {
return
}
opcode := b1 & 0x0F
masked := (b2 & 0x80) != 0
length := int64(b2 & 0x7F)
if length == 126 {
ext := make([]byte, 2)
if _, err := io.ReadFull(reader, ext); err != nil {
return
}
length = int64(binary.BigEndian.Uint16(ext))
} else if length == 127 {
ext := make([]byte, 8)
if _, err := io.ReadFull(reader, ext); err != nil {
return
}
length = int64(binary.BigEndian.Uint64(ext))
}
var maskKey [4]byte
if masked {
if _, err := io.ReadFull(reader, maskKey[:]); err != nil {
return
}
}
// Handle Ping -> Pong
if opcode == 0x9 && length <= 125 {
payload := make([]byte, length)
if _, err := io.ReadFull(reader, payload); err != nil {
return
}
// Unmask if masked
if masked {
for i := int64(0); i < length; i++ {
payload[i] ^= maskKey[i%4]
}
}
_ = writeWSFrame(c.conn, 0xA, payload) // best-effort pong
continue
}
// Close frame
if opcode == 0x8 {
// Drain payload if any, then exit
if _, err := io.CopyN(io.Discard, reader, length); err != nil {
return
}
return
}
// For other frames, just discard payload
if _, err := io.CopyN(io.Discard, reader, length); err != nil {
return
}
}
}
// writePump sends queued messages to the client and pings periodically to keep the connection alive.
func (c *Client) writePump() {
ticker := time.NewTicker(30 * time.Second)
defer func() {
ticker.Stop()
_ = c.conn.Close()
}()
for {
select {
case msg, ok := <-c.send:
if !ok {
// try to send close frame
_ = writeWSFrame(c.conn, 0x8, nil)
return
}
if err := writeWSFrame(c.conn, 0x1, msg); err != nil {
return
}
case <-ticker.C:
// Send a ping to keep connections alive behind proxies
_ = writeWSFrame(c.conn, 0x9, []byte("ping"))
}
}
}
+145 -1
View File
@@ -1,5 +1,149 @@
package main
import (
"context"
"errors"
"log"
"net/http"
"os"
"time"
actor "git.tornberg.me/go-cart-actor/pkg/actor"
"git.tornberg.me/go-cart-actor/pkg/cart"
"github.com/matst80/slask-finder/pkg/messaging"
amqp "github.com/rabbitmq/amqp091-go"
)
type CartFileInfo struct {
ID string `json:"id"`
CartId cart.CartId `json:"cartId"`
Size int64 `json:"size"`
Modified time.Time `json:"modified"`
System any `json:"system"`
}
func envOrDefault(key, def string) string {
if v := os.Getenv(key); v != "" {
return v
}
return def
}
func startMutationConsumer(ctx context.Context, conn *amqp.Connection, hub *Hub) error {
ch, err := conn.Channel()
if err != nil {
_ = conn.Close()
return err
}
msgs, err := messaging.DeclareBindAndConsume(ch, "cart", "mutation")
if err != nil {
_ = ch.Close()
return err
}
go func() {
defer ch.Close()
for {
select {
case <-ctx.Done():
return
case m, ok := <-msgs:
if !ok {
log.Fatalf("connection closed")
continue
}
// Log and broadcast to all websocket clients
log.Printf("mutation event: %s", string(m.Body))
if hub != nil {
select {
case hub.broadcast <- m.Body:
default:
// if hub queue is full, drop to avoid blocking
}
}
if err := m.Ack(false); err != nil {
log.Printf("error acknowledging message: %v", err)
}
}
}
}()
return nil
}
func main() {
// Your code here
dataDir := envOrDefault("DATA_DIR", "data")
addr := envOrDefault("ADDR", ":8080")
amqpURL := os.Getenv("AMQP_URL")
_ = os.MkdirAll(dataDir, 0755)
reg := cart.NewCartMultationRegistry()
diskStorage := actor.NewDiskStorage[cart.CartGrain](dataDir, reg)
fs := NewFileServer(dataDir, diskStorage)
hub := NewHub()
go hub.Run()
mux := http.NewServeMux()
mux.HandleFunc("GET /carts", fs.CartsHandler)
mux.HandleFunc("GET /cart/{id}", fs.CartHandler)
mux.HandleFunc("/ws", hub.ServeWS)
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("ok"))
})
mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("ok"))
})
mux.HandleFunc("/livez", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("ok"))
})
// Global CORS middleware allowing all origins and handling preflight
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS")
w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Requested-With")
w.Header().Set("Access-Control-Expose-Headers", "*")
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusNoContent)
return
}
mux.ServeHTTP(w, r)
})
srv := &http.Server{
Addr: addr,
Handler: handler,
ReadTimeout: 10 * time.Second,
WriteTimeout: 30 * time.Second,
IdleTimeout: 60 * time.Second,
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
if amqpURL != "" {
conn, err := amqp.Dial(amqpURL)
if err != nil {
log.Fatalf("failed to connect to RabbitMQ: %w", err)
}
if err := startMutationConsumer(ctx, conn, hub); err != nil {
log.Printf("AMQP listener disabled: %v", err)
} else {
log.Printf("AMQP listener connected")
}
}
log.Printf("backoffice HTTP listening on %s (dataDir=%s)", addr, dataDir)
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatalf("http server error: %v", err)
}
// server stopped
}
+3 -1
View File
@@ -3,6 +3,8 @@ package main
import (
"encoding/json"
"fmt"
"git.tornberg.me/go-cart-actor/pkg/cart"
)
// CheckoutMeta carries the external / URL metadata required to build a
@@ -33,7 +35,7 @@ type CheckoutMeta struct {
//
// If you later need to support different tax rates per line, you can extend
// CartItem / Delivery to expose that data and propagate it here.
func BuildCheckoutOrderPayload(grain *CartGrain, meta *CheckoutMeta) ([]byte, *CheckoutOrder, error) {
func BuildCheckoutOrderPayload(grain *cart.CartGrain, meta *CheckoutMeta) ([]byte, *CheckoutOrder, error) {
if grain == nil {
return nil, nil, fmt.Errorf("nil grain")
}
+23 -56
View File
@@ -13,6 +13,7 @@ import (
"time"
"git.tornberg.me/go-cart-actor/pkg/actor"
"git.tornberg.me/go-cart-actor/pkg/cart"
"git.tornberg.me/go-cart-actor/pkg/discovery"
messages "git.tornberg.me/go-cart-actor/pkg/messages"
"git.tornberg.me/go-cart-actor/pkg/proxy"
@@ -46,7 +47,7 @@ func init() {
}
type App struct {
pool *actor.SimpleGrainPool[CartGrain]
pool *actor.SimpleGrainPool[cart.CartGrain]
}
var podIp = os.Getenv("POD_IP")
@@ -98,63 +99,24 @@ type MutationContext struct {
VoucherService voucher.Service
}
type CartChangeEvent struct {
CartId cart.CartId `json:"cartId"`
Mutations []actor.ApplyResult `json:"mutations"`
}
func main() {
controlPlaneConfig := actor.DefaultServerConfig()
reg := actor.NewMutationRegistry()
reg.RegisterMutations(
actor.NewMutation(AddItem, func() *messages.AddItem {
return &messages.AddItem{}
}),
actor.NewMutation(ChangeQuantity, func() *messages.ChangeQuantity {
return &messages.ChangeQuantity{}
}),
actor.NewMutation(RemoveItem, func() *messages.RemoveItem {
return &messages.RemoveItem{}
}),
actor.NewMutation(InitializeCheckout, func() *messages.InitializeCheckout {
return &messages.InitializeCheckout{}
}),
actor.NewMutation(OrderCreated, func() *messages.OrderCreated {
return &messages.OrderCreated{}
}),
actor.NewMutation(RemoveDelivery, func() *messages.RemoveDelivery {
return &messages.RemoveDelivery{}
}),
actor.NewMutation(SetDelivery, func() *messages.SetDelivery {
return &messages.SetDelivery{}
}),
actor.NewMutation(SetPickupPoint, func() *messages.SetPickupPoint {
return &messages.SetPickupPoint{}
}),
actor.NewMutation(ClearCart, func() *messages.ClearCartRequest {
return &messages.ClearCartRequest{}
}),
actor.NewMutation(AddVoucher, func() *messages.AddVoucher {
return &messages.AddVoucher{}
}),
actor.NewMutation(RemoveVoucher, func() *messages.RemoveVoucher {
return &messages.RemoveVoucher{}
}),
)
diskStorage := actor.NewDiskStorage[CartGrain]("data", reg)
poolConfig := actor.GrainPoolConfig[CartGrain]{
reg := cart.NewCartMultationRegistry()
diskStorage := actor.NewDiskStorage[cart.CartGrain]("data", reg)
poolConfig := actor.GrainPoolConfig[cart.CartGrain]{
MutationRegistry: reg,
Storage: diskStorage,
Spawn: func(id uint64) (actor.Grain[CartGrain], error) {
Spawn: func(id uint64) (actor.Grain[cart.CartGrain], error) {
grainSpawns.Inc()
ret := &CartGrain{
lastItemId: 0,
lastDeliveryId: 0,
Deliveries: []*CartDelivery{},
Id: CartId(id),
Items: []*CartItem{},
TotalPrice: NewPrice(),
}
ret := cart.NewCartGrain(id, time.Now())
// Set baseline lastChange at spawn; replay may update it to last event timestamp.
ret.lastChange = time.Now()
ret.lastAccess = time.Now()
err := diskStorage.LoadEvents(id, ret)
@@ -181,17 +143,22 @@ func main() {
fmt.Errorf("failed to connect to RabbitMQ: %w", err)
}
amqpListener := actor.NewAmqpListener(conn)
amqpListener := actor.NewAmqpListener(conn, func(id uint64, msg []actor.ApplyResult) (any, error) {
return &CartChangeEvent{
CartId: cart.CartId(id),
Mutations: msg,
}, nil
})
amqpListener.DefineTopics()
pool.AddListener(amqpListener)
grpcSrv, err := actor.NewControlServer[*CartGrain](controlPlaneConfig, pool)
grpcSrv, err := actor.NewControlServer[*cart.CartGrain](controlPlaneConfig, pool)
if err != nil {
log.Fatalf("Error starting control plane gRPC server: %v\n", err)
}
defer grpcSrv.GracefulStop()
go diskStorage.SaveLoop(10 * time.Second)
// go diskStorage.SaveLoop(10 * time.Second)
go func(hw discovery.Discovery) {
if hw == nil {
@@ -282,14 +249,14 @@ func main() {
w.Write([]byte("no cart id to checkout is empty"))
return
}
parsed, ok := ParseCartId(cookie.Value)
parsed, ok := cart.ParseCartId(cookie.Value)
if !ok {
w.WriteHeader(http.StatusBadRequest)
w.Write([]byte("invalid cart id format"))
return
}
cartId := parsed
syncedServer.ProxyHandler(func(w http.ResponseWriter, r *http.Request, cartId CartId) error {
syncedServer.ProxyHandler(func(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error {
order, err = syncedServer.CreateOrUpdateCheckout(r.Host, cartId)
if err != nil {
return err
@@ -443,7 +410,7 @@ func triggerOrderCompleted(syncedServer *PoolServer, order *CheckoutOrder) error
OrderId: order.ID,
Status: order.Status,
}
cid, ok := ParseCartId(order.MerchantReference1)
cid, ok := cart.ParseCartId(order.MerchantReference1)
if !ok {
return fmt.Errorf("invalid cart id in order reference: %s", order.MerchantReference1)
}
+30 -29
View File
@@ -11,18 +11,19 @@ import (
"time"
"git.tornberg.me/go-cart-actor/pkg/actor"
"git.tornberg.me/go-cart-actor/pkg/cart"
messages "git.tornberg.me/go-cart-actor/pkg/messages"
"git.tornberg.me/go-cart-actor/pkg/voucher"
"github.com/gogo/protobuf/proto"
)
type PoolServer struct {
actor.GrainPool[*CartGrain]
actor.GrainPool[*cart.CartGrain]
pod_name string
klarnaClient *KlarnaClient
}
func NewPoolServer(pool actor.GrainPool[*CartGrain], pod_name string, klarnaClient *KlarnaClient) *PoolServer {
func NewPoolServer(pool actor.GrainPool[*cart.CartGrain], pod_name string, klarnaClient *KlarnaClient) *PoolServer {
return &PoolServer{
GrainPool: pool,
pod_name: pod_name,
@@ -30,11 +31,11 @@ func NewPoolServer(pool actor.GrainPool[*CartGrain], pod_name string, klarnaClie
}
}
func (s *PoolServer) ApplyLocal(id CartId, mutation ...proto.Message) (*actor.MutationResult[*CartGrain], error) {
func (s *PoolServer) ApplyLocal(id cart.CartId, mutation ...proto.Message) (*actor.MutationResult[*cart.CartGrain], error) {
return s.Apply(uint64(id), mutation...)
}
func (s *PoolServer) GetCartHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
func (s *PoolServer) GetCartHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
grain, err := s.Get(uint64(id))
if err != nil {
return err
@@ -43,7 +44,7 @@ func (s *PoolServer) GetCartHandler(w http.ResponseWriter, r *http.Request, id C
return s.WriteResult(w, grain)
}
func (s *PoolServer) AddSkuToCartHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
func (s *PoolServer) AddSkuToCartHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
sku := r.PathValue("sku")
msg, err := GetItemAddMessage(sku, 1, getCountryFromHost(r.Host), nil)
if err != nil {
@@ -73,7 +74,7 @@ func (s *PoolServer) WriteResult(w http.ResponseWriter, result any) error {
return err
}
func (s *PoolServer) DeleteItemHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
func (s *PoolServer) DeleteItemHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
itemIdString := r.PathValue("itemId")
itemId, err := strconv.ParseInt(itemIdString, 10, 64)
@@ -93,7 +94,7 @@ type SetDeliveryRequest struct {
PickupPoint *messages.PickupPoint `json:"pickupPoint,omitempty"`
}
func (s *PoolServer) SetDeliveryHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
func (s *PoolServer) SetDeliveryHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
delivery := SetDeliveryRequest{}
err := json.NewDecoder(r.Body).Decode(&delivery)
@@ -111,7 +112,7 @@ func (s *PoolServer) SetDeliveryHandler(w http.ResponseWriter, r *http.Request,
return s.WriteResult(w, data)
}
func (s *PoolServer) SetPickupPointHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
func (s *PoolServer) SetPickupPointHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
deliveryIdString := r.PathValue("deliveryId")
deliveryId, err := strconv.ParseInt(deliveryIdString, 10, 64)
@@ -138,7 +139,7 @@ func (s *PoolServer) SetPickupPointHandler(w http.ResponseWriter, r *http.Reques
return s.WriteResult(w, reply)
}
func (s *PoolServer) RemoveDeliveryHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
func (s *PoolServer) RemoveDeliveryHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
deliveryIdString := r.PathValue("deliveryId")
deliveryId, err := strconv.Atoi(deliveryIdString)
@@ -152,7 +153,7 @@ func (s *PoolServer) RemoveDeliveryHandler(w http.ResponseWriter, r *http.Reques
return s.WriteResult(w, reply)
}
func (s *PoolServer) QuantityChangeHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
func (s *PoolServer) QuantityChangeHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
changeQuantity := messages.ChangeQuantity{}
err := json.NewDecoder(r.Body).Decode(&changeQuantity)
if err != nil {
@@ -197,7 +198,7 @@ func getMultipleAddMessages(items []Item, country string) []proto.Message {
return msgs
}
func (s *PoolServer) SetCartItemsHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
func (s *PoolServer) SetCartItemsHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
setCartItems := SetCartItems{}
err := json.NewDecoder(r.Body).Decode(&setCartItems)
if err != nil {
@@ -215,7 +216,7 @@ func (s *PoolServer) SetCartItemsHandler(w http.ResponseWriter, r *http.Request,
return s.WriteResult(w, reply)
}
func (s *PoolServer) AddMultipleItemHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
func (s *PoolServer) AddMultipleItemHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
setCartItems := SetCartItems{}
err := json.NewDecoder(r.Body).Decode(&setCartItems)
if err != nil {
@@ -236,7 +237,7 @@ type AddRequest struct {
StoreId *string `json:"storeId"`
}
func (s *PoolServer) AddSkuRequestHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
func (s *PoolServer) AddSkuRequestHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
addRequest := AddRequest{Quantity: 1}
err := json.NewDecoder(r.Body).Decode(&addRequest)
if err != nil {
@@ -287,7 +288,7 @@ func getLocale(country string) string {
return "sv-se"
}
func (s *PoolServer) CreateOrUpdateCheckout(host string, id CartId) (*CheckoutOrder, error) {
func (s *PoolServer) CreateOrUpdateCheckout(host string, id cart.CartId) (*CheckoutOrder, error) {
country := getCountryFromHost(host)
meta := &CheckoutMeta{
Terms: fmt.Sprintf("https://%s/terms", host),
@@ -319,7 +320,7 @@ func (s *PoolServer) CreateOrUpdateCheckout(host string, id CartId) (*CheckoutOr
}
}
func (s *PoolServer) ApplyCheckoutStarted(klarnaOrder *CheckoutOrder, id CartId) (*actor.MutationResult[*CartGrain], error) {
func (s *PoolServer) ApplyCheckoutStarted(klarnaOrder *CheckoutOrder, id cart.CartId) (*actor.MutationResult[*cart.CartGrain], error) {
// Persist initialization state via mutation (best-effort)
return s.ApplyLocal(id, &messages.InitializeCheckout{
OrderId: klarnaOrder.ID,
@@ -341,12 +342,12 @@ func (s *PoolServer) ApplyCheckoutStarted(klarnaOrder *CheckoutOrder, id CartId)
// }
//
func CookieCartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
func CookieCartIdHandler(fn func(cartId cart.CartId, w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
var id CartId
var id cart.CartId
cookie, err := r.Cookie("cartid")
if err != nil || cookie.Value == "" {
id = MustNewCartId()
id = cart.MustNewCartId()
http.SetCookie(w, &http.Cookie{
Name: "cartid",
Value: id.String(),
@@ -358,9 +359,9 @@ func CookieCartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.R
})
w.Header().Set("Set-Cart-Id", id.String())
} else {
parsed, ok := ParseCartId(cookie.Value)
parsed, ok := cart.ParseCartId(cookie.Value)
if !ok {
id = MustNewCartId()
id = cart.MustNewCartId()
http.SetCookie(w, &http.Cookie{
Name: "cartid",
Value: id.String(),
@@ -388,7 +389,7 @@ func CookieCartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.R
// Removed leftover legacy block after CookieCartIdHandler (obsolete code referencing cid/legacy)
func (s *PoolServer) RemoveCartCookie(w http.ResponseWriter, r *http.Request, cartId CartId) error {
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{
Name: "cartid",
@@ -403,17 +404,17 @@ func (s *PoolServer) RemoveCartCookie(w http.ResponseWriter, r *http.Request, ca
return nil
}
func CartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
func CartIdHandler(fn func(cartId cart.CartId, w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
var id CartId
var id cart.CartId
raw := r.PathValue("id")
// If no id supplied, generate a new one
if raw == "" {
id := MustNewCartId()
id := cart.MustNewCartId()
w.Header().Set("Set-Cart-Id", id.String())
} else {
// Parse base62 cart id
if parsedId, ok := ParseCartId(raw); !ok {
if parsedId, ok := cart.ParseCartId(raw); !ok {
w.WriteHeader(http.StatusBadRequest)
w.Write([]byte("cart id is invalid"))
return
@@ -431,8 +432,8 @@ func CartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.Request
}
}
func (s *PoolServer) ProxyHandler(fn func(w http.ResponseWriter, r *http.Request, cartId CartId) error) func(cartId CartId, w http.ResponseWriter, r *http.Request) error {
return func(cartId CartId, w http.ResponseWriter, r *http.Request) error {
func (s *PoolServer) ProxyHandler(fn func(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error) func(cartId cart.CartId, w http.ResponseWriter, r *http.Request) error {
return func(cartId cart.CartId, w http.ResponseWriter, r *http.Request) error {
if ownerHost, ok := s.OwnerHost(uint64(cartId)); ok {
handled, err := ownerHost.Proxy(uint64(cartId), w, r)
if err == nil && handled {
@@ -449,7 +450,7 @@ type AddVoucherRequest struct {
VoucherCode string `json:"code"`
}
func (s *PoolServer) AddVoucherHandler(w http.ResponseWriter, r *http.Request, cartId CartId) error {
func (s *PoolServer) AddVoucherHandler(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error {
data := &AddVoucherRequest{}
json.NewDecoder(r.Body).Decode(data)
v := voucher.Service{}
@@ -469,7 +470,7 @@ func (s *PoolServer) AddVoucherHandler(w http.ResponseWriter, r *http.Request, c
return nil
}
func (s *PoolServer) RemoveVoucherHandler(w http.ResponseWriter, r *http.Request, cartId CartId) error {
func (s *PoolServer) RemoveVoucherHandler(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error {
idStr := r.PathValue("voucherId")
id, err := strconv.ParseInt(idStr, 10, 64)
+11 -3
View File
@@ -7,6 +7,7 @@ import (
"strconv"
"strings"
"git.tornberg.me/go-cart-actor/pkg/cart"
messages "git.tornberg.me/go-cart-actor/pkg/messages"
"github.com/matst80/slask-finder/pkg/index"
)
@@ -53,19 +54,19 @@ func ToItemAddMessage(item *index.DataItem, storeId *string, qty int, country st
return nil
}
stock := StockStatus(0)
stock := cart.StockStatus(0)
centralStockValue, ok := item.GetStringFieldValue(3)
if storeId == nil {
if ok {
pureNumber := strings.Replace(centralStockValue, "+", "", -1)
if centralStock, err := strconv.ParseInt(pureNumber, 10, 64); err == nil {
stock = StockStatus(centralStock)
stock = cart.StockStatus(centralStock)
}
}
} else {
storeStock, ok := item.Stock.GetStock()[*storeId]
if ok {
stock = StockStatus(storeStock)
stock = cart.StockStatus(storeStock)
}
}
@@ -119,3 +120,10 @@ func getTax(articleType string) int32 {
return 2500
}
}
func getInt(data float64, ok bool) (int, error) {
if !ok {
return 0, fmt.Errorf("invalid type")
}
return int(data), nil
}
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
Binary file not shown.
+123
View File
@@ -9,6 +9,94 @@ type: Opaque
---
apiVersion: apps/v1
kind: Deployment
metadata:
labels:
app: cart-backoffice
arch: amd64
name: cart-backoffice-x86
spec:
replicas: 1
selector:
matchLabels:
app: cart-backoffice
arch: amd64
template:
metadata:
labels:
app: cart-backoffice
arch: amd64
spec:
affinity:
nodeAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
nodeSelectorTerms:
- matchExpressions:
- key: kubernetes.io/arch
operator: NotIn
values:
- arm64
volumes:
- name: data
nfs:
path: /i-data/7a8af061/nfs/cart-actor
server: 10.10.1.10
imagePullSecrets:
- name: regcred
serviceAccountName: default
containers:
- image: registry.knatofs.se/go-cart-actor-amd64:latest
name: cart-actor-amd64
imagePullPolicy: Always
command: ["/go-cart-backoffice"]
lifecycle:
preStop:
exec:
command: ["sleep", "15"]
ports:
- containerPort: 8080
name: web
livenessProbe:
httpGet:
path: /livez
port: web
failureThreshold: 1
periodSeconds: 10
readinessProbe:
httpGet:
path: /readyz
port: web
failureThreshold: 2
initialDelaySeconds: 2
periodSeconds: 10
volumeMounts:
- mountPath: "/data"
name: data
resources:
limits:
memory: "768Mi"
requests:
memory: "70Mi"
cpu: "1200m"
env:
- name: TZ
value: "Europe/Stockholm"
- name: KLARNA_API_USERNAME
valueFrom:
secretKeyRef:
name: klarna-api-credentials
key: username
- name: KLARNA_API_PASSWORD
valueFrom:
secretKeyRef:
name: klarna-api-credentials
key: password
- name: AMQP_URL
value: "amqp://admin:12bananer@rabbitmq.dev:5672/"
# - name: BASE_URL
# value: "https://s10n-no.tornberg.me"
---
apiVersion: apps/v1
kind: Deployment
metadata:
labels:
app: cart-actor
@@ -222,6 +310,17 @@ spec:
- name: web
port: 8080
---
kind: Service
apiVersion: v1
metadata:
name: cart-backoffice
spec:
selector:
app: cart-backoffice
ports:
- name: web
port: 8080
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
@@ -250,3 +349,27 @@ spec:
name: cart-actor
port:
number: 8080
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: cart-backend-ingress
annotations:
cert-manager.io/cluster-issuer: letsencrypt-prod
spec:
ingressClassName: nginx
tls:
- hosts:
- slask-cart.tornberg.me
secretName: cart-backoffice-actor-tls-secret
rules:
- host: slask-cart.tornberg.me
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: cart-backoffice
port:
number: 8080
-42
View File
@@ -10,8 +10,6 @@ import (
"time"
"github.com/gogo/protobuf/proto"
"github.com/matst80/slask-finder/pkg/messaging"
amqp "github.com/rabbitmq/amqp091-go"
)
type QueueEvent struct {
@@ -31,46 +29,6 @@ type LogStorage[V any] interface {
AppendMutations(id uint64, msg ...proto.Message) error
}
type LogListener interface {
AppendMutations(id uint64, msg ...ApplyResult)
}
type AmqpListener struct {
conn *amqp.Connection
}
func NewAmqpListener(conn *amqp.Connection) *AmqpListener {
return &AmqpListener{
conn: conn,
}
}
func (l *AmqpListener) DefineTopics() {
ch, err := l.conn.Channel()
if err != nil {
log.Fatalf("Failed to open a channel: %v", err)
}
defer ch.Close()
if err := messaging.DefineTopic(ch, "cart", "mutation"); err != nil {
log.Fatalf("Failed to declare topic mutation: %v", err)
}
}
type CartEvent struct {
Id uint64 `json:"id"`
Mutations []ApplyResult `json:"mutations"`
}
func (l *AmqpListener) AppendMutations(id uint64, msg ...ApplyResult) {
err := messaging.SendChange(l.conn, "cart", "mutation", &CartEvent{
Id: id,
Mutations: msg,
})
if err != nil {
log.Printf("Failed to send mutation event: %v", err)
}
}
func NewDiskStorage[V any](path string, registry MutationRegistry) *DiskStorage[V] {
return &DiskStorage[V]{
StateStorage: NewState(registry),
+47
View File
@@ -0,0 +1,47 @@
package actor
import (
"log"
"github.com/matst80/slask-finder/pkg/messaging"
amqp "github.com/rabbitmq/amqp091-go"
)
type LogListener interface {
AppendMutations(id uint64, msg ...ApplyResult)
}
type AmqpListener struct {
conn *amqp.Connection
transformer func(id uint64, msg []ApplyResult) (any, error)
}
func NewAmqpListener(conn *amqp.Connection, transformer func(id uint64, msg []ApplyResult) (any, error)) *AmqpListener {
return &AmqpListener{
conn: conn,
transformer: transformer,
}
}
func (l *AmqpListener) DefineTopics() {
ch, err := l.conn.Channel()
if err != nil {
log.Fatalf("Failed to open a channel: %v", err)
}
defer ch.Close()
if err := messaging.DefineTopic(ch, "cart", "mutation"); err != nil {
log.Fatalf("Failed to declare topic mutation: %v", err)
}
}
func (l *AmqpListener) AppendMutations(id uint64, msg ...ApplyResult) {
data, err := l.transformer(id, msg)
if err != nil {
log.Printf("Failed to transform mutation event: %v", err)
return
}
err = messaging.SendChange(l.conn, "cart", "mutation", data)
if err != nil {
log.Printf("Failed to send mutation event: %v", err)
}
}
@@ -1,8 +1,7 @@
package main
package cart
import (
"encoding/json"
"fmt"
"slices"
"sync"
"time"
@@ -138,6 +137,22 @@ func (v *Voucher) AppliesTo(cart *CartGrain) ([]*CartItem, bool) {
return cart.Items, true
}
func NewCartGrain(id uint64, ts time.Time) *CartGrain {
return &CartGrain{
lastItemId: 0,
lastDeliveryId: 0,
lastVoucherId: 0,
lastAccess: ts,
lastChange: ts,
TotalDiscount: NewPrice(),
Vouchers: []*Voucher{},
Deliveries: []*CartDelivery{},
Id: CartId(id),
Items: []*CartItem{},
TotalPrice: NewPrice(),
}
}
func (c *CartGrain) GetId() uint64 {
return uint64(c.Id)
}
@@ -155,22 +170,6 @@ func (c *CartGrain) GetCurrentState() (*CartGrain, error) {
return c, nil
}
func getInt(data float64, ok bool) (int, error) {
if !ok {
return 0, fmt.Errorf("invalid type")
}
return int(data), nil
}
// func (c *CartGrain) AddItem(sku string, qty int, country string, storeId *string) (*CartGrain, error) {
// cartItem, err := getItemData(sku, qty, country)
// if err != nil {
// return nil, err
// }
// cartItem.StoreId = storeId
// return c.Apply(cartItem, false)
// }
func (c *CartGrain) GetState() ([]byte, error) {
return json.Marshal(c)
}
+48
View File
@@ -0,0 +1,48 @@
package cart
import (
"git.tornberg.me/go-cart-actor/pkg/actor"
messages "git.tornberg.me/go-cart-actor/pkg/messages"
)
func NewCartMultationRegistry() actor.MutationRegistry {
reg := actor.NewMutationRegistry()
reg.RegisterMutations(
actor.NewMutation(AddItem, func() *messages.AddItem {
return &messages.AddItem{}
}),
actor.NewMutation(ChangeQuantity, func() *messages.ChangeQuantity {
return &messages.ChangeQuantity{}
}),
actor.NewMutation(RemoveItem, func() *messages.RemoveItem {
return &messages.RemoveItem{}
}),
actor.NewMutation(InitializeCheckout, func() *messages.InitializeCheckout {
return &messages.InitializeCheckout{}
}),
actor.NewMutation(OrderCreated, func() *messages.OrderCreated {
return &messages.OrderCreated{}
}),
actor.NewMutation(RemoveDelivery, func() *messages.RemoveDelivery {
return &messages.RemoveDelivery{}
}),
actor.NewMutation(SetDelivery, func() *messages.SetDelivery {
return &messages.SetDelivery{}
}),
actor.NewMutation(SetPickupPoint, func() *messages.SetPickupPoint {
return &messages.SetPickupPoint{}
}),
actor.NewMutation(ClearCart, func() *messages.ClearCartRequest {
return &messages.ClearCartRequest{}
}),
actor.NewMutation(AddVoucher, func() *messages.AddVoucher {
return &messages.AddVoucher{}
}),
actor.NewMutation(RemoveVoucher, func() *messages.RemoveVoucher {
return &messages.RemoveVoucher{}
}),
)
return reg
}
@@ -1,4 +1,4 @@
package main
package cart
import (
"testing"
+1 -1
View File
@@ -1,4 +1,4 @@
package main
package cart
import (
"crypto/rand"
@@ -1,4 +1,4 @@
package main
package cart
import (
"encoding/json"
@@ -1,4 +1,4 @@
package main
package cart
import (
"fmt"
@@ -1,4 +1,4 @@
package main
package cart
import (
"slices"
@@ -1,4 +1,4 @@
package main
package cart
import (
"fmt"
@@ -1,4 +1,4 @@
package main
package cart
import (
"fmt"
@@ -1,4 +1,4 @@
package main
package cart
import (
"fmt"
@@ -1,4 +1,4 @@
package main
package cart
import (
"fmt"
@@ -1,4 +1,4 @@
package main
package cart
import (
"fmt"
@@ -1,4 +1,4 @@
package main
package cart
import (
"fmt"
@@ -1,4 +1,4 @@
package main
package cart
import (
"fmt"
+1 -1
View File
@@ -1,4 +1,4 @@
package main
package cart
import (
"encoding/json"