stream

type AppendValue struct {
	// Partition is the partition the whole batch was appended to.
	Partition int
	// First is the offset of the batch's first record.
	First Offset
	// Last is the offset of the batch's last record.
	Last Offset
}

type BoundsValue struct {
	// First is the lowest offset still readable. Retention and Trim raise
	// it, compaction never moves it; nothing lowers it.
	First Offset
	// Next is the offset the partition's next record will take.
	Next Offset
}

type Committer interface {
	// Commit records next as the offset group reads next on partition,
	// replacing whatever the group committed there before — moving it back is
	// legitimate: it is a replay. next past the partition's next offset is
	// [OffsetAhead]; below its first retained offset it is kept, and the
	// group's next Read reports [OffsetTrimmed]. An empty group is
	// [StreamMisconfigured]; a partition outside [0, Partitions()) is
	// [UnknownPartition].
	Commit(ctx context.Context, group string, partition int, next Offset) error
	// Committed returns the offset group committed last on partition.
	//
	// A group that never committed there is [UnknownGroup] — never a zero
	// offset, because "start at the beginning" and "start at the end" are both
	// legitimate first positions and only the reader knows which it wants.
	Committed(ctx context.Context, group string, partition int) (Offset, error)
}

type CompactValue struct {
	// Scanned is how many retained records were at least
	// [PolicyValue.CompactAfter] old, the only ones a compaction may drop.
	Scanned int64
	// Kept is how many of them survived.
	Kept int64
	// Dropped is how many of them were removed: superseded by a later
	// record of their key, or a tombstone past its second window.
	Dropped int64
}

type Compactor interface {
	// Compact removes from partition every keyed record at least
	// [PolicyValue.CompactAfter] old that a later record of the same key, as
	// old, supersedes, and every tombstone at least twice CompactAfter old with
	// them; a record younger than CompactAfter is never touched, and an unkeyed
	// record is left as it is. Every survivor keeps its offset: offsets are
	// never renumbered and [Stream.Bounds] does not move, so a reader inside a
	// compacted range reads on from the next survivor and is refused
	// [OffsetTrimmed] only below the first retained offset, never in a hole.
	//
	// It reports what it looked at — every retained record at least
	// CompactAfter old —, what it kept and what it dropped. A zero CompactAfter
	// never compacts: Compact returns a zero [CompactValue]. A partition outside
	// [0, Partitions()) is [UnknownPartition].
	Compact(ctx context.Context, partition int) (CompactValue, error)
}

type EntryValue struct {
	// AppendedAt is the engine's clock when the batch holding the record was
	// appended; every record of one batch carries the same instant. It is
	// what [Seeker.OffsetAt] and retention by age read.
	AppendedAt time.Time
	// Record is the record as it was appended.
	Record RecordValue
	// Partition is the partition the record was appended to.
	Partition int
	// Offset is the record's position in its partition, unique and never reused.
	Offset Offset
}

type Offset uint64

type PolicyValue struct {
	// Partitions is how many partitions the stream has, fixed for its life.
	// Zero means one; negative is refused.
	Partitions int
	// MaxRecordBytes bounds one record, its key and payload together. A
	// larger record is refused at [Stream.Append] with [RecordTooLarge]. Zero
	// means [DefaultMaxRecordBytes]; negative is refused.
	MaxRecordBytes int
	// RetainFor is how long a record is kept, measured from its
	// [EntryValue.AppendedAt]: older records are trimmed by the engine, a
	// file engine dropping whole segments only. Zero keeps every record for
	// good; negative is refused.
	RetainFor time.Duration
	// RetainBytes is how many bytes of records a partition keeps: above it
	// the engine trims the oldest. Zero is unbounded; negative is refused.
	RetainBytes int64
	// MaxBytes is how many bytes of records the stream may hold before
	// [Stream.Append] waits for room — the producers' backpressure, for a
	// stream whose retention is bounded by its readers rather than by time.
	// Zero is unbounded; negative is refused, and so is a positive value
	// below the normalized MaxRecordBytes, under which no record could fit.
	MaxBytes int64
	// BlockTimeout is how long an Append waits for room under MaxBytes before
	// it is refused with [StreamFull]. Zero refuses at once. Negative is
	// refused, and so is a positive value with no MaxBytes, which would wait
	// for nothing.
	BlockTimeout time.Duration
	// SegmentBytes is the size of the segments a file engine writes, and the
	// unit its retention releases. Zero means [DefaultSegmentBytes]; negative
	// is refused. Engines that keep no segments ignore it.
	SegmentBytes int64
	// CompactAfter is how old a record must be before a [Compactor] may
	// drop it, measured from its [EntryValue.AppendedAt]: a reader that
	// stays within that window behind the head sees every update, and a
	// tombstone is kept one more CompactAfter after it became compactable.
	// Zero never compacts; negative is refused.
	CompactAfter time.Duration
}

