[V0_FILE]typescriptreact:file="app/components/demo.tsx" "use client"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import type { DigestEvent, DigestLogEvent, DigestSnapshot, } from "@/app/api/mock/store"; import { DigestCodeWorkbench } from "./digest-code-workbench"; type LifecycleState = "idle" | "collecting" | "sending" | "completed" | "empty"; type HighlightTone = "amber" | "cyan" | "green" | "red"; type StartResponse = { runId: string; token: string; status: string; snapshot: DigestSnapshot; }; type DigestWorkflowLineMap = { createToken: number[]; raceLoop: number[]; pushEvent: number[]; sendDigest: number[]; returnSent: number[]; returnEmpty: number[]; }; type DigestStepLineMap = { sendBody: number[]; returnLine: number[]; }; type ScheduledDigestDemoProps = { workflowCode: string; workflowHtmlLines: string[]; workflowLineMap: DigestWorkflowLineMap; stepCode: string; stepHtmlLines: string[]; stepLineMap: DigestStepLineMap; }; const POLL_INTERVAL_MS = 120; const EVENT_PRESETS: Array<{ type: string; message: string; label: string }> = [ { type: "comment", message: "Alice commented on your post", label: "Comment" }, { type: "like", message: "Bob liked your photo", label: "Like" }, { type: "mention", message: "Carol mentioned you in a thread", label: "Mention" }, { type: "follow", message: "Dave started following you", label: "Follow" }, ]; function formatElapsedLabel(durationMs: number): string { const seconds = (durationMs / 1000).toFixed(2); return `${seconds}s`; } async function postJson( url: string, body: Record, signal: AbortSignal ): Promise { const response = await fetch(url, { method: "POST", headers: { "Content-Type": "application/json", }, body: JSON.stringify(body), signal, }); const payload = (await response .json() .catch(() => ({ error: "Failed to parse JSON response" }))) as | T | { error?: string | { message?: string } }; if (signal.aborted) { throw new DOMException("Request aborted", "AbortError"); } if (!response.ok) { const payloadError = typeof payload === "object" && payload && "error" in payload ? payload.error : undefined; const message = typeof payloadError === "string" ? payloadError : typeof payloadError === "object" && payloadError && "message" in payloadError && typeof payloadError.message === "string" ? payloadError.message : `Request failed (${response.status})`; throw new Error(message); } return payload as T; } export function ScheduledDigestDemo({ workflowCode, workflowHtmlLines, workflowLineMap, stepCode, stepHtmlLines, stepLineMap, }: ScheduledDigestDemoProps) { const [lifecycle, setLifecycle] = useState("idle"); const [runId, setRunId] = useState(null); const [token, setToken] = useState(null); const [snapshot, setSnapshot] = useState(null); const [error, setError] = useState(null); const pollTimerRef = useRef | null>(null); const inFlightRef = useRef(false); const abortRef = useRef(null); const startButtonRef = useRef(null); const stopPolling = useCallback(() => { if (pollTimerRef.current) { clearInterval(pollTimerRef.current); pollTimerRef.current = null; } inFlightRef.current = false; }, []); const ensureAbortController = useCallback((): AbortController => { if (!abortRef.current || abortRef.current.signal.aborted) { abortRef.current = new AbortController(); } return abortRef.current; }, []); useEffect(() => { return () => { stopPolling(); abortRef.current?.abort(); abortRef.current = null; }; }, [stopPolling]); const fetchStatusSnapshot = useCallback(async (targetRunId: string, signal: AbortSignal) => { const response = await fetch( `/api/mock/status?runId=${encodeURIComponent(targetRunId)}`, { cache: "no-store", signal, } ); const payload = (await response .json() .catch(() => ({ error: "Failed to parse status response" }))) as | DigestSnapshot | { error?: string | { message?: string } }; if (signal.aborted) { throw new DOMException("Request aborted", "AbortError"); } if (!response.ok) { const payloadError = typeof payload === "object" && payload && "error" in payload ? payload.error : undefined; const message = typeof payloadError === "string" ? payloadError : typeof payloadError === "object" && payloadError && "message" in payloadError && typeof payloadError.message === "string" ? payloadError.message : `Status request failed (${response.status})`; throw new Error(message); } return payload as DigestSnapshot; }, []); const beginPolling = useCallback( (targetRunId: string) => { stopPolling(); const poll = async () => { if (inFlightRef.current) { return; } inFlightRef.current = true; let signal: AbortSignal | null = null; try { const controller = ensureAbortController(); signal = controller.signal; const nextSnapshot = await fetchStatusSnapshot(targetRunId, signal); if (signal.aborted) { return; } setSnapshot(nextSnapshot); if (nextSnapshot.status === "completed") { if (signal.aborted) return; setLifecycle("completed"); stopPolling(); return; } if (nextSnapshot.status === "empty") { if (signal.aborted) return; setLifecycle("empty"); stopPolling(); return; } if (nextSnapshot.status === "sending") { if (signal.aborted) return; setLifecycle("sending"); } else { if (signal.aborted) return; setLifecycle("collecting"); } } catch (pollError) { if ((signal && signal.aborted) || (pollError instanceof Error && pollError.name === "AbortError")) { return; } stopPolling(); setLifecycle("idle"); const message = pollError instanceof Error ? pollError.message : "Failed to read digest status"; setError(message); } finally { inFlightRef.current = false; } }; void poll(); pollTimerRef.current = setInterval(() => { void poll(); }, POLL_INTERVAL_MS); }, [ensureAbortController, fetchStatusSnapshot, stopPolling] ); const hasScrolledRef = useRef(false); useEffect(() => { if (lifecycle !== "idle" && !hasScrolledRef.current) { hasScrolledRef.current = true; const heading = document.getElementById("try-it-heading"); if (heading) { const top = heading.getBoundingClientRect().top + window.scrollY; window.scrollTo({ top, behavior: "smooth" }); } } if (lifecycle === "idle") { hasScrolledRef.current = false; } }, [lifecycle]); const handleStart = useCallback(async () => { setError(null); stopPolling(); abortRef.current?.abort(); abortRef.current = null; const controller = ensureAbortController(); const signal = controller.signal; try { const payload = await postJson("/api/mock/start", { userId: `user_${Date.now().toString(36)}`, }, signal); if (signal.aborted) return; setRunId(payload.runId); if (signal.aborted) return; setToken(payload.token); if (signal.aborted) return; setSnapshot(payload.snapshot); if (signal.aborted) return; setLifecycle("collecting"); if (signal.aborted) return; beginPolling(payload.runId); } catch (startError) { if (signal.aborted || (startError instanceof Error && startError.name === "AbortError")) { return; } const message = startError instanceof Error ? startError.message : "Failed to start digest run"; setError(message); setLifecycle("idle"); } }, [beginPolling, ensureAbortController, stopPolling]); const handleAddEvent = useCallback(async (preset: typeof EVENT_PRESETS[number]) => { if (!token || lifecycle !== "collecting") return; const controller = ensureAbortController(); const signal = controller.signal; try { await postJson("/api/mock/event", { token, type: preset.type, message: preset.message, }, signal); } catch (eventError) { if (signal.aborted || (eventError instanceof Error && eventError.name === "AbortError")) { return; } } }, [token, lifecycle, ensureAbortController]); const handleReset = useCallback(() => { stopPolling(); abortRef.current?.abort(); abortRef.current = null; setLifecycle("idle"); setRunId(null); setToken(null); setSnapshot(null); setError(null); setTimeout(() => { startButtonRef.current?.focus(); }, 0); }, [stopPolling]); const isCollecting = lifecycle === "collecting"; const phaseExplainer = useMemo(() => { if (!snapshot) { return "Waiting to start a digest collection window."; } if (snapshot.status === "collecting") { const remaining = (snapshot.windowRemainingMs / 1000).toFixed(1); return `while (true) + Promise.race([hook, sleep('${snapshot.collectSliceMs}ms')]) is collecting for ${remaining}s more. The workflow is durably suspended between wakeups and consumes zero compute while waiting.`; } if (snapshot.status === "sending") { return `Window closed. sendDigestEmail() is executing with ${snapshot.events.length} event(s).`; } if (snapshot.status === "completed") { return `Digest sent with ${snapshot.events.length} event(s).`; } if (snapshot.status === "empty") { return "Window closed with no events. Digest skipped."; } return "Run is active."; }, [snapshot]); type GutterMarkKind = "success" | "fail"; const codeState = useMemo(() => { const wfMarks: Record = {}; const stepMarks: Record = {}; if (snapshot) { const hasWindowClosed = snapshot.executionLog.some( (e) => e.kind === "window_closed" ); const hasDigestSent = snapshot.executionLog.some( (e) => e.kind === "digest_sent" ); const hasDigestEmpty = snapshot.executionLog.some( (e) => e.kind === "digest_empty" ); const hasEvents = snapshot.events.length > 0; if (hasEvents && snapshot.status === "collecting") { for (const ln of workflowLineMap.pushEvent) wfMarks[ln] = "success"; for (const ln of workflowLineMap.createToken) wfMarks[ln] = "success"; } if (hasWindowClosed) { for (const ln of workflowLineMap.raceLoop) wfMarks[ln] = "success"; for (const ln of workflowLineMap.createToken) wfMarks[ln] = "success"; } if (hasDigestSent) { for (const ln of workflowLineMap.sendDigest) wfMarks[ln] = "success"; for (const ln of stepLineMap.sendBody) stepMarks[ln] = "success"; for (const ln of stepLineMap.returnLine) stepMarks[ln] = "success"; } if (snapshot.status === "completed") { for (const ln of workflowLineMap.returnSent) wfMarks[ln] = "success"; } if (hasDigestEmpty) { for (const ln of workflowLineMap.returnEmpty) wfMarks[ln] = "success"; } } if (!snapshot) { return { tone: "amber" as HighlightTone, workflowActiveLines: [] as number[], stepActiveLines: [] as number[], workflowGutterMarks: wfMarks, stepGutterMarks: stepMarks, }; } if (snapshot.status === "collecting") { return { tone: "cyan" as HighlightTone, workflowActiveLines: workflowLineMap.raceLoop, stepActiveLines: [] as number[], workflowGutterMarks: wfMarks, stepGutterMarks: stepMarks, }; } if (snapshot.status === "sending") { return { tone: "amber" as HighlightTone, workflowActiveLines: workflowLineMap.sendDigest, stepActiveLines: stepLineMap.sendBody, workflowGutterMarks: wfMarks, stepGutterMarks: stepMarks, }; } if (snapshot.status === "completed") { return { tone: "green" as HighlightTone, workflowActiveLines: workflowLineMap.returnSent, stepActiveLines: stepLineMap.returnLine, workflowGutterMarks: wfMarks, stepGutterMarks: stepMarks, }; } if (snapshot.status === "empty") { return { tone: "green" as HighlightTone, workflowActiveLines: workflowLineMap.returnEmpty, stepActiveLines: [] as number[], workflowGutterMarks: wfMarks, stepGutterMarks: stepMarks, }; } return { tone: "amber" as HighlightTone, workflowActiveLines: [] as number[], stepActiveLines: [] as number[], workflowGutterMarks: wfMarks, stepGutterMarks: stepMarks, }; }, [snapshot, stepLineMap, workflowLineMap]); return (

{error && (

{error}

)}

{lifecycle !== "idle" && ( )} {isCollecting && (

Add event: {EVENT_PRESETS.map((preset) => ( ))}

)}

{isCollecting && token && (

POST /api/mock/event with token {token}

)}

{phaseExplainer}

{runId && ( run: {runId} )} {token && ( token: {token} )}

{snapshot && snapshot.status === "collecting" && ( )}

sleep() + defineHook() → collect events over a time window with zero infrastructure

); } function WindowProgressBar({ progressPct, remainingMs, durationMs, }: { progressPct: number; remainingMs: number; durationMs: number; }) { const remainingSec = (remainingMs / 1000).toFixed(1); const totalSec = (durationMs / 1000).toFixed(0); return (

Collection window {remainingSec}s / {totalSec}s remaining

); } function EventList({ events, status, }: { events: DigestEvent[]; status: DigestSnapshot["status"] | null; }) { const scrollRef = useRef(null); useEffect(() => { const el = scrollRef.current; if (el) { el.scrollTop = el.scrollHeight; } }, [events.length]); const eventTypeIcon = (type: string) => { switch (type) { case "comment": return "💬"; case "like": return "❤️"; case "mention": return "@"; case "follow": return "👤"; default: return "•"; } }; return (

Events collected

{events.length}

{events.length === 0 && (

{status === "collecting" ? "Waiting for events\u2026 click a button above to add one." : "No events."}

)} {events.map((event, index) => { const tone = eventTypeTone(event.type); return (

{event.type}

{event.message}

+{event.arrivedAtMs}ms

); })}

); } function ExecutionLog({ events, elapsedMs, }: { events: DigestLogEvent[]; elapsedMs: number; }) { const scrollRef = useRef(null); useEffect(() => { const el = scrollRef.current; if (el) { el.scrollTop = el.scrollHeight; } }, [events.length]); return (

Execution log

{formatElapsedLabel(elapsedMs)}

{events.length === 0 && (

No events yet.

)} {events.map((event, index) => { const tone = logEventTone(event.kind); return (

); })}

); } function eventTypeTone(type: string): string { switch (type) { case "comment": return "text-blue-700"; case "like": return "text-red-700"; case "mention": return "text-amber-700"; case "follow": return "text-green-700"; default: return "text-gray-900"; } } function logEventTone(kind: DigestLogEvent["kind"]): { dotClass: string; labelClass: string; } { switch (kind) { case "window_open": return { dotClass: "bg-blue-700", labelClass: "text-blue-700", }; case "event_received": return { dotClass: "bg-amber-700", labelClass: "text-amber-700", }; case "sleep_tick": return { dotClass: "bg-cyan-700", labelClass: "text-cyan-700", }; case "window_closed": return { dotClass: "bg-violet-700", labelClass: "text-violet-700", }; case "sending_digest": return { dotClass: "bg-amber-700", labelClass: "text-amber-700", }; case "digest_sent": return { dotClass: "bg-green-700", labelClass: "text-green-700", }; case "digest_empty": return { dotClass: "bg-gray-900", labelClass: "text-gray-900", }; default: return { dotClass: "bg-gray-500", labelClass: "text-gray-900", }; } } [V0_FILE]typescriptreact:file="app/components/digest-code-workbench.tsx" "use client"; import { useEffect, useMemo, useRef, useState } from "react"; type HighlightTone = "amber" | "cyan" | "green" | "red"; type CopyState = "idle" | "copied" | "failed"; type HighlightStyle = { border: string; bg: string; text: string; }; const HIGHLIGHT_STYLES: Record = { amber: { border: "border-amber-700", bg: "bg-amber-700/15", text: "text-amber-700", }, cyan: { border: "border-cyan-700", bg: "bg-cyan-700/15", text: "text-cyan-700", }, green: { border: "border-green-700", bg: "bg-green-700/15", text: "text-green-700", }, red: { border: "border-red-700", bg: "bg-red-700/15", text: "text-red-700", }, }; const CHECKMARK_POINTS = "3,8.5 7,12.5 14,4.5"; type GutterMarkKind = "success" | "fail"; type CodePaneProps = { code: string; htmlLines: string[]; activeLines: number[]; gutterMarks: Record; filename: string; subtitle: string; tone: HighlightTone; }; function CodePane({ code, htmlLines, activeLines, gutterMarks, filename, subtitle, tone, }: CodePaneProps) { const [copyState, setCopyState] = useState("idle"); const activeLineSet = useMemo(() => new Set(activeLines), [activeLines]); const gutterMarkMap = useMemo(() => new Map(Object.entries(gutterMarks).map(([k, v]) => [Number(k), v])), [gutterMarks]); const prevMarkRef = useRef>(new Map()); const isFullReset = gutterMarkMap.size === 0 && activeLines.length === 0; useEffect(() => { if (isFullReset) { prevMarkRef.current.clear(); return; } gutterMarkMap.forEach((kind, line) => { prevMarkRef.current.set(line, kind); }); }, [gutterMarkMap, isFullReset]); const highlightStyle = HIGHLIGHT_STYLES[tone]; const handleCopy = async () => { try { await navigator.clipboard.writeText(code); setCopyState("copied"); window.setTimeout(() => setCopyState("idle"), 1200); } catch { setCopyState("failed"); window.setTimeout(() => setCopyState("idle"), 1200); } }; return (

{filename}

{subtitle}

          
            {htmlLines.map((lineHtml, index) => {
              const lineNumber = index + 1;
              const isActive = activeLineSet.has(lineNumber);
              const markKind = gutterMarkMap.get(lineNumber);
              const displayKind = isFullReset
                ? undefined
                : markKind ?? prevMarkRef.current.get(lineNumber);

              return (
                
                  
                    
                  
                  
                    {lineNumber}
                  
                   0 ? lineHtml : " ",
                    }}
                  />
                
              );
            })}

); } export type DigestCodeWorkbenchProps = { workflowCode: string; workflowHtmlLines: string[]; workflowActiveLines: number[]; workflowGutterMarks: Record; stepCode: string; stepHtmlLines: string[]; stepActiveLines: number[]; stepGutterMarks: Record; tone: HighlightTone; }; export function DigestCodeWorkbench({ workflowCode, workflowHtmlLines, workflowActiveLines, workflowGutterMarks, stepCode, stepHtmlLines, stepActiveLines, stepGutterMarks, tone, }: DigestCodeWorkbenchProps) { return (

); } [V0_FILE]typescript:file="app/components/code-highlight-server.ts" import { highlight } from "sugar-high"; export function highlightCodeToHtmlLines(code: string): string[] { return highlight(code).split("\n"); } [V0_FILE]typescriptreact:file="app/layout.tsx" import type { Metadata } from "next"; import { Geist, Geist_Mono } from "next/font/google"; import "./globals.css"; const geistSans = Geist({ variable: "--font-geist-sans", subsets: ["latin"], }); const geistMono = Geist_Mono({ variable: "--font-geist-mono", subsets: ["latin"], }); export const metadata: Metadata = { title: "Scheduled Digest - Workflow DevKit Example", description: "Demonstrates collecting events over a time window and batch-processing them with sleep() in Workflow DevKit", }; export default function RootLayout({ children, }: Readonly<{ children: React.ReactNode; }>) { return ( Skip to main content {children} ); } [V0_FILE]typescript:file="app/api/mock/start/route.ts" import { createApiErrorResponse, createApiSuccessResponse } from "../api"; import { buildDigestSnapshot, createMockDigestRun } from "../store"; type StartRequestBody = { userId?: unknown; }; const DEFAULT_USER_ID = "user_123"; function generateRunId(): string { return `digest_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 8)}`; } function parseUserId(value: unknown): { value: string } | { error: string } { if (value === undefined) { return { value: DEFAULT_USER_ID }; } if (typeof value !== "string") { return { error: "userId must be a string" }; } const normalized = value.trim(); if (!normalized) { return { error: "userId cannot be empty" }; } return { value: normalized }; } export async function POST(request: Request): Promise { let body: StartRequestBody; try { body = (await request.json()) as StartRequestBody; } catch { return createApiErrorResponse( { code: "INVALID_REQUEST", message: "Invalid JSON body", }, { status: 400 } ); } const userIdResult = parseUserId(body.userId); if ("error" in userIdResult) { console.warn("[mock-scheduled-digest] start_validation_failed", { reason: userIdResult.error, body, }); return createApiErrorResponse( { code: "INVALID_REQUEST", message: userIdResult.error, }, { status: 400 } ); } const run = createMockDigestRun({ runId: generateRunId(), userId: userIdResult.value, }); const snapshot = buildDigestSnapshot(run, run.startedAtMs); return createApiSuccessResponse({ runId: run.id, token: run.token, userId: run.userId, status: snapshot.status, snapshot, }); } [V0_FILE]typescript:file="app/api/mock/api.ts" import { NextResponse } from "next/server"; type ApiError = { code: string; message: string; }; export function createApiSuccessResponse>( fields: T, init?: ResponseInit ) { return NextResponse.json( { ok: true, ...fields, }, init ); } export function createApiErrorResponse(error: ApiError, init?: ResponseInit) { return NextResponse.json( { ok: false, error, }, init ); } [V0_FILE]typescript:file="app/api/mock/status/route.ts" import { createApiErrorResponse, createApiSuccessResponse } from "../api"; import { buildDigestSnapshot, getMockDigestRun } from "../store"; export async function GET(request: Request): Promise { const searchParams = new URL(request.url).searchParams; const runId = searchParams.get("runId"); if (!runId) { return createApiErrorResponse( { code: "INVALID_REQUEST", message: "runId is required", }, { status: 400 } ); } const run = getMockDigestRun(runId); if (!run) { console.warn("[mock-scheduled-digest] status_lookup_failed", { attempted: "mock_digest_status_lookup", reason: "run_not_found", runId, }); return createApiErrorResponse( { code: "NOT_FOUND", message: "Run not found", }, { status: 404 } ); } return createApiSuccessResponse({ ...buildDigestSnapshot(run) }); } [V0_FILE]typescript:file="app/api/mock/event/route.ts" import { createApiErrorResponse, createApiSuccessResponse } from "../api"; import { addEventToDigestToken, getMockDigestRunByToken } from "../store"; type EventRequestBody = { token?: unknown; type?: unknown; message?: unknown; }; export async function POST(request: Request): Promise { let body: EventRequestBody; try { body = (await request.json()) as EventRequestBody; } catch { return createApiErrorResponse( { code: "INVALID_REQUEST", message: "Invalid JSON body", }, { status: 400 } ); } if (!body.token || typeof body.token !== "string") { return createApiErrorResponse( { code: "INVALID_REQUEST", message: "token is required", }, { status: 400 } ); } if (!body.type || typeof body.type !== "string") { return createApiErrorResponse( { code: "INVALID_REQUEST", message: "type is required", }, { status: 400 } ); } if (!body.message || typeof body.message !== "string") { return createApiErrorResponse( { code: "INVALID_REQUEST", message: "message is required", }, { status: 400 } ); } const run = getMockDigestRunByToken(body.token); if (!run) { console.warn("[mock-scheduled-digest] event_lookup_failed", { attempted: "mock_digest_event_lookup", reason: "token_not_found", token: body.token, }); return createApiErrorResponse( { code: "NOT_FOUND", message: "Token not found", }, { status: 404 } ); } const event = addEventToDigestToken(body.token, { type: body.type, message: body.message, }); if (!event) { return createApiErrorResponse( { code: "WINDOW_CLOSED", message: "Collection window has already closed", }, { status: 409 } ); } return createApiSuccessResponse({ event, totalEvents: run.events.length, }); } [V0_FILE]typescript:file="app/api/mock/store.ts" export const WINDOW_DURATION_MS = 6_000; export const COLLECT_SLICE_MS = 500; export const SEND_STEP_MS = 500; export type MockRunStatus = "collecting" | "sending" | "completed" | "empty"; export type DigestEvent = { type: string; message: string; arrivedAtMs: number; }; export type DigestLogEventKind = | "window_open" | "event_received" | "sleep_tick" | "window_closed" | "sending_digest" | "digest_sent" | "digest_empty"; export interface DigestLogEvent { atMs: number; kind: DigestLogEventKind; message: string; } export interface MockDigestRun { id: string; token: string; userId: string; windowDurationMs: number; collectSliceMs: number; startedAt: string; startedAtMs: number; events: DigestEvent[]; } export interface DigestSnapshot { runId: string; token: string; userId: string; status: MockRunStatus; windowDurationMs: number; collectSliceMs: number; startedAt: string; startedAtMs: number; elapsedMs: number; windowRemainingMs: number; windowProgressPct: number; events: DigestEvent[]; executionLog: DigestLogEvent[]; } declare global { // eslint-disable-next-line no-var var __mockDigestRunsById: Map | undefined; // eslint-disable-next-line no-var var __mockDigestRunIdsByToken: Map | undefined; } function toIso(timestampMs: number): string { return new Date(timestampMs).toISOString(); } export function buildDigestToken(runId: string): string { return `digest:${runId}`; } export function getRunStore(): Map { return ( globalThis.__mockDigestRunsById ?? (globalThis.__mockDigestRunsById = new Map()) ); } export function getTokenStore(): Map { return ( globalThis.__mockDigestRunIdsByToken ?? (globalThis.__mockDigestRunIdsByToken = new Map()) ); } export function resetMockDigestStore(): void { getRunStore().clear(); getTokenStore().clear(); } export function createMockDigestRun(input: { runId: string; userId: string; nowMs?: number; }): MockDigestRun { const nowMs = input.nowMs ?? Date.now(); const token = buildDigestToken(input.runId); const run: MockDigestRun = { id: input.runId, token, userId: input.userId, windowDurationMs: WINDOW_DURATION_MS, collectSliceMs: COLLECT_SLICE_MS, startedAt: toIso(nowMs), startedAtMs: nowMs, events: [], }; console.info("[mock-scheduled-digest] run_created", { runId: run.id, token: run.token, userId: run.userId, windowDurationMs: run.windowDurationMs, collectSliceMs: run.collectSliceMs, }); getRunStore().set(run.id, run); getTokenStore().set(run.token, run.id); return run; } export function getMockDigestRun(runId: string): MockDigestRun | undefined { return getRunStore().get(runId); } export function getMockDigestRunByToken( token: string ): MockDigestRun | undefined { const runId = getTokenStore().get(token); if (!runId) { return undefined; } return getRunStore().get(runId); } export function addEventToDigestToken( token: string, event: { type: string; message: string }, nowMs: number = Date.now() ): DigestEvent | null { const run = getMockDigestRunByToken(token); if (!run) { console.warn("[mock-scheduled-digest] add_event_failed", { attempted: "add_event_by_token", reason: "token_not_found", token, }); return null; } const elapsedMs = nowMs - run.startedAtMs; if (elapsedMs >= run.windowDurationMs) { console.warn("[mock-scheduled-digest] add_event_failed", { attempted: "add_event_by_token", reason: "window_closed", runId: run.id, token, elapsedMs, windowDurationMs: run.windowDurationMs, }); return null; } const digestEvent: DigestEvent = { type: event.type, message: event.message, arrivedAtMs: elapsedMs, }; run.events.push(digestEvent); console.info("[mock-scheduled-digest] event_added", { runId: run.id, token, type: event.type, arrivedAtMs: elapsedMs, totalEvents: run.events.length, }); return digestEvent; } function buildExecutionTimeline( run: MockDigestRun, elapsedMs: number ): DigestLogEvent[] { const events: DigestLogEvent[] = []; events.push({ atMs: 0, kind: "window_open", message: `Collection loop opened (token=${run.token})`, }); for (const digestEvent of run.events) { if (digestEvent.arrivedAtMs <= elapsedMs) { events.push({ atMs: digestEvent.arrivedAtMs, kind: "event_received", message: `${digestEvent.type}: ${digestEvent.message}`, }); } } for ( let tick = run.collectSliceMs; tick < run.windowDurationMs; tick += run.collectSliceMs ) { if (tick <= elapsedMs) { const remaining = Math.max(0, run.windowDurationMs - tick); events.push({ atMs: tick, kind: "sleep_tick", message: `Promise.race(hook, sleep(${run.collectSliceMs}ms)) - ${(remaining / 1000).toFixed(1)}s remaining`, }); } } if (elapsedMs >= run.windowDurationMs) { events.push({ atMs: run.windowDurationMs, kind: "window_closed", message: `Window closed - ${run.events.length} event(s) collected`, }); } if (run.events.length > 0 && elapsedMs >= run.windowDurationMs) { events.push({ atMs: run.windowDurationMs, kind: "sending_digest", message: `Sending digest with ${run.events.length} event(s)`, }); } if (run.events.length > 0 && elapsedMs >= run.windowDurationMs + SEND_STEP_MS) { events.push({ atMs: run.windowDurationMs + SEND_STEP_MS, kind: "digest_sent", message: `Digest email sent to ${run.userId}`, }); } if (run.events.length === 0 && elapsedMs >= run.windowDurationMs) { events.push({ atMs: run.windowDurationMs, kind: "digest_empty", message: "No events in inbox - skipping digest", }); } return events.sort((a, b) => a.atMs - b.atMs); } export function buildDigestSnapshot( run: MockDigestRun, nowMs: number = Date.now() ): DigestSnapshot { const elapsedMs = Math.max(0, nowMs - run.startedAtMs); const windowRemainingMs = Math.max(0, run.windowDurationMs - elapsedMs); const windowProgressPct = Math.min(100, (elapsedMs / run.windowDurationMs) * 100); let status: MockRunStatus; if (elapsedMs < run.windowDurationMs) { status = "collecting"; } else if (run.events.length === 0) { status = "empty"; } else if (elapsedMs < run.windowDurationMs + SEND_STEP_MS) { status = "sending"; } else { status = "completed"; } const visibleEvents = run.events.filter((event) => event.arrivedAtMs <= elapsedMs); return { runId: run.id, token: run.token, userId: run.userId, status, windowDurationMs: run.windowDurationMs, collectSliceMs: run.collectSliceMs, startedAt: run.startedAt, startedAtMs: run.startedAtMs, elapsedMs, windowRemainingMs, windowProgressPct, events: visibleEvents, executionLog: buildExecutionTimeline(run, elapsedMs), }; } [V0_FILE]typescriptreact:file="app/page.tsx" import { highlightCodeToHtmlLines } from "./components/code-highlight-server"; import { ScheduledDigestDemo } from "./components/demo"; const directiveUseWorkflow = `"use ${"workflow"}"`; const directiveUseStep = `"use ${"step"}"`; const workflowCode = `import { defineHook, sleep } from "workflow"; export const digestInbox = defineHook<{ type: string; message: string }>(); export async function collectAndSendDigest(runId: string, userId: string, windowMs: number) { ${directiveUseWorkflow}; const token = \`digest:\${runId}\`; const hook = digestInbox.create({ token }); const windowClosed = sleep(\`\${windowMs}ms\`).then(() => ({ kind: "window_closed" as const })); const events: Array<{ type: string; message: string }> = []; while (true) { const outcome = await Promise.race([ hook.then((payload) => ({ kind: "event" as const, payload })), windowClosed, ]); if (outcome.kind === "window_closed") break; events.push(outcome.payload); } if (events.length === 0) { return { runId, token, status: "empty", eventCount: 0 }; } await sendDigestEmail(userId, events); return { runId, token, status: "sent", eventCount: events.length }; }`; const stepCode = `async function sendDigestEmail(userId: string, events: DigestEvent[]) { ${directiveUseStep}; const digest = events.map(e => \`[\${e.type}] \${e.message}\`).join("\\n"); await fetch("https://email.example.com/send", { method: "POST", body: JSON.stringify({ to: userId, subject: "Your Digest", body: digest }), }); }`; function buildWorkflowLineMap(code: string) { const lines = code.split("\n"); const createToken = lines .map((line, index) => line.includes("const token =") || line.includes("digestInbox.create(") ? index + 1 : null ) .filter((line): line is number => line !== null); const raceLoop = lines .map((line, index) => line.includes("const outcome = await Promise.race([") ? index + 1 : null ) .filter((line): line is number => line !== null); const pushEvent = lines .map((line, index) => line.includes("events.push(outcome.payload)") ? index + 1 : null ) .filter((line): line is number => line !== null); const sendDigest = lines .map((line, index) => line.includes("await sendDigestEmail(") ? index + 1 : null ) .filter((line): line is number => line !== null); const returnSent = lines .map((line, index) => line.includes('status: "sent"') ? index + 1 : null ) .filter((line): line is number => line !== null); const returnEmpty = lines .map((line, index) => line.includes('status: "empty"') ? index + 1 : null ) .filter((line): line is number => line !== null); return { createToken, raceLoop, pushEvent, sendDigest, returnSent, returnEmpty, }; } function buildStepLineMap(code: string) { const lines = code.split("\n"); const sendBody = lines .map((line, index) => line.includes('await fetch("https://email.example.com') ? index + 1 : null ) .filter((line): line is number => line !== null); const returnLine = lines .map((line, index) => line.includes("body: JSON.stringify") ? index + 1 : null ) .filter((line): line is number => line !== null); return { sendBody, returnLine }; } const workflowHtmlLines = highlightCodeToHtmlLines(workflowCode); const stepHtmlLines = highlightCodeToHtmlLines(stepCode); const workflowLineMap = buildWorkflowLineMap(workflowCode); const stepLineMap = buildStepLineMap(stepCode); export default function Home() { return (

Workflow DevKit Example

Scheduled Digest

Collecting events over a time window and batch-processing them normally requires a message queue, a cron job, and a database to accumulate state. With Workflow DevKit, it is just{" "} sleep(){" "} +{" "} defineHook(){" "} plus a deterministic token inbox and one send step.

Try It

Why Not Just Use a Cron Job?

Traditional

You wire up a message queue (SQS, RabbitMQ) to buffer incoming events, a cron job to flush the window on a fixed schedule, and a database table to accumulate state across ticks. If the cron fires before events arrive or after the window drifts, you get partial or duplicate digests. The {"\u201C"}collect then send{"\u201D"} logic is scattered across three systems.

Workflow Digest

A defineHook() collects events into the workflow while{" "} sleep() holds the window open{"\u2014"}durably, at zero compute. When the timer fires, a single{" "} step sends the digest. No queue, no cron, no database rows. The entire pipeline is one function.

In production, adjust the window duration per user preference and add deduplication logic inside the collector loop.

); } [V0_FILE]css:file="app/globals.css" @import "tailwindcss"; @theme { --color-background-100: #0a0a0a; --color-background-200: #171717; --color-gray-100: #1a1a1a; --color-gray-200: #1f1f1f; --color-gray-300: #292929; --color-gray-400: #737373; --color-gray-500: #525252; --color-gray-600: #404040; --color-gray-700: #262626; --color-gray-800: #171717; --color-gray-900: #a1a1a1; --color-gray-1000: #ededed; --color-blue-700: #0070F3; --color-red-700: #E5484D; --color-amber-700: #FFB224; --color-green-700: #46A758; --color-violet-700: #8B5CF6; --color-cyan-700: #06B6D4; --font-sans: var(--font-geist-sans); --font-mono: var(--font-geist-mono); } body { background: #0a0a0a; color: #ededed; font-family: var(--font-sans), -apple-system, BlinkMacSystemFont, sans-serif; } /* sugar-high syntax tokens */ :root, .dark { --sh-class: #52a8ff; --sh-identifier: #ededed; --sh-sign: #a1a1a1; --sh-property: #ff6166; --sh-entity: #62c073; --sh-jsxliterals: #ededed; --sh-string: #62c073; --sh-keyword: #f75f8f; --sh-comment: #a1a1a1; } /* Prism token colors - Geist dark-mode palette */ .token.keyword, .token.keyword.control-flow { color: #f75f8f; } .token.plain, .token.parameter, .token.punctuation, .token.plain-text, .token.property-access, .token.operator, .token.tag.punctuation { color: #ededed; } .token.function, .token.function.maybe-class-name { color: #bf7af0; } .token.tag { color: #62c073; } .token.literal-property.property { color: #ff6166; } .token.string, .token.boolean { color: #62c073; } .token.comment { color: #a1a1a1; } .token.class-name { color: #52a8ff; } .token.number { color: #62c073; } .token.attr-name { color: #ff6166; } .token.attr-value { color: #62c073; } .token.builtin { color: #52a8ff; } .token.template-string { color: #62c073; } .token.interpolation { color: #ededed; } [V0_FILE]javascript:file="postcss.config.mjs" export default { plugins: { "@tailwindcss/postcss": {}, }, }; [V0_FILE]typescript:file="workflows/scheduled-digest.ts" import { sleep, defineHook } from "workflow"; export type DigestEventPayload = { type: string; message: string; }; export const digestEvent = defineHook(); export interface DigestResult { runId: string; userId: string; token: string; status: "sent" | "empty"; eventCount: number; windowMs: number; } export async function collectAndSendDigest( runId: string, userId: string, windowMs: number = 3_600_000 ): Promise { "use workflow"; const token = `digest:${runId}`; const hook = digestEvent.create({ token }); const windowClosed = sleep(`${windowMs}ms`).then(() => ({ kind: "window_closed" as const, })); const events: DigestEventPayload[] = []; while (true) { const outcome = await Promise.race([ hook.then((payload) => ({ kind: "event" as const, payload, })), windowClosed, ]); if (outcome.kind === "window_closed") { break; } events.push(outcome.payload); } if (events.length === 0) { return { runId, userId, token, status: "empty", eventCount: 0, windowMs, }; } await sendDigestEmail(userId, events); return { runId, userId, token, status: "sent", eventCount: events.length, windowMs, }; } async function sendDigestEmail( userId: string, events: DigestEventPayload[] ): Promise { "use step"; console.info("[scheduled-digest] send_digest", { userId, eventCount: events.length, types: events.map((e) => e.type), }); } [V0_FILE]json:file="package.json" { "name": "scheduled-digest", "version": "0.1.0", "private": true, "scripts": { "dev": "next dev", "build": "next build", "start": "next start", "lint": "next lint" }, "dependencies": { "next": "16.0.10", "react": "19.2.3", "react-dom": "19.2.3", "sugar-high": "^0.9.5", "workflow": "4.1.0-beta.53" }, "devDependencies": { "@tailwindcss/postcss": "^4", "@types/node": "^20", "@types/react": "^19", "@types/react-dom": "^19", "tailwindcss": "^4", "typescript": "^5" } } [V0_FILE]json:file="tsconfig.json" { "compilerOptions": { "target": "ES2017", "lib": ["dom", "dom.iterable", "esnext"], "allowJs": true, "skipLibCheck": true, "strict": true, "noEmit": true, "esModuleInterop": true, "module": "esnext", "moduleResolution": "bundler", "resolveJsonModule": true, "isolatedModules": true, "jsx": "react-jsx", "incremental": true, "plugins": [ { "name": "next" }, { "name": "workflow" } ], "paths": { "@/*": ["./*"] } }, "include": [ "next-env.d.ts", "**/*.ts", "**/*.tsx", ".next/types/**/*.ts", ".next/dev/types/**/*.ts", "**/*.mts" ], "exclude": ["node_modules"] } [V0_FILE]typescript:file="next.config.ts" import { withWorkflow } from "workflow/next"; import type { NextConfig } from "next"; const nextConfig: NextConfig = {}; export default withWorkflow(nextConfig);