@hyghstreet/utils

Generic runtime helpers shared across hyghstreet packages: a queue, a mutex, a keyed registry, a leaky-bucket rate limiter, and a Shopify GraphQL query-cost calculator. Published as a library (dist/index.js + dist/index.d.ts) — everything below is exported from the package root.

import { Queue, Mutex, Registry, LeakyBucketProcessor, calculatePoints } from "@hyghstreet/utils";

Install / build

npm run build   # tsc -> dist/
npm run watch   # tsc --watch
npm test        # runs src/index.test.ts via tsx

There's no test framework — *.test.ts files are plain scripts that assert/throw on failure. Run a single one directly with npx tsx src/<name>.test.ts.

Queue<T>

Plain FIFO queue, no concurrency control.

const queue = new Queue<number>();
queue.enqueue(1);
queue.enqueue(2);
queue.peek();      // 1 (does not remove)
queue.dequeue();   // 1
queue.length;      // 1
queue.isEmpty();   // false
queue.toArray();   // snapshot array, front to back
queue.clear();     // empties the queue

Mutex

Async mutual-exclusion lock. Waiters queue up and are woken in order.

const mutex = new Mutex();

// manual acquire/release
const lock = await mutex.acquire();
try {
  // critical section
} finally {
  lock.release();
}

// or let it handle release for you
await mutex.runExclusive(async () => {
  // critical section
});

Registry<T>

Keyed store for singletons/instances. Throws on duplicate registration instead of silently overwriting.

const registry = new Registry<Connection>();

registry.register("shop-a", connectionA);
registry.register("shop-a", connectionA); // throws: Key "shop-a" already exists.

registry.has("shop-a");      // true
registry.get("shop-a");      // connectionA
registry.unregister("shop-a"); // true, removes it
registry.getKeys();          // string[]
registry.getValues();        // T[]
registry.getEntries();       // [string, T][]
registry.clear();            // empties the registry

LeakyBucketProcessor

Leaky-bucket rate limiter for async jobs, built for throttling Shopify GraphQL API calls (where each request has a query cost and Shopify replenishes a points bucket over time).

How it works:

• You enqueue(executor, estimatedCost) a job with its estimated point cost. It returns a promise that resolves/rejects with the job's result.
• A background loop (driven by a 1s interval that restores restoreRate points, capped at maximumAvailable) dequeues and runs jobs only once enough points (currentlyAvailable) exist; otherwise it waits for points to regenerate.
• After a job runs, an optional actualCostExtractor reconciles the estimated cost against the real cost returned by the API (Shopify's query cost is dynamic) and adjusts currentlyAvailable by the difference.
• Failed jobs are retried by re-enqueuing, up to maxRetries times — except when the error message contains "Error: Access denied", which rejects immediately without retrying.
• Call stop() to clear the restore interval and stop processing.

const processor = new LeakyBucketProcessor({
  throttleStatus: { maximumAvailable: 1000, currentlyAvailable: 1000, restoreRate: 50 },
  maxRetries: 3,
  debug: false,
  prometheusPrefix: "leaky_bucket_processor",
  actualCostExtractor: (response, estimatedCost) => response.extensions.cost.actualQueryCost,
});

const result = await processor.enqueue(
  () => shopifyClient.request(query, variables),
  calculatePoints(query, variables) // estimated cost, see below
);

processor.throttleStatus;   // { maximumAvailable, currentlyAvailable, restoreRate }
processor.queueLength;      // pending jobs
processor.isStopped;        // boolean
processor.getPrometheusMetrics(); // Prometheus text-format metrics string

processor.stop();

Config options (all optional, sensible defaults applied):

| Option | Default | Purpose | |---|---|---| | throttleStatus | { maximumAvailable: 100, currentlyAvailable: 100, restoreRate: 50 } | Initial bucket state | | maxRetries | 3 | Retries per job before rejecting | | debug | false | Console logging of processor internals | | prometheusPrefix | "leaky_bucket_processor" | Metric name prefix | | actualCostExtractor | none | (response, estimatedCost) => actualCost, used to reconcile bucket after each job |

calculatePoints(query, variables?)

Estimates a Shopify Admin GraphQL query's cost in points before executing it, so you can pass that as estimatedCost into LeakyBucketProcessor.enqueue(). Mirrors Shopify's calculated-cost model (shopify.dev/docs/api/usage/rate-limits):

• Scalar/enum fields: 0 points.
• Object fields: 1 point.
• Connection fields (first/last arguments): cost scales linearly with the requested page size.
• Mutations: flat 10 points, except productSet, which uses Shopify's dynamic complexity formula (10 + 0.2×variants + 0.6×variant files + 0.4×variant metafields + 0.4×product metafields + 1.9×product files).

calculatePoints(`query { shop { name } }`); // 1

calculatePoints(
  `query getProducts($first: Int!) { products(first: $first) { edges { node { id } } } }`,
  { first: 50 }
); // 52 -> 1 (products) + 1 (edges) + 50 (node, sized by first)

calculatePoints(
  `mutation createProduct($productSet: ProductSetInput!) { productSet(input: $productSet) { product { id } } }`,
  { productSet: { variants: [{ file: {} }] } }
); // 10 (base) + 0.2 (1 variant) + 0.6 (1 variant file)

Variables passed for connection arguments must be numbers (or the argument must be a literal IntValue) — otherwise that field isn't counted as a connection.