pupsourcing

pupsourcing

A production-ready Event Sourcing library for Go with clean architecture principles.

Overview

pupsourcing provides minimal, reliable infrastructure for event sourcing in Go applications. State changes are stored as an immutable sequence of events, providing a complete audit trail and enabling event replay, temporal queries, and flexible read models.

Key Features

• Clean Architecture - Core interfaces are datastore-agnostic; no "infrastructure creep" into your domain model (no annotations, no framework-specific base classes)
• Multiple Database Adapters - PostgreSQL, SQLite, and MySQL/MariaDB
• Bounded Context Support - Events are scoped to bounded contexts for domain-driven design alignment
• Optimistic Concurrency - Automatic conflict detection via database constraints
• Projection System - Pull-based event processing with checkpoints, supporting both global and context-scoped projections
• Horizontal Scaling - Hash-based partitioning for projection workers
• Code Generation - Optional tool for strongly-typed domain event mapping
• Minimal Dependencies - Go standard library plus database driver

Installation

go get github.com/getpup/pupsourcing

Choose your database driver:

# PostgreSQL (recommended for production)
go get github.com/lib/pq

# SQLite (embedded, ideal for testing)
go get modernc.org/sqlite

# MySQL/MariaDB
go get github.com/go-sql-driver/mysql

Quick Start

1. Generate Database Schema

go run github.com/getpup/pupsourcing/cmd/migrate-gen -output migrations

Apply the generated migrations using your preferred migration tool.

2. (Optional) Generate Event Mapping Code

If you want type-safe mapping between domain events and event sourcing types:

go run github.com/getpup/pupsourcing/cmd/eventmap-gen \
  -input internal/domain/events \
  -output internal/infrastructure/generated

See Event Mapping Documentation for details.

3. Append Events

import (
    "github.com/getpup/pupsourcing/es"
    "github.com/getpup/pupsourcing/es/adapters/postgres"
    "github.com/google/uuid"
)

// Create store
store := postgres.NewStore(postgres.DefaultStoreConfig())

// Create event with bounded context
event := es.Event{
    BoundedContext: "Identity",       // Required: bounded context for the event
    AggregateType:  "User",
    AggregateID:    uuid.New().String(),  // String-based ID for flexibility
    EventID:        uuid.New(),
    EventType:      "UserCreated",
    EventVersion:   1,
    Payload:        []byte(`{"email":"alice@example.com","name":"Alice"}`),
    Metadata:       []byte(`{}`),
    CreatedAt:      time.Now(),
}

// Append within transaction with optimistic concurrency control
tx, _ := db.BeginTx(ctx, nil)
// Use NoStream() for creating a new aggregate
result, err := store.Append(ctx, tx, es.NoStream(), []es.Event{event})
if err != nil {
    tx.Rollback()
    log.Fatal(err)
}
tx.Commit()

// Access the result
fmt.Printf("Events appended at positions: %v\n", result.GlobalPositions)
fmt.Printf("Aggregate is now at version: %d\n", result.ToVersion())

4. Read Aggregate Streams

// Read all events for an aggregate (with bounded context)
aggregateID := uuid.New().String()
stream, err := store.ReadAggregateStream(ctx, tx, "Identity", "User", aggregateID, nil, nil)

// Access stream information
fmt.Printf("Stream has %d events\n", stream.Len())
fmt.Printf("Current aggregate version: %d\n", stream.Version())
fmt.Printf("Bounded context: %s\n", stream.BoundedContext)

// Read from a specific version
fromVersion := int64(5)
stream, err = store.ReadAggregateStream(ctx, tx, "Identity", "User", aggregateID, &fromVersion, nil)

// Process events
for _, event := range stream.Events {
    // Handle event - all events have the same bounded context
    fmt.Printf("Event %s in context %s\n", event.EventType, event.BoundedContext)
}

5. Run Projections

Projections transform events into query-optimized read models. Use scoped projections for read models that only care about specific aggregate types and/or bounded contexts, or global projections for integration publishers that need all events.

import "github.com/getpup/pupsourcing/es/projection"

// Scoped projection - only receives User aggregate events from Identity context
type UserReadModelProjection struct{}

func (p *UserReadModelProjection) Name() string {
    return "user_read_model"
}

// AggregateTypes filters events by aggregate type
func (p *UserReadModelProjection) AggregateTypes() []string {
    return []string{"User"}  // Only receives User events
}

// BoundedContexts filters events by bounded context
func (p *UserReadModelProjection) BoundedContexts() []string {
    return []string{"Identity"}  // Only receives events from Identity context
}

func (p *UserReadModelProjection) Handle(ctx context.Context, event es.PersistedEvent) error {
    // Update read model based on User events from Identity context only
    // Projection manages its own persistence
    return nil
}

// Run projection with adapter-specific processor
store := postgres.NewStore(postgres.DefaultStoreConfig())
config := projection.DefaultProcessorConfig()
processor := postgres.NewProcessor(db, store, &config)
err := processor.Run(ctx, &UserReadModelProjection{})

Documentation

Comprehensive documentation is available at https://pupsourcing.gopup.dev:

• Getting Started - Installation, setup, and first steps
• Core Concepts - Understanding event sourcing principles
• Database Adapters - Choosing the right database
• Projections & Scaling - Horizontal scaling and production patterns
• Event Mapping Code Generation - Type-safe domain event mapping
• Observability - Logging, tracing, and monitoring
• API Reference - Complete API documentation

Examples

Complete runnable examples are available in the examples/ directory:

• Single Worker - Basic projection pattern
• Multiple Projections - Running different projections concurrently
• Worker Pool - Multiple workers in the same process
• Partitioned - Horizontal scaling across processes
• With Logging - Observability and debugging

See the examples README for more details.

Development

Running Tests

Unit tests:

make test-unit

Integration tests locally (requires Docker):

make test-integration-local

This command automatically:

1. Starts PostgreSQL and MySQL containers via docker compose
2. Runs all integration tests
3. Cleans up containers

Manual integration testing:

# Start databases
docker compose up -d

# Run integration tests
make test-integration

# Stop databases
docker compose down

Contributing

Contributions are welcome! Please submit a Pull Request.

License

This project is licensed under the MIT License - see the LICENSE file for details.