@plinthjs/pulse

A headless application performance recorder for Mason, the Laravel Pulse equivalent (minus the dashboard). Pulse records small { type, key, value?, tags? } measurements into a swappable PulseStore, and rolls them up with count, min, max, sum, avg and p95 aggregates, optionally bucketed into fixed time windows. Recorder helpers turn slow queries, slow requests, exceptions, cache hits, user activity and queue jobs into entries.

The clock is injected (epoch milliseconds), so timestamps and buckets are deterministic in tests.

Install

npm install @plinthjs/pulse

Usage

import { Pulse } from '@plinthjs/pulse'

const pulse = new Pulse() // ArrayPulseStore + Date.now by default

pulse.record('slow_query', 'select * from users', 1450)
pulse.record('slow_query', 'select * from users', 1200)
pulse.record('slow_query', 'select * from orders', 2100)

pulse.aggregate('slow_query', { aggregate: 'max', limit: 5 })
// [{ key: 'select * from orders', value: 2100, count: 1 }, { key: 'select * from users', value: 1450, count: 2 }]

pulse.aggregate('slow_query', { aggregate: 'sum', period: 60 }) // one row per key per 60s bucket
pulse.values('slow_query') // [1450, 1200, 2100]
pulse.entries({ type: 'slow_query', since: Date.now() - 3_600_000 }) // most recent first

aggregate() defaults to count, ordered by value descending (orderBy: 'asc' flips it), and accepts since / before bounds. period is in seconds, and bucketed rows carry a bucket start instant.

Recorders

import {
  cacheInteraction,
  exceptionEntry,
  PulseType,
  queueEntry,
  slowQuery,
  slowRequest,
  userRequest,
} from '@plinthjs/pulse'

// Threshold recorders return undefined below the threshold (default 1000 ms).
const draft = slowQuery(sql, durationMs, { threshold: 500 })
if (draft) pulse.ingestDraft(draft)

pulse.ingestDraft(slowRequest('GET /reports', 1800))
pulse.ingestDraft(exceptionEntry(new TypeError('boom'))) // keyed by error name
pulse.ingestDraft(cacheInteraction('user:1', true)) // cache_hit or cache_miss
pulse.ingestDraft(userRequest(42))
pulse.ingestDraft(queueEntry('emails', 320))

pulse.aggregate(PulseType.CacheMiss) // most-missed keys

Filtering, buffering and toggling

const pulse = new Pulse({ lazy: true }) // buffer entries until flush()

pulse.filter((entry) => !entry.key.startsWith('/health')) // drop matching entries

pulse.record('user_request', '7')
pulse.pending() // 1
pulse.flush() // writes the buffer to the store, returns the count

pulse.disable() // record() becomes a no-op
pulse.recording() // false
pulse.enable()

Reads (aggregate, values, entries) flush the buffer first, so they always see every recorded entry.

Storage and retention

import { ArrayPulseStore, Pulse } from '@plinthjs/pulse'

let now = 0
const pulse = new Pulse({ store: new ArrayPulseStore(), now: () => now })

pulse.prune(now - 7 * 24 * 3_600_000) // remove entries older than a week, returns the count
pulse.clear() // drop the buffer and every stored entry

Implement the PulseStore interface (record, entries, aggregate, prune, clear) to persist entries elsewhere.