Quickstart

Install

npm install watukuy      # pnpm add watukuy / yarn add watukuy / bun add watukuy

• Node >= 22.12. CI runs Node 22, 24, and 26.
• ESM-only. CommonJS projects (including NestJS apps compiled to CJS) use require('watukuy'), which works natively on Node 22.12+ via require(esm). There is no dual build.
• Optional peers only for the subpath you import: pg for watukuy/store-postgres; redis or ioredis for watukuy/store-redis; @opentelemetry/api for watukuy/otel; @nestjs/common and @nestjs/core (>=11 <13) for watukuy/nestjs. watukuy/store-sqlite uses node:sqlite and needs nothing.
• Zero runtime dependencies in the published package. No install scripts. Published with npm provenance.

Define a poller and start the engine

import { createWatukuy, definePoller } from 'watukuy';
import { SqliteStore } from 'watukuy/store-sqlite';
import { z } from 'zod';

const Order = z.object({
  id: z.string(),
  updatedAt: z.iso.datetime(),
  status: z.enum(['open', 'paid', 'cancelled']),
  total: z.number(),
});

export const orders = definePoller({
  name: 'orders',
  schema: Order,                          // any Standard Schema v1 validator; infers the item type
  identity: (o) => o.id,                  // stable id per item
  version: (o) => o.updatedAt,            // optional; defaults to the content hash
  fingerprint: (o) => ({ status: o.status, total: o.total }), // optional; what counts as a change
  schemaVersion: 1,                       // bump deliberately when your fingerprint changes

  cursor: {
    strategy: 'timestamp',
    field: 'updatedAt',
    tieBreak: 'id',                       // composite keyset (updatedAt, id): no skipped ties
    initial: '2026-01-01T00:00:00Z',
    lag: '30s',                           // never read past now - lag (late commits)
    overlap: '2m',                        // re-scan this window each cycle; dedup by version
  },

  fetch: async ({ cursor, http, signal }) => {
    const res = await http.get('https://erp.example.com/orders', {
      query: { updated_since: cursor.value, after_id: cursor.tieBreak, limit: 500 },
      signal,
    });
    if (res.notModified) return { items: [] };   // ETag 304: nothing to diff, counts as idle
    const body = await res.json<{ data: unknown[]; has_more: boolean }>();
    return { items: body.data, hasMore: body.has_more };
  },

  schedule: { min: '5s', max: '5m', adaptive: true },
  budget: 'erp',                          // shared token bucket
});

const engine = createWatukuy({
  store: new SqliteStore({ path: './watukuy.db' }), // durable, zero dependencies
  budgets: { erp: { requests: 100, per: '1m' } },
  pollers: { orders },                    // keyed object: engine.on('orders') is fully typed
});

engine.on('orders', async (event) => {
  // event.type: 'created' | 'updated' | 'deleted'
  // event.data: Order        event.previous?: Order (when retain: 'payload')
  await queue.add('order-sync', event, { jobId: event.id }); // at-least-once + dedup by id
});

await engine.start();

That is the whole integration. Crash it, scale it to three pods, hit a 429, get a page with fifty identical updatedAt values: the stream stays correct.

Test it without a network

import { createWatukuy, definePoller } from 'watukuy';
import { FakeApi, VirtualClock, SeededRandom, fakeItems } from 'watukuy/testing';
import { MemoryStore } from 'watukuy';

const clock = new VirtualClock('2026-01-01T00:00:00Z');
const api = new FakeApi({ clock, identity: (o) => o.id, timestampField: 'updatedAt', items: fakeItems(20) });

const items = definePoller({
  name: 'items',
  identity: (o: { id: string; updatedAt: string; value: number }) => o.id,
  version: (o) => o.updatedAt,
  cursor: { strategy: 'timestamp', field: 'updatedAt', tieBreak: 'id', initial: null },
  fetch: async ({ cursor }) => api.listSince({ since: cursor.value, afterId: cursor.tieBreak }),
  schedule: { min: '5s', max: '1m', jitter: 0 },
});

const engine = createWatukuy({ store: new MemoryStore(), pollers: { items }, clock, random: new SeededRandom(1) });
const seen: string[] = [];
engine.on('items', (e) => void seen.push(`${e.type}:${e.subject}`));

await engine.tick();                 // 20 created
api.update('item-001', { value: 99 });
await clock.advance(5_000);          // time only moves when you say so
await engine.tick();                 // 1 updated

FakeApi also serves a fetch-compatible function (api.fetchImpl()) with faults, Retry-After, ETags, rate-limit headers, and latency, so the HTTP helper and scheduler are testable end to end without a socket.

Next steps

• How it works: the poll cycle, the commit protocol, kill points, lanes.
• Cursor strategies: which strategy fits your API and the timestamp pitfalls.
• Delivery: ordering keys, retries, parking, and dedup at the consumer.
• Stores: SQLite for one node, Postgres for many, Redis for shared budgets.
• Serverless: tick() from Cloudflare Workers, Lambda, Vercel, or a CronJob.
• Examples: runnable demos, no credentials required.