Digest

@visulima/notification/digest collapses a burst of events into one notification — the antidote to notification fatigue ("50 people liked your post" instead of 50 pushes). Events are grouped by a key into a time (or cron) window; when the window closes, every collected event is flushed together.

It is store-backed and edge-safe; croner is an optional peer used for cron windows:

npm install @visulima/notification croner

Usage

import { createDigester } from "@visulima/notification/digest";

const digester = createDigester<{ subscriberId: string; postId: string }>({
    // Group events into one window per subscriber + post.
    key: (event) => `${event.subscriberId}:${event.postId}`,
    // Collect for 10 minutes from the first event.
    window: { amount: 10, unit: "minutes" },
    // Called once when the window closes, with every event.
    onFlush: (events, key) => runtime.trigger(summaryWorkflow, { count: events.length, events }),
});

// On each incoming event:
await digester.add({ subscriberId: "u1", postId: "p1" }); // opens a window
await digester.add({ subscriberId: "u1", postId: "p1" }); // folds into the open window

// Flush closed windows on a timer (cron job, Cloudflare alarm, alongside runtime.sweep):
await digester.sweep();

add returns a boolean — true when it opened a new window for the key. sweep(now?, limit?) flushes every window whose wake-at has passed (capped at limit, the rest carry to the next sweep) and returns how many it flushed; it rejects a non-positive limit.

Flushing is at-least-once: a window is removed only after its onFlush resolves, so a throwing onFlush is retried on the next sweep (and may run again if the subsequent removal fails) — make onFlush idempotent.

Windows

window is a Duration: a number of milliseconds, a structured { amount, unit }, or a { cron } expression. Pass a function (event) => Duration to vary the window per event. A window's close time is fixed by its first event and later events do not extend it — including for cron: a { cron: "0 9 * * *" } window closes at the next 9am after the first event of each burst, not on an absolute daily schedule. Non-finite windows (NaN/Infinity) and an impossible cron are rejected at add time.

Stores

createDigester defaults to an in-process store. For durability across restarts/instances pass an UnstorageDigestStore (any unstorage driver — Cloudflare KV/D1, Redis, filesystem):

import { createDigester, UnstorageDigestStore } from "@visulima/notification/digest";
import { createStorage } from "unstorage";

const digester = createDigester({
    key: (event) => event.subscriberId,
    window: { amount: 1, unit: "hours" },
    onFlush: (events) => sendSummary(events),
    store: new UnstorageDigestStore(createStorage()),
});

Each window is a self-contained document (no shared index), so concurrent adds for different keys can't lose-update each other; writes are still not transactional, so for high-contention multi-writer setups prefer a store with atomic guarantees.

Digest vs. workflow step.digest

This digester aggregates before triggering — it batches raw events and fires one workflow/notification per window. That keeps the model simple and the engine store-agnostic. Combine it with workflows by having onFlush trigger a notification workflow with the batched events.