Frequently Asked Questions

Install the CLI, scaffold a service, implement the methods you want to expose, run it, and generate a typed client from the running process. Five commands, and the only prerequisites are Node.js 22.12+ and a reachable Redis 6.2+.

npm i -g @imqueue/cli
mkdir user-service && cd user-service
imq service create
npm run dev                                     # terminal 1: the service
imq client generate UserService ./src/clients   # terminal 2: the client

What the scaffold gives you is a class extending IMQService with @expose() on each remotely callable method. There is no schema file and no IDL: the generated client is built by asking the running service to describe itself, which is why step four has to be running before step five.

Reference: IMQService · expose() · IMQClient.create(). Worked walkthroughs: Get started and the Tutorial.

Decorate it with @expose() and give it a JSDoc block with typed @param and @returns tags. A method without the decorator is absent from the service description and stays callable in-process only — a remote call to it is rejected with IMQ_RPC_NO_ACCESS.

import { IMQService, expose } from '@imqueue/rpc';

export class UserService extends IMQService {
    /**
     * Returns how many users are active
     *
     * @return {Promise<number>} - the number of active users
     */
    @expose()
    public async countActive(): Promise<number> {
        return this.users.filter(user => user.isActive).length;
    }
}

Four things decide whether it actually works. The doc-block is not decoration: it is the only type source the client generator has, and the documented @param list is also what the service's argument-count check validates — it must match the method's real arity or calls fail with IMQ_RPC_INVALID_ARGS_COUNT. Combined with @lock(), @cache or @logged(), @expose() must be the innermost decorator, listed last: those replace the method with a (...args) wrapper, and registering the wrapper records its rest parameter as the method's only argument. It applies to instance methods only — on a static method it silently registers under the pseudo-class name Function and the method stays unreachable. And under standard decorators registration is deferred to an initializer that runs on first construction, so the description is empty until an instance exists.

Reference: expose() · IMQService · lock() · cache · logged()

Declare the type as a class, put @classType() on the class and @property() on each field that crosses the wire. @property()'s first argument is the type in TypeScript notation, and a second argument of true marks the field optional. That pair is what lets the service and the generated client agree on the shape.

import { classType, property } from '@imqueue/rpc';

@classType()
export class UserObject {
    @property('string')
    id: string;

    @property('boolean')
    isActive: boolean;

    // optional — pass true as the second argument
    @property('string', true)
    nickname?: string;
}

Under standard (TC39) decorators — the protocol @imqueue/rpc targets — a field decorator cannot see its own class, so @property() only collects field metadata and @classType() is what flushes it under the class name. Omitting it raises no error: the type is silently missing from the RPC type description and generated clients then reference a type nobody declared. @indexed() performs the same flush plus an index signature, so a class carrying it does not also need @classType(). A field may name another complex type ('UserCarObject') or an array of one ('UserCarObject[]'); everything crossing the queue is JSON, so returning a plain object literal instead of a declared class works at runtime and types the field any on the client.

Reference: classType() · property() · indexed()

Because the JSDoc block on an exposed method is the runtime type source. Standard decorators provide no runtime type reflection, so the client generator reads the doc-block and nothing else — strip comments and it sees no types at all: parameters and return values degrade to any, and the @param list the service's argument-count check validates against disappears with them.

{
  "compilerOptions": {
    "target": "es2024",
    "lib": ["es2024", "esnext.decorators"],
    "experimentalDecorators": false,

    // doc-blocks are the only type source the generator reads,
    // so stripping comments leaves it nothing
    "removeComments": false
  }
}

This is a property of the consuming project, not of the packages: your service compiles fine with comments stripped, and the damage shows up later as an untyped or empty generated client.

Reference: expose() · @imqueue/rpc package reference · migration from 2.x to 3.x

Run imq client generate <ServiceName> <outDir> while the service is up with Redis reachable. Generation works by asking the running service to describe itself, so there is no schema file to keep in step — and nothing to generate from if the process is not running.

imq client generate UserService ./src/clients

