basic.go

package main

import (
	"fmt"
	"iter"

	bolt "go.etcd.io/bbolt"

	"github.com/lymar/physalis"
)

// AppEvent is the application's event envelope (a union of all event kinds).
type AppEvent struct {
	Win  *Win
	Loss *Loss
}

// Win represents a "player won points" event.
type Win struct {
	Player string
	Points int
}

// Loss represents a "player lost points" event.
type Loss struct {
	Player string
	Points int
}

// AppReducerState is the reducer state for a single group (player).
//
// Reducers are the core building block of Physalis: they build state by replaying
// (reducing) an append-only stream of events.
type AppReducerState struct {
	Points int
}

// AppReducer is a reducer that maintains per-player points.
type AppReducer struct {
	version string
}

// Version returns the reducer version.
//
// When the version changes, Physalis rebuilds the reducer state from scratch by
// replaying the event log. This is how you can evolve your state model safely:
// change reducer logic, bump the version, and let Physalis recompute.
func (rd *AppReducer) Version() string {
	return rd.version
}

// Prepare returns a group key and (optionally) a transformed event.
//
// It acts as a filter.
//
// Reducer state is maintained independently per group. In this example, the
// grouping key is the player name.
func (rd *AppReducer) Prepare(
	ev *physalis.Event[AppEvent],
) (string, *physalis.Event[AppEvent]) {
	if ev.Payload.Win != nil {
		return ev.Payload.Win.Player, nil
	} else if ev.Payload.Loss != nil {
		return ev.Payload.Loss.Player, nil
	}
	panic("invalid event")
}

// Apply applies events to the reducer state.
//
// This is where you implement your state-building logic. In this example we:
// - accumulate player's total points
// - track "lucky" minutes (minutes when the player gained points)
func (rd *AppReducer) Apply(
	runtime *physalis.ReducerRuntime,
	state *AppReducerState,
	groupKey string,
	evs iter.Seq2[uint64, *physalis.Event[AppEvent]]) *AppReducerState {

	// In addition to the main reducer state, Physalis can persist per-group
	// key/value datasets (KV stores). Here we record how many points were earned
	// in each "lucky" minute.
	luckyMinutes := physalis.OpenKV[uint32, int](runtime, "lucky_minutes")

	if state == nil {
		state = &AppReducerState{}
	}

	for _, ev := range evs {
		// Use the event's recorded timestamp.
		minute := uint32(ev.ReadTimestamp().Minute())
		if ev.Payload.Win != nil {
			state.Points += ev.Payload.Win.Points

			prevVal := luckyMinutes.Get(minute)
			if prevVal == nil {
				luckyMinutes.Put(minute, &ev.Payload.Win.Points)
			} else {
				newVal := *prevVal + ev.Payload.Win.Points
				luckyMinutes.Put(minute, &newVal)
			}
		} else if ev.Payload.Loss != nil {
			state.Points -= ev.Payload.Loss.Points
		}
	}

	return state
}

func main() {
	registry := physalis.NewReducerRegistry[AppEvent]()
	// Register a reducer in the registry (you can have many reducers).
	// AddReducer returns a reader used to query that reducer's state.
	reader, err := physalis.AddReducer(registry, "points",
		&AppReducer{version: "v1"})
	if err != nil {
		panic(err)
	}

	// Open a Physalis database (internally it's a bbolt DB).
	phs, err := physalis.Open("basic.db", registry)
	if err != nil {
		panic(err)
	}
	defer phs.Close()

	// Write events in a single transaction.
	// Reducer states are updated automatically and synchronously.
	err = phs.Write(physalis.Transaction[AppEvent]{
		Events: []*physalis.Event[AppEvent]{
			{Payload: AppEvent{Win: &Win{Player: "Alice", Points: 10}}},
			{Payload: AppEvent{Win: &Win{Player: "Bob", Points: 5}}},
			{Payload: AppEvent{Loss: &Loss{Player: "Alice", Points: 4}}},
		},
	})

	if err != nil {
		panic(err)
	}

	// Read reducer state.
	phs.View(func(tx *bolt.Tx) error {
		alice, err := reader.State(tx, "Alice")
		if err != nil {
			return err
		}
		fmt.Println("Alice points:", alice.Points)

		bob, err := reader.State(tx, "Bob")
		if err != nil {
			return err
		}
		fmt.Println("Bob points:", bob.Points)

		// Open a KV view to read the per-group key/value dataset.
		aliceLuckyMinutes := physalis.OpenKVView[uint32, int](reader, tx,
			"Alice", "lucky_minutes")
		for k, v := range aliceLuckyMinutes.Ascend() {
			fmt.Printf("Alice lucky minute %d: %d points\n", k, *v)
		}

		return nil
	})
}