package main

import "fmt"

// Log is a single-partition, append-only teaching model, not a broker.
type Log struct {
	records []string
	next    map[string]int
}

func NewLog(records ...string) *Log {
	return &Log{records: records, next: make(map[string]int)}
}

func (l *Log) Fetch(group string) (int, string, bool) {
	i := l.next[group]
	if i >= len(l.records) {
		return i, "", false
	}
	return i, l.records[i], true
}

func (l *Log) Commit(group string, next int) error {
	if next < l.next[group] || next > len(l.records) {
		return fmt.Errorf("invalid next offset: %d", next)
	}
	l.next[group] = next
	return nil
}

// Frontier tracks a contiguous completed prefix in an ordered worker batch.
type Frontier struct {
	Next int
	done map[int]bool
}

func (f *Frontier) Complete(offset int) int {
	if f.done == nil {
		f.done = make(map[int]bool)
	}
	if offset >= f.Next {
		f.done[offset] = true
	}
	for f.done[f.Next] {
		delete(f.done, f.Next)
		f.Next++
	}
	return f.Next
}

func demo() {
	l := NewLog("order-1", "order-2")
	i, v, _ := l.Fetch("billing")
	fmt.Println("billing:", i, v)
	i, v, _ = l.Fetch("billing")
	fmt.Println("without commit:", i, v)
	if err := l.Commit("billing", i+1); err != nil {
		panic(err)
	}
	i, v, _ = l.Fetch("billing")
	fmt.Println("after commit:", i, v)
	i, v, _ = l.Fetch("audit")
	fmt.Println("audit:", i, v)
	f := Frontier{Next: 10}
	fmt.Println("finish 12 -> next:", f.Complete(12))
	fmt.Println("finish 10 -> next:", f.Complete(10))
	fmt.Println("finish 11 -> next:", f.Complete(11))
	jobs := make(chan string, 2)
	for _, job := range []string{"a", "b", "c"} {
		select {
		case jobs <- job:
			fmt.Println("accepted:", job)
		default:
			fmt.Println("backpressure:", job)
		}
	}
}

func main() { demo() }