eventsourcing

package module
v0.1.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 16, 2026 License: MIT Imports: 11 Imported by: 0

README

kite-mcp-eventsourcing

Go Reference

Event sourcing primitives — domain event aggregate roots (alerts, orders, positions, sessions), outbox pattern for at-least-once event delivery, read-side projections, and event store contract.

Used by Sundeepg98/kite-mcp-server to back the kc/manager state machine + the kc/usecases CQRS write side.

Why a separate module?

Event sourcing is a foundational primitive — any algo2go consumer needing the aggregate-root + outbox + projection pattern can adopt it without depending on kite-mcp-server's broker integration. Hosting as its own module:

  • Centralizes the EventStore + Aggregate + Projection contracts
  • Lets event-payload schemas version independently
  • Keeps the outbox runtime decoupled from any one broker

Stability promise

v0.x — unstable. Pin v0.1.0 deliberately.

Install

go get github.com/algo2go/kite-mcp-eventsourcing@v0.1.0

Public API

  • Aggregate — base interface for event-sourced aggregates
  • AlertAggregate, OrderAggregate, PositionAggregate, SessionAggregate — domain aggregate roots
  • Store — event store with append + load
  • Projection — read-side materialized view contract
  • Outbox — at-least-once event delivery integration

Dependencies

  • github.com/algo2go/kite-mcp-alerts v0.1.0
  • github.com/algo2go/kite-mcp-broker v0.1.0
  • github.com/algo2go/kite-mcp-domain v0.1.0
  • github.com/google/uuid — event/aggregate IDs
  • github.com/stretchr/testify — test assertions

All algo2go deps published; no upstream replace directives needed.

Reference consumer

