aether/examples/cross_node_broadcasting.go

//go:build integration
// +build integration

package examples

import (
	"context"
	"fmt"
	"log"
	"time"

	"git.flowmade.one/flowmade-one/aether"
	"git.flowmade.one/flowmade-one/aether/store"
	"github.com/nats-io/nats.go"
)

// CrossNodeBroadcasting demonstrates how to implement cross-node event broadcasting
// using NATSEventBus with JetStreamEventStore. This example shows how events persist
// to JetStream and are then broadcast to other nodes in the cluster via NATS.
//
// Key Concepts:
// 1. NATSEventBus wraps EventBus to add NATS publishing
// 2. JetStreamEventStore with broadcaster publishes EventStored to NATS
// 3. Other nodes receive these events via NATS subscription
// 4. Version cache is updated on remote events to maintain consistency
//
// Usage:
//   go run examples/cross_node_broadcasting.go
func CrossNodeBroadcastingExample() {
	// Connect to NATS
	nc, err := nats.Connect(nats.DefaultURL)
	if err != nil {
		log.Fatalf("Failed to connect to NATS: %v", err)
	}
	defer nc.Close()

	// Create NATS event bus (this will broadcast to all nodes)
	natsBus, err := aether.NewNATSEventBus(nc)
	if err != nil {
		log.Fatalf("Failed to create NATS event bus: %v", err)
	}
	defer natsBus.Stop()

	// Create JetStream event store WITH broadcaster
	// This enables EventStored events to be published to NATS
	store, err := store.NewJetStreamEventStoreWithBroadcaster(
		nc,
		"events",
		natsBus,
		"tenant-abc", // Optional namespace for isolation
	)
	if err != nil {
		log.Fatalf("Failed to create event store: %v", err)
	}

	// Subscribe to EventStored events to update version cache
	// This keeps the version cache synchronized across nodes
	eventStoredCh := natsBus.SubscribeWithFilter(
		"tenant-abc",
		&aether.SubscriptionFilter{
			EventTypes: []string{aether.EventTypeEventStored},
		},
	)

	go func() {
		for event := range eventStoredCh {
			actorID := event.Data["actorId"].(string)
			version := int64(event.Data["version"].(float64))
			store.UpdateVersionCache(actorID, version)
		}
	}()

	// Now save an event - it will be:
	// 1. Persisted to JetStream
	// 2. Published to NATS as EventStored
	// 3. Received by other nodes via NATS
	// 4. Used to update version cache

	event := &aether.Event{
		ID:        "event-1",
		EventType: "OrderPlaced",
		ActorID:   "order-123",
		Version:   1,
		Data: map[string]interface{}{
			"total": 100.00,
			"item":  "widget",
		},
		Timestamp: time.Now(),
	}

	err = store.SaveEvent(event)
	if err != nil {
		log.Fatalf("Failed to save event: %v", err)
	}

	log.Println("Event saved to JetStream and broadcast to NATS")

	// Other nodes in the cluster will receive this event via NATS
	// and update their version cache accordingly

	// Subscribe to events in this namespace
	eventCh := natsBus.Subscribe("tenant-abc")

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	select {
	case receivedEvent := <-eventCh:
		log.Printf("Received event via NATS: %s (version %d)",
			receivedEvent.EventType, receivedEvent.Version)
	case <-ctx.Done():
		log.Println("Timeout waiting for event")
	}
}

// CrossNodeBroadcastingMultiNode demonstrates a multi-node cluster setup.
// This simulates multiple nodes connecting to the same NATS cluster.
// In production, these would be separate processes/machines.
//
// This example requires a running NATS server with JetStream enabled.
func CrossNodeBroadcastingMultiNode() {
	// Connect to NATS
	nc, err := nats.Connect(nats.DefaultURL)
	if err != nil {
		log.Fatalf("Failed to connect to NATS: %v", err)
	}
	defer nc.Close()

	nodeID := "node-1"
	log.Printf("Starting %s", nodeID)

	// Each node creates its own NATS event bus and event store
	natsBus, err := aether.NewNATSEventBus(nc)
	if err != nil {
		log.Fatalf("Failed to create NATS event bus: %v", err)
	}
	defer natsBus.Stop()

	store, err := store.NewJetStreamEventStoreWithBroadcaster(
		nc,
		"events",
		natsBus,
		"tenant-abc",
	)
	if err != nil {
		log.Fatalf("Failed to create event store: %v", err)
	}

	// Setup EventStored subscription for cache synchronization
	eventStoredCh := natsBus.SubscribeWithFilter(
		"tenant-abc",
		&aether.SubscriptionFilter{
			EventTypes: []string{aether.EventTypeEventStored},
		},
	)

	go func() {
		for event := range eventStoredCh {
			actorID := event.Data["actorId"].(string)
			version := int64(event.Data["version"].(float64))
			store.UpdateVersionCache(actorID, version)
			log.Printf("[%s] Received EventStored for %s v%d", nodeID, actorID, version)
		}
	}()

	// Subscribe to actual events
	eventCh := natsBus.Subscribe("tenant-abc")
	log.Printf("[%s] Subscribed to tenant-abc", nodeID)

	// Save an event
	savedEvent := &aether.Event{
		ID:        fmt.Sprintf("event-%s-1", nodeID),
		EventType: "OrderPlaced",
		ActorID:   "order-123",
		Version:   1,
		Data: map[string]interface{}{
			"node": nodeID,
			"total": 100.00,
		},
		Timestamp: time.Now(),
	}

	err = store.SaveEvent(savedEvent)
	if err != nil {
		log.Fatalf("[%s] Failed to save event: %v", nodeID, err)
	}

	log.Printf("[%s] Saved event to JetStream", nodeID)

	// Wait to receive the event (either from local or remote)
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	select {
	case receivedEvent := <-eventCh:
		log.Printf("[%s] Received: %s v%d from %s",
			nodeID,
			receivedEvent.EventType,
			receivedEvent.Version,
			receivedEvent.GetCorrelationID(),
		)
	case <-ctx.Done():
		log.Printf("[%s] Timeout waiting for event", nodeID)
	}
}

