package main

import (
	"bytes"
	"compress/gzip"
	"fmt"
	"strings"
)

// canAck models a fixed ISR snapshot, not the Kafka replication protocol.
func canAck(minISR, requiredNext int, isrNext ...int) bool {
	if len(isrNext) < minISR {
		return false
	}
	for _, next := range isrNext {
		if next < requiredNext {
			return false
		}
	}
	return true
}

type Record struct {
	Offset int
	State  string
}

func readCommitted(log []Record) []int {
	var result []int
	for _, r := range log {
		if r.State == "open" {
			break
		} // first unstable record blocks later records
		if r.State == "committed" {
			result = append(result, r.Offset)
		}
	}
	return result
}

func compressedSize(s string) int {
	var b bytes.Buffer
	w := gzip.NewWriter(&b)
	if _, err := w.Write([]byte(s)); err != nil {
		panic(err)
	}
	if err := w.Close(); err != nil {
		panic(err)
	}
	return b.Len()
}

func demo() {
	fmt.Println("ISR A,B caught up:", canAck(2, 101, 101, 101))
	fmt.Println("ISR A,B, B behind:", canAck(2, 101, 101, 100))
	fmt.Println("only A remains:", canAck(2, 101, 101))
	committedNext, staleReplicaNext := 101, 98
	fmt.Println("stale leader missing slots:", committedNext-staleReplicaNext)
	log := []Record{{0, "committed"}, {1, "aborted"}, {2, "open"}, {3, "committed"}}
	fmt.Println("open transaction:", readCommitted(log))
	log[2].State = "committed"
	fmt.Println("transaction finished:", readCommitted(log))
	msg := `{"type":"order.created","status":"new"}` + "\n"
	fmt.Println("gzip separately:", 100*compressedSize(msg))
	fmt.Println("gzip one batch:", compressedSize(strings.Repeat(msg, 100)))
}

func main() { demo() }