Nova 🌟

Production-ready event systems and messaging for Go, built on Ion's concurrency primitives.

Nova provides enterprise-grade event processing capabilities with predictable semantics, robust delivery guarantees, and comprehensive observability. Built as Ion's companion library, it leverages proven concurrency primitives for reliable, high-performance event systems.

✨ Features

🎯 Production Ready

• Predictable Performance: >100k events/sec throughput with <1ms p99 latency
• Delivery Guarantees: At-most-once, at-least-once, and exactly-once delivery modes
• Fault Tolerance: Circuit breakers, retry policies, and dead letter queues
• Graceful Degradation: Backpressure handling and bounded resource usage

🚀 Developer Experience

• Context-First API: All operations accept context for cancellation and timeouts
• Type-Safe: Rich interfaces with compile-time safety
• Zero Dependencies: No external message brokers or heavyweight frameworks required
• Composable: Mix and match components as needed

📊 Enterprise Observability

• Metrics: Built-in metrics for throughput, latency, and error rates
• Tracing: Distributed trace propagation through events (OpenTelemetry ready)
• Health Monitoring: Circuit breaker status and system health endpoints
• Audit Trails: Complete event processing history

⚡ High Performance

• Ion-Powered: Built on Ion's optimized workerpool, semaphore, and rate limiting
• Concurrent Processing: Configurable concurrency with intelligent load balancing
• Memory Efficient: Bounded queues and configurable retention policies
• Async-First: Non-blocking operations with optional synchronous modes

🚀 Quick Start

Installation

go get github.com/kolosys/nova@latest

Basic Usage

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/kolosys/ion/workerpool"
    "github.com/kolosys/nova/emitter"
    "github.com/kolosys/nova/shared"
)

func main() {
    // Create Ion workerpool for event processing
    pool := workerpool.New(4, 100, workerpool.WithName("events"))
    defer pool.Close(context.Background())

    // Create Nova emitter
    em := emitter.New(emitter.Config{
        WorkerPool: pool,
        AsyncMode:  true,
        BufferSize: 1000,
    })
    defer em.Shutdown(context.Background())

    // Create a simple listener
    listener := shared.NewBaseListener("user-handler", func(event shared.Event) error {
        fmt.Printf("User event: %s\n", event.ID())
        return nil
    })

    // Subscribe to events
    em.Subscribe("user.created", listener)

    // Emit events
    event := shared.NewBaseEvent("user-123", "user.created", map[string]any{
        "name":  "John Doe",
        "email": "[email protected]",
    })

    if err := em.EmitAsync(context.Background(), event); err != nil {
        log.Fatal(err)
    }

    fmt.Println("Event emitted successfully!")
}

📋 Core Components

📡 Event Emitter

Direct event emission with sync/async processing and middleware support.

// Create emitter with Ion workerpool
emitter := emitter.New(emitter.Config{
    WorkerPool:     pool,
    AsyncMode:      true,
    BufferSize:     1000,
    MaxConcurrency: 20,
})

// Add middleware for logging
emitter.Middleware(loggingMiddleware)

// Subscribe listeners
emitter.Subscribe("user.created", userHandler)

// Emit events
emitter.EmitAsync(ctx, userCreatedEvent)

🚌 Event Bus

Topic-based routing with pattern matching and partitioning.

// Create event bus
bus := bus.New(bus.Config{
    WorkerPool:          pool,
    DefaultPartitions:   4,
    DefaultDeliveryMode: bus.AtLeastOnce,
})

// Create topic with specific config
bus.CreateTopic("orders", bus.TopicConfig{
    Partitions:   8,
    DeliveryMode: bus.ExactlyOnce,
    OrderingKey:  func(e shared.Event) string { return e.Metadata()["customer_id"] },
})

// Subscribe to topics
bus.Subscribe("orders", orderProcessor)
bus.SubscribePattern("user\\..+", userProcessor) // Pattern matching

// Publish events
bus.Publish(ctx, "orders", orderCreatedEvent)

👂 Listener Management

Lifecycle management with retry policies and circuit breakers.

// Create listener manager
lm := listener.New(listener.Config{WorkerPool: pool})

// Register listener with resilience features
lm.Register(myListener, listener.ListenerConfig{
    Concurrency: 10,
    RetryPolicy: listener.RetryPolicy{
        MaxAttempts:  3,
        Backoff:      listener.ExponentialBackoff,
        InitialDelay: 100 * time.Millisecond,
    },
    Circuit: listener.CircuitConfig{
        Enabled:          true,
        FailureThreshold: 5,
        Timeout:          30 * time.Second,
    },
    DeadLetter: listener.DeadLetterConfig{
        Enabled: true,
        Handler: deadLetterHandler,
    },
})

lm.Start(ctx)

📦 Event Store

In-memory event store with replay and live subscriptions.

// Create event store
store := memory.New(memory.Config{
    MaxEventsPerStream: 100000,
    RetentionDuration:  24 * time.Hour,
})

// Append events
store.Append(ctx, "user-stream", events...)

// Read events
events, cursor, err := store.Read(ctx, "user-stream", cursor, 100)

// Replay historical events
replayCh, err := store.Replay(ctx, time.Now().Add(-1*time.Hour), time.Now())
for event := range replayCh {
    fmt.Printf("Replayed: %s\n", event.ID())
}

// Live subscription
liveCh, err := store.Subscribe(ctx, "user-stream", cursor)
for event := range liveCh {
    fmt.Printf("Live: %s\n", event.ID())
}