// DistributedOrderProcessing demonstrates a realistic scenario where
// multiple nodes process orders for the same actor. This shows how
// cross-node broadcasting ensures consistency.
func DistributedOrderProcessing() {
	// Connect to NATS
	nc, err := nats.Connect(nats.DefaultURL)
	if err != nil {
		log.Fatalf("Failed to connect to NATS: %v", err)
	}
	defer nc.Close()

	// Node 1: Order creation
	node1Bus, _ := aether.NewNATSEventBus(nc)
	node1Store, _ := store.NewJetStreamEventStoreWithBroadcaster(
		nc, "events", node1Bus, "orders",
	)
	defer node1Bus.Stop()

	// Setup EventStored subscription for Node 1
	eventStoredCh1 := node1Bus.SubscribeWithFilter(
		"orders",
		&aether.SubscriptionFilter{
			EventTypes: []string{aether.EventTypeEventStored},
		},
	)
	go func() {
		for event := range eventStoredCh1 {
			actorID := event.Data["actorId"].(string)
			version := int64(event.Data["version"].(float64))
			node1Store.UpdateVersionCache(actorID, version)
		}
	}()

	// Node 2: Order processing (different node)
	node2Bus, _ := aether.NewNATSEventBus(nc)
	node2Store, _ := store.NewJetStreamEventStoreWithBroadcaster(
		nc, "events", node2Bus, "orders",
	)
	defer node2Bus.Stop()

	// Setup EventStored subscription for Node 2
	eventStoredCh2 := node2Bus.SubscribeWithFilter(
		"orders",
		&aether.SubscriptionFilter{
			EventTypes: []string{aether.EventTypeEventStored},
		},
	)
	go func() {
		for event := range eventStoredCh2 {
			actorID := event.Data["actorId"].(string)
			version := int64(event.Data["version"].(float64))
			node2Store.UpdateVersionCache(actorID, version)
		}
	}()

	// Node 2 also subscribes to events to receive updates
	node2EventCh := node2Bus.Subscribe("orders")

	// Node 1 creates an order
	orderPlaced := &aether.Event{
		ID:        "order-created",
		EventType: "OrderPlaced",
		ActorID:   "order-456",
		Version:   1,
		Data: map[string]interface{}{
			"total": 100.00,
		},
		Timestamp: time.Now(),
	}

	if err := node1Store.SaveEvent(orderPlaced); err != nil {
		log.Fatalf("Failed to create order: %v", err)
	}
	log.Println("Node 1: Created order-456")

	// Node 2 receives the OrderPlaced event via NATS
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	select {
	case event := <-node2EventCh:
		log.Printf("Node 2: Received %s v%d", event.EventType, event.Version)

		// Node 2 processes the order (must use version 2)
		orderProcessed := &aether.Event{
			ID:        "order-processed",
			EventType: "OrderProcessed",
			ActorID:   "order-456",
			Version:   2, // Must be > 1
			Data: map[string]interface{}{
				"status": "shipped",
			},
			Timestamp: time.Now(),
		}

		if err := node2Store.SaveEvent(orderProcessed); err != nil {
			log.Fatalf("Node 2: Failed to process order: %v", err)
		}
		log.Println("Node 2: Processed order-456")

	case <-ctx.Done():
		log.Println("Node 2: Timeout waiting for order event")
	}

	// Verify event stream consistency
	events, err := node1Store.GetEvents("order-456", 0)
	if err != nil {
		log.Fatalf("Failed to get events: %v", err)
	}
	log.Printf("Node 1: Event stream has %d events", len(events))
}