// Package plog is a partitioned commit log — Kafka in miniature. It
// builds the model behind every modern streaming platform: an append-only
// log split into partitions (the unit of ordering and parallelism),
// monotonic per-partition offsets, key-based partitioning, retention, and
// (in groups.go) consumer-group offset tracking + partition assignment.
// Stdlib-only.
package plog

import "sync"

// Record is one entry in a partition. Offset is absolute and stable: it
// never changes or repeats, even after retention truncates older records.
type Record struct {
	Offset int64
	Key    string
	Value  []byte
}

// PartitionedLog is an append-only log of `n` partitions.
type PartitionedLog struct {
	mu    sync.RWMutex
	parts [][]Record // parts[p] holds that partition's records (absolute offsets)
	next  []int64    // next offset to assign per partition (monotonic)
}

func New(numPartitions int) *PartitionedLog {
	if numPartitions < 1 {
		numPartitions = 1
	}
	return &PartitionedLog{
		parts: make([][]Record, numPartitions),
		next:  make([]int64, numPartitions),
	}
}

// Partitions returns the partition count (fixed at creation; changing it
// in real Kafka reshuffles key→partition mapping — a migration).
func (l *PartitionedLog) Partitions() int { return len(l.parts) }

// PartitionFor maps a key to a partition by hashing — so all records for
// the same key land in the same partition and are therefore ORDERED
// relative to each other. Keyless records would round-robin.
func (l *PartitionedLog) PartitionFor(key string) int {
	return int(fnv1a32(key) % uint32(len(l.parts)))
}

// Produce appends a keyed record and returns its (partition, offset).
func (l *PartitionedLog) Produce(key string, value []byte) (partition int, offset int64) {
	p := l.PartitionFor(key)
	l.mu.Lock()
	defer l.mu.Unlock()
	off := l.next[p]
	l.next[p]++
	l.parts[p] = append(l.parts[p], Record{Offset: off, Key: key, Value: value})
	return p, off
}

// Consume returns up to `max` records from a partition with Offset >=
// fromOffset (a consumer polling from where it left off). max<=0 means
// "all available".
func (l *PartitionedLog) Consume(partition int, fromOffset int64, max int) []Record {
	l.mu.RLock()
	defer l.mu.RUnlock()
	var out []Record
	for _, r := range l.parts[partition] {
		if r.Offset < fromOffset {
			continue
		}
		out = append(out, r)
		if max > 0 && len(out) >= max {
			break
		}
	}
	return out
}

// HighWatermark is the next offset that will be assigned in a partition
// (== one past the last record). A consumer is "caught up" when its
// committed offset equals the high watermark.
func (l *PartitionedLog) HighWatermark(partition int) int64 {
	l.mu.RLock()
	defer l.mu.RUnlock()
	return l.next[partition]
}

// Truncate applies retention: drop records with Offset < beforeOffset.
// Offsets of the remaining records are UNCHANGED (absolute), so committed
// consumer offsets stay valid; a consumer that committed below the new
// base simply resumes at the earliest retained record.
func (l *PartitionedLog) Truncate(partition int, beforeOffset int64) {
	l.mu.Lock()
	defer l.mu.Unlock()
	recs := l.parts[partition]
	i := 0
	for i < len(recs) && recs[i].Offset < beforeOffset {
		i++
	}
	l.parts[partition] = append([]Record(nil), recs[i:]...)
}

func fnv1a32(s string) uint32 {
	var h uint32 = 2166136261
	for i := 0; i < len(s); i++ {
		h ^= uint32(s[i])
		h *= 16777619
	}
	return h
}