The generated file exports exactly one thing: a namespace named after the service with a lower-case first letter, holding a client class whose name is the service's with a trailing Service replaced by Client. So UserService gives you userService.UserClient, and there is no top-level UserClient to import. Treat the file as a build artefact — regenerate it when the interface changes, -o overwrites without prompting. IMQClient.create() does the same job in-process when you would rather not shell out.

import { userService } from './src/clients/UserService.js';

const client = new userService.UserClient();

await client.start();
try {
    console.log(await client.countActive());
} finally {
    await client.destroy();   // or the process will not exit
}

Reference: IMQClient.create() · IMQClient · Clients & Versioning

Declare the rules on the input class — @validate() per field, @validatable() on the class to seal them — and guard the method with @validated(...). The check runs before the method body, and the body then receives exactly what the caller passed.

import { z } from 'zod';
import { validatable, validate, validated } from '@imqueue/validation';

@validatable()
class Credentials {
    @validate(z.string().email())
    email!: string;

    @validate(z.string().min(8))
    password!: string;
}

class AuthService {
    @validated(Credentials)
    async signIn(creds: Credentials): Promise<string> {
        return `token-for-${creds.email}`;
    }
}

@validatable() is not optional bookkeeping. Field validators are buffered until a class decorator claims them, so a class that uses @validate() without it hands its fields to the next class that is sealed — which then rejects valid input over properties it does not declare, while the class with the real mistake validates nothing. Note also that arguments are checked but never replaced, so transforming schemas (z.coerce.number(), .trim(), .default(...)) validate as expected and change nothing the method sees. And a failure does not reach a remote caller as a ZodError: @imqueue/rpc converts whatever a method throws into its own error payload, so the caller sees IMQ_RPC_CALL_ERROR with Zod's issue list as the message string.

Reference: validated() · validate() · validatable() · schemaOf()

Decorate the service class with @PgCache() and the method with @cacheWith(), naming the tables the result depends on. PostgreSQL then decides when the entry dies: @PgCache() installs a change-notify trigger on each declared table and subscribes to one LISTEN/NOTIFY channel per table, and a row change drops every entry tagged with that table. So the entry lives exactly as long as the data behind it is unchanged — no guessed TTL, no manual del().

import { PgCache, cacheWith } from '@imqueue/pg-cache';

@PgCache({
    postgres: process.env.DB_URL!,
    redis: { host: 'localhost', port: 6379 },
})
class UserService extends IMQService {
    @cacheWith({ channels: ['users'] })
    public async list(): Promise<User[]> {
        return this.db.query('SELECT * FROM users');
    }
}

Two things to know before you rely on it. The triggers and the subscription are established in start(), so a service that never starts is never cached. And a ChannelFilter given as an array of ChannelOperation is an exclusion list — the operations named in it are the ones that do not invalidate — which reads the opposite way round from how it looks.

Reference: PgCache() · cacheWith() · CacheWithOptions · ChannelFilter · ChannelOperation

@cacheWith() names its tables literally; @cacheBy() derives them from a model and narrows invalidation using the field map the caller actually asked for, so a change to a column nobody selected does not drop the entry. It is the same mechanism with a smaller blast radius.

import { cacheBy } from '@imqueue/pg-cache';

class UserService extends IMQService {
    // fields is the 2nd argument, which is where cacheBy looks by default
    @cacheBy(User, { ttl: 30000 })
    public async list(filter: UserFilter, fields?: any): Promise<User[]> {
        return this.repo.find(filter, fields);
    }
}

fieldsArg is the zero-based position of that field map among the method's runtime arguments. Omit it and the decorator looks at the second argument — the position @imqueue/pg-sequelize services pass it in — and pass -1 to disable the analysis explicitly. The field map itself is the kind of thing fieldsMap() from graphql-fields-list produces from an incoming GraphQL request, which is where most callers get one.

Reference: cacheBy() · CacheByOptions · CacheByOptions.fieldsArg · channelsOf() · cacheWith()

Store it with a set of tags and invalidate a tag rather than a key. @imqueue/tag-cache keeps values under their own keys and additionally adds each key to a Redis set per tag, so dropping a tag drops everything stored under it — which is exactly what plain key-based caching cannot express.

import { RedisCache } from '@imqueue/rpc';
import { TagCache } from '@imqueue/tag-cache';

const cache = new TagCache(await new RedisCache().init({ prefix: 'app' }));