Sundeepg98/kite-mcp-server — consumed across kc/eventing_service.go, kc/manager_init.go, kc/manager_orders_fallback.go, kc/manager_reconstitution.go, kc/manager_struct.go, kc/usecases/*, app/adapters.go, app/app.go, app/wire.go, mcp/alert_history_tool_test.go, mcp/order_history_tool_test.go, mcp/position_history_tool_test.go.

License

MIT — see LICENSE.

Authors

Original design: Sundeepg98 (Zerodha Tech). Multi-module promotion (2026-05-10): algo2go contributors.

Documentation

Overview

Package eventsourcing provides an append-only domain audit log backed by SQLite and aggregate patterns for modeling domain state transitions.

ARCHITECTURE NOTE: In production, the EventStore functions as a **domain audit log**, not a true event-sourced system. Domain events are appended by makeEventPersister (in app/app.go) whenever use cases dispatch events, but they are never read back to reconstitute state. Reads go to broker APIs (orders, positions) or CRUD stores (alerts, users). The aggregate types (OrderAggregate, PositionAggregate, AlertAggregate) and their Load*FromEvents functions exist as test infrastructure for verifying event replay correctness — they are not instantiated in production code paths.

Events are immutable once appended — no UPDATE or DELETE is ever issued. The domain_events table serves as a compliance/debugging audit trail.

Index

Constants

View Source 
const (
	AlertStatusActive    = "ACTIVE"
	AlertStatusTriggered = "TRIGGERED"
	AlertStatusDeleted   = "DELETED"
)

Alert status constants.

View Source 
const (
	OrderStatusNew       = "NEW"
	OrderStatusPlaced    = "PLACED"
	OrderStatusModified  = "MODIFIED"
	OrderStatusFilled    = "FILLED"
	OrderStatusCancelled = "CANCELLED"
)

Order status constants model the lifecycle states.

View Source 
const (
	PositionStatusOpen   = "OPEN"
	PositionStatusClosed = "CLOSED"
)

Position status constants.

View Source 
const (
	SessionStatusActive      = "ACTIVE"
	SessionStatusCleared     = "CLEARED"
	SessionStatusInvalidated = "INVALIDATED"
)

Session status constants — the small set of lifecycle states a session moves through over its lifetime. Active is the default starting state after a session.created event lands. Cleared means the broker attachment was torn down but the session record lives on. Invalidated is terminal.

Variables

This section is empty.

Functions

func MarshalPayload 

func MarshalPayload(v any) ([]byte, error)

MarshalPayload is a convenience helper that JSON-encodes a value for event payloads.

Types

type AggregateRoot 

type AggregateRoot interface {
	// AggregateID returns the unique identifier for this aggregate instance.
	AggregateID() string
	// AggregateType returns the type name (e.g., "Order", "Alert").
	AggregateType() string
	// Version returns the number of events applied to this aggregate.
	Version() int
	// Apply replays a single domain event to reconstitute state.
	Apply(event domain.Event)
	// PendingEvents returns the uncommitted events emitted by command methods.
	PendingEvents() []domain.Event
	// ClearPendingEvents removes all uncommitted events (called after persistence).
	ClearPendingEvents()
}

AggregateRoot is the interface that all event-sourced aggregates must satisfy. An aggregate reconstitutes its state by replaying events, and emits new events when commands mutate it.

ARCHITECTURE NOTE: Aggregates are currently used as test infrastructure only. In production, order/position/alert state comes from broker APIs and CRUD stores, not from event replay. The aggregates model domain invariants and lifecycle transitions, which is valuable for testing correctness of event schemas and state machine logic. They may be wired into production use cases in the future if temporal queries or multi-broker replay become requirements.

type AlertAggregate 

type AlertAggregate struct {
	BaseAggregate
	Email       string
	Instrument  domain.InstrumentKey
	TargetPrice domain.Money
	Direction   string // above, below, drop_pct, rise_pct
	Status      string // ACTIVE, TRIGGERED, DELETED
	CreatedAt   time.Time
	TriggeredAt time.Time
	DeletedAt   time.Time
}

AlertAggregate models the lifecycle of a price alert through events. State is only mutated via Apply, which processes domain events.

NOTE: Not instantiated in production. Alert state is managed by alerts.Store (CRUD). This aggregate exists for testing event replay correctness.

func LoadAlertFromEvents 

func LoadAlertFromEvents(events []StoredEvent) (*AlertAggregate, error)

LoadAlertFromEvents replays a sequence of stored events to reconstitute an AlertAggregate.

func NewAlertAggregate 

func NewAlertAggregate(id string) *AlertAggregate

NewAlertAggregate creates a new alert aggregate.

func (*AlertAggregate) AggregateType 

func (a *AlertAggregate) AggregateType() string

AggregateType returns "Alert".

func (*AlertAggregate) Apply 

func (a *AlertAggregate) Apply(event domain.Event)

Apply processes a domain event and updates aggregate state.

Both internal event types (emitted by Create/Trigger/Delete command methods) and the public domain.*Event types dispatched on the domain.EventDispatcher are handled. The latter lets the projection pipeline feed live alert events from the bus into the aggregate directly.

func (*AlertAggregate) CanDelete 

func (a *AlertAggregate) CanDelete() error

CanDelete returns an error if the alert cannot be deleted.

func (*AlertAggregate) CanTrigger 

func (a *AlertAggregate) CanTrigger() error

CanTrigger returns an error if the alert cannot be triggered.

func (*AlertAggregate) Create 

func (a *AlertAggregate) Create(email, symbol, exchange string, targetPrice float64, direction string) error

Create emits an AlertCreated event.

func (*AlertAggregate) Delete 

func (a *AlertAggregate) Delete() error

Delete emits an AlertDeleted event.

func (*AlertAggregate) Trigger 

func (a *AlertAggregate) Trigger(currentPrice float64) error

Trigger emits an AlertTriggered event.

type AlertCreatedPayload 

type AlertCreatedPayload struct {
	Email       string  `json:"email"`
	Symbol      string  `json:"symbol"`
	Exchange    string  `json:"exchange"`
	TargetPrice float64 `json:"target_price"`
	Direction   string  `json:"direction"`
}

AlertCreatedPayload is the JSON payload for an AlertCreated event.

type AlertDeletedPayload 

type AlertDeletedPayload struct{}

AlertDeletedPayload is the JSON payload for an AlertDeleted event.

type AlertTriggeredPayload 

type AlertTriggeredPayload struct {
	CurrentPrice float64 `json:"current_price"`
}

AlertTriggeredPayload is the JSON payload for an AlertTriggered event.

type BaseAggregate 

type BaseAggregate struct {
	// contains filtered or unexported fields
}

BaseAggregate provides the common bookkeeping for all aggregates: ID tracking, version counter, and pending event accumulation.

func (*BaseAggregate) AggregateID 

func (b *BaseAggregate) AggregateID() string

AggregateID returns the aggregate's unique identifier.

func (*BaseAggregate) ClearPendingEvents 

func (b *BaseAggregate) ClearPendingEvents()

ClearPendingEvents removes all uncommitted events after they have been persisted.

func (*BaseAggregate) PendingEvents 

func (b *BaseAggregate) PendingEvents() []domain.Event

PendingEvents returns all uncommitted events.

func (*BaseAggregate) Version 

func (b *BaseAggregate) Version() int

Version returns the number of events that have been applied.

type EventStore 

type EventStore struct {
	// contains filtered or unexported fields
}

EventStore provides append-only persistence for domain events backed by SQLite. In production, this acts as a domain audit log — events are appended but not read back for state reconstitution. LoadEvents and LoadEventsSince are available for debugging, compliance queries, and test infrastructure. It reuses the existing alerts.DB handle for database access.

func NewEventStore 

func NewEventStore(db *alerts.DB) *EventStore

NewEventStore creates an EventStore using the given database handle.

func (*EventStore) Append 

func (s *EventStore) Append(events ...StoredEvent) error

Append persists one or more events atomically. Each event is assigned a UUID if its ID is empty. The Sequence field must be set by the caller (typically derived from the aggregate version).

func (*EventStore) AppendToOutbox 

func (s *EventStore) AppendToOutbox(evt StoredEvent) error

AppendToOutbox inserts an event into the staging outbox. The row is processed asynchronously by OutboxPump.Drain into domain_events. Use this on hot mutation paths (place_order, modify_order, cancel_order, create_alert) where the audit-loss window must be minimised.

Failure here = caller should treat the event as not durably recorded (same as Append failing). The use case currently logs and continues; outbox semantics are preserved.

func (*EventStore) Drain 

func (s *EventStore) Drain(ctx context.Context) (int, error)

Drain processes all pending outbox rows, appending them to domain_events and marking them processed. Returns the number of rows successfully drained. Errors short-circuit the loop — partial progress is the whole point of an outbox pattern (next tick picks up where this one stopped).

ctx is honoured between rows so a long backlog doesn't tie up shutdown.

func (*EventStore) InitOutboxTable 

func (s *EventStore) InitOutboxTable() error

InitOutboxTable creates event_outbox if missing. Call once during startup alongside InitTable. Idempotent.

func (*EventStore) InitTable 

func (s *EventStore) InitTable() error

InitTable creates the domain_events table and indexes if they do not exist.

func (*EventStore) LoadEvents 

func (s *EventStore) LoadEvents(aggregateID string) ([]StoredEvent, error)

LoadEvents retrieves all events for a given aggregate, ordered by sequence.

func (*EventStore) LoadEventsByEmailHash 

func (s *EventStore) LoadEventsByEmailHash(emailHash string, since time.Time) ([]StoredEvent, error)

LoadEventsByEmailHash retrieves all events associated with the given hashed email, ordered by occurred_at. Used by the data-portability export (PR-D Item 3) to find every domain event a Data Principal accumulated over the retention window.

Empty emailHash returns no rows (defensive — never matches the empty-string default for system events).

func (*EventStore) LoadEventsSince 

func (s *EventStore) LoadEventsSince(since time.Time) ([]StoredEvent, error)

LoadEventsSince retrieves all events that occurred after the given time, ordered by occurred_at. Useful for building projections and read models.

func (*EventStore) NextSequence 

func (s *EventStore) NextSequence(aggregateID string) (int64, error)

NextSequence returns the next sequence number for an aggregate.

type OrderAggregate 

type OrderAggregate struct {
	BaseAggregate
	Status          string
	Email           string
	Instrument      domain.InstrumentKey
	TransactionType string
	OrderType       string
	Product         string
	Quantity        domain.Quantity
	Price           domain.Money
	FilledPrice     domain.Money
	FilledQuantity  domain.Quantity
	ModifyCount     int
	PlacedAt        time.Time
	ModifiedAt      time.Time
	CancelledAt     time.Time
	FilledAt        time.Time
}

OrderAggregate models the full lifecycle of a trading order through events. State is never set directly — only via Apply, which processes domain events.

NOTE: Not instantiated in production. PlaceOrderUseCase calls broker.Client directly and dispatches domain events to the audit log. This aggregate exists for testing event replay correctness and modeling order state machine invariants.

func LoadOrderFromEvents 

func LoadOrderFromEvents(events []StoredEvent) (*OrderAggregate, error)

LoadOrderFromEvents replays a sequence of stored events to reconstitute an OrderAggregate. The aggregate ID is taken from the first event.

func NewOrderAggregate 

func NewOrderAggregate(id string) *OrderAggregate

NewOrderAggregate creates a new order aggregate in the NEW state.

func (*OrderAggregate) AggregateType 

func (o *OrderAggregate) AggregateType() string

AggregateType returns "Order".

func (*OrderAggregate) Apply 

func (o *OrderAggregate) Apply(event domain.Event)

Apply processes a domain event and updates aggregate state. This is the only method that mutates fields — called during both command execution and historical replay.

Both internal event types (emitted by Place/Modify/Cancel/Fill command methods) and the public domain.*Event types dispatched on the domain.EventDispatcher by production use cases are handled. This lets the projection pipeline feed live order events from the bus into the aggregate without constructing internal events.

func (*OrderAggregate) CanCancel 

func (o *OrderAggregate) CanCancel() error

CanCancel returns an error if the order cannot be cancelled in its current state.

func (*OrderAggregate) CanFill 

func (o *OrderAggregate) CanFill() error

CanFill returns an error if the order cannot be filled in its current state.

func (*OrderAggregate) CanModify 

func (o *OrderAggregate) CanModify() error

CanModify returns an error if the order cannot be modified in its current state.

func (*OrderAggregate) Cancel 

func (o *OrderAggregate) Cancel() error

Cancel emits an OrderCancelled event after validating the order is in PLACED or MODIFIED state.

func (*OrderAggregate) Fill 

func (o *OrderAggregate) Fill(price float64, qty int) error

Fill emits an OrderFilled event after validating the order is in PLACED or MODIFIED state.

func (*OrderAggregate) Modify 

func (o *OrderAggregate) Modify(newQty int, newPrice float64, newOrderType string) error

Modify emits an OrderModified event after validating the order can be modified. At least one of newQty or newPrice must differ from current values.

func (*OrderAggregate) Place 

func (o *OrderAggregate) Place(params broker.OrderParams, email string) error

Place emits an OrderPlaced event after validating the order is in NEW state.

type OrderCancelledPayload 

type OrderCancelledPayload struct {
	Reason string `json:"reason,omitempty"`
}

OrderCancelledPayload is the JSON payload for an OrderCancelled event.

type OrderFilledPayload 

type OrderFilledPayload struct {
	FilledPrice    float64 `json:"filled_price"`
	FilledQuantity int     `json:"filled_quantity"`
}

OrderFilledPayload is the JSON payload for an OrderFilled event.

type OrderModifiedPayload 

type OrderModifiedPayload struct {
	NewQuantity  int     `json:"new_quantity,omitempty"`
	NewPrice     float64 `json:"new_price,omitempty"`
	NewOrderType string  `json:"new_order_type,omitempty"`
}

OrderModifiedPayload is the JSON payload for an OrderModified event.

type OrderPlacedPayload 

type OrderPlacedPayload struct {
	Email           string  `json:"email"`
	Exchange        string  `json:"exchange"`
	Tradingsymbol   string  `json:"tradingsymbol"`
	TransactionType string  `json:"transaction_type"`
	OrderType       string  `json:"order_type"`
	Product         string  `json:"product"`
	Quantity        int     `json:"quantity"`
	Price           float64 `json:"price"`
}

OrderPlacedPayload is the JSON payload for an OrderPlaced event.

type OutboxPump 

type OutboxPump struct {
	// contains filtered or unexported fields
}

OutboxPump asynchronously drains event_outbox into domain_events. One pump per EventStore. Stop() must be called at shutdown so the goroutine terminates cleanly.

func NewOutboxPump 

func NewOutboxPump(store *EventStore, logger *slog.Logger) *OutboxPump

NewOutboxPump creates and starts a pump for the given store. The pump runs an immediate Drain() then enters a polling loop.

func (*OutboxPump) Stop 

func (p *OutboxPump) Stop()

Stop signals the pump to exit. Idempotent. Returns after the goroutine has confirmed shutdown so callers can rely on no further DB activity from this pump after Stop returns.

type PositionAggregate 

type PositionAggregate struct {
	BaseAggregate
	Email           string
	Instrument      domain.InstrumentKey
	Product         string // MIS, CNC, NRML — part of natural aggregate key
	TransactionType string // original direction: BUY or SELL
	Quantity        domain.Quantity
	AvgPrice        domain.Money
	Status          string // OPEN, CLOSED
	OpenedAt        time.Time
	ClosedAt        time.Time
}

PositionAggregate models the lifecycle of a trading position through events. State is only mutated via Apply, which processes domain events.

NOTE: Not instantiated in production. Position state comes from broker API (GetPositions). This aggregate exists for testing event replay correctness.

func LoadPositionFromEvents 

func LoadPositionFromEvents(events []StoredEvent) (*PositionAggregate, error)

LoadPositionFromEvents replays a sequence of stored events to reconstitute a PositionAggregate.

func NewPositionAggregate 

func NewPositionAggregate(id string) *PositionAggregate

NewPositionAggregate creates a new position aggregate in the OPEN state.

func (*PositionAggregate) AggregateType 

func (p *PositionAggregate) AggregateType() string

AggregateType returns "Position".

func (*PositionAggregate) Apply 

func (p *PositionAggregate) Apply(event domain.Event)

Apply processes a domain event and updates aggregate state.

Both internal event types (emitted by Open/Close command methods) and the public domain.*Event types dispatched on the domain.EventDispatcher are handled. The latter lets the projection pipeline feed live position events from the bus into the aggregate directly.

func (*PositionAggregate) CanClose 

func (p *PositionAggregate) CanClose() error

CanClose returns an error if the position cannot be closed in its current state.

func (*PositionAggregate) Close 

func (p *PositionAggregate) Close(closeOrderID, closeTxnType string) error

Close emits a PositionClosed event after validating the position can be closed.

func (*PositionAggregate) Open 

func (p *PositionAggregate) Open(email, symbol, exchange, txnType string, qty int, avgPrice float64) error

Open emits a PositionOpened event.

type PositionClosedPayload 

type PositionClosedPayload struct {
	CloseOrderID    string `json:"close_order_id"`
	TransactionType string `json:"transaction_type"` // opposite direction
}

PositionClosedPayload is the JSON payload for a PositionClosed event.

type PositionOpenedPayload 

type PositionOpenedPayload struct {
	Email           string  `json:"email"`
	Symbol          string  `json:"symbol"`
	Exchange        string  `json:"exchange"`
	TransactionType string  `json:"transaction_type"`
	Quantity        int     `json:"quantity"`
	AvgPrice        float64 `json:"avg_price"`
}

PositionOpenedPayload is the JSON payload for a PositionOpened event.

type Projector 

type Projector struct {
	// contains filtered or unexported fields
}

Projector is a read-side projection that maintains live OrderAggregate, AlertAggregate, and PositionAggregate snapshots by subscribing to the domain.EventDispatcher. Each incoming public domain event is replayed into the matching aggregate via Apply.

This is how the three aggregates are kept wired into production: a use case dispatches (e.g.) domain.OrderPlacedEvent on the shared dispatcher, the projector Subscribe handler looks up (or constructs) the aggregate for the event's aggregate ID, and calls aggregate.Apply(event) to update state.

The projection is in-process, synchronous, and cleared on restart. It is a read-side only: it does not persist events (that is done by the audit log subscribers in app/wire.go) and does not replay from history.

func NewProjector 

func NewProjector() *Projector

NewProjector constructs an empty Projector with initialized maps.

func (*Projector) AlertCount 

func (p *Projector) AlertCount() int

AlertCount returns the total number of projected alerts.

func (*Projector) GetAlert 

func (p *Projector) GetAlert(id string) (*AlertAggregate, bool)

GetAlert returns a snapshot of the projected alert aggregate with the given alert ID.

func (*Projector) GetOrder 

func (p *Projector) GetOrder(id string) (*OrderAggregate, bool)

GetOrder returns a snapshot of the projected order aggregate with the given order ID. The bool result is false if the projector has never seen an event for that order.

func (*Projector) GetPosition 

func (p *Projector) GetPosition(id string) (*PositionAggregate, bool)

GetPosition returns a snapshot of the projected position aggregate with the given ID (PositionID for opens, OrderID for closes).

func (*Projector) ListActiveAlerts 

func (p *Projector) ListActiveAlerts() []*AlertAggregate

ListActiveAlerts returns all projected alerts currently in ACTIVE state, sorted by created-at (oldest first).

func (*Projector) ListActiveOrders 

func (p *Projector) ListActiveOrders() []*OrderAggregate

ListActiveOrders returns all projected orders currently in PLACED or MODIFIED state, sorted by placed-at (oldest first).

func (*Projector) ListOrdersForEmail 

func (p *Projector) ListOrdersForEmail(email string) []*OrderAggregate

ListOrdersForEmail returns all projected orders belonging to the given email regardless of status (PLACED, MODIFIED, FILLED, CANCELLED), sorted by placed-at (oldest first). Intended for the optimistic-fallback path in GetOrdersQuery when Kite is rate-limited or unavailable — callers get the full current-day lifecycle, not just active ones.

func (*Projector) OrderCount 

func (p *Projector) OrderCount() int

OrderCount returns the total number of projected orders.

func (*Projector) PositionCount 

func (p *Projector) PositionCount() int

PositionCount returns the total number of projected positions.

func (*Projector) Subscribe 

func (p *Projector) Subscribe(d *domain.EventDispatcher)

Subscribe registers handlers on the dispatcher for every order/alert/ position event type. Safe to call once per projector.

type SessionAggregate 

type SessionAggregate struct {
	BaseAggregate
	Email              string
	Broker             string
	Status             string // ACTIVE, CLEARED, INVALIDATED
	CreatedAt          time.Time
	LastClearedAt      time.Time // reset each time session.cleared lands
	InvalidatedAt      time.Time
	LastClearReason    string
	InvalidationReason string
}

SessionAggregate models the lifecycle of an MCP session through events. State is only mutated via Apply, which processes domain events.

NOTE: Not instantiated in production — session state lives in SessionRegistry (in-memory) with SQLite restore. The aggregate exists for testing event replay correctness and for compliance queries that walk session.created / session.cleared / session.invalidated streams.

func LoadSessionFromEvents 

func LoadSessionFromEvents(events []StoredEvent) (*SessionAggregate, error)

LoadSessionFromEvents replays a sequence of stored events to reconstitute a SessionAggregate. Returns an error if the stream is empty or an event type is unknown — the caller can treat that as a corrupted audit log.

func NewSessionAggregate 

func NewSessionAggregate(sessionID string) *SessionAggregate

NewSessionAggregate creates a new session aggregate keyed by sessionID.

func (*SessionAggregate) AggregateType 

func (a *SessionAggregate) AggregateType() string

AggregateType returns "Session".

func (*SessionAggregate) Apply 

func (a *SessionAggregate) Apply(event domain.Event)

Apply processes a domain event and updates aggregate state. Both the public domain.Session*Event types (dispatched on the bus and persisted via makeEventPersister) and any internal event types are handled.

type SessionClearedPayload 

type SessionClearedPayload struct {
	SessionID string `json:"session_id"`
	Reason    string `json:"reason"`
}

SessionClearedPayload is the JSON payload for a session.cleared event.

type SessionCreatedPayload 

type SessionCreatedPayload struct {
	Email     string `json:"email"`
	SessionID string `json:"session_id"`
	Broker    string `json:"broker"`
}

SessionCreatedPayload is the JSON payload for a session.created event when persisted by the use case path (not makeEventPersister).

type SessionInvalidatedPayload 

type SessionInvalidatedPayload struct {
	SessionID string `json:"session_id"`
	Reason    string `json:"reason"`
}

SessionInvalidatedPayload is the JSON payload for a session.invalidated event.

type StoredEvent 

type StoredEvent struct {
	ID            string    `json:"id"`
	AggregateID   string    `json:"aggregate_id"`   // e.g., order ID
	AggregateType string    `json:"aggregate_type"` // "Order", "Alert", etc.
	EventType     string    `json:"event_type"`     // "OrderPlaced", "OrderCancelled", etc.
	Payload       []byte    `json:"payload"`        // JSON-serialized event data
	OccurredAt    time.Time `json:"occurred_at"`
	Sequence      int64     `json:"sequence"`             // auto-incremented per aggregate
	EmailHash     string    `json:"email_hash,omitempty"` // SHA-256 hex of lowercased email; empty for system events
}

StoredEvent is the persistent representation of a domain event. All events are stored in a single table, keyed by aggregate ID and type.

EmailHash is the SHA-256 hex digest of the lowercased email associated with the event (PR-D Item 2). Empty for events with no user-association (system events, alert.triggered without recipient context). The audit log uses email_hash as the canonical user identifier so the domain_events table never carries plaintext PII — DPDP minimisation principle.

func ToAlertStoredEvents 

func ToAlertStoredEvents(agg *AlertAggregate, startSequence int64) ([]StoredEvent, error)

ToAlertStoredEvents converts pending domain events from an AlertAggregate into StoredEvents ready for persistence.

func ToPositionStoredEvents 

func ToPositionStoredEvents(agg *PositionAggregate, startSequence int64) ([]StoredEvent, error)

ToPositionStoredEvents converts pending domain events from a PositionAggregate into StoredEvents ready for persistence.

func ToStoredEvents 

func ToStoredEvents(agg *OrderAggregate, startSequence int64) ([]StoredEvent, error)

ToStoredEvents converts pending domain events from an OrderAggregate into StoredEvents ready for persistence. startSequence is the first sequence number to assign.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.