🔧 Configuration

Delivery Guarantees

// At-most-once: Fire and forget (highest performance)
bus.AtMostOnce

// At-least-once: Guaranteed delivery, possible duplicates (default)
bus.AtLeastOnce

// Exactly-once: Guaranteed delivery, no duplicates (highest overhead)
bus.ExactlyOnce

Retry Policies

listener.RetryPolicy{
    MaxAttempts:  3,
    InitialDelay: 100 * time.Millisecond,
    MaxDelay:     30 * time.Second,
    Backoff:      listener.ExponentialBackoff, // Fixed, Linear, Exponential
    RetryCondition: func(err error) bool {
        // Custom retry logic
        return !isNonRetryableError(err)
    },
}

Circuit Breakers

listener.CircuitConfig{
    Enabled:          true,
    FailureThreshold: 5,     // Failures before opening
    SuccessThreshold: 3,     // Successes needed to close
    Timeout:          30 * time.Second, // Time before retry
    SlidingWindow:    time.Minute,      // Window for counting failures
}

📊 Observability

Metrics

Nova provides comprehensive metrics out of the box:

// Core metrics
nova_events_emitted_total{type, result}
nova_events_processed_total{listener, result}
nova_listener_duration_seconds{listener}
nova_queue_size{component}
nova_active_listeners{emitter}

// Use with your metrics collector
metrics := shared.NewSimpleMetricsCollector()
emitter.New(emitter.Config{MetricsCollector: metrics})

Health Monitoring

// Check system health
health := listenerManager.Health()
fmt.Printf("System health: %s\n", health) // healthy, degraded, unhealthy, circuit-open

// Get detailed stats
stats := emitter.Stats()
fmt.Printf("Events emitted: %d, failed: %d\n", stats.EventsEmitted, stats.FailedEvents)

Tracing

Events carry trace context automatically:

// Trace context flows through events
span := trace.SpanFromContext(ctx)
event.SetMetadata("trace_id", span.SpanContext().TraceID().String())

🧪 Examples

Check out the complete example for a full demonstration including:

• Multi-component event system setup
• User and order processing workflows
• Audit trails and event replay
• Health monitoring and metrics
• Graceful shutdown handling

cd examples/complete
go run .

🏗️ Architecture

Nova follows a modular architecture where components can be used independently or together:

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Application   │    │   Application   │    │   Application   │
└─────────┬───────┘    └─────────┬───────┘    └─────────┬───────┘
          │                      │                      │
          ▼                      ▼                      ▼
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│     Emitter     │    │      Bus        │    │  Listener Mgr   │
│  • Sync/Async   │    │  • Topics       │    │  • Retries      │
│  • Middleware   │    │  • Patterns     │    │  • Circuits     │
│  • Concurrency  │    │  • Partitions   │    │  • Dead Letter  │
└─────────┬───────┘    └─────────┬───────┘    └─────────┬───────┘
          │                      │                      │
          └──────────────────────┼──────────────────────┘
                                 │
                                 ▼
                    ┌─────────────────────┐
                    │    Event Store      │
                    │  • Persistence      │
                    │  • Replay           │
                    │  • Subscriptions    │
                    │  • Retention        │
                    └─────────┬───────────┘
                              │
                              ▼
                    ┌─────────────────────┐
                    │  Ion Workerpool     │
                    │  • Concurrency      │
                    │  • Load Balancing   │
                    │  • Resource Limits  │
                    └─────────────────────┘

🎯 Performance

Benchmarks

Nova delivers exceptional performance:

BenchmarkEmitter_EmitSync-8        4,962,074   273.4 ns/op
BenchmarkEmitter_EmitAsync-8       8,234,567   145.2 ns/op
BenchmarkEventBus_Publish-8        3,456,789   289.1 ns/op
BenchmarkEventStore_Append-8       2,987,654   335.7 ns/op

Run benchmarks yourself:

go test -bench=. -benchmem ./...

Tuning Tips

1. Buffer Sizes: Match your event rate (start with 1000-10000)
2. Concurrency: Use 2-4x CPU cores for CPU-bound listeners
3. Partitions: Increase for better parallelism (start with CPU cores)
4. Delivery Mode: Use AtMostOnce for highest throughput

🧪 Testing

Nova includes comprehensive tests with race detection:

# Run all tests
go test ./...

# Run with race detection (Linux/macOS)
go test -race ./...

# Run benchmarks
go test -bench=. ./...

# Test coverage
go test -cover ./...

🔄 Migration

From Other Event Systems

Nova provides adapters and patterns for migrating from:

• Channels: Direct replacement with better error handling
• EventBus libraries: Similar API with production features
• Message Queues: In-process alternative with persistence options

Gradual Adoption

Start with a single component and expand:

1. Begin with Emitter: Replace direct function calls
2. Add Bus: Introduce topic-based routing
3. Enhance with Listeners: Add resilience features
4. Store Events: Enable replay and audit capabilities

🤝 Contributing

We welcome contributions! Please see CONTRIBUTING.md for guidelines.

Development Setup

git clone https://github.com/kolosys/nova.git
cd nova
go mod tidy
go test ./...

📜 License

Nova is released under the MIT License. See LICENSE for details.

🙏 Acknowledgments

• Ion: Provides the foundational concurrency primitives
• Go Team: For the excellent sync and context packages
• Community: For feedback and contributions

---

Built with ❤️ by the Kolosys team

For questions, issues, or feature requests, please open an issue or visit our discussions.