await cache.set('user:1:invoices', invoices, ['user:1', 'invoices'], 60000);

// when user 1 changes: drops the entry above, and anything else
// tagged user:1, whatever key it was stored under
await cache.invalidate('user:1');

Reads and writes never throw on a Redis failure — they log and report it in the return value, so an outage degrades to cache misses instead of taking the caller down. Two consequences worth holding on to: get() returning null means "not cached or the lookup failed", and invalidate() resolves once the work is issued, not once the keys are gone.

Reference: TagCache · TagCache.set() · TagCache.invalidate() · TagCache.get()

Push it with { delay } and return a positive number from the handler to have it come back. The retry policy lives in the handler rather than in queue configuration, because the handler's return value is the re-scheduling instruction.

import JobQueue from '@imqueue/job';

const queue = new JobQueue<Email>({ name: 'Email' });

queue.onPop(async (email: Email) => {
    try {
        await send(email);
        return -1;              // done; no re-schedule
    } catch (err) {
        return 60000;           // try again in a minute
    }
});

await queue.start();

queue.push({ to: 'a@b.c', subject: 'Later' }, {
    delay: 3600000,      // not before an hour from now
    ttl: 86400000,       // and stop re-scheduling after a day
});

That last row is the trap: a job pushed with no delay and then throwing is dropped rather than retried. Catch your own errors and return a number. ttl bounds how long a job stays worth re-scheduling, counted from the push. Delivery is at-least-once, so handlers must be idempotent, and job data travels as JSON — class instances, Date and undefined properties do not arrive as they left.

Reference: JobQueue · JobQueuePopHandler · PushOptions.delay · PushOptions.ttl · JobQueueWorker · JobQueuePublisher

Use PgPubSub and leave singleListener at its default of true. LISTEN/NOTIFY is a broadcast — every listening connection receives every notification, so a service scaled to N replicas handles each message N times. In single-listener mode the replicas compete for a per-channel lock held as a row in PostgreSQL, only the holder listens, and the rest stay connected as hot standbys.

import { type AnyJson, PgPubSub } from '@imqueue/pg-pubsub';

const pubSub = new PgPubSub({ connectionString: process.env.DB_URL });

pubSub.on('connect', async () => {
    await pubSub.listen('UserChanged');
});
pubSub.on('message', (channel: string, payload: AnyJson) =>
    handle(channel, payload),
);

await pubSub.connect();

The trade is worth stating plainly: this makes delivery at-most-once. NOTIFY has no backlog, so anything published while no process holds the lock is gone. A clean shutdown releases the lock and a standby takes over at once; an unclean exit leaves the channel unhandled until the next acquire retry. PostgreSQL also caps payloads at 8000 bytes. Where losing a message is unacceptable, pair this with a durable queue rather than replacing one.

Reference: PgPubSub · PgPubSubOptions.singleListener · PgPubSub.listen() · PgPubSub.notify()

PgIpLock holds the lock for one channel as a row in PostgreSQL and retries acquiring it on an interval; it also installs SIGINT, SIGTERM and SIGABRT handlers so a shutdown releases the lock and another process can pick the channel up immediately. Turn it off — singleListener: false, which swaps in NoLock — only when every replica genuinely needs to see every notification, such as invalidating an in-process cache in each one.

const pubSub = new PgPubSub({
    connectionString: process.env.DB_URL,
    singleListener: false,   // every replica handles every notification
});

acquireInterval is the tuning knob while the lock is on, and it is a real trade-off in both directions: too short, with many replicas, floods the database with lock-acquire requests; too long widens the window in which a silent disconnect leaves a channel unhandled. You do not normally construct PgIpLock yourself — PgPubSub does it — but it is exported so the mechanism can be reused elsewhere.

Reference: PgIpLock · NoLock · PgPubSubOptions.singleListener · PgPubSubOptions.acquireInterval

Take the loading out of the field resolvers. Declare a bulk loader per type and the requirements between types once at start-up, then make a single load() call in the top-level resolver: it walks the fields the client actually asked for, merges everything that needs the same type into one filter, and calls each loader once per level instead of once per parent object.

