package main

import (
	"errors"
	"fmt"
	"slices"
	"sync"
)

type Event struct {
	ID, Task, Kind string
	Seq, Schema    int
	Name, Title    string
}
type Canonical struct {
	ID, Task, Kind, Title string
	Seq                   int
}

func upcast(e Event) (Canonical, error) {
	c := Canonical{ID: e.ID, Task: e.Task, Kind: e.Kind, Seq: e.Seq}
	switch e.Schema {
	case 1:
		c.Title = e.Name
	case 2:
		c.Title = e.Title
	default:
		return Canonical{}, fmt.Errorf("unsupported schema %d", e.Schema)
	}
	return c, nil
}

type State struct {
	Title   string
	Done    bool
	Version int
}

func reduce(s State, e Canonical) (State, error) {
	if e.Seq != s.Version+1 {
		return s, errors.New("sequence gap")
	}
	switch e.Kind {
	case "Created":
		if s.Version != 0 || e.Title == "" {
			return s, errors.New("invalid create")
		}
		s.Title = e.Title
	case "Renamed":
		if s.Version == 0 || s.Done || e.Title == "" {
			return s, errors.New("invalid rename")
		}
		s.Title = e.Title
	case "Completed":
		if s.Version == 0 || s.Done {
			return s, errors.New("invalid complete")
		}
		s.Done = true
	default:
		return s, errors.New("unknown event kind")
	}
	s.Version = e.Seq
	return s, nil
}

// Atomic in-memory append and expected-version check. No disk persistence.
type Store struct {
	mu      sync.Mutex
	History map[string][]Event
}

func (s *Store) Append(expected int, e Event) (Event, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	hs := s.History[e.Task]
	if expected != len(hs) {
		return Event{}, errors.New("version conflict")
	}
	e.Seq = len(hs) + 1
	e.ID = fmt.Sprintf("%s:%d", e.Task, e.Seq)
	state := State{}
	for _, old := range hs {
		c, err := upcast(old)
		if err != nil {
			return Event{}, err
		}
		state, err = reduce(state, c)
		if err != nil {
			return Event{}, err
		}
	}
	c, err := upcast(e)
	if err != nil {
		return Event{}, err
	}
	if _, err = reduce(state, c); err != nil {
		return Event{}, err
	}
	s.History[e.Task] = append(hs, e)
	return e, nil
}
func (s *Store) Events(task string) []Event {
	s.mu.Lock()
	defer s.mu.Unlock()
	return slices.Clone(s.History[task])
}

// This projection is called serially. Production needs transactional update+offset.
type Projection struct {
	Rows map[string]State
	Seen map[string]Event
}

func NewProjection() *Projection {
	return &Projection{Rows: map[string]State{}, Seen: map[string]Event{}}
}
func (p *Projection) Apply(e Event) error {
	if old, ok := p.Seen[e.ID]; ok {
		if old != e {
			return errors.New("event ID payload conflict")
		}
		return nil
	}
	c, err := upcast(e)
	if err != nil {
		return err
	}
	next, err := reduce(p.Rows[e.Task], c)
	if err != nil {
		return err
	}
	// Logical atomic boundary; no external calls in this reducer.
	p.Rows[e.Task] = next
	p.Seen[e.ID] = e
	return nil
}
func rebuild(es []Event) (*Projection, error) {
	p := NewProjection()
	for _, e := range es {
		if err := p.Apply(e); err != nil {
			return nil, err
		}
	}
	return p, nil
}

// An independent, explicitly invoked effect consumer. Rebuild never invokes it.
func notifyCompleted(es []Event, delivered map[string]bool) int {
	n := 0
	for _, e := range es {
		if e.Kind == "Completed" && !delivered[e.ID] {
			delivered[e.ID] = true
			n++
		}
	}
	return n
}
func must(err error) {
	if err != nil {
		panic(err)
	}
}
func main() {
	s := &Store{History: map[string][]Event{}}
	p := NewProjection()
	e1, err := s.Append(0, Event{Task: "t1", Kind: "Created", Schema: 1, Name: "初稿"})
	must(err)
	must(p.Apply(e1))
	e2, err := s.Append(1, Event{Task: "t1", Kind: "Renamed", Schema: 2, Title: "审校"})
	must(err)
	fmt.Printf("write_version=%d read_version=%d title=%s\n", e2.Seq, p.Rows["t1"].Version, p.Rows["t1"].Title)
	_, err = s.Append(1, Event{Task: "t1", Kind: "Renamed", Schema: 2, Title: "旧页面"})
	fmt.Println("stale command:", err)
	must(p.Apply(e2))
	must(p.Apply(e2))
	fmt.Printf("after replay: version=%d title=%s\n", p.Rows["t1"].Version, p.Rows["t1"].Title)
	e3, err := s.Append(2, Event{Task: "t1", Kind: "Completed", Schema: 2})
	must(err)
	fmt.Println("gap:", NewProjection().Apply(e3))
	must(p.Apply(e3))
	delivered := map[string]bool{}
	sent := notifyCompleted(s.Events("t1"), delivered)
	rebuilt, err := rebuild(s.Events("t1"))
	must(err)
	fmt.Printf("rebuild equal=%v done=%v notifications=%d\n", rebuilt.Rows["t1"] == p.Rows["t1"], rebuilt.Rows["t1"].Done, sent)
	fmt.Printf("retry notifications=%d\n", notifyCompleted(s.Events("t1"), delivered))
}