orderedlog

type Applier func(tx Txn, op Op) ([]Mutation, error)

type BackendOptions struct {
	// WatchBuffer is the per-watcher channel capacity. Defaults to 1024.
	// A watcher that fails to drain within capacity is dropped (its
	// channel is closed) so a single slow consumer cannot stall others.
	WatchBuffer int

	// RetentionAge is how long committed events are kept in the log. A
	// watcher requesting a sequence older than this window receives
	// ErrCompacted. Defaults to 24h. Set to 0 to disable age-based GC.
	RetentionAge time.Duration

	// RetentionMaxEvents is the hard cap on the number of events kept in
	// the log. The pruner keeps at least this many, regardless of age.
	// Defaults to 100_000. Set to 0 to disable count-based GC.
	RetentionMaxEvents uint64

	// RetentionInterval is how often the background pruner runs.
	// Defaults to 5 minutes.
	RetentionInterval time.Duration

	// Logger is the structured logger to use. Defaults to the global
	// "orderedlog" logger.
	Logger log.Logger
}

type BadgerBackend struct {
	// contains filtered or unexported fields
}

type Event struct {
	// Seq is the strictly monotonic sequence assigned by the backend.
	// Starts at 1; never zero, never reused.
	Seq uint64
	// OpType is the Op type that produced this Event.
	OpType string
	// Mutations are the resource changes the Op committed.
	Mutations []Mutation
}

type Mutation struct {
	// Kind classifies the change.
	Kind MutationKind
	// ResourceType is a short, stable identifier (e.g. "service",
	// "endpoints", "clusternetwork"). Used by watchers to filter.
	ResourceType string
	// Namespace scopes the mutation. Empty for cluster-scoped resources.
	Namespace string
	// Name uniquely identifies the resource within (ResourceType, Namespace).
	Name string
	// Payload is the marshalled new value for Put mutations and nil for
	// Delete mutations. Format is the resource type's chosen encoding;
	// consumers and the backend must agree on it out of band.
	Payload []byte
}

type MutationKind uint8

type Op interface {
	// OpType returns a stable identifier used by the codec registry.
	// Stable across releases; treat as a wire identifier.
	OpType() string

	// Marshal returns the canonical byte representation of the Op.
	// Two equivalent Ops MUST marshal to byte-equal output.
	Marshal() ([]byte, error)
}

type OpUnmarshaler func(data []byte) (Op, error)

type OrderedLog interface {
	// Register binds an Applier and OpUnmarshaler to an OpType. It MUST
	// be called for every OpType the consumer intends to Propose, before
	// the first Propose for that type. Returns ErrAlreadyRegistered if
	// called twice for the same type.
	Register(opType string, applier Applier, unmarshal OpUnmarshaler) error

	// Propose serializes the Op, applies it via the registered Applier
	// inside a backend transaction, assigns a sequence number, persists
	// the resulting Event, and publishes it to watchers.
	//
	// Propose returns the assigned sequence number on success. On error,
	// the transaction is rolled back and no sequence number is consumed.
	//
	// Propose is strictly serialized: at most one Op is in-flight per
	// OrderedLog at any time.
	Propose(ctx context.Context, op Op) (seq uint64, err error)

	// Watch subscribes to the event stream starting at fromSeq+1.
	//
	// Pass fromSeq=0 to receive every event from the start of the
	// retained log. If fromSeq is older than the retention window,
	// Watch returns ErrCompacted; the caller should Snapshot and resume.
	//
	// The returned channel is closed when ctx is cancelled or the
	// OrderedLog is closed. Slow consumers may be dropped (channel
	// closed early) if they fall too far behind; the buffer size is
	// implementation-defined.
	Watch(ctx context.Context, fromSeq uint64) (<-chan Event, error)

	// Snapshot returns a point-in-time view of committed state and the
	// sequence number it was taken at. Use after ErrCompacted.
	Snapshot(ctx context.Context) (snap Snapshot, seq uint64, err error)

	// Close shuts the OrderedLog down. In-flight Propose calls return
	// ErrClosed; existing watchers' channels are closed.
	Close() error
}

type Snapshot interface {
	// Close releases any resources held by the snapshot. Safe to call
	// multiple times.
	Close() error

	// Range iterates committed key/value pairs sharing prefix and
	// invokes visit for each. visit must not retain the byte slices
	// past its return. A non-nil error from visit aborts iteration
	// and is returned to the caller.
	Range(prefix []byte, visit func(key, value []byte) error) error
}

type Txn interface {
	// Get returns the value stored at key, or os.ErrNotExist (wrapped)
	// if the key has never been written.
	Get(key []byte) ([]byte, error)
	// Set writes value at key, creating the key if needed.
	Set(key, value []byte) error
	// Delete removes the key. Idempotent.
	Delete(key []byte) error
}