import { Dependency } from '@imqueue/graphql-dependency';
import { fieldsMap } from 'graphql-fields-list';

// at start-up, next to the type definitions
Dependency(UserType).defineLoader(async (context, filter, fields) =>
    (await context.user.listUser(filter, fields)).data,
);

Dependency(CompanyType).require(UserType, () => ({
    as: CompanyType.getFields().employees,
    filter: {
        [UserType.getFields().companyId.name]: CompanyType.getFields().id,
    },
}));

// in the top-level company resolver
async function companies(source, args, context, info) {
    const data = await context.company.listCompany(args);

    return Dependency(CompanyType).load(data, context, fieldsMap(info));
}

Every object taking part must carry an id — loaded rows are matched back onto their parents by id and by nothing else. Batching is per level rather than global: siblings of one type run concurrently, and the next level down waits, because a child's filter is built from values the parent level has just loaded. The resolution cache lives for the duration of one load() call, so no request can serve another request's stale data. On a type-graphql schema use @DependencyFor() on the decorated class instead — and remember to run every hook in schemaHooks after building the schema, because nothing else does, and missing that step leaves the dependency fields silently empty.

Reference: Dependency · defineLoader() · require() · load() · DependencyFor() · schemaHooks

Register ImqueueInstrumentation at start-up, before any client or service is constructed. Every RPC through @imqueue/rpc then produces a CLIENT span on the calling side and a SERVER span on the handling side, linked into one trace by context carried in the IMQ request metadata — with no changes to service or client code.

import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
import { registerInstrumentations } from '@opentelemetry/instrumentation';
import { ImqueueInstrumentation } from '@imqueue/opentelemetry';

new NodeTracerProvider().register();

registerInstrumentations({
    instrumentations: [new ImqueueInstrumentation()],
});

"Before any client or service is constructed" is the whole reason this belongs at start-up. The instrumentation works by patching @imqueue/rpc's exported default option singletons rather than by hooking module loading, so anything built before enable() has already copied those defaults and is not traced. The same mechanism has a second failure mode: if the @imqueue/rpc this package resolves is a different copy from the one your application imported — duplicate installs at different tree depths — the patch lands on the wrong singletons and no spans appear at all. Note also that this package only produces spans; exporting them is the host application's job.

Reference: ImqueueInstrumentation · traced() · traceStart() · traceEnd()

Declare a file transport in LOGGER_TRANSPORTS and import the default logger. There is nothing to construct and nothing to wire: the default export is already configured from the environment, and its records go to a winston File transport, which writes one JSON object per line unless you hand it a different format.

export LOGGER_TRANSPORTS='[{
  "type": "file",
  "options": { "filename": "/var/log/user-service.log" },
  "enabled": true
}]'
export LOGGER_METADATA='{"service":"%name %version"}'

import logger from '@imqueue/async-logger';

logger.info('service started on port %s', port);

%name and %version are substituted from the running service's own package.json, so one configuration can be shared across services. With no transports configured the logger still works, console only — that is the intended local-development mode, not a misconfiguration. One caveat that follows from the package's name: console writes are deferred with setTimeout, so a process that exits immediately after logging may lose the tail. Log a tick before exiting if the last lines matter.

Reference: Logger · TransportOptions · getTransport() · defaultMetadata()

LOGGER_TRANSPORTS is a JSON array of transport declarations, each one { type, options, enabled }. type is 'file' or 'http' and any other value is rejected at construction time; options is handed straight to the winston transport constructor; and enabled: false skips a transport entirely, so it can be left in the configuration and switched off per environment rather than deleted.

export LOGGER_TRANSPORTS='[{
  "type": "file",
  "options": { "filename": "/var/log/user-service.log" },
  "enabled": true
}, {
  "type": "http",
  "options": {
    "ssl": true,
    "port": 443,
    "host": "http-intake.logs.datadoghq.com",
    "path": "/v1/input/<API_KEY>"
  },
  "enabled": true
}]'

options is typed as winston's wide LoggerOptions for historical reasons, so the compiler will not catch a mismatch for you: treat it as FileTransportOptions or HttpTransportOptions according to type — filename for a file, host/port/path/ssl for HTTP. LOGGER_METADATA is separate: a JSON object of default fields attached to every record, with the same %name/%version expansion.

