From fad06b580717be81d4ff9fb0536152e15e2c3c64 Mon Sep 17 00:00:00 2001 From: Deepso Date: Fri, 3 Jul 2026 16:39:26 +0530 Subject: [PATCH 1/6] Redesign SDK: scoped Effect core, non-blocking track, Standard Schema support Breaking rewrite of both entry points: - Delivery moved to a background fiber (Latch + Queue + forkScoped); track() now only validates and enqueues, never waits on the sink - trashlytics/effect: Tracker.make is a scoped constructor; closing the scope stops the worker and flushes remaining events (replaces shutdown) - Unified tagged errors: EventValidationError, UnknownEventError, TrackerClosedError, QueueFullError, SinkError - event() accepts Effect schemas, Schema.Struct fields, any Standard Schema v1 validator (zod/valibot/arktype), or no schema (payload-less) - Root entry: close() + Symbol.asyncDispose (await using), auto-flush on page hide/unload, sinks may return void/Promise/Effect - httpSink defaults to keepalive: true; new beaconSink for browsers - New options: context (meta enrichment), retry.jitter, maxQueueSize (renamed from bufferSize); retries renamed to retry Co-Authored-By: Claude Fable 5 --- README.md | 152 +++++--- src/effect.ts | 911 +++++++++++++++++++++++++++---------------- src/index.ts | 246 +++++++----- test/effect.test.ts | 192 +++++---- test/tracker.test.ts | 190 ++++++++- tsconfig.json | 2 +- 6 files changed, 1085 insertions(+), 608 deletions(-) diff --git a/README.md b/README.md index 89beb62..c3bd735 100644 --- a/README.md +++ b/README.md @@ -1,12 +1,10 @@ # trashlytics -A lightweight, generic event tracking library with type-safe events, runtime validation, batching, and retries. +A lightweight, type-safe event tracking library with runtime validation, background batching, and retries. Works in Node.js, browsers, and any modern JavaScript runtime. -Effect powers validation and delivery internally. App code uses a plain TypeScript API. +Effect powers validation and delivery internally. App code uses a plain TypeScript API — no Effect knowledge required. Use `trashlytics/effect` when your app is already Effect-based. -Use `trashlytics/effect` when your app is already Effect-based and you want tracker operations as `Effect` values. - -## Usage +## Quick Start ```ts import { Schema } from "effect" @@ -21,7 +19,9 @@ const events = { purchase: event("purchase.completed", { orderId: Schema.String, total: Schema.Number - }) + }), + + pageview: event("page.viewed") } const tracker = createTracker({ @@ -29,24 +29,38 @@ const tracker = createTracker({ sink: httpSink("/api/events"), batchSize: 20, flushInterval: 5000, - retries: { - attempts: 3, - delay: 250, - factor: 2 - }, + retry: { attempts: 3, delay: 250, factor: 2, jitter: true }, + context: () => ({ sessionId: getSessionId() }), onError(error, batch) { console.warn("event delivery failed", error, batch) } }) -tracker.track("signup", { - userId: "u_123", - plan: "free" -}) +tracker.track("signup", { userId: "u_123", plan: "free" }) +tracker.track("pageview") // payload-less event + +await tracker.close() // flush everything and release resources +``` + +`track` is fire-and-forget: it validates, stamps, and queues the event, and **never blocks on the network**. Delivery happens on a background fiber — when the batch size is reached, on the flush interval, on `flush()`, and on `close()`. + +## Schemas: Effect or Standard Schema + +Event payloads can be validated with Effect schemas **or any [Standard Schema v1](https://standardschema.dev) validator** (zod, valibot, arktype, ...): + +```ts +import { z } from "zod" -await tracker.flush() +const events = { + signup: event("user.signup", z.object({ + userId: z.string(), + plan: z.enum(["free", "pro"]) + })) +} ``` +Payload types are inferred from the schema either way. + ## Type-Safe Batches The sink receives a discriminated union based on your event map. @@ -57,20 +71,18 @@ const tracker = createTracker({ sink: async (batch) => { for (const item of batch) { if (item.key === "signup") { - item.payload.plan - // "free" | "pro" + item.payload.plan // "free" | "pro" } if (item.key === "purchase") { - item.payload.total - // number + item.payload.total // number } } } }) ``` -Each event includes both the local typed key and the external event name. +Each event includes the local typed key, the external event name, a timestamp, and merged metadata: ```ts type Event = { @@ -82,70 +94,94 @@ type Event = { } ``` +`meta` is the tracker-level `context` (static object or lazy function) merged with per-event metadata: + +```ts +tracker.track("signup", payload, { meta: { experiment: "b" } }) +``` + +## Sinks + +A sink is just a function receiving batches. It can return `void`, a `Promise`, or an Effect. + +- `httpSink(url, options?)` — POSTs JSON batches with `fetch`. `keepalive` defaults to `true` so requests survive page unloads. +- `beaconSink(url)` — delivers with `navigator.sendBeacon` (browsers). +- `consoleSink()` — logs batches. +- Any custom function: `sink: async (batch) => { ... }`. + +Failed deliveries are retried per the `retry` policy; batches that still fail are reported to `onError` and dropped. + ## Immediate Delivery -Use `trackNow` when the caller needs to wait for delivery. +Use `trackNow` when the caller needs to wait for delivery (it bypasses the queue): ```ts -await tracker.trackNow("purchase", { - orderId: "o_123", - total: 49 -}) +await tracker.trackNow("purchase", { orderId: "o_123", total: 49 }) ``` +## Lifecycle + +`close()` stops background delivery, flushes all remaining events, and releases resources. Trackers also implement `AsyncDisposable`: + +```ts +await using tracker = createTracker({ events, sink }) +// tracker.close() runs automatically at scope exit +``` + +In browsers, the tracker automatically flushes when the page is hidden or unloading (`visibilitychange`/`pagehide`). Disable with `flushOnHide: false`. + +## Errors + +All failures are tagged: `EventValidationError`, `UnknownEventError`, `TrackerClosedError`, `QueueFullError`, `SinkError`. Validation and background delivery failures are reported through `onError`; `trackNow` and `flush` reject with the failure. + ## Effect-Native API ```ts import { Effect, Schema } from "effect" -import { createTracker, event, httpSink } from "trashlytics/effect" +import * as Tracker from "trashlytics/effect" const events = { - signup: event("user.signup", { + signup: Tracker.event("user.signup", { userId: Schema.String, plan: Schema.Literals(["free", "pro"]) }) } -const tracker = createTracker({ - events, - sink: httpSink("/api/events"), - retries: { attempts: 3, delay: 250, factor: 2 } -}) - -const program = Effect.gen(function*() { - yield* tracker.track("signup", { - userId: "u_123", - plan: "free" +const program = Effect.gen(function* () { + const tracker = yield* Tracker.make({ + events, + sink: Tracker.httpSink("/api/events"), + retry: { attempts: 3 } }) - yield* tracker.flush() -}) + yield* tracker.track("signup", { userId: "u_123", plan: "free" }) + yield* tracker.flush +}).pipe(Effect.scoped) ``` -The root `trashlytics` entry point wraps this API with `Promise`/`void` methods. The `trashlytics/effect` entry point does not hide the Effect boundary. +`Tracker.make` is scoped: closing the scope interrupts the background delivery fiber and flushes all remaining events. Errors are fully typed in the failure channel (`TrackError` for `track`, the sink's error type for `flush`/`trackNow`). -## Custom Sinks - -Core delivery is sink-based, so you can send events anywhere. +To share a tracker across your app, wrap it in a Layer: ```ts -const tracker = createTracker({ - events, - sink: async (batch) => { - await fetch("/analytics", { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify(batch) - }) - } -}) -``` +import { Layer, ServiceMap } from "effect" -## Browser Support +class Analytics extends ServiceMap.Key>()("Analytics") {} -The core uses browser-safe timers and no Node-only APIs. `httpSink` uses `globalThis.fetch`. +const AnalyticsLayer = Layer.effect(Analytics, Tracker.make({ events, sink })) +``` -For page unloads, use `flush()` on lifecycle events when possible. Browsers may still terminate pending async work during tab close. +## Configuration + +| Option | Default | Description | +| --- | --- | --- | +| `batchSize` | `20` | Max events per sink call. Reaching it triggers background delivery. | +| `flushInterval` | `5000` | Auto-flush interval in ms. `0` disables interval flushing. | +| `maxQueueSize` | `1000` | Max queued events; beyond it new events are rejected. | +| `retry` | none | Retry count or `{ attempts, delay, factor, jitter }`. | +| `context` | none | Static or lazy metadata merged into every event's `meta`. | +| `onError` | none | Observes validation and delivery failures. | +| `flushOnHide` | `true` | (Root entry only) flush on page hide/unload in browsers. | ## License diff --git a/src/effect.ts b/src/effect.ts index cf584f5..d4fc3f0 100644 --- a/src/effect.ts +++ b/src/effect.ts @@ -1,61 +1,156 @@ import { + Cause, + Data, Duration, Effect, - Fiber, + Latch, Option, Queue, Schedule, Schema, + type Scope, Semaphore, } from "effect"; -type AnySchema = Schema.Decoder; +// ----------------------------------------------------------------------------- +// Schemas +// ----------------------------------------------------------------------------- + +/** + * Minimal Standard Schema v1 interface (https://standardschema.dev). + * + * Any spec-compliant validator (zod, valibot, arktype, ...) is accepted as an + * event payload schema, in addition to Effect schemas. + */ +export interface StandardSchemaV1 { + readonly "~standard": { + readonly version: 1; + readonly vendor: string; + readonly validate: ( + value: unknown + ) => StandardResult | Promise>; + readonly types?: { + readonly input: Input; + readonly output: Output; + }; + }; +} + +/** + * Result returned by a Standard Schema `validate` call. + */ +export type StandardResult = + | { readonly value: Output; readonly issues?: undefined } + | { readonly issues: readonly StandardIssue[] }; + +/** + * Issue reported by a Standard Schema `validate` call. + */ +export interface StandardIssue { + readonly message: string; + readonly path?: + | readonly (PropertyKey | { readonly key: PropertyKey })[] + | undefined; +} + +type AnyEffectSchema = Schema.ConstraintDecoder; type EventFields = Schema.Struct.Fields; +/** + * Any schema accepted as an event payload validator. + */ +export type PayloadSchema = AnyEffectSchema | StandardSchemaV1; + +// ----------------------------------------------------------------------------- +// Errors +// ----------------------------------------------------------------------------- + +/** + * A single normalized validation issue. + */ +export interface ValidationIssue { + readonly message: string; + readonly path?: readonly PropertyKey[] | undefined; +} + +/** + * Error returned when an event payload fails schema validation. + */ +export class EventValidationError extends Data.TaggedError( + "EventValidationError" +)<{ + readonly key: string; + readonly issues: readonly ValidationIssue[]; + readonly cause?: unknown; +}> { + override get message() { + return `Invalid payload for event "${this.key}": ${this.issues + .map((issue) => issue.message) + .join("; ")}`; + } +} + /** * Error returned when an event key is not present in a tracker's event registry. */ -export class UnknownEventError extends Schema.TaggedErrorClass()( - "UnknownEventError", - { - key: Schema.String, +export class UnknownEventError extends Data.TaggedError("UnknownEventError")<{ + readonly key: string; +}> { + override get message() { + return `Unknown event "${this.key}"`; } -) {} +} /** - * Error returned when a tracker operation is attempted after shutdown. + * Error returned when a tracker operation is attempted after the tracker's + * scope has been closed. */ -export class TrackerShutdownError extends Schema.TaggedErrorClass()( - "TrackerShutdownError", - {} -) {} +export class TrackerClosedError extends Data.TaggedError("TrackerClosedError") { + override get message() { + return "Tracker has been closed"; + } +} /** - * Error returned when a queued event cannot be accepted because the buffer is full. + * Error returned when a queued event cannot be accepted because the queue is + * at capacity. */ -export class BufferFullError extends Schema.TaggedErrorClass()( - "BufferFullError", - { - size: Schema.Number, +export class QueueFullError extends Data.TaggedError("QueueFullError")<{ + readonly capacity: number; +}> { + override get message() { + return `Event queue is full (capacity ${this.capacity})`; } -) {} +} /** * Wraps failures raised while delivering a batch to a sink. */ -export class SinkDeliveryError extends Schema.TaggedErrorClass()( - "SinkDeliveryError", - { - cause: Schema.Unknown, - } -) {} +export class SinkError extends Data.TaggedError("SinkError")<{ + readonly cause: unknown; +}> {} /** - * Defines a trackable event name and the schema used to validate its payload. + * Errors that can be raised while accepting an event for tracking. + */ +export type TrackError = + | EventValidationError + | UnknownEventError + | TrackerClosedError + | QueueFullError; + +// ----------------------------------------------------------------------------- +// Events +// ----------------------------------------------------------------------------- + +/** + * Defines a trackable event: its public name and the schema used to validate + * its payload. Created with {@link event}. */ export interface EventDefinition { + readonly _payload?: Payload; readonly name: Name; - readonly schema: Schema.Decoder; + readonly schema: PayloadSchema | undefined; } /** @@ -69,8 +164,16 @@ export type EventPayload = */ export type EventsMap = Record>; +/** + * Optional metadata attached to a tracked event. + */ +export type EventMeta = Record; + /** * Event object delivered to sinks after validation and timestamping. + * + * The union is discriminated by `key`, so narrowing on `key` narrows + * `name` and `payload` accordingly. */ export type TrackedEvent< Events extends EventsMap, @@ -85,10 +188,76 @@ export type TrackedEvent< }; }[Key]; +type InferFields = Schema.Schema.Type< + Schema.Struct +>; + +type StandardOutput = + S extends StandardSchemaV1 ? Output : never; + /** - * Optional metadata attached to an individual tracked event. + * Creates a payload-less event definition. + * + * @param name - Public event name delivered to sinks. */ -export type EventMeta = Record; +export function event( + name: Name +): EventDefinition; +/** + * Creates a typed event definition from `Schema.Struct` fields. + * + * @param name - Public event name delivered to sinks. + * @param fields - Struct fields used to validate and type the event payload. + */ +export function event< + const Name extends string, + const Fields extends EventFields, +>(name: Name, fields: Fields): EventDefinition>; +/** + * Creates a typed event definition from an Effect schema. + * + * @param name - Public event name delivered to sinks. + * @param schema - Effect schema used to validate and type the event payload. + */ +export function event< + const Name extends string, + const EventSchema extends AnyEffectSchema, +>( + name: Name, + schema: EventSchema +): EventDefinition>; +/** + * Creates a typed event definition from any Standard Schema v1 validator + * (zod, valibot, arktype, ...). + * + * @param name - Public event name delivered to sinks. + * @param schema - Standard Schema used to validate and type the event payload. + */ +export function event< + const Name extends string, + const EventSchema extends StandardSchemaV1, +>( + name: Name, + schema: EventSchema +): EventDefinition>; +export function event( + name: string, + schemaOrFields?: PayloadSchema | EventFields +) { + if (schemaOrFields === undefined) { + return { name, schema: undefined }; + } + + if (isEffectSchema(schemaOrFields) || isStandardSchema(schemaOrFields)) { + return { name, schema: schemaOrFields }; + } + + return { name, schema: Schema.Struct(schemaOrFields) }; +} + +// ----------------------------------------------------------------------------- +// Sinks +// ----------------------------------------------------------------------------- /** * Effect-native sink that receives validated events in batches. @@ -102,19 +271,120 @@ export type Sink< ) => Effect.Effect; /** - * Retry configuration for failed sink deliveries. + * Creates a sink that logs each delivered batch. + * + * @param log - Logger implementation to receive delivered batches. + */ +export function consoleSink( + log: Pick = console +): Sink { + return (batch) => + Effect.sync(() => { + log.log(batch); + }); +} + +/** + * Fetch options accepted by {@link httpSink}. + */ +export type HttpSinkOptions = Omit & { + /** HTTP method used to deliver batches. Defaults to `POST`. */ + readonly method?: "POST" | "PUT" | "PATCH"; + /** Custom fetch implementation. Defaults to `globalThis.fetch`. */ + readonly fetch?: typeof globalThis.fetch; +}; + +/** + * Creates a sink that posts JSON-encoded batches to an HTTP endpoint. + * + * `keepalive` defaults to `true` so in-flight batches survive page unloads in + * browsers. Note that browsers cap keepalive request bodies at ~64KB. + * + * @param url - HTTP endpoint that receives event batches. + * @param options - Fetch options and optional delivery method. + */ +export function httpSink( + url: string | URL, + options: HttpSinkOptions = {} +): Sink { + const { fetch: fetchImpl, method, ...init } = options; + + return (batch) => + Effect.tryPromise({ + try: async () => { + const headers = new Headers(init.headers); + + if (!headers.has("content-type")) { + headers.set("content-type", "application/json"); + } + + const response = await (fetchImpl ?? globalThis.fetch)(url, { + keepalive: true, + ...init, + headers, + method: method ?? "POST", + body: JSON.stringify(batch), + }); + + if (!response.ok) { + throw new Error(`HTTP sink failed with status ${response.status}`); + } + }, + catch: (cause) => new SinkError({ cause }), + }); +} + +/** + * Creates a sink that delivers batches with `navigator.sendBeacon`. + * + * Beacon requests survive page unloads, making this a good fit for + * browser-side trackers. Fails with {@link SinkError} outside browsers or when + * the user agent refuses to queue the payload. + * + * @param url - HTTP endpoint that receives event batches. + */ +export function beaconSink( + url: string | URL +): Sink { + return (batch) => + Effect.try({ + try: () => { + if (typeof navigator === "undefined" || !navigator.sendBeacon) { + throw new Error("navigator.sendBeacon is not available"); + } + + const body = new Blob([JSON.stringify(batch)], { + type: "application/json", + }); + + if (!navigator.sendBeacon(url, body)) { + throw new Error("navigator.sendBeacon refused the payload"); + } + }, + catch: (cause) => new SinkError({ cause }), + }); +} + +// ----------------------------------------------------------------------------- +// Tracker +// ----------------------------------------------------------------------------- + +/** + * Retry policy for failed sink deliveries. */ -export interface RetryOptions { +export interface RetryPolicy { /** Number of retry attempts after the initial delivery attempt. */ readonly attempts?: number; - /** Initial retry delay in milliseconds. */ + /** Initial retry delay in milliseconds. Defaults to 250. */ readonly delay?: number; - /** Exponential backoff multiplier. */ + /** Exponential backoff multiplier. Defaults to 2. */ readonly factor?: number; + /** Applies random jitter to retry delays. Defaults to false. */ + readonly jitter?: boolean; } /** - * Configuration used to create an Effect-native tracker. + * Configuration used to create a tracker. */ export interface TrackerOptions< Events extends EventsMap, @@ -124,21 +394,33 @@ export interface TrackerOptions< /** Maximum number of events delivered in one sink call. Defaults to 20. */ readonly batchSize?: number; /** - * Maximum number of queued events before new events are dropped. Defaults to - * 1000. + * Static or lazily computed metadata merged into every event's `meta`. + * Per-event metadata wins on key conflicts. */ - readonly bufferSize?: number; + readonly context?: EventMeta | (() => EventMeta); /** Event definitions accepted by this tracker. */ readonly events: Events; /** - * Automatic flush interval in milliseconds. Set to 0 to disable. Defaults to - * 5000. + * Automatic flush interval in milliseconds. Set to 0 to flush only when the + * batch size is reached or `flush` is called. Defaults to 5000. */ readonly flushInterval?: number; - /** Called when background delivery fails. */ - readonly onError?: (error: unknown) => void; - /** Retry policy for failed sink deliveries. */ - readonly retries?: number | RetryOptions; + /** + * Maximum number of queued events before new events are rejected. Defaults + * to 1000. + */ + readonly maxQueueSize?: number; + /** + * Observes every delivery failure (after retries are exhausted). Receives + * the batch that could not be delivered, when one exists. Failed batches + * are dropped. + */ + readonly onError?: ( + error: unknown, + batch?: readonly TrackedEvent[] + ) => void; + /** Retry policy (or retry count) for failed sink deliveries. */ + readonly retry?: number | RetryPolicy; /** Destination for validated event batches. */ readonly sink: Sink; } @@ -147,7 +429,7 @@ export interface TrackerOptions< * Per-event options accepted by {@link Tracker.track} and {@link Tracker.trackNow}. */ export interface TrackOptions { - /** Metadata copied onto the tracked event. */ + /** Metadata merged onto the tracked event (over tracker `context`). */ readonly meta?: EventMeta; /** * Event timestamp in milliseconds since the Unix epoch. Defaults to @@ -157,377 +439,322 @@ export interface TrackOptions { } /** - * Errors that can be raised while accepting an event for tracking. + * Argument list for `track`/`trackNow`: the payload argument is omittable for + * payload-less events. */ -export type TrackError = - | Schema.SchemaError - | UnknownEventError - | TrackerShutdownError - | BufferFullError; +export type TrackArgs = + EventPayload extends void + ? [payload?: undefined, options?: TrackOptions] + : [payload: EventPayload, options?: TrackOptions]; /** * Effect-native tracker for validating, queueing, and delivering typed events. + * + * Created with {@link make}. Delivery happens on a background fiber, so + * `track` never waits on the sink. Closing the surrounding scope stops the + * background fiber and delivers all remaining events. */ export interface Tracker< Events extends EventsMap, Error = never, Requirements = never, > { - /** Delivers all currently queued events. */ - readonly flush: () => Effect.Effect; - /** Stops background flushing and delivers remaining queued events. */ - readonly shutdown: () => Effect.Effect; - /** Validates and queues an event for batched delivery. */ + /** Delivers all currently queued events and waits for completion. */ + readonly flush: Effect.Effect; + /** Number of events currently queued. */ + readonly size: Effect.Effect; + /** + * Validates and queues an event for batched background delivery. Returns as + * soon as the event is queued; it never waits on the sink. + */ readonly track: ( key: Key, - payload: EventPayload, - options?: TrackOptions - ) => Effect.Effect; - /** Validates an event and delivers it immediately without queueing. */ + ...args: TrackArgs + ) => Effect.Effect; + /** Validates an event and delivers it immediately, bypassing the queue. */ readonly trackNow: ( key: Key, - payload: EventPayload, - options?: TrackOptions + ...args: TrackArgs ) => Effect.Effect< void, - Schema.SchemaError | UnknownEventError | TrackerShutdownError | Error, + Exclude | Error, Requirements >; } -type InferFields = Schema.Schema.Type< - Schema.Struct ->; +const DEFAULT_BATCH_SIZE = 20; +const DEFAULT_MAX_QUEUE_SIZE = 1000; +const DEFAULT_FLUSH_INTERVAL = 5000; +const DEFAULT_RETRY_DELAY = 250; +const DEFAULT_RETRY_FACTOR = 2; /** - * Creates a typed event definition from a public event name and struct fields. - * - * @param name - Public event name delivered to sinks. - * @param fields - Struct fields used to validate and type the event payload. - * @returns A typed event definition for use in a tracker event registry. - */ -export function event< - const Name extends string, - const Fields extends EventFields, ->(name: Name, fields: Fields): EventDefinition>; -/** - * Creates a typed event definition from a public event name and a schema. + * Creates a tracker that validates event payloads and delivers them to the + * configured sink in batches on a background fiber. * - * @param name - Public event name delivered to sinks. - * @param schema - Schema used to validate and type the event payload. - * @returns A typed event definition for use in a tracker event registry. - */ -export function event< - const Name extends string, - const EventSchema extends AnySchema, ->( - name: Name, - schema: EventSchema -): EventDefinition>; -/** - * Creates a typed event definition from a public event name and an Effect schema. - * - * The second argument may be either a full schema or `Schema.Struct` fields. - * - * @param name - Public event name delivered to sinks. - * @param schemaOrFields - Full schema or struct fields used for payload validation. - * @returns A typed event definition for use in a tracker event registry. - */ -export function event(name: string, schemaOrFields: AnySchema | EventFields) { - return { - name, - schema: isSchema(schemaOrFields) - ? schemaOrFields - : Schema.Struct(schemaOrFields), - }; -} - -/** - * Creates an Effect-native tracker that validates event payloads before delivering - * them to the configured sink. + * The tracker is scoped: closing the scope interrupts the background fiber + * and flushes all remaining events through the sink. * * @param options - Tracker configuration, including event definitions and sink. - * @returns A tracker whose operations return Effect values. + * @returns A scoped Effect producing the tracker. */ -export function createTracker< +export function make< const Events extends EventsMap, Error = never, Requirements = never, >( options: TrackerOptions -): Tracker { - const batchSize = options.batchSize ?? 20; - const bufferSize = options.bufferSize ?? 1000; - const flushInterval = options.flushInterval ?? 5000; - const retryOptions = normalizeRetries(options.retries); - const queue = Effect.runSync( - Queue.dropping>(bufferSize) - ); - const flushSemaphore = Semaphore.makeUnsafe(1); - const intervalSemaphore = Semaphore.makeUnsafe(1); - let intervalFiber: Fiber.Fiber | undefined; - let closed = false; - - const stopFlushInterval = Effect.fn("trashlytics.stopFlushInterval")( - function* () { - yield* intervalSemaphore.withPermit( - Effect.gen(function* () { - if (intervalFiber !== undefined) { - const fiber = intervalFiber; - intervalFiber = undefined; - yield* Fiber.interrupt(fiber); - } - }) - ); - } - ); - - const takeBatch = Effect.fn("trashlytics.takeBatch")(function* () { - const batch: TrackedEvent[] = []; - - while (batch.length < batchSize) { - const item = yield* Queue.poll(queue); +): Effect.Effect< + Tracker, + never, + Scope.Scope | Requirements +> { + return Effect.gen(function* () { + const batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE; + const maxQueueSize = options.maxQueueSize ?? DEFAULT_MAX_QUEUE_SIZE; + const flushInterval = options.flushInterval ?? DEFAULT_FLUSH_INTERVAL; + const retry = normalizeRetry(options.retry); + const queue = yield* Queue.dropping>(maxQueueSize); + const wakeWorker = Latch.makeUnsafe(false); + const deliveryLock = Semaphore.makeUnsafe(1); + let closed = false; + + const deliver = (batch: readonly TrackedEvent[]) => + Effect.retry(options.sink(batch), { + times: retry.attempts, + schedule: retry.jitter + ? Schedule.jittered( + Schedule.exponential(Duration.millis(retry.delay), retry.factor) + ) + : Schedule.exponential(Duration.millis(retry.delay), retry.factor), + }); + + const takeBatch = Effect.gen(function* () { + const batch: TrackedEvent[] = []; + + while (batch.length < batchSize) { + const item = yield* Queue.poll(queue); + + if (Option.isNone(item)) { + break; + } - if (Option.isNone(item)) { - break; + batch.push(item.value); } - batch.push(item.value); - } - - return batch; - }); - - const drainQueue = Effect.fn("trashlytics.drainQueue")(function* () { - yield* flushSemaphore.withPermit( - Effect.uninterruptible( - Effect.gen(function* () { - while (true) { - const batch = yield* takeBatch(); + return batch; + }); - if (batch.length === 0) { - break; - } + // Serialized with trackNow so batches reach the sink in order. A batch + // that fails after all retries is reported via onError and dropped; + // events still in the queue stay queued for the next attempt. + const drain = deliveryLock.withPermit( + Effect.gen(function* () { + while (true) { + const batch = yield* takeBatch; - yield* sendWithRetries(options.sink, batch, retryOptions); + if (batch.length === 0) { + return; } - }) - ) + + yield* deliver(batch).pipe( + Effect.tapCause((cause) => + Effect.sync(() => { + options.onError?.(Cause.squash(cause), batch); + }) + ) + ); + } + }) ); - }); - const flush = Effect.fn("trashlytics.flush")(function* () { - yield* drainQueue(); - }); + const drainSilently = drain.pipe(Effect.catchCause(() => Effect.void)); + + const worker = Effect.gen(function* () { + while (true) { + yield* flushInterval > 0 + ? Effect.timeoutOption( + wakeWorker.await, + Duration.millis(flushInterval) + ) + : wakeWorker.await; + yield* wakeWorker.close; + yield* drainSilently; + } + }); - const ensureFlushInterval = Effect.fn("trashlytics.ensureFlushInterval")( - function* () { - yield* intervalSemaphore.withPermit( - Effect.uninterruptible( - Effect.gen(function* () { - if (closed || flushInterval <= 0 || intervalFiber !== undefined) { - return; - } - - intervalFiber = yield* Effect.schedule( - Effect.void, - Schedule.duration(Duration.millis(flushInterval)) - ).pipe( - Effect.andThen(drainQueue()), - Effect.tapError((error) => - Effect.sync(() => { - options.onError?.(error); - }) - ), - Effect.ignore, - Effect.forever, - Effect.forkDetach - ); - }) - ) - ); - } - ); + // Finalizers run in reverse order: mark closed, interrupt the worker + // (registered by forkScoped), then flush whatever is still queued. + yield* Effect.addFinalizer(() => drainSilently); + yield* Effect.forkScoped(worker); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + closed = true; + }) + ); - const flushIfBatchSizeReached = Effect.fn( - "trashlytics.flushIfBatchSizeReached" - )(function* () { - const queueSize = yield* Queue.size(queue); + const makeEvent = Effect.fn("trashlytics.makeEvent")(function* ( + key: keyof Events & string, + payload: unknown, + trackOptions: TrackOptions | undefined + ) { + if (closed) { + return yield* new TrackerClosedError(); + } - if (queueSize < batchSize) { - return; - } + const definition = options.events[key]; - yield* drainQueue(); - }); + if (definition === undefined) { + return yield* new UnknownEventError({ key }); + } - const makeEvent = ( - key: Key, - payload: EventPayload, - trackOptions?: TrackOptions - ) => { - const definition = options.events[key]; - - if (definition === undefined) { - return Effect.fail(new UnknownEventError({ key })); - } - - return Schema.decodeUnknownEffect(definition.schema)(payload).pipe( - Effect.map( - (decodedPayload): TrackedEvent => ({ - key, - name: definition.name, - payload: decodedPayload as EventPayload, - timestamp: trackOptions?.timestamp ?? Date.now(), - ...(trackOptions?.meta === undefined - ? {} - : { meta: trackOptions.meta }), - }) - ) - ); - }; + const decoded = yield* validatePayload(definition.schema, key, payload); + const meta = mergeMeta(options.context, trackOptions?.meta); - return { - track: Effect.fn("trashlytics.track")( - function* (key, payload, trackOptions) { - if (closed) { - return yield* new TrackerShutdownError(); - } + const trackedEvent: TrackedEvent = { + key, + name: definition.name, + payload: decoded as EventPayload, + timestamp: trackOptions?.timestamp ?? Date.now(), + ...(meta === undefined ? {} : { meta }), + }; - const trackedEvent = yield* makeEvent(key, payload, trackOptions); - const wasQueued = yield* Queue.offer(queue, trackedEvent); + return trackedEvent; + }); - if (!wasQueued) { - return yield* new BufferFullError({ size: bufferSize }); - } + const track = Effect.fn("trashlytics.track")(function* ( + key: keyof Events & string, + payload?: unknown, + trackOptions?: TrackOptions + ) { + const trackedEvent = yield* makeEvent(key, payload, trackOptions); - yield* ensureFlushInterval(); - yield* flushIfBatchSizeReached(); + if (!Queue.offerUnsafe(queue, trackedEvent)) { + return yield* new QueueFullError({ capacity: maxQueueSize }); } - ), - trackNow: Effect.fn("trashlytics.trackNow")( - function* (key, payload, trackOptions) { - if (closed) { - return yield* new TrackerShutdownError(); - } - - const trackedEvent = yield* makeEvent(key, payload, trackOptions); - - yield* sendWithRetries(options.sink, [trackedEvent], retryOptions); + if (Queue.sizeUnsafe(queue) >= batchSize) { + wakeWorker.openUnsafe(); } - ), - - flush, + }); - shutdown: Effect.fn("trashlytics.shutdown")(function* () { - closed = true; - yield* stopFlushInterval(); - yield* flush(); - }), - }; -} + const trackNow = Effect.fn("trashlytics.trackNow")(function* ( + key: keyof Events & string, + payload?: unknown, + trackOptions?: TrackOptions + ) { + const trackedEvent = yield* makeEvent(key, payload, trackOptions); -/** - * Creates a sink that logs each delivered batch with `console.log`. - * - * @param log - Logger implementation to receive delivered batches. - * @returns An Effect-native sink for tracker configuration. - */ -export function consoleSink( - log: Pick = console -): Sink { - return (batch) => - Effect.sync(() => { - log.log(batch); + yield* deliveryLock.withPermit(deliver([trackedEvent])); }); + + const tracker: Tracker = { + track: track as Tracker["track"], + trackNow: trackNow as Tracker["trackNow"], + flush: drain, + size: Effect.sync(() => Queue.sizeUnsafe(queue)), + }; + + return tracker; + }); } -/** - * Fetch options accepted by {@link httpSink}. - */ -export type HttpSinkOptions = Omit & { - /** HTTP method used to deliver batches. Defaults to `POST`. */ - readonly method?: "POST" | "PUT" | "PATCH"; -}; +// ----------------------------------------------------------------------------- +// Internals +// ----------------------------------------------------------------------------- + +function validatePayload( + schema: PayloadSchema | undefined, + key: string, + payload: unknown +): Effect.Effect { + if (schema === undefined) { + return Effect.void; + } -/** - * Creates a sink that posts JSON-encoded batches to an HTTP endpoint. - * - * @param url - HTTP endpoint that receives event batches. - * @param options - Fetch options and optional delivery method. - * @returns An Effect-native sink that fails with `SinkDeliveryError`. - */ -export function httpSink( - url: string | URL, - options: HttpSinkOptions = {} -): Sink { - return (batch) => - Effect.tryPromise({ - try: async () => { - const headers = new Headers(options.headers); + if (isEffectSchema(schema)) { + return Schema.decodeUnknownEffect(schema)(payload).pipe( + Effect.mapError( + (error) => + new EventValidationError({ + key, + issues: [{ message: error.message }], + cause: error, + }) + ) + ); + } - if (!headers.has("content-type")) { - headers.set("content-type", "application/json"); - } + const toEffect = ( + result: StandardResult + ): Effect.Effect => + result.issues === undefined + ? Effect.succeed(result.value) + : Effect.fail( + new EventValidationError({ + key, + issues: result.issues.map(normalizeIssue), + }) + ); - const response = await globalThis.fetch(url, { - ...options, - headers, - method: options.method ?? "POST", - body: Schema.encodeUnknownSync(Schema.UnknownFromJsonString)(batch), - }); + return Effect.suspend(() => { + const result = schema["~standard"].validate(payload); - if (!response.ok) { - throw new Error(`HTTP sink failed with status ${response.status}`); - } - }, - catch: (cause) => new SinkDeliveryError({ cause }), - }); + return result instanceof Promise + ? Effect.promise(() => result).pipe(Effect.flatMap(toEffect)) + : toEffect(result); + }); } -/** - * Delivers a batch through a sink using the provided retry policy. - * - * @param sink - Sink used to deliver the batch. - * @param batch - Validated events to deliver. - * @param retries - Expanded retry policy. - * @returns An Effect that completes when delivery succeeds or retries are exhausted. - */ -export function sendWithRetries( - sink: Sink, - batch: readonly TrackedEvent[], - retries: Required -) { - return Effect.retry(sink(batch), { - times: retries.attempts, - schedule: Schedule.exponential( - Duration.millis(retries.delay), - retries.factor +function normalizeIssue(issue: StandardIssue): ValidationIssue { + return { + message: issue.message, + path: issue.path?.map((segment) => + typeof segment === "object" && segment !== null && "key" in segment + ? segment.key + : segment ), - }); + }; } -/** - * Expands shorthand retry configuration into explicit retry defaults. - * - * @param retries - Retry count, partial retry options, or undefined. - * @returns Retry options with attempts, delay, and factor populated. - */ -export function normalizeRetries(retries: number | RetryOptions | undefined) { - if (typeof retries === "number") { - return { attempts: retries, delay: 250, factor: 2 }; +function mergeMeta( + context: EventMeta | (() => EventMeta) | undefined, + meta: EventMeta | undefined +): EventMeta | undefined { + const contextMeta = typeof context === "function" ? context() : context; + + if (contextMeta === undefined) { + return meta; + } + + return { ...contextMeta, ...meta }; +} + +function normalizeRetry(retry: number | RetryPolicy | undefined) { + if (typeof retry === "number") { + return { + attempts: retry, + delay: DEFAULT_RETRY_DELAY, + factor: DEFAULT_RETRY_FACTOR, + jitter: false, + }; } return { - attempts: retries?.attempts ?? 0, - delay: retries?.delay ?? 250, - factor: retries?.factor ?? 2, + attempts: retry?.attempts ?? 0, + delay: retry?.delay ?? DEFAULT_RETRY_DELAY, + factor: retry?.factor ?? DEFAULT_RETRY_FACTOR, + jitter: retry?.jitter ?? false, }; } -const isSchema = (value: unknown): value is AnySchema => +const isEffectSchema = (value: unknown): value is AnyEffectSchema => (typeof value === "object" || typeof value === "function") && value !== null && "ast" in value && "rebuild" in value; + +const isStandardSchema = (value: unknown): value is StandardSchemaV1 => + (typeof value === "object" || typeof value === "function") && + value !== null && + "~standard" in value; diff --git a/src/index.ts b/src/index.ts index a1b6993..b4f467a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,19 +1,18 @@ -import { Effect } from "effect"; +import { Effect, Exit, Scope } from "effect"; import type { - EventPayload as EffectEventPayload, - EventsMap as EffectEventsMap, - HttpSinkOptions as EffectHttpSinkOptions, - TrackedEvent as EffectTrackedEvent, + Sink as EffectSink, TrackerOptions as EffectTrackerOptions, - TrackOptions as EffectTrackOptions, -} from "./effect"; -import { - createTracker as createEffectTracker, - consoleSink as effectConsoleSink, - event as effectEvent, - httpSink as effectHttpSink, - SinkDeliveryError, + EventsMap, + TrackArgs, + TrackedEvent, } from "./effect"; +import { make, SinkError } from "./effect"; + +// Runtime support for `await using` on platforms that predate the explicit +// resource management proposal. +(Symbol as { asyncDispose?: symbol }).asyncDispose ??= Symbol.for( + "Symbol.asyncDispose" +); export type { EventDefinition, @@ -21,144 +20,185 @@ export type { EventPayload, EventsMap, HttpSinkOptions, - RetryOptions, + RetryPolicy, + StandardIssue, + StandardResult, + StandardSchemaV1, + TrackError, TrackedEvent, TrackOptions, + ValidationIssue, +} from "./effect"; +// biome-ignore lint/performance/noBarrelFile: the root entry intentionally shares the event/sink/error API with the Effect entry. +export { + beaconSink, + consoleSink, + EventValidationError, + event, + httpSink, + QueueFullError, + SinkError, + TrackerClosedError, + UnknownEventError, } from "./effect"; /** - * Creates a typed event definition from a public event name and an Effect schema - * or `Schema.Struct` fields. + * Sink that receives validated events in batches. * - * @param name - Public event name delivered to sinks. - * @param schemaOrFields - Full schema or struct fields used for payload validation. - * @returns A typed event definition for use in a tracker event registry. + * May return `void`, a `Promise`, or an Effect — so the sinks exported from + * this module ({@link httpSink}, {@link beaconSink}, {@link consoleSink}) and + * plain async functions both work. */ -export const event = effectEvent; - -/** - * Promise-style sink that receives validated events in batches. - */ -export type Sink = ( - batch: readonly EffectTrackedEvent[] -) => void | Promise; +export type Sink = ( + batch: readonly TrackedEvent[] +) => void | Promise | Effect.Effect; /** * Configuration used to create a Promise-style tracker. */ -export type TrackerOptions = Omit< - EffectTrackerOptions, - "onError" | "sink" +export type TrackerOptions = Omit< + EffectTrackerOptions, + "sink" > & { /** Destination for validated event batches. */ readonly sink: Sink; - /** Automatic flush interval in milliseconds. Set to 0 to disable. */ - readonly flushInterval?: number; - /** Called when asynchronous tracking or background delivery fails. */ - readonly onError?: ( - error: unknown, - batch?: readonly EffectTrackedEvent[] - ) => void; + /** + * Flushes pending events when the page is hidden or unloading (browsers + * only; ignored elsewhere). Defaults to true. + */ + readonly flushOnHide?: boolean; }; /** * Promise-style tracker for validating, queueing, and delivering typed events. + * + * Supports `await using tracker = createTracker(...)` for automatic cleanup. */ -export interface Tracker { - /** Delivers all currently queued events. */ +export interface Tracker extends AsyncDisposable { + /** + * Stops background delivery, flushes all remaining events, and releases + * resources. Idempotent. Tracking after `close` reports + * `TrackerClosedError` through `onError`. + */ + readonly close: () => Promise; + /** Delivers all currently queued events and waits for completion. */ readonly flush: () => Promise; - /** Stops background flushing and delivers remaining queued events. */ - readonly shutdown: () => Promise; - /** Validates and queues an event for batched delivery. */ + /** + * Validates and queues an event for batched background delivery. Fire and + * forget: it never throws and never waits on the sink. Validation and + * delivery failures are reported through `onError`. + */ readonly track: ( key: Key, - payload: EffectEventPayload, - options?: EffectTrackOptions + ...args: TrackArgs ) => void; - /** Validates an event and delivers it immediately without queueing. */ + /** Validates an event and delivers it immediately, bypassing the queue. */ readonly trackNow: ( key: Key, - payload: EffectEventPayload, - options?: EffectTrackOptions + ...args: TrackArgs ) => Promise; } /** * Creates a Promise-style tracker that validates event payloads before - * delivering them to the configured sink. + * delivering them to the configured sink in batches on a background fiber. * * @param options - Tracker configuration, including event definitions and sink. * @returns A tracker whose operations use Promise-style APIs. */ -export function createTracker( +export function createTracker( options: TrackerOptions ): Tracker { - const tracker = createEffectTracker({ - events: options.events, - sink: (batch) => - Effect.tryPromise({ - try: async () => { - await options.sink(batch); - }, - catch: (cause) => new SinkDeliveryError({ cause }), - }), - batchSize: options.batchSize, - bufferSize: options.bufferSize, - flushInterval: options.flushInterval, - onError: options.onError, - retries: options.retries, - }); - - const flush = async () => { - try { - await Effect.runPromise(tracker.flush()); - } catch (error) { - options.onError?.(error); - throw error; + const scope = Scope.makeUnsafe(); + const tracker = Effect.runSync( + Scope.provide(make({ ...options, sink: adaptSink(options.sink) }), scope) + ); + + const flush = () => Effect.runPromise(tracker.flush); + + const detachLifecycle = attachLifecycleFlush( + options.flushOnHide ?? true, + () => { + flush().catch(() => { + // Delivery failures are already reported through onError. + }); } + ); + + let closing: Promise | undefined; + const close = () => { + closing ??= (() => { + detachLifecycle(); + return Effect.runPromise(Scope.close(scope, Exit.void)); + })(); + + return closing; }; return { - track: (key, payload, trackOptions) => { - Effect.runPromise(tracker.track(key, payload, trackOptions)).catch( - (error) => options.onError?.(error) - ); + track: (key, ...args) => { + Effect.runPromise(tracker.track(key, ...args)).catch((error) => { + options.onError?.(error); + }); }, - trackNow: (key, payload, trackOptions) => - Effect.runPromise(tracker.trackNow(key, payload, trackOptions)), + trackNow: (key, ...args) => + Effect.runPromise(tracker.trackNow(key, ...args)), flush, - shutdown: async () => { - await Effect.runPromise(tracker.shutdown()); - }, + close, + + [Symbol.asyncDispose]: close, }; } -/** - * Creates a sink that logs each delivered batch with `console.log`. - * - * @param log - Logger implementation to receive delivered batches. - * @returns A Promise-style sink for tracker configuration. - */ -export function consoleSink( - log: Pick = console -): Sink { - return (batch) => Effect.runSync(effectConsoleSink(log)(batch)); +function adaptSink( + sink: Sink +): EffectSink { + return (batch) => + Effect.suspend(() => { + let result: ReturnType>; + + try { + result = sink(batch); + } catch (cause) { + return Effect.fail(new SinkError({ cause })); + } + + if (Effect.isEffect(result)) { + return result; + } + + if (result instanceof Promise) { + return Effect.tryPromise({ + try: () => result as Promise, + catch: (cause) => new SinkError({ cause }), + }); + } + + return Effect.void; + }); } -/** - * Creates a sink that posts JSON-encoded batches to an HTTP endpoint. - * - * @param url - HTTP endpoint that receives event batches. - * @param options - Fetch options and optional delivery method. - * @returns A Promise-style sink for tracker configuration. - */ -export function httpSink( - url: string | URL, - options: EffectHttpSinkOptions = {} -): Sink { - return (batch) => - Effect.runPromise(effectHttpSink(url, options)(batch)); +function attachLifecycleFlush(enabled: boolean, flush: () => void): () => void { + if (!enabled || typeof document === "undefined") { + return () => { + // Nothing to detach outside browsers. + }; + } + + const onVisibilityChange = () => { + if (document.visibilityState === "hidden") { + flush(); + } + }; + + document.addEventListener("visibilitychange", onVisibilityChange); + addEventListener("pagehide", flush); + + return () => { + document.removeEventListener("visibilitychange", onVisibilityChange); + removeEventListener("pagehide", flush); + }; } diff --git a/test/effect.test.ts b/test/effect.test.ts index b317d90..0628ce0 100644 --- a/test/effect.test.ts +++ b/test/effect.test.ts @@ -1,9 +1,11 @@ import { Effect, Schema } from "effect"; import { describe, expect, it } from "vitest"; import { - createTracker, + EventValidationError, event, - SinkDeliveryError, + make, + type Sink, + SinkError, type TrackedEvent, } from "../src/effect"; @@ -14,22 +16,29 @@ const events = { }), }; +const collectingSink = () => { + const batches: (readonly TrackedEvent[])[] = []; + const sink: Sink = (batch) => + Effect.sync(() => { + batches.push(batch); + }); + + return { batches, sink }; +}; + describe("effect tracker", () => { it("exposes Effect-native tracker operations", async () => { - const batches: (readonly TrackedEvent[])[] = []; - const tracker = createTracker({ - events, - sink: (batch) => - Effect.sync(() => { - batches.push(batch); - }), - }); + const { batches, sink } = collectingSink(); await Effect.runPromise( - Effect.gen(function* () { - yield* tracker.track("signup", { userId: "u_1", plan: "free" }); - yield* tracker.flush(); - }) + Effect.scoped( + Effect.gen(function* () { + const tracker = yield* make({ events, sink, flushInterval: 0 }); + + yield* tracker.track("signup", { userId: "u_1", plan: "free" }); + yield* tracker.flush; + }) + ) ); expect(batches).toHaveLength(1); @@ -42,98 +51,111 @@ describe("effect tracker", () => { ]); }); - it("retries Effect sink failures", async () => { + it("fails track with EventValidationError on invalid payloads", async () => { + const { sink } = collectingSink(); + + const error = await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const tracker = yield* make({ events, sink, flushInterval: 0 }); + + return yield* tracker + .track("signup", { userId: "u_1", plan: "enterprise" } as never) + .pipe(Effect.flip); + }) + ) + ); + + expect(error).toBeInstanceOf(EventValidationError); + expect(error._tag).toBe("EventValidationError"); + expect(error._tag === "EventValidationError" && error.key).toBe("signup"); + }); + + it("retries sink failures", async () => { let attempts = 0; - const tracker = createTracker({ - events, - retries: { attempts: 2, delay: 1, factor: 1 }, - sink: () => - Effect.sync(() => { - attempts += 1; - }).pipe( - Effect.andThen(() => - attempts < 3 - ? Effect.fail(new SinkDeliveryError({ cause: "not yet" })) - : Effect.void - ) - ), - }); + const sink: Sink = () => + Effect.suspend(() => { + attempts += 1; + + return attempts < 3 + ? Effect.fail(new SinkError({ cause: "not yet" })) + : Effect.void; + }); await Effect.runPromise( - tracker.trackNow("signup", { userId: "u_1", plan: "free" }) + Effect.scoped( + Effect.gen(function* () { + const tracker = yield* make({ + events, + sink, + flushInterval: 0, + retry: { attempts: 2, delay: 1, factor: 1 }, + }); + + yield* tracker.trackNow("signup", { userId: "u_1", plan: "free" }); + }) + ) ); expect(attempts).toBe(3); }); - it("flushes queued events after the flush interval", async () => { - const batches: (readonly TrackedEvent[])[] = []; - const tracker = createTracker({ - events, - flushInterval: 1, - sink: (batch) => - Effect.sync(() => { - batches.push([...batch]); - }), - }); + it("flushes remaining events when the scope closes", async () => { + const { batches, sink } = collectingSink(); await Effect.runPromise( - tracker.track("signup", { userId: "u_1", plan: "free" }) + Effect.scoped( + Effect.gen(function* () { + const tracker = yield* make({ + events, + sink, + flushInterval: 10_000, + }); + + yield* tracker.track("signup", { userId: "u_1", plan: "free" }); + + expect(batches).toHaveLength(0); + }) + ) ); - await new Promise((resolve) => setTimeout(resolve, 10)); - await Effect.runPromise(tracker.shutdown()); expect(batches).toHaveLength(1); - expect(batches[0]).toMatchObject([ - { - key: "signup", - name: "user.signup", - payload: { userId: "u_1", plan: "free" }, - }, - ]); }); - it("does not interrupt an in-flight interval delivery on shutdown", async () => { - const batches: (readonly TrackedEvent[])[] = []; - let deliveryStarted!: () => void; - let resumeDelivery!: (effect: Effect.Effect) => void; - let shutdownCompleted = false; - let deliveryInterrupted = false; - const deliveryStartedPromise = new Promise((resolve) => { - deliveryStarted = resolve; - }); - const tracker = createTracker({ - events, - flushInterval: 1, - sink: (batch) => - Effect.callback((resume) => { - batches.push([...batch]); - resumeDelivery = resume; - deliveryStarted(); - - return Effect.sync(() => { - deliveryInterrupted = true; - }); - }), - }); + it("reports queue size", async () => { + const { sink } = collectingSink(); - await Effect.runPromise( - tracker.track("signup", { userId: "u_1", plan: "free" }) + const size = await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const tracker = yield* make({ events, sink, flushInterval: 0 }); + + yield* tracker.track("signup", { userId: "u_1", plan: "free" }); + yield* tracker.track("signup", { userId: "u_2", plan: "pro" }); + + return yield* tracker.size; + }) + ) ); - await deliveryStartedPromise; - const shutdownPromise = Effect.runPromise(tracker.shutdown()).then(() => { - shutdownCompleted = true; - }); - await new Promise((resolve) => setTimeout(resolve, 0)); + expect(size).toBe(2); + }); - expect(shutdownCompleted).toBe(false); - expect(deliveryInterrupted).toBe(false); + it("delivers on the flush interval without an explicit flush", async () => { + const { batches, sink } = collectingSink(); - resumeDelivery(Effect.void); - await shutdownPromise; + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const tracker = yield* make({ events, sink, flushInterval: 5 }); + + yield* tracker.track("signup", { userId: "u_1", plan: "free" }); + + yield* Effect.sleep(50); + }) + ) + ); - expect(deliveryInterrupted).toBe(false); expect(batches).toHaveLength(1); }); }); diff --git a/test/tracker.test.ts b/test/tracker.test.ts index bc02514..f5bca58 100644 --- a/test/tracker.test.ts +++ b/test/tracker.test.ts @@ -1,6 +1,12 @@ import { Schema } from "effect"; import { describe, expect, it } from "vitest"; -import { createTracker, event, type TrackedEvent } from "../src/index"; +import { + createTracker, + event, + type StandardResult, + type StandardSchemaV1, + type TrackedEvent, +} from "../src/index"; const events = { signup: event("user.signup", { @@ -11,12 +17,25 @@ const events = { orderId: Schema.String, total: Schema.Number, }), + pageview: event("page.viewed"), +}; + +const waitFor = async (predicate: () => boolean, timeout = 1000) => { + const start = Date.now(); + + while (!predicate()) { + if (Date.now() - start > timeout) { + throw new Error("condition not met in time"); + } + + await new Promise((resolve) => setTimeout(resolve, 5)); + } }; describe("tracker", () => { it("sends typed batches", async () => { const batches: (readonly TrackedEvent[])[] = []; - const tracker = createTracker({ + await using tracker = createTracker({ events, flushInterval: 0, sink: (batch) => { @@ -56,10 +75,29 @@ describe("tracker", () => { ]); }); + it("tracks payload-less events", async () => { + const batches: (readonly TrackedEvent[])[] = []; + await using tracker = createTracker({ + events, + flushInterval: 0, + sink: (batch) => { + batches.push(batch); + }, + }); + + tracker.track("pageview"); + + await tracker.flush(); + + expect(batches[0]).toMatchObject([ + { key: "pageview", name: "page.viewed" }, + ]); + }); + it("does not queue invalid payloads", async () => { const errors: unknown[] = []; const batches: (readonly TrackedEvent[])[] = []; - const tracker = createTracker({ + await using tracker = createTracker({ events, flushInterval: 0, onError: (error) => errors.push(error), @@ -70,13 +108,13 @@ describe("tracker", () => { tracker.track("signup", { userId: "u_1", plan: "enterprise" } as never); + await waitFor(() => errors.length === 1); await tracker.flush(); - expect(errors).toHaveLength(1); expect(batches).toHaveLength(0); }); - it("accepts full v4 schemas as event definitions", async () => { + it("accepts full effect schemas as event definitions", async () => { const schemaEvents = { identified: event( "user.identified", @@ -84,7 +122,7 @@ describe("tracker", () => { ), }; const batches: (readonly TrackedEvent[])[] = []; - const tracker = createTracker({ + await using tracker = createTracker({ events: schemaEvents, flushInterval: 0, sink: (batch) => { @@ -105,9 +143,79 @@ describe("tracker", () => { ]); }); - it("splits flushes by batch size", async () => { + it("accepts standard schemas (zod-style) as event definitions", async () => { + const userIdSchema: StandardSchemaV1 = { + "~standard": { + version: 1, + vendor: "test", + validate: (value): StandardResult<{ userId: string }> => { + if ( + typeof value === "object" && + value !== null && + "userId" in value && + typeof value.userId === "string" + ) { + return { value: { userId: value.userId } }; + } + + return { issues: [{ message: "expected { userId: string }" }] }; + }, + }, + }; + + const standardEvents = { + identified: event("user.identified", userIdSchema), + }; + const errors: unknown[] = []; + const batches: (readonly TrackedEvent[])[] = []; + await using tracker = createTracker({ + events: standardEvents, + flushInterval: 0, + onError: (error) => errors.push(error), + sink: (batch) => { + batches.push(batch); + }, + }); + + tracker.track("identified", { userId: "u_1" }); + tracker.track("identified", { userId: 42 } as never); + + await waitFor(() => errors.length === 1); + await tracker.flush(); + + expect(batches[0]).toMatchObject([ + { key: "identified", payload: { userId: "u_1" } }, + ]); + }); + + it("merges tracker context into event meta", async () => { const batches: (readonly TrackedEvent[])[] = []; - const tracker = createTracker({ + await using tracker = createTracker({ + events, + flushInterval: 0, + context: () => ({ sessionId: "s_1", source: "context" }), + sink: (batch) => { + batches.push(batch); + }, + }); + + tracker.track( + "signup", + { userId: "u_1", plan: "free" }, + { meta: { source: "event" } } + ); + + await tracker.flush(); + + expect(batches[0]?.[0]?.meta).toEqual({ + sessionId: "s_1", + source: "event", + }); + }); + + it("delivers in the background when the batch size is reached", async () => { + const batches: (readonly TrackedEvent[])[] = []; + await using tracker = createTracker({ events, batchSize: 2, flushInterval: 0, @@ -117,20 +225,45 @@ describe("tracker", () => { }); tracker.track("signup", { userId: "u_1", plan: "free" }); + tracker.track("signup", { userId: "u_2", plan: "pro" }); + + await waitFor(() => batches.length === 1); + + expect(batches[0]).toHaveLength(2); + }); + + it("splits flushes by batch size", async () => { + const batches: (readonly TrackedEvent[])[] = []; + await using tracker = createTracker({ + events, + batchSize: 2, + flushInterval: 1_000_000, + sink: (batch) => { + batches.push([...batch]); + }, + }); + + tracker.track("signup", { userId: "u_1", plan: "free" }); + + await tracker.flush(); + tracker.track("signup", { userId: "u_2", plan: "pro" }); tracker.track("purchase", { orderId: "o_1", total: 42 }); + tracker.track("purchase", { orderId: "o_2", total: 7 }); + await waitFor(() => batches.length >= 2); await tracker.flush(); - expect(batches.map((batch) => batch.length)).toEqual([2, 1]); + expect(batches.flat()).toHaveLength(4); + expect(batches.every((batch) => batch.length <= 2)).toBe(true); }); it("retries failed deliveries", async () => { let attempts = 0; - const tracker = createTracker({ + await using tracker = createTracker({ events, flushInterval: 0, - retries: { attempts: 2, delay: 1, factor: 1 }, + retry: { attempts: 2, delay: 1, factor: 1 }, sink: () => { attempts += 1; @@ -145,7 +278,7 @@ describe("tracker", () => { expect(attempts).toBe(3); }); - it("flushes remaining events on shutdown", async () => { + it("flushes remaining events on close", async () => { const batches: (readonly TrackedEvent[])[] = []; const tracker = createTracker({ events, @@ -157,26 +290,45 @@ describe("tracker", () => { tracker.track("signup", { userId: "u_1", plan: "free" }); - await tracker.shutdown(); + await tracker.close(); expect(batches).toHaveLength(1); }); - it("reports interval flush failures", async () => { - const errors: unknown[] = []; + it("reports interval flush failures with the failed batch", async () => { + const errors: [unknown, unknown][] = []; const tracker = createTracker({ events, flushInterval: 1, + onError: (error, batch) => errors.push([error, batch]), + sink: () => { + throw new Error("delivery down"); + }, + }); + + tracker.track("signup", { userId: "u_1", plan: "free" }); + + await waitFor(() => errors.length >= 1); + await tracker.close(); + + expect(errors[0]?.[1]).toMatchObject([{ key: "signup" }]); + }); + + it("reports tracking after close", async () => { + const errors: unknown[] = []; + const tracker = createTracker({ + events, + flushInterval: 0, onError: (error) => errors.push(error), sink: () => { - throw new Error("not yet"); + // Discard. }, }); + await tracker.close(); + tracker.track("signup", { userId: "u_1", plan: "free" }); - await new Promise((resolve) => setTimeout(resolve, 10)); - await tracker.shutdown(); - expect(errors).toHaveLength(1); + await waitFor(() => errors.length === 1); }); }); diff --git a/tsconfig.json b/tsconfig.json index 431474a..fa8d425 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -23,7 +23,7 @@ "noEmit": true, /* If your code runs in the DOM: */ - "lib": ["es2022", "dom", "dom.iterable"], + "lib": ["es2022", "esnext.disposable", "dom", "dom.iterable"], "plugins": [ { From 1d002823588cfd5e75f211f9b7c8b0e95b468808 Mon Sep 17 00:00:00 2001 From: Deepso Date: Fri, 3 Jul 2026 16:45:01 +0530 Subject: [PATCH 2/6] Add changeset for SDK redesign (minor bump) Co-Authored-By: Claude Fable 5 --- .changeset/nervous-pandas-refactor.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 .changeset/nervous-pandas-refactor.md diff --git a/.changeset/nervous-pandas-refactor.md b/.changeset/nervous-pandas-refactor.md new file mode 100644 index 0000000..008d467 --- /dev/null +++ b/.changeset/nervous-pandas-refactor.md @@ -0,0 +1,13 @@ +--- +"trashlytics": minor +--- + +Redesign the SDK: scoped Effect core, non-blocking track, Standard Schema support. Breaking changes: + +- Delivery moved to a background fiber; `track()` only validates and enqueues, never waits on the sink +- `trashlytics/effect`: `make` replaces `createTracker` and returns a scoped Effect; closing the scope stops the worker and flushes remaining events (replaces `shutdown`) +- Root entry: `close()` replaces `shutdown()`, trackers support `await using` (`Symbol.asyncDispose`), and pending events auto-flush on page hide/unload in browsers (`flushOnHide`) +- `event()` accepts Effect schemas, `Schema.Struct` fields, any Standard Schema v1 validator (zod/valibot/arktype), or no schema for payload-less events +- Unified tagged errors: `EventValidationError`, `UnknownEventError`, `TrackerClosedError`, `QueueFullError`, `SinkError` (replaces `SinkDeliveryError`) +- New options: `context` (meta enrichment), `retry.jitter`, `maxQueueSize` (renamed from `bufferSize`); `retries` renamed to `retry` +- `httpSink` defaults to `keepalive: true`; new `beaconSink` for browsers From 9a8abea2c5c537f2a3fff637eace3aded33fc94d Mon Sep 17 00:00:00 2001 From: Deepso Date: Fri, 3 Jul 2026 16:49:28 +0530 Subject: [PATCH 3/6] Fix close()/flush() racing in-flight fire-and-forget track() calls track() runs its effect without awaiting, so close() or flush() could close the scope and run the final flush before a pending track() (e.g. one awaiting async Standard Schema validation) enqueued its event, silently dropping it. The wrapper now tracks in-flight track() promises and settles them before flushing or closing. Co-Authored-By: Claude Fable 5 --- src/index.ts | 34 ++++++++++++++++++++++++++++------ test/tracker.test.ts | 31 +++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 6 deletions(-) diff --git a/src/index.ts b/src/index.ts index b4f467a..0207a6a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -114,7 +114,20 @@ export function createTracker( Scope.provide(make({ ...options, sink: adaptSink(options.sink) }), scope) ); - const flush = () => Effect.runPromise(tracker.flush); + // In-flight fire-and-forget track() calls (e.g. awaiting async Standard + // Schema validation). flush() and close() wait for these so events tracked + // before the call cannot be lost. + const inFlight = new Set>(); + const settleInFlight = async () => { + while (inFlight.size > 0) { + await Promise.allSettled([...inFlight]); + } + }; + + const flush = async () => { + await settleInFlight(); + await Effect.runPromise(tracker.flush); + }; const detachLifecycle = attachLifecycleFlush( options.flushOnHide ?? true, @@ -127,9 +140,10 @@ export function createTracker( let closing: Promise | undefined; const close = () => { - closing ??= (() => { + closing ??= (async () => { detachLifecycle(); - return Effect.runPromise(Scope.close(scope, Exit.void)); + await settleInFlight(); + await Effect.runPromise(Scope.close(scope, Exit.void)); })(); return closing; @@ -137,9 +151,17 @@ export function createTracker( return { track: (key, ...args) => { - Effect.runPromise(tracker.track(key, ...args)).catch((error) => { - options.onError?.(error); - }); + const pending: Promise = Effect.runPromise( + tracker.track(key, ...args) + ) + .catch((error) => { + options.onError?.(error); + }) + .finally(() => { + inFlight.delete(pending); + }); + + inFlight.add(pending); }, trackNow: (key, ...args) => diff --git a/test/tracker.test.ts b/test/tracker.test.ts index f5bca58..677e85b 100644 --- a/test/tracker.test.ts +++ b/test/tracker.test.ts @@ -295,6 +295,37 @@ describe("tracker", () => { expect(batches).toHaveLength(1); }); + it("waits for async validation before flush and close", async () => { + const asyncSchema: StandardSchemaV1 = { + "~standard": { + version: 1, + vendor: "test", + validate: async (value) => { + await new Promise((resolve) => setTimeout(resolve, 20)); + + return { value: value as { userId: string } }; + }, + }, + }; + + const asyncEvents = { identified: event("user.identified", asyncSchema) }; + const batches: (readonly TrackedEvent[])[] = []; + const tracker = createTracker({ + events: asyncEvents, + flushInterval: 0, + sink: (batch) => { + batches.push(batch); + }, + }); + + tracker.track("identified", { userId: "u_1" }); + + await tracker.close(); + + expect(batches).toHaveLength(1); + expect(batches[0]).toMatchObject([{ payload: { userId: "u_1" } }]); + }); + it("reports interval flush failures with the failed batch", async () => { const errors: [unknown, unknown][] = []; const tracker = createTracker({ From 54a0fc07bd8fd9bd6c82e31c54ccdb83f47570e7 Mon Sep 17 00:00:00 2001 From: Deepso Date: Fri, 3 Jul 2026 16:53:46 +0530 Subject: [PATCH 4/6] Make close() a hard barrier for new tracking work track() and trackNow() now check the closing flag before starting: once close() has been called, track() deterministically reports TrackerClosedError through onError and trackNow() rejects, instead of racing the final flush. This also guarantees the in-flight settling loop in close() terminates, since no new in-flight work can appear. Co-Authored-By: Claude Fable 5 --- src/index.ts | 13 +++++++++++-- test/tracker.test.ts | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/src/index.ts b/src/index.ts index 0207a6a..4c88f2d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -6,7 +6,7 @@ import type { TrackArgs, TrackedEvent, } from "./effect"; -import { make, SinkError } from "./effect"; +import { make, SinkError, TrackerClosedError } from "./effect"; // Runtime support for `await using` on platforms that predate the explicit // resource management proposal. @@ -151,6 +151,13 @@ export function createTracker( return { track: (key, ...args) => { + // Hard barrier: once close() has been called, no new tracking work is + // started, so close() cannot race a late enqueue. + if (closing !== undefined) { + options.onError?.(new TrackerClosedError()); + return; + } + const pending: Promise = Effect.runPromise( tracker.track(key, ...args) ) @@ -165,7 +172,9 @@ export function createTracker( }, trackNow: (key, ...args) => - Effect.runPromise(tracker.trackNow(key, ...args)), + closing === undefined + ? Effect.runPromise(tracker.trackNow(key, ...args)) + : Promise.reject(new TrackerClosedError()), flush, diff --git a/test/tracker.test.ts b/test/tracker.test.ts index 677e85b..3149b7b 100644 --- a/test/tracker.test.ts +++ b/test/tracker.test.ts @@ -6,6 +6,7 @@ import { type StandardResult, type StandardSchemaV1, type TrackedEvent, + TrackerClosedError, } from "../src/index"; const events = { @@ -361,5 +362,36 @@ describe("tracker", () => { tracker.track("signup", { userId: "u_1", plan: "free" }); await waitFor(() => errors.length === 1); + expect(errors[0]).toBeInstanceOf(TrackerClosedError); + }); + + it("rejects tracking while close is in progress", async () => { + const errors: unknown[] = []; + const delivered: TrackedEvent[] = []; + const tracker = createTracker({ + events, + flushInterval: 0, + onError: (error) => errors.push(error), + sink: async (batch) => { + await new Promise((resolve) => setTimeout(resolve, 10)); + delivered.push(...batch); + }, + }); + + tracker.track("signup", { userId: "u_1", plan: "free" }); + + const closed = tracker.close(); + + tracker.track("signup", { userId: "u_2", plan: "pro" }); + await expect( + tracker.trackNow("signup", { userId: "u_3", plan: "pro" }) + ).rejects.toBeInstanceOf(TrackerClosedError); + + await closed; + + expect(errors).toHaveLength(1); + expect(errors[0]).toBeInstanceOf(TrackerClosedError); + expect(delivered).toHaveLength(1); + expect(delivered[0]).toMatchObject({ payload: { userId: "u_1" } }); }); }); From 0af3325a33326ec1e4f05a6775c838a09e2cd5db Mon Sep 17 00:00:00 2001 From: Deepso Date: Fri, 3 Jul 2026 17:28:53 +0530 Subject: [PATCH 5/6] Address review: in-flight batch loss on close, README fixes, onError consistency - Make the drain loop uninterruptible (restoring the guarantee the old drainQueue had): closing the scope can no longer interrupt the worker between dequeuing a batch and delivering it, so in-flight batches always complete (or exhaust retries) before shutdown - Report trackNow delivery failures through onError like every other delivery path (callers still get the rejection) - Replace the Symbol.asyncDispose global mutation with a pure module constant so "sideEffects": false remains truthful for bundlers - Fix README Layer example: ServiceMap does not exist in effect 4.0.0-beta.93; use Context.Service (example verified against tsc) Co-Authored-By: Claude Fable 5 --- README.md | 6 +++--- src/effect.ts | 37 ++++++++++++++++++++--------------- src/index.ts | 14 ++++++++------ test/effect.test.ts | 31 ++++++++++++++++++++++++++++- test/tracker.test.ts | 46 ++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 108 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index c3bd735..0edbc86 100644 --- a/README.md +++ b/README.md @@ -132,7 +132,7 @@ In browsers, the tracker automatically flushes when the page is hidden or unload ## Errors -All failures are tagged: `EventValidationError`, `UnknownEventError`, `TrackerClosedError`, `QueueFullError`, `SinkError`. Validation and background delivery failures are reported through `onError`; `trackNow` and `flush` reject with the failure. +All failures are tagged: `EventValidationError`, `UnknownEventError`, `TrackerClosedError`, `QueueFullError`, `SinkError`. `onError` observes every delivery failure (from background flushing, `flush`, and `trackNow`) plus validation failures from fire-and-forget `track`; `trackNow` and `flush` additionally reject with the failure so callers can react. ## Effect-Native API @@ -164,9 +164,9 @@ const program = Effect.gen(function* () { To share a tracker across your app, wrap it in a Layer: ```ts -import { Layer, ServiceMap } from "effect" +import { Context, Layer } from "effect" -class Analytics extends ServiceMap.Key>()("Analytics") {} +class Analytics extends Context.Service>()("Analytics") {} const AnalyticsLayer = Layer.effect(Analytics, Tracker.make({ events, sink })) ``` diff --git a/src/effect.ts b/src/effect.ts index d4fc3f0..03d5ee5 100644 --- a/src/effect.ts +++ b/src/effect.ts @@ -527,7 +527,13 @@ export function make< Schedule.exponential(Duration.millis(retry.delay), retry.factor) ) : Schedule.exponential(Duration.millis(retry.delay), retry.factor), - }); + }).pipe( + Effect.tapCause((cause) => + Effect.sync(() => { + options.onError?.(Cause.squash(cause), batch); + }) + ) + ); const takeBatch = Effect.gen(function* () { const batch: TrackedEvent[] = []; @@ -548,24 +554,23 @@ export function make< // Serialized with trackNow so batches reach the sink in order. A batch // that fails after all retries is reported via onError and dropped; // events still in the queue stay queued for the next attempt. + // Uninterruptible so that closing the scope cannot interrupt the worker + // between taking a batch off the queue and delivering it — an in-flight + // batch always completes (or exhausts its retries) before shutdown. const drain = deliveryLock.withPermit( - Effect.gen(function* () { - while (true) { - const batch = yield* takeBatch; + Effect.uninterruptible( + Effect.gen(function* () { + while (true) { + const batch = yield* takeBatch; - if (batch.length === 0) { - return; - } + if (batch.length === 0) { + return; + } - yield* deliver(batch).pipe( - Effect.tapCause((cause) => - Effect.sync(() => { - options.onError?.(Cause.squash(cause), batch); - }) - ) - ); - } - }) + yield* deliver(batch); + } + }) + ) ); const drainSilently = drain.pipe(Effect.catchCause(() => Effect.void)); diff --git a/src/index.ts b/src/index.ts index 4c88f2d..05227a1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,11 +8,13 @@ import type { } from "./effect"; import { make, SinkError, TrackerClosedError } from "./effect"; -// Runtime support for `await using` on platforms that predate the explicit -// resource management proposal. -(Symbol as { asyncDispose?: symbol }).asyncDispose ??= Symbol.for( - "Symbol.asyncDispose" -); +// Key used for `await using` support. Falls back to the registered symbol on +// platforms that predate explicit resource management — the same fallback +// TypeScript's downlevel helpers use — without mutating the Symbol global +// (this module declares `sideEffects: false`). +const asyncDispose: typeof Symbol.asyncDispose = + Symbol.asyncDispose ?? + (Symbol.for("Symbol.asyncDispose") as typeof Symbol.asyncDispose); export type { EventDefinition, @@ -180,7 +182,7 @@ export function createTracker( close, - [Symbol.asyncDispose]: close, + [asyncDispose]: close, }; } diff --git a/test/effect.test.ts b/test/effect.test.ts index 0628ce0..308db66 100644 --- a/test/effect.test.ts +++ b/test/effect.test.ts @@ -1,4 +1,4 @@ -import { Effect, Schema } from "effect"; +import { Effect, Latch, Schema } from "effect"; import { describe, expect, it } from "vitest"; import { EventValidationError, @@ -100,6 +100,35 @@ describe("effect tracker", () => { expect(attempts).toBe(3); }); + it("does not drop an in-flight batch when the scope closes", async () => { + const delivered: TrackedEvent[] = []; + const sinkStarted = Latch.makeUnsafe(false); + + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const tracker = yield* make({ + events, + batchSize: 1, + flushInterval: 0, + sink: (batch) => + Effect.gen(function* () { + sinkStarted.openUnsafe(); + yield* Effect.sleep(30); + delivered.push(...batch); + }), + }); + + yield* tracker.track("signup", { userId: "u_1", plan: "free" }); + // Leave the scope while the background worker is mid-delivery. + yield* sinkStarted.await; + }) + ) + ); + + expect(delivered).toHaveLength(1); + }); + it("flushes remaining events when the scope closes", async () => { const { batches, sink } = collectingSink(); diff --git a/test/tracker.test.ts b/test/tracker.test.ts index 3149b7b..29071d5 100644 --- a/test/tracker.test.ts +++ b/test/tracker.test.ts @@ -279,6 +279,52 @@ describe("tracker", () => { expect(attempts).toBe(3); }); + it("reports trackNow delivery failures to onError", async () => { + const errors: [unknown, unknown][] = []; + await using tracker = createTracker({ + events, + flushInterval: 0, + onError: (error, batch) => errors.push([error, batch]), + sink: () => { + throw new Error("delivery down"); + }, + }); + + await expect( + tracker.trackNow("signup", { userId: "u_1", plan: "free" }) + ).rejects.toThrow(); + + expect(errors).toHaveLength(1); + expect(errors[0]?.[1]).toMatchObject([{ key: "signup" }]); + }); + + it("does not drop an in-flight batch when closed mid-delivery", async () => { + const delivered: TrackedEvent[] = []; + let signalStarted = () => { + // Reassigned below. + }; + const sinkStarted = new Promise((resolve) => { + signalStarted = resolve; + }); + const tracker = createTracker({ + events, + batchSize: 1, + flushInterval: 0, + sink: async (batch) => { + signalStarted(); + await new Promise((resolve) => setTimeout(resolve, 30)); + delivered.push(...batch); + }, + }); + + tracker.track("signup", { userId: "u_1", plan: "free" }); + + await sinkStarted; + await tracker.close(); + + expect(delivered).toHaveLength(1); + }); + it("flushes remaining events on close", async () => { const batches: (readonly TrackedEvent[])[] = []; const tracker = createTracker({ From c12d010bd35940e11a19de20973f3f415bc789c1 Mon Sep 17 00:00:00 2001 From: Deepso Date: Fri, 3 Jul 2026 21:25:57 +0530 Subject: [PATCH 6/6] Address review: bounded shutdown with never-settling sinks, safe onError MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add deliveryTimeout (default 30s): each sink call is bounded and a timed-out attempt fails with SinkError, subject to the retry policy, so flush() and close() can no longer hang on a sink that never settles. The timeout window is explicitly interruptible inside the otherwise-uninterruptible drain loop; if an interrupt lands there the dequeued batch is re-offered so it is not lost. - Reorder shutdown finalizers to mark closed, drain remaining events (waiting on the delivery lock for any in-flight batch), and only then interrupt the now-idle worker — shutdown never cancels a delivery mid-flight and no longer contends with an uninterruptible worker. - Guard the onError observer with try/catch so a throwing callback can no longer defect the delivery path out of trackNow/flush. Co-Authored-By: Claude Fable 5 --- .changeset/nervous-pandas-refactor.md | 2 +- README.md | 1 + src/effect.ts | 72 ++++++++++++++++++++++----- test/effect.test.ts | 23 +++++++++ test/tracker.test.ts | 26 ++++++++++ 5 files changed, 111 insertions(+), 13 deletions(-) diff --git a/.changeset/nervous-pandas-refactor.md b/.changeset/nervous-pandas-refactor.md index 008d467..e5cddde 100644 --- a/.changeset/nervous-pandas-refactor.md +++ b/.changeset/nervous-pandas-refactor.md @@ -9,5 +9,5 @@ Redesign the SDK: scoped Effect core, non-blocking track, Standard Schema suppor - Root entry: `close()` replaces `shutdown()`, trackers support `await using` (`Symbol.asyncDispose`), and pending events auto-flush on page hide/unload in browsers (`flushOnHide`) - `event()` accepts Effect schemas, `Schema.Struct` fields, any Standard Schema v1 validator (zod/valibot/arktype), or no schema for payload-less events - Unified tagged errors: `EventValidationError`, `UnknownEventError`, `TrackerClosedError`, `QueueFullError`, `SinkError` (replaces `SinkDeliveryError`) -- New options: `context` (meta enrichment), `retry.jitter`, `maxQueueSize` (renamed from `bufferSize`); `retries` renamed to `retry` +- New options: `context` (meta enrichment), `retry.jitter`, `deliveryTimeout` (bounds each sink call, default 30s), `maxQueueSize` (renamed from `bufferSize`); `retries` renamed to `retry` - `httpSink` defaults to `keepalive: true`; new `beaconSink` for browsers diff --git a/README.md b/README.md index 0edbc86..2a32fb4 100644 --- a/README.md +++ b/README.md @@ -177,6 +177,7 @@ const AnalyticsLayer = Layer.effect(Analytics, Tracker.make({ events, sink })) | --- | --- | --- | | `batchSize` | `20` | Max events per sink call. Reaching it triggers background delivery. | | `flushInterval` | `5000` | Auto-flush interval in ms. `0` disables interval flushing. | +| `deliveryTimeout` | `30000` | Max ms per sink call before the attempt is failed (and retried per `retry`). Keeps `flush`/`close` bounded even if a sink never settles. `0` disables. | | `maxQueueSize` | `1000` | Max queued events; beyond it new events are rejected. | | `retry` | none | Retry count or `{ attempts, delay, factor, jitter }`. | | `context` | none | Static or lazy metadata merged into every event's `meta`. | diff --git a/src/effect.ts b/src/effect.ts index 03d5ee5..fdf6e9b 100644 --- a/src/effect.ts +++ b/src/effect.ts @@ -398,6 +398,13 @@ export interface TrackerOptions< * Per-event metadata wins on key conflicts. */ readonly context?: EventMeta | (() => EventMeta); + /** + * Maximum time in milliseconds a single sink call may take before it is + * interrupted and treated as a failed delivery attempt (subject to the + * retry policy). Keeps `flush` and shutdown bounded even when a sink never + * settles. Set to 0 to disable. Defaults to 30000. + */ + readonly deliveryTimeout?: number; /** Event definitions accepted by this tracker. */ readonly events: Events; /** @@ -459,8 +466,12 @@ export interface Tracker< Error = never, Requirements = never, > { - /** Delivers all currently queued events and waits for completion. */ - readonly flush: Effect.Effect; + /** + * Delivers all currently queued events and waits for completion. Fails with + * the sink's error, or `SinkError` when a delivery attempt exceeds + * `deliveryTimeout`. + */ + readonly flush: Effect.Effect; /** Number of events currently queued. */ readonly size: Effect.Effect; /** @@ -477,7 +488,7 @@ export interface Tracker< ...args: TrackArgs ) => Effect.Effect< void, - Exclude | Error, + Exclude | Error | SinkError, Requirements >; } @@ -485,6 +496,7 @@ export interface Tracker< const DEFAULT_BATCH_SIZE = 20; const DEFAULT_MAX_QUEUE_SIZE = 1000; const DEFAULT_FLUSH_INTERVAL = 5000; +const DEFAULT_DELIVERY_TIMEOUT = 30_000; const DEFAULT_RETRY_DELAY = 250; const DEFAULT_RETRY_FACTOR = 2; @@ -513,14 +525,36 @@ export function make< const batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE; const maxQueueSize = options.maxQueueSize ?? DEFAULT_MAX_QUEUE_SIZE; const flushInterval = options.flushInterval ?? DEFAULT_FLUSH_INTERVAL; + const deliveryTimeout = options.deliveryTimeout ?? DEFAULT_DELIVERY_TIMEOUT; const retry = normalizeRetry(options.retry); const queue = yield* Queue.dropping>(maxQueueSize); const wakeWorker = Latch.makeUnsafe(false); const deliveryLock = Semaphore.makeUnsafe(1); let closed = false; + // A single delivery attempt. Bounded by deliveryTimeout so a sink that + // never settles cannot hang flush or shutdown; the attempt window is + // explicitly interruptible so the timeout works even inside the + // uninterruptible drain loop. + const attemptDelivery = ( + batch: readonly TrackedEvent[] + ): Effect.Effect => + deliveryTimeout > 0 + ? Effect.interruptible( + Effect.timeoutOrElse(options.sink(batch), { + duration: Duration.millis(deliveryTimeout), + orElse: () => + new SinkError({ + cause: new Error( + `Sink did not complete within ${deliveryTimeout}ms` + ), + }), + }) + ) + : options.sink(batch); + const deliver = (batch: readonly TrackedEvent[]) => - Effect.retry(options.sink(batch), { + Effect.retry(attemptDelivery(batch), { times: retry.attempts, schedule: retry.jitter ? Schedule.jittered( @@ -530,7 +564,11 @@ export function make< }).pipe( Effect.tapCause((cause) => Effect.sync(() => { - options.onError?.(Cause.squash(cause), batch); + try { + options.onError?.(Cause.squash(cause), batch); + } catch { + // onError is an observer; its failures must not affect delivery. + } }) ) ); @@ -554,9 +592,10 @@ export function make< // Serialized with trackNow so batches reach the sink in order. A batch // that fails after all retries is reported via onError and dropped; // events still in the queue stay queued for the next attempt. - // Uninterruptible so that closing the scope cannot interrupt the worker - // between taking a batch off the queue and delivering it — an in-flight - // batch always completes (or exhausts its retries) before shutdown. + // Uninterruptible so an in-flight batch always completes (or exhausts its + // retries) before shutdown. The delivery-timeout window inside deliver is + // the one interruptible gap; if an interrupt lands there (e.g. a caller + // interrupts flush), the dequeued batch is put back so it is not lost. const drain = deliveryLock.withPermit( Effect.uninterruptible( Effect.gen(function* () { @@ -567,7 +606,13 @@ export function make< return; } - yield* deliver(batch); + yield* deliver(batch).pipe( + Effect.onInterrupt(() => + Effect.sync(() => { + Queue.offerAllUnsafe(queue, batch); + }) + ) + ); } }) ) @@ -588,10 +633,13 @@ export function make< } }); - // Finalizers run in reverse order: mark closed, interrupt the worker - // (registered by forkScoped), then flush whatever is still queued. - yield* Effect.addFinalizer(() => drainSilently); + // Finalizers run in reverse order of registration: mark closed (so no new + // events are accepted), flush everything still queued (the delivery lock + // makes this wait for any in-flight worker delivery first), and only then + // interrupt the now-idle worker. Interrupting last means shutdown never + // cancels a delivery mid-flight. yield* Effect.forkScoped(worker); + yield* Effect.addFinalizer(() => drainSilently); yield* Effect.addFinalizer(() => Effect.sync(() => { closed = true; diff --git a/test/effect.test.ts b/test/effect.test.ts index 308db66..059a11f 100644 --- a/test/effect.test.ts +++ b/test/effect.test.ts @@ -100,6 +100,29 @@ describe("effect tracker", () => { expect(attempts).toBe(3); }); + it("fails trackNow with the typed sink error even when onError throws", async () => { + const error = await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const tracker = yield* make({ + events, + flushInterval: 0, + onError: () => { + throw new Error("observer boom"); + }, + sink: () => Effect.fail(new SinkError({ cause: "down" })), + }); + + return yield* tracker + .trackNow("signup", { userId: "u_1", plan: "free" }) + .pipe(Effect.flip); + }) + ) + ); + + expect(error).toBeInstanceOf(SinkError); + }); + it("does not drop an in-flight batch when the scope closes", async () => { const delivered: TrackedEvent[] = []; const sinkStarted = Latch.makeUnsafe(false); diff --git a/test/tracker.test.ts b/test/tracker.test.ts index 29071d5..c434358 100644 --- a/test/tracker.test.ts +++ b/test/tracker.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from "vitest"; import { createTracker, event, + SinkError, type StandardResult, type StandardSchemaV1, type TrackedEvent, @@ -298,6 +299,31 @@ describe("tracker", () => { expect(errors[0]?.[1]).toMatchObject([{ key: "signup" }]); }); + it("close() stays bounded when the sink never settles", async () => { + const errors: unknown[] = []; + const tracker = createTracker({ + events, + batchSize: 1, + deliveryTimeout: 20, + flushInterval: 0, + onError: (error) => errors.push(error), + sink: () => + new Promise(() => { + // Never settles. + }), + }); + + tracker.track("signup", { userId: "u_1", plan: "free" }); + + await tracker.close(); + + expect(errors.length).toBeGreaterThanOrEqual(1); + expect(errors[0]).toBeInstanceOf(SinkError); + expect(String((errors[0] as SinkError).cause)).toContain( + "did not complete within 20ms" + ); + }); + it("does not drop an in-flight batch when closed mid-delivery", async () => { const delivered: TrackedEvent[] = []; let signalStarted = () => {