package main

import (
	"encoding/json"
	"errors"
	"fmt"
)

type Event struct {
	ID      string `json:"id"`
	Type    string `json:"type"`
	Version int    `json:"version"`
	OrderID string `json:"order_id"`
}
type Row struct {
	Event Event
	Sent  bool
}
type DB struct {
	orders map[string]bool
	outbox []Row
}

// Create is a single-threaded atomic-state model, not a durable database.
func (d *DB) Create(order string, abort bool) error {
	if order == "" || d.orders[order] {
		return errors.New("invalid or duplicate order")
	}
	e := Event{ID: "created:" + order, Type: "order.created", Version: 1, OrderID: order}
	if abort {
		return errors.New("transaction aborted")
	}
	if d.orders == nil {
		d.orders = make(map[string]bool)
	}
	d.orders[order] = true
	d.outbox = append(d.outbox, Row{Event: e})
	return nil
}

type Broker struct{ Delivered []Event }

func (d *DB) Relay(b *Broker, crashAfterPublish bool) {
	for i := range d.outbox {
		if d.outbox[i].Sent {
			continue
		}
		b.Delivered = append(b.Delivered, d.outbox[i].Event)
		if crashAfterPublish {
			return
		} // restart retains DB state but loses execution point
		d.outbox[i].Sent = true
	}
}

type Projection struct {
	seen  map[string]bool
	count int
}

func (p *Projection) Apply(e Event) {
	if p.seen[e.ID] {
		return
	}
	if p.seen == nil {
		p.seen = make(map[string]bool)
	}
	p.seen[e.ID] = true
	p.count++
}

func Decode(b []byte) (Event, error) {
	var e Event
	if err := json.Unmarshal(b, &e); err != nil {
		return e, err
	}
	if e.Type != "order.created" || e.Version != 1 {
		return e, errors.New("unsupported event contract")
	}
	if e.ID == "" || e.OrderID == "" {
		return e, errors.New("missing event identity")
	}
	return e, nil
}

type Lease struct {
	Token int
	Owner string
	Until int
	Done  bool
}

func (l *Lease) Claim(owner string, now, duration int) (int, bool) {
	if l.Done || now < l.Until {
		return 0, false
	}
	l.Token++
	l.Owner, l.Until = owner, now+duration
	return l.Token, true
}
func (l *Lease) Complete(token, now int) bool {
	if l.Done || token != l.Token || now >= l.Until {
		return false
	}
	l.Done = true
	return true
}

func demo() {
	d := new(DB)
	fmt.Println("create failed:", d.Create("o0", true))
	fmt.Println("after abort orders/outbox:", len(d.orders), len(d.outbox))
	if err := d.Create("o1", false); err != nil {
		panic(err)
	}
	b := new(Broker)
	d.Relay(b, true)
	fmt.Println("after publish crash deliveries/sent:", len(b.Delivered), d.outbox[0].Sent)
	d.Relay(b, false)
	p := new(Projection)
	for _, e := range b.Delivered {
		p.Apply(e)
	}
	fmt.Println("after restart deliveries/effects:", len(b.Delivered), p.count)
	_, err := Decode([]byte(`{"id":"e1","type":"order.created","version":1,"order_id":"o1","region":"eu"}`))
	fmt.Println("additive field:", err)
	_, err = Decode([]byte(`{"id":"e2","type":"order.created","version":2,"order_id":"o1"}`))
	fmt.Println("new contract:", err)
	l := new(Lease)
	old, _ := l.Claim("worker-a", 0, 5)
	fresh, _ := l.Claim("worker-b", 6, 5)
	fmt.Println("stale completion:", l.Complete(old, 7))
	fmt.Println("current completion:", l.Complete(fresh, 7))
}
func main() { demo() }