Reference: TransportOptions.type · TransportOptions.options · TransportOptions.enabled · AsyncLoggerOptions.transports · Logger

Scale on queue depth, and let the service report it. Every IMQService can serve GET /metrics with the number of messages waiting in its queue — one option away, no exporter to install — which is the signal an autoscaler actually wants: work waiting for this service, rather than the CPU it happens to be burning.

import { IMQService, expose } from '@imqueue/rpc';

export class UserService extends IMQService {
    // ...exposed methods
}

const service = new UserService({
    metricsServer: {
        enabled: true,   // off by default
        port: 9090,      // the default
    },
});

await service.start();   // the listener comes up with it

The listener answers exactly one route, in Prometheus exposition format, and 404 for anything else:

$ curl -s localhost:9090/metrics
queue_length{} 17

Point Prometheus at it, expose queue_length as an external metric through prometheus-adapter, and a HorizontalPodAutoscaler reads it like any other:

metrics:
  - type: External
    external:
      metric:
        name: queue_length
      target:
        type: Value
        value: "20"      # scale out while more than 20 wait

Four properties of the number decide whether that loop behaves. It counts messages waiting in the queue's main list: delayed messages not yet due, and messages already leased to a worker under safeDelivery, are not included, so it is a backlog gauge and not the amount of outstanding work. It is 0 while the queue's writer is disconnected, which makes a broker outage indistinguishable from an empty queue — keep minReplicas above zero so a Redis blip cannot scale the service to nothing. Every replica of the service reads the same queue and so reports the same figure, which is why an External target on the value is a better fit than a per-pod average. And every process that starts the service binds the port, so under multiProcess the primary plus N workers all try: either leave it off there or expect N of the N+1 binds to fail. On a ClusteredRedisQueue the figure is summed across every broker in the fleet.

One shutdown detail: the signal handlers close the listener on SIGINT/SIGTERM, but destroy() does not — close service.metricsServer yourself there, or the open listener keeps the process alive.

Reference: IMQServiceOptions.metricsServer · IMQMetricsServerOptions · IMQMetricsServerOptions.enabled · IMQMetricsServerOptions.port · IMQMetricsServerOptions.queueLengthFormatter · DEFAULT_IMQ_METRICS_SERVER_OPTIONS · IMQService.metricsServer · IMessageQueue.queueLength() · ClusteredRedisQueue.queueLength() · IMQServiceOptions.multiProcess

Run each Redis with one of the two announcer modules and give every service and client a UDPClusterManager. The module makes a broker announce itself over UDP; the manager folds announced brokers into a ClusteredRedisQueue round-robin as they appear and drops them when they stop announcing. Starting or stopping a redis-server then is the scaling operation — nothing is restarted or reconfigured.

Which module depends on one question: does your network deliver broadcast?

# L2 segment — bare metal, VMs, a docker bridge, your laptop
REDIS_BROADCAST_NAME=imq-broker \
REDIS_BROADCAST_INTERVAL=1 \
redis-server --port 6379 --loadmodule $PWD/promoter.so

# Kubernetes on GCP or any cloud VPC, where broadcast is dropped:
# the same datagram, unicast to every pod the K8s API lists
DEPLOYMENT_ENV=production \
SELECTED_INTERFACES=10. \
redis-server --port 6379 --loadmodule $PWD/unicaster.so

The client side does not care which one is running, because both emit the same datagram to the same port:

import { IMQService, UDPClusterManager } from '@imqueue/rpc';

const options = +(process.env.DISABLE_CLUSTER_MANAGER || 0)
    ? { cluster: [{ host: 'localhost', port: 6379 }] }  // static
    : { clusterManagers: [new UDPClusterManager()] };  // discovery

const service = new UserService(options);

Apply those same options to every service and every client in the fleet. Replies travel back through whichever broker the request round-robined onto, so a client pinned to one Redis silently misses responses that landed elsewhere.

