Chronicle

An event.AnyTransformer is a type alias for event.Transformer[event.Any]. It is used to create generic transformers that can operate on events from any aggregate, rather than being tied to a specific event type. This is ideal for cross-cutting concerns like global logging, auditing, or telemetry that should apply to all events in the system.

To implement one, you must satisfy the TransformForWrite and TransformForRead methods using the event.Any interface.

type LoggingTransformer struct{}

func (t *LoggingTransformer) TransformForWrite(
	_ context.Context,
	events []event.Any,
) ([]event.Any, error) {
	for _, event := range events {
		fmt.Printf("[LOG] Writing event: %s\n", event.EventName())
	}
	return events, nil
}

func (t *LoggingTransformer) TransformForRead(
	_ context.Context,
	events []event.Any,
) ([]event.Any, error) {
	for _, event := range events {
		fmt.Printf("[LOG] Reading event: %s\n", event.EventName())
	}
	return events, nil
}

Chronicle uses Optimistic Concurrency Control to prevent race conditions in event-sourced systems. This is managed automatically via the versioning system built into aggregate.Base, which is the struct you embed in your aggregates.

The Workflow

1. Versioning: Every aggregate has a version number. When you perform actions (like Open or DepositMoney), the version increments with each event.
2. Loading: When you load an aggregate using a repository (e.g., accountRepo.Get), you receive a copy of the aggregate at its current version.
3. Concurrent Updates: If two users load the same aggregate at the same version (e.g., Version 2), they both hold a copy that expects the next version to be 3.
4. Conflict Detection: The first user to call repo.Save() succeeds and increments the version in the database (e.g., to Version 3). When the second user calls repo.Save(), Chronicle detects that the version in the database (3) does not match the version the user's copy expected (2), and it rejects the update.

This ensures that business logic is never applied to stale data without the system being aware of the conflict.

// Example of the versioning lifecycle
acc, _ := account.Open(accID, time.Now(), "John Smith")
_ = acc.DepositMoney(200) 

// After saving, the version increments based on the number of events applied
_, _, _ = accountRepo.Save(ctx, acc)
fmt.Printf("Initial account saved. Balance: 200, Version: %d\n", acc.Version())

Chronicle distinguishes between per-aggregate ordering and system-wide ordering through two primary interfaces. Choosing the right one depends on your projection requirements and the capabilities of your storage backend.

event.Log (Per-Aggregate Ordering)

• Guarantee: Events for a single aggregate (e.g., a specific AccountID) are stored and retrieved in the exact order they occurred.
• Use Case: Rehydrating an aggregate's state or building projections that only process one stream of events at a time.

event.GlobalLog (System-Wide Ordering)

• Guarantee: Provides a single, globally ordered stream of all events across the entire system.
• Mechanism: Backends like Postgres and Sqlite implement this by assigning a unique, monotonically increasing global_version to every committed event, alongside the per-aggregate version.
• Use Case: Building system-wide audit trails, projections that aggregate data from multiple different aggregate types, or projections requiring cross-aggregate data (e.g., calculating total money withdrawn across all Account aggregates).

Backend Considerations

• SQL Databases (Postgres, Sqlite): Support event.GlobalLog efficiently using IDENTITY or AUTOINCREMENT primary keys.
• Distributed Systems (e.g., Kafka): Typically only support the base event.Log because they only guarantee strict ordering within a specific partition, not globally across the system.
• Key-Value Stores: Can implement GlobalLog via secondary indexes, but may require global locks during writes to safely assign global_version, which can impact write performance under high concurrency.

An aggregate is the central entity in Chronicle. To implement one, follow these requirements:

1. Embed aggregate.Base: This handles versioning automatically.
2. Implement an ID: Declare a type for the ID that implements fmt.Stringer and add an ID() method to the aggregate.
3. Define Events: Use a sum type (interface) for events. Each event must implement event.Any (by providing an EventName() string method) and a private marker method to satisfy the sum type (e.g., isAccountEvent()).
4. Bind Events: Implement EventFuncs() event.FuncsFor[T] to provide constructor functions for each event type, allowing the library to decode them from the log.
5. Implement Apply: Define an Apply(evt T) error method to recompute the aggregate's state by replaying events. This method should only update state and should not contain business logic.

Example implementation:

type AccountID string
func (a AccountID) String() string { return string(a) }

type Account struct {
	aggregate.Base
	id AccountID
	openedAt   time.Time
	balance    int
	holderName string
}

func (a *Account) ID() AccountID { return a.id }

func (a *Account) EventFuncs() event.FuncsFor[AccountEvent] {
	return event.FuncsFor[AccountEvent]{
		func() AccountEvent { return new(accountOpened) },
		func() AccountEvent { return new(moneyDeposited) },
		func() AccountEvent { return new(moneyWithdrawn) },
	}
}

func (a *Account) Apply(evt AccountEvent) error {
	switch event := evt.(type) {
	case *accountOpened:
		a.id = event.ID
		a.openedAt = event.OpenedAt
		a.holderName = event.HolderName
	case *moneyWithdrawn:
		a.balance -= event.Amount
	case *moneyDeposited:
		a.balance += event.Amount
	default:
		return fmt.Errorf("unexpected event kind: %T", event)
	}
	return nil
}