@absolutejs/sync-pack-digest

bun add @absolutejs/sync-pack-digest

bun add @absolutejs/sync-pack-digest

import { createSyncEngine } from '@absolutejs/sync/engine';
import { createDigestPack } from '@absolutejs/sync-pack-digest';

const engine = createSyncEngine();
engine.registerPack(
	createDigestPack({
		// Cron pattern for the digest fire. Default '0 8 * * 1' (Mon 8am).
		cron: '0 8 * * 1',

		// REQUIRED: yield the actor ids to iterate per fire. The host's
		// data layer is the source of truth for "who exists" — the pack
		// just maintains per-actor cursors.
		listActors: () => db.users.allActiveIds(),

		// REQUIRED: build the payload for one actor. `since` is the actor's
		// last successful send (or null on first-ever). Return null to skip
		// this actor silently (no content this week).
		buildDigest: async (actorId, since) => {
			const items = await db.feed.since(actorId, since ?? new Date(0));
			if (items.length === 0) return null;
			return {
				to: await db.users.email(actorId),
				subject: `Your weekly digest`,
				body: render(items),
			};
		},

		// REQUIRED: host's transport adapter. Pack catches throws per
		// actor so one bad send doesn't block the rest.
		send: async (msg) => await resend.send(msg),

		// OPTIONAL: back-pressure. Default 1000.
		maxActorsPerFire: 1000,

		// OPTIONAL: per-actor cool-down. Default 168 hours = 1 week. Match
		// to your cron — weekly cron + 168h = each actor gets one digest
		// per fire; daily cron + 168h = each actor gets at most one per week.
		minHoursBetweenDigests: 168,

		// OPTIONAL: surface failures (default logs to console.error).
		onActorFailure: ({ actorId, phase, error }) => {
			logger.warn({ actorId, phase, error }, 'digest failure');
		},

		// OPTIONAL: outer schedule retry (sync 1.9.0+). Defaults to no
		// retry — per-actor failures are handled internally; this is for
		// transient infra failures (DB, etc) of the whole fire. The next
		// cron fire catches up anyway.
		// retry: { maxAttempts: 3, backoff: exponentialBackoff() },
	})
);

type DigestCursor = {
	id: string; // = actorId
	actorId: string;
	lastSentAt: number; // epoch ms
	lastSubject: string;
};

import {
	createDigestPack,
	type DigestStore,
} from '@absolutejs/sync-pack-digest';

const store: DigestStore = {
	getById: (id) => /* SELECT * FROM digest_cursors WHERE id = $1 */,
	reader: { all: () => /* SELECT * FROM digest_cursors */ },
	writer: { insert, update, delete },
};