digestr

README

digestr

A pure-Go notification digest / batching engine: many events per recipient in, one batched notification out, on a window you control. Zero dependencies.

Every notification product eventually needs "email me a digest of the 12 things that happened, not 12 emails." The category leaders sell that as a hosted workflow step — six containers, MongoDB and Redis, the digest computed on their servers. Their Go artifacts are generated REST clients: Trigger and Cancel, nothing local. General-purpose Go debouncers don't close the gap either — their callbacks are func() with no payload and no key, so they collapse repeated calls last-wins but cannot accumulate N events per recipient into a batch. Single-value debounce is not a digest.

digestr is just the batching core, extracted: it does the grouping-and- windowing computation and only that. It never sends, stores, templates, or owns a channel — which is exactly why it needs no database, no broker, and no running process.

• Pure and clock-driven. No goroutines, no timers. The caller advances time by passing now to Add and Due, so the engine is deterministic, trivial to test, and safe to persist.
• Two triggers. A window flushes on time (it elapsed) or size (a count ceiling — Knock's batch_size, which bounds latency for a hot key), whichever comes first.
• Two strategies. Fixed digests every event; Backoff passes a lone event straight through and only digests when events repeat within a look-back horizon (Novu's "when events repeat").
• Dedup within a window by a caller-supplied idempotency id.
• Deterministic flush ordering — same construction, same now, identical output every run.
• Serializable state. Snapshot the open windows, persist them however you like, and restore to resume exactly across a restart.
• Concurrency-safe, race-clean, generic over your event type.
• Zero dependencies. Go 1.23+.

import "github.com/zkrebbekx/digestr"

Example

d, _ := digestr.New(digestr.Config[Event]{
    Key:    func(e Event) string { return e.Recipient },
    Window: time.Minute,
})

d.Add(t0, Event{Recipient: "amy", Text: "liked your post"})
d.Add(t0.Add(10*time.Second), Event{Recipient: "amy", Text: "commented"})
d.Add(t0.Add(20*time.Second), Event{Recipient: "amy", Text: "followed you"})

for _, b := range d.Due(t0.Add(time.Minute)) {
    send(b.Key, b.Events) // one digest of three events, Reason Elapsed
}

The model — a clock-driven state machine

digestr holds one open window per active key. The first event for a key opens a window [T, T+Window); later events for that key join it. A window flushes exactly once, into a Batch, on whichever trigger fires first:

• Time. Due(now) flushes every window whose DueAt is at or before now — Reason Elapsed. The boundary is inclusive: a window opened at T is due at exactly T+Window. Due closes what it flushes, sorts the batches by key, and is idempotent (a second Due(now) returns nothing).
• Size. If MaxSize > 0 and a window reaches that many distinct events, the Add that fills it flushes it immediately — Reason SizeReached, before the time trigger. A later Due does not re-emit those events.

type Config[E any] struct {
    Key      func(E) string // grouping key; required
    Window   time.Duration  // how long a window stays open; required, > 0
    Strategy Strategy       // Fixed (default) or Backoff
    MaxSize  int            // 0 = no size trigger; N>0 flushes early at N
    Lookback time.Duration  // Backoff only: the "recent send" horizon
    Dedup    func(E) string // optional; "" id is never deduplicated
}

The Add contract

Add(now, e) returns (immediate []E, flushed *Batch[E]) and sets at most one of them:

Result	Meaning
(nil, nil)	The event joined an open window, or was dropped as a duplicate. The common case.
([e], nil)	A Backoff passthrough: no window, deliver e now.
(nil, &batch)	This Add filled a window to MaxSize; deliver batch now.

They are never both set: a passthrough opens no window to flush, and a size flush requires an open window that a passthrough would not have created. Add never time-flushes — that is Due's job.

Strategies

Fixed (Novu "Regular") digests every event: the first opens a window and the rest join.

Backoff (Novu "when events repeat") digests only repeats. A lone event for a key with no recent send passes straight through, returned as immediate, and opens no window; only a follow-up within Lookback opens a digest. Both a passthrough delivery and a batch flush count as a "send" that starts the look-back horizon, so a cold key's second event within Lookback digests — see the correction note in docs/DESIGN.md. The recency horizon is half-open: a send exactly Lookback ago is no longer recent.

d, _ := digestr.New(digestr.Config[Event]{
    Key: recipient, Window: time.Minute,
    Strategy: digestr.Backoff, Lookback: 5 * time.Minute,
})
imm, _ := d.Add(t0, Event{Recipient: "amy", Text: "first"})       // len(imm)==1: delivered now
imm2, _ := d.Add(t0.Add(30*time.Second), Event{Recipient: "amy"}) // len(imm2)==0: digesting

Dedup

A non-nil Dedup collapses duplicates within a single window: an event whose non-empty id a still-open window has already seen is dropped, and the batch keeps the one copy. An empty ("") id is never deduplicated, and dedup never spans windows.

Out-of-order time

The caller drives the clock, so digestr defends against a non-monotonic one. It tracks the highest now it has observed; any earlier now handed to Add or Due is clamped forward to it. Windows therefore never carry a negative span, Due never un-fires, and no event is lost. Call Due on a cadence no coarser than your smallest Window; between Due calls an overdue window keeps accepting joins (it flushes lazily on the next Due).

Persistence — the crash-safety boundary

An open window is state: if the process dies, un-flushed events must not be lost or double-sent. digestr keeps that state in memory and makes it yours to persist, rather than dragging in a database.

snap := d.Snapshot()               // serializable open windows + markers + clock
blob, _ := json.Marshal(snap)      // persist however you like
// ... restart ...
var s digestr.Snapshot[Event]
json.Unmarshal(blob, &s)
d, _ := digestr.Restore(cfg, s)    // resumes exactly; Config supplied fresh

Only data persists — never the Config's function fields, which you supply afresh to Restore. For a Snapshot[E] to survive a JSON round-trip, your event type E must have exported fields, as with any encoding/json value.

Concurrency

A *Digester is safe for concurrent Add and Due; every method takes an internal mutex, and Snapshot is a consistent point-in-time copy. A notification hot path is inherently concurrent, so this is the default, not an option.

Errors

New and Restore return typed sentinels matchable with errors.Is: ErrNoKey, ErrBadWindow, ErrBadSize, ErrBadLookback and ErrBadStrategy. A MaxSize of zero is valid (no size trigger); a negative one is an error.

Non-goals

• No delivery. digestr tells you what to send and when; the channel is yours. This is why it needs no SMTP, no push gateway, no broker.
• No templating, channels, or preferences. Rendering the batch and picking email vs push are the caller's.
• No storage or running process. digestr is a computation, not a daemon. It owns no goroutines and no timers; crash-safe persistence is delegated to the caller via snapshot/restore.
• No quiet hours. A scheduling concern the caller composes on top.

License

MIT

Documentation

Overview

Package digestr is a pure-Go notification digest / batching engine: many events per recipient in, one batched notification out, on a window the caller controls. It has zero dependencies — only the standard library.

digestr does the grouping-and-windowing computation that notification platforms monetize as a workflow step, and only that. It does not send, store, template, or own channels. It owns no goroutines and no timers: the caller advances the clock by passing now to Digester.Add and Digester.Due, which makes the engine deterministic, trivially testable, and safe to persist.

d, _ := digestr.New(digestr.Config[Event]{
	Key:    func(e Event) string { return e.Recipient },
	Window: time.Minute,
})
d.Add(t0, Event{Recipient: "amy", Text: "liked your post"})
d.Add(t0.Add(10*time.Second), Event{Recipient: "amy", Text: "commented"})
for _, b := range d.Due(t0.Add(time.Minute)) {
	send(b.Key, b.Events) // one digest of two events
}

The model

digestr holds one open Window per active key. The first event for a key opens a window [T, T+Window); later events for that key join it. A window flushes exactly once, into a Batch, on whichever trigger fires first:

• Time. Digester.Due(now) flushes every window whose DueAt is at or before now (Reason Elapsed). The boundary is inclusive: a window opened at T is due at exactly T+Window.
• Size. If Config.MaxSize is positive and a window reaches that many distinct events, the Digester.Add that fills it flushes it immediately (Reason SizeReached) — Knock's batch_size, which bounds latency for a hot key. A later Due does not re-emit those events.

Strategies

Fixed digests every event: the first opens a window and the rest join. Backoff digests only repeats — a lone event for a key with no recent send passes straight through undigested (returned by Add as immediate), and only a follow-up within Config.Lookback opens a window. The recency horizon is half-open: a send exactly Lookback ago is not recent. Both a passthrough delivery and a batch flush count as a "send" that starts the horizon, so a cold key's second event within Lookback digests — see the correction note in docs/DESIGN.md.

The Add contract

Digester.Add returns at most one of two results, never both. immediate carries a Backoff passthrough (nil under Fixed, or whenever a window is already open); flushed carries a Batch only when that Add reached MaxSize. Both nil is the ordinary case: the event joined an open window, or was dropped as a duplicate. Add never time-flushes.

Dedup

A non-nil Config.Dedup collapses duplicates within one window: an event whose non-empty id a still-open window has already seen is dropped, and the batch keeps the single copy. An empty ("") id is never deduplicated, and dedup never spans windows.

Determinism

Digester.Due sorts flushed batches by key, and events keep arrival order, so the same construction over the same now produces identical output on every run — no map-iteration nondeterminism. Draining is idempotent: a second Due with the same now returns nothing.

Out-of-order time

The caller drives the clock, so digestr defends against a non-monotonic one. It tracks the highest now it has observed; any earlier now handed to Add or Due is clamped forward to it. Windows therefore never carry a negative span, Due never un-fires, and no event is lost.

Persistence

An open window is state that must survive a crash. digestr keeps it in memory and makes it the caller's to persist rather than dragging in a database. Digester.Snapshot returns a serializable point-in-time copy of the open windows, the Backoff markers, and the clock; Restore rebuilds a Digester from a fresh Config and that snapshot, resuming exactly. Only data persists — never the Config's function fields.

Concurrency

A *Digester is safe for concurrent Add and Due; every method takes an internal mutex, and Snapshot is a consistent point-in-time copy.

Errors

New and Restore return typed sentinels matchable with errors.Is: ErrNoKey, ErrBadWindow, ErrBadSize, ErrBadLookback and ErrBadStrategy.

Not in scope

No delivery, channels, templates, preferences, quiet hours, goroutines, timers, or database. digestr computes batches; everything else is the caller's. See docs/DESIGN.md, including the "Phase 1 as built" note recording the decisions this implementation made where the design left a choice open.

Example

Example digests three activities for one recipient into a single batch when the window elapses.

package main

import (
	"fmt"
	"time"

	"github.com/zkrebbekx/digestr"
)

// Event is a caller's notification payload. digestr never inspects it beyond
// the key and dedup id the caller derives from it.
type Event struct {
	Recipient string
	ID        string
	Text      string
}

func recipient(e Event) string { return e.Recipient }

var base = time.Date(2026, time.January, 1, 9, 0, 0, 0, time.UTC)

func main() {
	d, _ := digestr.New(digestr.Config[Event]{
		Key:    recipient,
		Window: time.Minute,
	})

	d.Add(base, Event{Recipient: "amy", Text: "liked your post"})
	d.Add(base.Add(10*time.Second), Event{Recipient: "amy", Text: "commented"})
	d.Add(base.Add(20*time.Second), Event{Recipient: "amy", Text: "followed you"})

	for _, b := range d.Due(base.Add(time.Minute)) {
		fmt.Printf("%s: %d updates (%s)\n", b.Key, len(b.Events), b.Reason)
	}
}

Output:
amy: 3 updates (Elapsed)

Example (Backoff)

Example_backoff delivers a lone event immediately and only digests once events repeat within the look-back horizon.

package main

import (
	"fmt"
	"time"

	"github.com/zkrebbekx/digestr"
)

// Event is a caller's notification payload. digestr never inspects it beyond
// the key and dedup id the caller derives from it.
type Event struct {
	Recipient string
	ID        string
	Text      string
}

func recipient(e Event) string { return e.Recipient }

var base = time.Date(2026, time.January, 1, 9, 0, 0, 0, time.UTC)

func main() {
	d, _ := digestr.New(digestr.Config[Event]{
		Key:      recipient,
		Window:   time.Minute,
		Strategy: digestr.Backoff,
		Lookback: 5 * time.Minute,
	})

	imm, _ := d.Add(base, Event{Recipient: "amy", Text: "first"})
	fmt.Printf("lone event delivered now: %d\n", len(imm))

	repeat, _ := d.Add(base.Add(30*time.Second), Event{Recipient: "amy", Text: "repeat"})
	fmt.Printf("repeat delivered now: %d (it is digesting instead)\n", len(repeat))
}

Output:
lone event delivered now: 1
repeat delivered now: 0 (it is digesting instead)

Example (Dedup)

Example_dedup collapses events that share a non-empty id within one window.

package main

import (
	"fmt"
	"time"

	"github.com/zkrebbekx/digestr"
)

// Event is a caller's notification payload. digestr never inspects it beyond
// the key and dedup id the caller derives from it.
type Event struct {
	Recipient string
	ID        string
	Text      string
}

func recipient(e Event) string { return e.Recipient }

var base = time.Date(2026, time.January, 1, 9, 0, 0, 0, time.UTC)

func main() {
	d, _ := digestr.New(digestr.Config[Event]{
		Key:    recipient,
		Window: time.Minute,
		Dedup:  func(e Event) string { return e.ID },
	})

	d.Add(base, Event{Recipient: "amy", ID: "order-42", Text: "shipped"})
	d.Add(base.Add(time.Second), Event{Recipient: "amy", ID: "order-42", Text: "shipped"})
	d.Add(base.Add(2*time.Second), Event{Recipient: "amy", ID: "order-43", Text: "delivered"})

	for _, b := range d.Due(base.Add(time.Minute)) {
		fmt.Printf("%d distinct events\n", len(b.Events))
	}
}

Output:
2 distinct events

Example (SizeTrigger)

Example_sizeTrigger flushes early once a window reaches MaxSize, without waiting for the (here, one-hour) time trigger.

package main

import (
	"fmt"
	"time"

	"github.com/zkrebbekx/digestr"
)

// Event is a caller's notification payload. digestr never inspects it beyond
// the key and dedup id the caller derives from it.
type Event struct {
	Recipient string
	ID        string
	Text      string
}

func recipient(e Event) string { return e.Recipient }

var base = time.Date(2026, time.January, 1, 9, 0, 0, 0, time.UTC)

func main() {
	d, _ := digestr.New(digestr.Config[Event]{
		Key:     recipient,
		Window:  time.Hour,
		MaxSize: 3,
	})

	for i, txt := range []string{"one", "two", "three"} {
		_, flushed := d.Add(base.Add(time.Duration(i)*time.Second), Event{Recipient: "amy", Text: txt})
		if flushed != nil {
			fmt.Printf("flushed %d on %s\n", len(flushed.Events), flushed.Reason)
		}
	}
}

Output:
flushed 3 on SizeReached

Index

• Variables
• type Batch
• type Config
• type Digester
  • func New[E any](cfg Config[E]) (*Digester[E], error)
  • func Restore[E any](cfg Config[E], s Snapshot[E]) (*Digester[E], error)
  • func (d *Digester[E]) Add(now time.Time, e E) (immediate []E, flushed *Batch[E])
  • func (d *Digester[E]) Due(now time.Time) []Batch[E]
  • func (d *Digester[E]) Pending() []Window
  • func (d *Digester[E]) Snapshot() Snapshot[E]
• type FlushReason
  • func (r FlushReason) String() string
• type KeyTime
• type Snapshot
• type Strategy
  • func (s Strategy) String() string
• type Window
• type WindowState

Examples

• Package
• Package (Backoff)
• Package (Dedup)
• Package (SizeTrigger)

Variables

var (
	// ErrNoKey is returned when a Config has a nil Key function. digestr groups
	// events by the string that Key returns for each; without it there is no
	// grouping and nothing to do.
	ErrNoKey = errors.New("digestr: nil Key function")

	// ErrBadWindow is returned when a Config's Window is not strictly positive.
	// A window is the span an open digest stays open for; a zero or negative
	// span cannot hold events.
	ErrBadWindow = errors.New("digestr: Window must be positive")

	// ErrBadSize is returned when a Config's MaxSize is negative. Zero is valid
	// and means "no size trigger"; a positive value is the count ceiling that
	// flushes a window early.
	ErrBadSize = errors.New("digestr: MaxSize must not be negative")

	// ErrBadLookback is returned when a Config selects the Backoff strategy but
	// its Lookback is not strictly positive. Backoff decides passthrough versus
	// digest by whether a send happened within the last Lookback, so a zero or
	// negative horizon is meaningless.
	ErrBadLookback = errors.New("digestr: Backoff requires a positive Lookback")

	// ErrBadStrategy is returned when a Config names a Strategy that is neither
	// [Fixed] nor [Backoff].
	ErrBadStrategy = errors.New("digestr: unknown strategy")

	// ErrBadSnapshot is returned by [Restore] when a Snapshot is malformed — for
	// example, two windows sharing one key, which would silently drop events. A
	// Snapshot produced by [Digester.Snapshot] is always well-formed; this
	// guards against a corrupt or hand-built one.
	ErrBadSnapshot = errors.New("digestr: malformed snapshot")
)

Sentinel errors returned by New and Restore when a Config is unusable. Match them with errors.Is rather than by comparing error strings.

Types

type Batch

type Batch[E any] struct {
	// Key is the grouping key these events shared.
	Key string

	// Events are the digested events in arrival order, with duplicates already
	// collapsed if a [Config.Dedup] was set. The slice is the caller's to keep
	// and mutate.
	Events []E

	// OpenedAt is when the window's first event arrived. DueAt is OpenedAt plus
	// the window; for a SizeReached flush it is the time trigger the batch beat.
	OpenedAt time.Time
	DueAt    time.Time

	// Reason is why the batch flushed.
	Reason FlushReason
}

Batch is one flushed digest: the aggregated events for a key, ready for the caller to render and deliver. digestr never delivers it.

type Config

type Config[E any] struct {
	// Key groups events; it is required. All events sharing a key digest
	// together. A common key is the recipient id, or recipient plus a
	// sub-topic (Knock's concat(recipient_id, batch_key)).
	Key func(E) string

	// Window is how long a window stays open. A window opened at instant T is
	// due, and flushes, at exactly T+Window. Required to be positive.
	Window time.Duration

	// Strategy is [Fixed] (default) or [Backoff].
	Strategy Strategy

	// MaxSize, when positive, is a count ceiling (Knock's batch_size): a window
	// that reaches MaxSize distinct events flushes immediately on [Digester.Add]
	// without waiting for the time trigger. Zero disables the size trigger.
	MaxSize int

	// Lookback is the Backoff recency horizon and is used only when Strategy is
	// [Backoff]. A key is "recent" — and so a new event opens a digest rather
	// than passing through — when a send happened strictly less than Lookback
	// ago. Required to be positive under Backoff.
	Lookback time.Duration

	// Dedup, when non-nil, collapses duplicates within a single window: if it
	// returns a non-empty id that a still-open window has already seen, the
	// event is dropped and the batch keeps the one copy. An empty ("") id is
	// never deduplicated. Dedup does not span windows.
	Dedup func(E) string
}

Config describes a Digester. E is the caller's event payload; digestr is generic over it and never inspects it beyond the key and optional dedup id the caller derives from it.

The function fields are behaviour, not data: they are supplied fresh at New (or Restore) and are never serialized by Snapshot. Two Digesters restored from the same snapshot with the same Config behave identically.

type Digester

type Digester[E any] struct {
	// contains filtered or unexported fields
}

Digester is a pure, clock-driven digest state machine. It holds one open window per active key and starts no goroutines or timers: the caller advances time by passing now to Digester.Add and Digester.Due. The zero value is not usable; construct one with New or Restore.

A *Digester is safe for concurrent use — every method takes an internal mutex — so it can sit on a notification hot path shared by many producers.

func New

func New[E any](cfg Config[E]) (*Digester[E], error)

New builds a Digester from cfg, returning a typed sentinel (see the package errors) if the configuration is unusable.

func Restore

func Restore[E any](cfg Config[E], s Snapshot[E]) (*Digester[E], error)

Restore rebuilds a Digester from a fresh cfg and a Snapshot, resuming the exact window state the snapshot captured. cfg is validated as in New and its error is returned if it is unusable; the snapshot itself is trusted.

The Config must be supplied afresh because a Snapshot carries no functions. Restoring with a different Config (a wider Window, a different Dedup) resumes the persisted events under the new behaviour — that is the caller's choice.

func (*Digester[E]) Add

func (d *Digester[E]) Add(now time.Time, e E) (immediate []E, flushed *Batch[E])

Add offers event e at instant now.

It returns at most one of two results, never both:

• immediate holds the event when Backoff passes it straight through (a lone event with no recent send for its key); it is nil otherwise.
• flushed holds a Batch when this Add filled a window to Config.MaxSize and flushed it early (Reason SizeReached); it is nil otherwise.

When both are nil the event was buffered into an open window — the common case — or dropped as a duplicate of one already in that window. Add never time-flushes; the time trigger belongs to Digester.Due.

If now precedes an instant the Digester has already observed, it is clamped forward, so windows never carry a negative span and events are never lost.

func (*Digester[E]) Due

func (d *Digester[E]) Due(now time.Time) []Batch[E]

Due flushes and returns every window whose time trigger has fired at or before now — that is, whose DueAt is at or before now — closing each. Results are ordered by key (one window is open per key), so a fixed now over identical history yields identical output every run. Draining is idempotent: calling Due again with the same now returns nothing, because the due windows are already gone.

now is clamped forward if it precedes an instant already observed.

func (*Digester[E]) Pending

func (d *Digester[E]) Pending() []Window

Pending reports every open window as a metadata-only Window, ordered by key. Use it for observability or to decide when next to call [Due].

func (*Digester[E]) Snapshot

func (d *Digester[E]) Snapshot() Snapshot[E]

Snapshot returns a consistent point-in-time copy of the Digester, taken under the lock. Every slice is freshly allocated, so the caller may marshal or mutate the result while the Digester keeps running.

type FlushReason

type FlushReason int

FlushReason records why a Batch closed.

const (
	// Elapsed means the window's time trigger fired: [Digester.Due] was called
	// at or after the window's DueAt.
	Elapsed FlushReason = iota

	// SizeReached means the window hit [Config.MaxSize] and [Digester.Add]
	// flushed it early, before its time trigger.
	SizeReached
)

func (FlushReason) String

func (r FlushReason) String() string

String returns the reason's name for diagnostics.

type KeyTime

type KeyTime struct {
	Key  string    `json:"key"`
	Time time.Time `json:"time"`
}

KeyTime is a per-key instant in a Snapshot — used for the Backoff last-send markers.

type Snapshot

type Snapshot[E any] struct {
	Windows  []WindowState[E] `json:"windows,omitempty"`
	LastSend []KeyTime        `json:"lastSend,omitempty"`
	Clock    time.Time        `json:"clock"`
}

Snapshot is a serializable, point-in-time copy of a Digester's state: its open windows, its Backoff recency markers, and its clock. It is a plain exported struct with stable JSON tags, so a caller can persist it with encoding/json (or anything else) on shutdown and hand it to Restore on boot to resume exactly where it left off.

Only data is captured — never the Config's function fields. The caller supplies a fresh Config to Restore. If the event type E is to survive a JSON round-trip, its own fields must be exported, as with any encoding/json value.

The slices are sorted deterministically by key, so two snapshots of equal state serialize byte-for-byte alike.

type Strategy

type Strategy int

Strategy selects how digestr decides whether an event begins (or joins) a digest window at all.

const (
	// Fixed opens a window on the first event for a key and digests every
	// event for that key until the window flushes — Novu's "Regular" digest.
	// Every event is digested; nothing passes through.
	Fixed Strategy = iota

	// Backoff digests only when events repeat: a lone event for a key with no
	// recent send passes straight through undigested, and only a follow-up
	// within Lookback opens a window — Novu's "when events repeat". It requires
	// a positive [Config.Lookback].
	Backoff
)

func (Strategy) String

func (s Strategy) String() string

String returns the strategy's name for diagnostics.

type Window

type Window struct {
	// Key is the grouping key.
	Key string

	// Size is how many events the window currently holds (post-dedup).
	Size int

	// OpenedAt is when the window opened; DueAt is when its time trigger fires.
	OpenedAt time.Time
	DueAt    time.Time
}

Window is an observability view of one open digest, returned by Digester.Pending. It carries no events — only metadata — so it is cheap and never generic.

type WindowState

type WindowState[E any] struct {
	Key      string    `json:"key"`
	Events   []E       `json:"events"`
	OpenedAt time.Time `json:"openedAt"`
	DueAt    time.Time `json:"dueAt"`
}

WindowState is one open window inside a Snapshot. The dedup set is not stored: it is rebuilt from Events on Restore using the fresh Config's Dedup, since the surviving events already are exactly the distinct ids seen.