The announce protocol is worth knowing because it sets the timings you observe. A broker announces up every REDIS_BROADCAST_INTERVAL seconds (default 1) to port 63000, and down on graceful shutdown, which removes it at once; a broker that dies silently is dropped when its advertised liveness timeout plus aliveTimeoutCorrection (5 s by default) passes without an announcement. So a join takes about one interval, a graceful exit is immediate, and a crash costs a few seconds during which sends may still be routed at it. Announcements are not filtered by queue name — every cluster registered with a manager on that address and port receives every server announced there, so unrelated fleets need distinct addresses or ports. The datagrams are plain, unauthenticated UDP: keep port 63000 inside the cluster, and give every broker the same credentials, since any service may connect to any broker it discovers.

The full recipes — building both modules, the RBAC a unicaster pod needs, and how the fleet behaves as brokers come and go — are in Auto-scaling Redis broker: with and without broadcast.

Reference: UDPClusterManager · UDPClusterManagerOptions · UDPClusterManagerOptions.port · UDPClusterManagerOptions.useAliveCheck · UDPClusterManagerOptions.aliveTimeoutCorrection · IMQOptions.clusterManagers · IMQOptions.cluster · ClusteredRedisQueue

Mount HttpProtect. It counts requests per client IP in Redis and, past two configurable thresholds, first answers 429 and then adds the address to a persistent block list that is answered with 418.

import HttpProtect from '@imqueue/http-protect';

const protect = new HttpProtect({ ttl: 60, maxRequests: 600, banLimit: 5000 });

app.use(protect.jsonMiddleware());

Three things here are load-bearing and none is visible in a signature. A ban is permanent — addresses go into a Redis set that is never given an expiry, and no method in the package removes one, so an address stays banned until something outside it deletes the key. The counter measures a continuous stream, not a fixed window — its TTL is pushed back to ttl on every request, so the count only resets after a full ttl of silence; the default maxRequests of 200 therefore stops a steady one-request-per-second client after about 200 seconds, not just a 200-request burst. And everything is keyed by the client IP, which by default comes from proxy headers a client can set, so set getClientIp before exposing this behind a proxy you do not control.

Reference: HttpProtect · HttpProtect.maxRequests · HttpProtect.banLimit · HttpProtect.ttl · HttpProtectOptions.getClientIp

One app.use(), before the routes it protects, with the middleware that matches how your gateway answers errors: jsonMiddleware() for a JSON API, textMiddleware() for plain text, middleware() for the bare form.

import HttpProtect from '@imqueue/http-protect';

// 429 then 418, as JSON, on default thresholds and a local Redis
app.use(new HttpProtect().jsonMiddleware());

If you would rather decide what happens yourself — a custom body, a redirect, a metric — skip the middlewares and call verify() on the request, then act on the VerificationStatus and httpCode it hands back:

import HttpProtect, { VerificationStatus } from '@imqueue/http-protect';

const protect = new HttpProtect();
const { status, httpCode } = await protect.verify(req);

if (status !== VerificationStatus.SAFE) {
    res.status(httpCode).end();
}

Construct one instance and share it: each one holds its own Redis connection.

Reference: HttpProtect.jsonMiddleware() · HttpProtect.textMiddleware() · HttpProtect.middleware() · HttpProtect.verify() · VerificationStatus

Build a Networks from your CIDR records and ask it. It covers IPv4 and IPv6 in one object, dispatching on the address it is given, and answers from sorted binary ranges rather than by comparing against each network in turn.

import { Networks } from '@imqueue/net';

const allowed = new Networks(['10.0.0.0/8', '192.168.0.0/16', '2001:db8::/32']);

allowed.includes('10.1.2.3');     // true
allowed.includes('8.8.8.8');      // false
allowed.includes('2001:db8::1');  // true

Every record needs an explicit prefix length: a bare address is rejected, so a single host is 203.0.113.7/32 or 2001:db8::1/128. Anything invalid throws while parsing rather than being skipped, so one bad entry fails the whole list — validate with isValid() first when the input is untrusted. cidrToRange() and ipToInt() are exported for building something else on the same primitives; this is the package @imqueue/http-protect uses for its own allow-list.

import { isValid, Networks } from '@imqueue/net';

const records = input.filter(r => isValid(r.split('/')[0]));
const networks = new Networks(records);

Reference: Networks · Networks.includes() · isValid() · cidrToRange() · ipToInt()