From b16820bf626935c837cb424b014cd28affba0a09 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mats=20T=C3=B6rnberg?= Date: Wed, 26 Aug 2026 16:42:05 +0200 Subject: [PATCH] Bump go-cart-actor: scheduling cluster service + booking-hold cart/checkout seam MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deploys cmd/scheduling as a 3-replica actor cluster (shared RWX commerce-data event logs, gRPC mesh via POD_IP + actor-pool label, edge /api/scheduling pass-through, fail-closed mcpauth on admin/MCP). cart/checkout get SCHEDULING_SERVICE_URL so line removal releases holds; backoffice gets SCHEDULING_URL for the admin proxy. Platform seam: order.Line.Properties carries booking_* fields through order.created; scheduling.booking_confirmed/ cancelled event types. Compose + prep-dev-data wire the schedules dir. 🤖 Generated with Codebuff Co-Authored-By: Codebuff --- event/event.go | 4 + order/order.go | 25 +++-- scheduling/client.go | 221 +++++++++++++++++++++++++++++++++++++++++++ scheduling/policy.go | 145 ++++++++++++++++++++++++++++ scheduling/types.go | 157 ++++++++++++++++++++++++++++++ 5 files changed, 542 insertions(+), 10 deletions(-) create mode 100644 scheduling/client.go create mode 100644 scheduling/policy.go create mode 100644 scheduling/types.go diff --git a/event/event.go b/event/event.go index da4e6a2..9d90b12 100644 --- a/event/event.go +++ b/event/event.go @@ -88,6 +88,10 @@ const ( // credit notes (invoice.credited), so RefundIssued never fires for them. RefundIssued Type = "refund.issued" + // Scheduling (go-cart-actor cmd/scheduling). Emitted when timeslots are + // confirmed from an order capture or cancelled. + SchedulingBookingConfirmed Type = "scheduling.booking_confirmed" + SchedulingBookingCancelled Type = "scheduling.booking_cancelled" // Analytics (tracking). Replaces the integer enum 0–6 that finder and // slask-tracking use on the `tracking` topic today. Producer: slask-finder diff --git a/order/order.go b/order/order.go index 1d71f54..b5f0023 100644 --- a/order/order.go +++ b/order/order.go @@ -28,6 +28,12 @@ type Line struct { // DropShip marks items fulfilled directly by the supplier. The inventory // reactor must skip the stock decrement for these lines. DropShip bool `json:"drop_ship,omitempty"` + // Properties carries per-line domain attributes a reactor needs that don't + // fit the fixed fields — e.g. booking_hold_id / booking_start / booking_end + // for the scheduling service's order.created commit reactor (wire-safe + // addition, same precedent as Created.CheckoutID). Optional; reactors must + // tolerate absent properties on every line. + Properties map[string]string `json:"properties,omitempty"` } // Created is the payload of an event.OrderCreated envelope. @@ -40,15 +46,15 @@ type Created struct { OrderReference string `json:"order_reference,omitempty"` CartID string `json:"cart_id,omitempty"` // CheckoutID links the originating checkout grain (wire-safe addition). - CheckoutID string `json:"checkout_id,omitempty"` - Country string `json:"country"` - Currency string `json:"currency,omitempty"` - CustomerEmail string `json:"customer_email,omitempty"` + CheckoutID string `json:"checkout_id,omitempty"` + Country string `json:"country"` + Currency string `json:"currency,omitempty"` + CustomerEmail string `json:"customer_email,omitempty"` // CustomerID is the customer profile ID for identity linking (B0.4). Reactors // (e.g. profile service) use this to link the order to the customer profile. - CustomerID string `json:"customer_id,omitempty"` - Lines []Line `json:"lines"` - PlacedAtMs int64 `json:"placed_at_ms,omitempty"` + CustomerID string `json:"customer_id,omitempty"` + Lines []Line `json:"lines"` + PlacedAtMs int64 `json:"placed_at_ms,omitempty"` } // Encode serialises the payload for event.Event.Payload. @@ -122,8 +128,8 @@ func DecodeInvoicePaid(raw []byte) (InvoicePaid, error) { // (1510 C / 304x D / 261x D per rate), idempotent per credit number. type InvoiceCredited struct { OrderID string `json:"order_id"` - InvoiceNumber string `json:"invoice_number"` // the original invoice - CreditNumber string `json:"credit_number"` // from the same gap-free series + InvoiceNumber string `json:"invoice_number"` // the original invoice + CreditNumber string `json:"credit_number"` // from the same gap-free series CompanyID string `json:"company_id,omitempty"` OcrReference string `json:"ocr_reference,omitempty"` AmountIncVat int64 `json:"amount_inc_vat"` @@ -186,4 +192,3 @@ func DecodeRefundIssued(raw []byte) (RefundIssued, error) { err := json.Unmarshal(raw, &r) return r, err } - diff --git a/scheduling/client.go b/scheduling/client.go new file mode 100644 index 0000000..7198e1e --- /dev/null +++ b/scheduling/client.go @@ -0,0 +1,221 @@ +package scheduling + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +// HTTPSchedulingPolicy implements SchedulingPolicy by calling the scheduling service over HTTP. +type HTTPSchedulingPolicy struct { + baseURL string + client *http.Client +} + +// NewHTTPSchedulingPolicy creates a client against the scheduling HTTP endpoint (e.g. "http://scheduling:8080"). +func NewHTTPSchedulingPolicy(baseURL string, client *http.Client) *HTTPSchedulingPolicy { + if client == nil { + client = &http.Client{Timeout: 5 * time.Second} + } + return &HTTPSchedulingPolicy{ + baseURL: strings.TrimRight(baseURL, "/"), + client: client, + } +} + +func (c *HTTPSchedulingPolicy) GetAvailability(ctx context.Context, resourceID string, from, to time.Time, durationMinutes int) ([]Slot, error) { + u := fmt.Sprintf("%s/api/scheduling/resources/%s/availability?from=%s&to=%s", + c.baseURL, + url.PathEscape(resourceID), + url.QueryEscape(from.Format(time.RFC3339)), + url.QueryEscape(to.Format(time.RFC3339)), + ) + if durationMinutes > 0 { + u += "&duration=" + strconv.Itoa(durationMinutes) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return nil, err + } + req.Header.Set("Accept", "application/json") + + resp, err := c.client.Do(req) + if err != nil { + return nil, fmt.Errorf("scheduling client: GetAvailability failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusNotFound { + return nil, ErrResourceNotFound + } + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("scheduling client: unexpected status %d: %s", resp.StatusCode, string(body)) + } + + var slots []Slot + if err := json.NewDecoder(resp.Body).Decode(&slots); err != nil { + return nil, fmt.Errorf("scheduling client: decode slots: %w", err) + } + return slots, nil +} + +func (c *HTTPSchedulingPolicy) HoldSlot(ctx context.Context, reqData HoldRequest) (HoldResult, error) { + data, err := json.Marshal(reqData) + if err != nil { + return HoldResult{}, err + } + + u := fmt.Sprintf("%s/api/scheduling/holds", c.baseURL) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewReader(data)) + if err != nil { + return HoldResult{}, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + + resp, err := c.client.Do(req) + if err != nil { + return HoldResult{}, fmt.Errorf("scheduling client: HoldSlot failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusConflict { + return HoldResult{}, ErrSlotUnavailable + } + if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return HoldResult{}, fmt.Errorf("scheduling client: HoldSlot status %d: %s", resp.StatusCode, string(body)) + } + + var res HoldResult + if err := json.NewDecoder(resp.Body).Decode(&res); err != nil { + return HoldResult{}, fmt.Errorf("scheduling client: decode hold result: %w", err) + } + return res, nil +} + +func (c *HTTPSchedulingPolicy) ReleaseHold(ctx context.Context, holdID string) error { + u := fmt.Sprintf("%s/api/scheduling/holds/%s/release", c.baseURL, url.PathEscape(holdID)) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, nil) + if err != nil { + return err + } + + resp, err := c.client.Do(req) + if err != nil { + return fmt.Errorf("scheduling client: ReleaseHold failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotFound { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("scheduling client: ReleaseHold status %d: %s", resp.StatusCode, string(body)) + } + return nil +} + +func (c *HTTPSchedulingPolicy) ExtendHold(ctx context.Context, holdID string, extensionSeconds int) (HoldResult, error) { + payload := map[string]int{"extension_seconds": extensionSeconds} + data, _ := json.Marshal(payload) + + u := fmt.Sprintf("%s/api/scheduling/holds/%s/extend", c.baseURL, url.PathEscape(holdID)) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewReader(data)) + if err != nil { + return HoldResult{}, err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.client.Do(req) + if err != nil { + return HoldResult{}, fmt.Errorf("scheduling client: ExtendHold failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone { + return HoldResult{}, ErrHoldExpired + } + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return HoldResult{}, fmt.Errorf("scheduling client: ExtendHold status %d: %s", resp.StatusCode, string(body)) + } + + var res HoldResult + if err := json.NewDecoder(resp.Body).Decode(&res); err != nil { + return HoldResult{}, fmt.Errorf("scheduling client: decode extend result: %w", err) + } + return res, nil +} + +func (c *HTTPSchedulingPolicy) ConfirmBooking(ctx context.Context, holdID, orderID, customerEmail, customerName, notes string) (BookingDetails, error) { + payload := map[string]string{ + "hold_id": holdID, + "order_id": orderID, + "customer_email": customerEmail, + "customer_name": customerName, + "notes": notes, + } + data, _ := json.Marshal(payload) + + u := fmt.Sprintf("%s/api/scheduling/bookings", c.baseURL) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewReader(data)) + if err != nil { + return BookingDetails{}, err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.client.Do(req) + if err != nil { + return BookingDetails{}, fmt.Errorf("scheduling client: ConfirmBooking failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusGone || resp.StatusCode == http.StatusConflict { + return BookingDetails{}, ErrHoldExpired + } + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + body, _ := io.ReadAll(resp.Body) + return BookingDetails{}, fmt.Errorf("scheduling client: ConfirmBooking status %d: %s", resp.StatusCode, string(body)) + } + + var details BookingDetails + if err := json.NewDecoder(resp.Body).Decode(&details); err != nil { + return BookingDetails{}, fmt.Errorf("scheduling client: decode booking details: %w", err) + } + return details, nil +} + +func (c *HTTPSchedulingPolicy) CancelBooking(ctx context.Context, bookingID, reason string) error { + payload := map[string]string{"reason": reason} + data, _ := json.Marshal(payload) + + u := fmt.Sprintf("%s/api/scheduling/bookings/%s/cancel", c.baseURL, url.PathEscape(bookingID)) + req, err := http.NewRequestWithContext(ctx, http.MethodPatch, u, bytes.NewReader(data)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.client.Do(req) + if err != nil { + return fmt.Errorf("scheduling client: CancelBooking failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusNotFound { + return ErrBookingNotFound + } + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("scheduling client: CancelBooking status %d: %s", resp.StatusCode, string(body)) + } + return nil +} diff --git a/scheduling/policy.go b/scheduling/policy.go new file mode 100644 index 0000000..ba50990 --- /dev/null +++ b/scheduling/policy.go @@ -0,0 +1,145 @@ +package scheduling + +import ( + "context" + "sync" + "time" +) + +// SchedulingPolicy is the in-process interface seam for scheduling operations. +// Callers (cart, checkout, backoffice) depend on this interface rather than +// concrete network transports or internal storage. +type SchedulingPolicy interface { + // GetAvailability queries available slots for a resource between from and to. + GetAvailability(ctx context.Context, resourceID string, from, to time.Time, durationMinutes int) ([]Slot, error) + + // HoldSlot requests a temporary TTL hold on a slot. + HoldSlot(ctx context.Context, req HoldRequest) (HoldResult, error) + + // ReleaseHold explicitly cancels a temporary hold. + ReleaseHold(ctx context.Context, holdID string) error + + // ExtendHold extends the TTL on an active hold. + ExtendHold(ctx context.Context, holdID string, extensionSeconds int) (HoldResult, error) + + // ConfirmBooking confirms a booking from an existing hold. + ConfirmBooking(ctx context.Context, holdID string, orderID string, customerEmail, customerName, notes string) (BookingDetails, error) + + // CancelBooking cancels a confirmed booking. + CancelBooking(ctx context.Context, bookingID, reason string) error +} + +// InMemorySchedulingPolicy is an in-memory test double for unit testing without network dependencies. +type InMemorySchedulingPolicy struct { + mu sync.RWMutex + holds map[string]HoldResult + bookings map[string]BookingDetails + availability map[string][]Slot +} + +// NewInMemorySchedulingPolicy creates a test double. +func NewInMemorySchedulingPolicy() *InMemorySchedulingPolicy { + return &InMemorySchedulingPolicy{ + holds: make(map[string]HoldResult), + bookings: make(map[string]BookingDetails), + availability: make(map[string][]Slot), + } +} + +func (m *InMemorySchedulingPolicy) SetSlots(resourceID string, slots []Slot) { + m.mu.Lock() + defer m.mu.Unlock() + m.availability[resourceID] = slots +} + +func (m *InMemorySchedulingPolicy) GetAvailability(ctx context.Context, resourceID string, from, to time.Time, durationMinutes int) ([]Slot, error) { + m.mu.RLock() + defer m.mu.RUnlock() + slots, ok := m.availability[resourceID] + if !ok { + return []Slot{}, nil + } + var res []Slot + for _, s := range slots { + if !s.StartTime.Before(from) && !s.EndTime.After(to) && s.Remaining > 0 { + res = append(res, s) + } + } + return res, nil +} + +func (m *InMemorySchedulingPolicy) HoldSlot(ctx context.Context, req HoldRequest) (HoldResult, error) { + m.mu.Lock() + defer m.mu.Unlock() + ttl := req.TTLSeconds + if ttl <= 0 { + ttl = 900 + } + hold := HoldResult{ + HoldID: "hold_" + time.Now().Format("20060102150405.000000"), + ResourceID: req.ResourceID, + StartTime: req.StartTime, + EndTime: req.EndTime, + Quantity: req.Quantity, + ExpiresAt: time.Now().Add(time.Duration(ttl) * time.Second), + } + m.holds[hold.HoldID] = hold + return hold, nil +} + +func (m *InMemorySchedulingPolicy) ReleaseHold(ctx context.Context, holdID string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.holds, holdID) + return nil +} + +func (m *InMemorySchedulingPolicy) ExtendHold(ctx context.Context, holdID string, extensionSeconds int) (HoldResult, error) { + m.mu.Lock() + defer m.mu.Unlock() + h, ok := m.holds[holdID] + if !ok { + return HoldResult{}, ErrHoldExpired + } + h.ExpiresAt = h.ExpiresAt.Add(time.Duration(extensionSeconds) * time.Second) + m.holds[holdID] = h + return h, nil +} + +func (m *InMemorySchedulingPolicy) ConfirmBooking(ctx context.Context, holdID string, orderID string, customerEmail, customerName, notes string) (BookingDetails, error) { + m.mu.Lock() + defer m.mu.Unlock() + h, ok := m.holds[holdID] + if !ok { + return BookingDetails{}, ErrHoldExpired + } + delete(m.holds, holdID) + b := BookingDetails{ + BookingID: "bk_" + time.Now().Format("20060102150405.000000"), + HoldID: holdID, + OrderID: orderID, + ResourceID: h.ResourceID, + StartTime: h.StartTime, + EndTime: h.EndTime, + Quantity: h.Quantity, + CustomerEmail: customerEmail, + CustomerName: customerName, + Notes: notes, + Status: BookingStatusConfirmed, + CreatedAt: time.Now(), + } + m.bookings[b.BookingID] = b + return b, nil +} + +func (m *InMemorySchedulingPolicy) CancelBooking(ctx context.Context, bookingID, reason string) error { + m.mu.Lock() + defer m.mu.Unlock() + b, ok := m.bookings[bookingID] + if !ok { + return ErrBookingNotFound + } + b.Status = BookingStatusCancelled + m.bookings[bookingID] = b + return nil +} diff --git a/scheduling/types.go b/scheduling/types.go new file mode 100644 index 0000000..76a7b3c --- /dev/null +++ b/scheduling/types.go @@ -0,0 +1,157 @@ +// Package scheduling defines the platform-wide scheduling and timeslot seam: +// wire shapes, DTOs, domain events, and client interfaces for booking services, +// appointments, and capacity-based timeslots. +// +// Like platform/inventory and platform/order, this package depends only on the +// stdlib and shared platform types (money, uid, event). +package scheduling + +import ( + "encoding/json" + "errors" + "time" +) + +var ( + // ErrSlotUnavailable is returned when a requested slot has no remaining capacity. + ErrSlotUnavailable = errors.New("scheduling: requested timeslot is not available") + // ErrHoldExpired is returned when attempting to confirm or extend an expired hold. + ErrHoldExpired = errors.New("scheduling: hold has expired") + // ErrBookingNotFound is returned when the specified booking is not found. + ErrBookingNotFound = errors.New("scheduling: booking not found") + // ErrResourceNotFound is returned when the specified resource or schedule is not found. + ErrResourceNotFound = errors.New("scheduling: resource not found") + // ErrOutsideWorkingHours is returned when a hold/booking window does not + // lie inside the resource's configured working hours (weekly rules or a + // date override). Guards the hold/direct-booking/reschedule appliers so + // API clients cannot reserve capacity outside bookable windows. + ErrOutsideWorkingHours = errors.New("scheduling: window is outside working hours") +) + +// Slot represents a discrete bookable window of time. +type Slot struct { + StartTime time.Time `json:"start_time"` + EndTime time.Time `json:"end_time"` + Capacity int `json:"capacity"` + Remaining int `json:"remaining"` +} + +// ResourceConfig defines a bookable resource (staff member, room, bay, equipment). +type ResourceConfig struct { + ID string `json:"id"` + Name string `json:"name"` + LocationID string `json:"location_id,omitempty"` + Timezone string `json:"timezone"` + Capacity int `json:"capacity"` // max concurrent bookings per slot (default 1) + SlotDuration time.Duration `json:"slot_duration"` // standard duration if not overridden + BufferBefore time.Duration `json:"buffer_before"` // buffer time before appointments + BufferAfter time.Duration `json:"buffer_after"` // buffer time after appointments + MinLeadTime time.Duration `json:"min_lead_time"` // minimum advance notice required + MaxAdvanceTime time.Duration `json:"max_advance_time"` // maximum booking window into the future +} + +// WorkingHours defines recurring weekly schedule rules (e.g. Mon-Fri 09:00-17:00). +type WorkingHours struct { + DayOfWeek int `json:"day_of_week"` // 0 = Sunday, 1 = Monday, ..., 6 = Saturday + StartTime string `json:"start_time"` // "09:00" (HH:MM) + EndTime string `json:"end_time"` // "17:00" (HH:MM) +} + +// ScheduleOverride defines a specific blackout (holiday, sick leave) or custom opening. +type ScheduleOverride struct { + Date string `json:"date"` // "2026-09-01" (YYYY-MM-DD) + IsClosed bool `json:"is_closed"` // true for complete closure / blackout + CustomStart string `json:"custom_start,omitempty"` // custom opening if not closed + CustomEnd string `json:"custom_end,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// HoldRequest is the payload to request a temporary TTL reservation on a slot. +type HoldRequest struct { + ResourceID string `json:"resource_id"` + StartTime time.Time `json:"start_time"` + EndTime time.Time `json:"end_time"` + Quantity int `json:"quantity"` // number of capacity units (default 1) + CartID string `json:"cart_id,omitempty"` + TTLSeconds int `json:"ttl_seconds,omitempty"` // hold duration in seconds (default 900 = 15m) +} + +// HoldResult is returned when a slot hold is successfully placed. +type HoldResult struct { + HoldID string `json:"hold_id"` + ResourceID string `json:"resource_id"` + StartTime time.Time `json:"start_time"` + EndTime time.Time `json:"end_time"` + Quantity int `json:"quantity"` + ExpiresAt time.Time `json:"expires_at"` +} + +// BookingStatus represents the lifecycle state of a booking. +type BookingStatus string + +const ( + BookingStatusConfirmed BookingStatus = "confirmed" + BookingStatusCancelled BookingStatus = "cancelled" + BookingStatusCompleted BookingStatus = "completed" + BookingStatusNoShow BookingStatus = "no_show" +) + +// BookingDetails holds a confirmed booking record. +type BookingDetails struct { + BookingID string `json:"booking_id"` + HoldID string `json:"hold_id,omitempty"` + OrderID string `json:"order_id,omitempty"` + ResourceID string `json:"resource_id"` + ResourceName string `json:"resource_name,omitempty"` + SKU string `json:"sku,omitempty"` + ServiceName string `json:"service_name,omitempty"` + StartTime time.Time `json:"start_time"` + EndTime time.Time `json:"end_time"` + Quantity int `json:"quantity"` + CustomerEmail string `json:"customer_email,omitempty"` + CustomerName string `json:"customer_name,omitempty"` + CustomerPhone string `json:"customer_phone,omitempty"` + Notes string `json:"notes,omitempty"` + Status BookingStatus `json:"status"` + CreatedAt time.Time `json:"created_at"` +} + +// BookingConfirmedEvent is the payload for event.SchedulingBookingConfirmed. +type BookingConfirmedEvent struct { + BookingID string `json:"booking_id"` + OrderID string `json:"order_id"` + ResourceID string `json:"resource_id"` + SKU string `json:"sku,omitempty"` + StartTime time.Time `json:"start_time"` + EndTime time.Time `json:"end_time"` + Quantity int `json:"quantity"` + CustomerEmail string `json:"customer_email,omitempty"` + CustomerName string `json:"customer_name,omitempty"` +} + +// Encode serializes the event payload. +func (e BookingConfirmedEvent) Encode() ([]byte, error) { return json.Marshal(e) } + +// DecodeBookingConfirmed parses the event payload. +func DecodeBookingConfirmed(raw []byte) (BookingConfirmedEvent, error) { + var e BookingConfirmedEvent + err := json.Unmarshal(raw, &e) + return e, err +} + +// BookingCancelledEvent is the payload for event.SchedulingBookingCancelled. +type BookingCancelledEvent struct { + BookingID string `json:"booking_id"` + OrderID string `json:"order_id,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// Encode serializes the event payload. +func (e BookingCancelledEvent) Encode() ([]byte, error) { return json.Marshal(e) } + +// DecodeBookingCancelled parses the event payload. +func DecodeBookingCancelled(raw []byte) (BookingCancelledEvent, error) { + var e BookingCancelledEvent + err := json.Unmarshal(raw, &e) + return e, err +}