type RecordValue struct {
	// Key places the record: records of one key go to one partition
	// ([PartitionOf]) and are read in the order they were appended. It may
	// be empty, for a record whose order relative to others does not matter.
	Key []byte
	// Payload is the record's bytes. It may be empty; under a key, an empty
	// payload is a tombstone, the key's deletion for a [Compactor].
	Payload []byte
}

type Seeker interface {
	// OffsetAt returns the first offset of partition whose [EntryValue.AppendedAt]
	// is at or after t: the partition's first retained offset when t precedes
	// every record it retains, its next offset when t follows every one. A
	// partition outside [0, Partitions()) is [UnknownPartition].
	OffsetAt(ctx context.Context, partition int, t time.Time) (Offset, error)
}

type Stream interface {
	// Append writes records to one partition as one batch and returns where
	// they landed.
	//
	// The partition is the records' keys': [PartitionOf] of the first keyed
	// record, every other keyed record required to hash to the same partition
	// ([PartitionsMixed] otherwise, nothing written), and an unkeyed record
	// following the batch. A batch with no key at all goes to the partitions in
	// turn. The records take consecutive offsets in their order.
	//
	// When it returns nil, a durable implementation has flushed the whole batch
	// to its device under one sync: a crash one instruction later loses none of
	// it, and a crash before loses all of it, never a part. An implementation
	// that joins the transaction ctx carries (ADR 0151) has instead made the
	// batch part of that transaction.
	//
	// An empty batch is [InvalidBatchSize]; a record whose key and payload
	// together exceed [PolicyValue.MaxRecordBytes] is [RecordTooLarge], the
	// whole batch refused. A stream holding [PolicyValue.MaxBytes] waits up to
	// [PolicyValue.BlockTimeout] for retention to make room, then refuses with
	// [StreamFull]; ctx ends the wait earlier with its own cause.
	Append(ctx context.Context, records []RecordValue) (AppendValue, error)
	// Read returns up to max entries of partition, from offset from on, in
	// offset order. It does NOT block waiting for records: a read at the
	// partition's next offset yields an empty slice and a nil error, and
	// waiting is the caller's ([Waker] tells it when to look again). The
	// offsets a [Compactor] emptied are skipped: the entries carry their own
	// offsets, which need not be consecutive.
	//
	// from below the partition's first retained offset is [OffsetTrimmed] —
	// the records are gone, and silently starting later would hide the gap; one
	// past the next offset is [OffsetAhead]. A partition outside
	// [0, Partitions()) is [UnknownPartition], and a non-positive max is
	// [InvalidBatchSize].
	Read(ctx context.Context, partition int, from Offset, max int) ([]EntryValue, error)
	// Bounds returns the first offset partition still retains and the next one
	// it will assign. A partition outside [0, Partitions()) is
	// [UnknownPartition].
	Bounds(ctx context.Context, partition int) (BoundsValue, error)
	// Partitions returns how many partitions the stream has, at least one. It
	// never changes for the life of a stream: a key's partition is its hash
	// modulo this number, and changing it would move every key.
	Partitions() int
}

type Trimmer interface {
	// Trim makes before the first offset partition retains: every record below
	// it becomes unreadable at once ([OffsetTrimmed]) and its storage is
	// released when the engine can — a file engine releases whole segments only.
	// A before at or below the first offset changes nothing; one past the next
	// offset is [OffsetAhead]; a partition outside [0, Partitions()) is
	// [UnknownPartition]. Offsets are never reused: the next offset does not
	// move.
	Trim(ctx context.Context, partition int, before Offset) error
}

type Waker interface {
	// Wake returns a channel closed by the next Append through this stream value
	// in this process, to any partition. Call it BEFORE the Read that may come
	// back empty: an append landing between that Read and the wait closes the
	// channel already handed out, so it is not missed.
	Wake() <-chan struct{}
}