diff --git a/src/api/index.ts b/src/api/index.ts index 7573a7f..5648ec7 100644 --- a/src/api/index.ts +++ b/src/api/index.ts @@ -4,6 +4,7 @@ import { v4 as uuidv4 } from 'uuid'; import express, { Express } from 'express'; import bodyParser from 'body-parser'; import promBundle from 'express-prom-bundle'; +import promClient from 'prom-client'; import logger from '../logger'; import Arnavon from '../'; @@ -11,7 +12,8 @@ import Arnavon from '../'; * Creates an express app, reusing a previous prometheus registry if provided * if not, a new one is created */ -export default ({ agent = 'arnavon' } = {}): Express => { +export default ({ agent = 'arnavon', registry }: { agent?: string, registry?: promClient.Registry } = {}): Express => { + const reg = registry || Arnavon.registry; const app = express(); app.use((req, res, next) => { @@ -26,7 +28,7 @@ export default ({ agent = 'arnavon' } = {}): Express => { const metricsMiddleware = promBundle({ includeMethod: true, includePath: true, - promRegistry: Arnavon.registry, + promRegistry: reg, }); app.use(metricsMiddleware); diff --git a/src/app.ts b/src/app.ts new file mode 100644 index 0000000..aab585c --- /dev/null +++ b/src/app.ts @@ -0,0 +1,207 @@ +import promClient from 'prom-client'; +import Queue from './queue'; +import { QueueConfig } from './queue'; +import ArnavonConfig, { + QueueDefinition, + JobDefinition, + ConsumerDefinition, +} from './config'; +import { JobDispatcher } from './jobs'; +import Consumer from './consumer'; +import Server from './server'; +import { ValidatorFn } from './jobs/validator'; +import { HandlerFn } from './jobs/runners/function'; + +export interface JobOptions { + validate?: ValidatorFn + inputSchema?: string + invalidJobExchange?: string +} + +export interface ConsumerOptions { + queue: string + handler?: HandlerFn + runner?: { + type: string + mode?: string + config?: Record + } +} + +/** + * ArnavonApp - instance-based API for using Arnavon as a library. + * + * Usage: + * const app = new ArnavonApp({ queue: { driver: 'amqp', config: { ... } } }) + * app.job('send-email', { validate: (d) => schema.parse(d) }) + * app.consumer('mailer', { queue: 'emails', handler: async (job) => { ... } }) + * await app.startApi({ port: 3000 }) + */ +export default class ArnavonApp { + + #queueDefinition: QueueDefinition; + #jobDefinitions: JobDefinition[] = []; + #consumerDefinitions: ConsumerDefinition[] = []; + #queue: Queue; + #registry: promClient.Registry; + #cwd: string; + #server?: Server; + #consumer?: Consumer; + + constructor(options: { queue: QueueDefinition, cwd?: string }) { + this.#queueDefinition = options.queue; + this.#cwd = options.cwd || process.cwd(); + this.#registry = new promClient.Registry(); + promClient.collectDefaultMetrics({ register: this.#registry }); + this.#queue = Queue.create(this.#queueDefinition as QueueConfig); + } + + /** + * Define a job type. + */ + job(name: string, options: JobOptions = {}) { + const inputSchema = options.validate || options.inputSchema || '.'; + this.#jobDefinitions.push({ + name, + inputSchema, + invalidJobExchange: options.invalidJobExchange, + }); + return this; + } + + /** + * Define a consumer. + */ + consumer(name: string, options: ConsumerOptions) { + const def: ConsumerDefinition = { + name, + queue: options.queue, + }; + if (options.handler) { + def.handler = options.handler; + } else if (options.runner) { + def.runner = options.runner; + } + this.#consumerDefinitions.push(def); + return this; + } + + /** + * Build the ArnavonConfig from accumulated definitions. + */ + #buildConfig(): ArnavonConfig { + return ArnavonConfig.from({ + queue: this.#queueDefinition, + jobs: this.#jobDefinitions, + consumers: this.#consumerDefinitions, + }, this.#cwd); + } + + /** + * Start the REST API server for job submission. + */ + async startApi({ port = 3000 } = {}) { + const config = this.#buildConfig(); + this.#server = new Server(config, this.#registry, this.#queue); + await this.#server.start(port); + return this; + } + + /** + * Start a specific consumer by name. + */ + async startConsumer(name: string, { port = 3000 } = {}) { + const config = this.#buildConfig(); + const consumerConfig = config.consumers.find(c => c.name === name); + if (!consumerConfig) { + throw new Error(`No consumer with name '${name}' found`); + } + const dispatcher = new JobDispatcher(config, this.#registry, this.#queue); + this.#consumer = new Consumer([consumerConfig], dispatcher, this.#registry, this.#queue, this.#cwd); + await this.#consumer.start(port); + return this; + } + + /** + * Start all consumers (or all except specified ones). + */ + async startAllConsumers({ port = 3000, except = [] as string[] } = {}) { + const config = this.#buildConfig(); + let configs = config.consumers; + if (except.length > 0) { + configs = configs.filter(c => !except.includes(c.name)); + } + if (configs.length === 0) { + throw new Error('No consumers to start'); + } + const dispatcher = new JobDispatcher(config, this.#registry, this.#queue); + this.#consumer = new Consumer(configs, dispatcher, this.#registry, this.#queue, this.#cwd); + await this.#consumer.start(port); + return this; + } + + /** + * Stop the running server or consumer. + */ + async stop() { + if (this.#server) { + await this.#server.stop(); + } + if (this.#consumer) { + await this.#consumer.stop(); + } + } + + /** + * Access the prometheus registry (for custom metrics or embedding). + */ + get registry() { + return this.#registry; + } + + /** + * Access the queue instance. + */ + get queue() { + return this.#queue; + } + + /** + * Access the queue definition (driver + config). + */ + get queueDefinition() { + return this.#queueDefinition; + } + + /** + * Create an ArnavonApp from a YAML config file. + * Convenience for migrating from CLI to library usage. + */ + static fromYaml(fname = 'config.yaml'): ArnavonApp { + const config = ArnavonConfig.fromFile(fname); + const app = new ArnavonApp({ + queue: config.queue as unknown as QueueDefinition, + cwd: config.cwd, + }); + + // Register all jobs from config + for (const job of config.jobs) { + app.#jobDefinitions.push({ + name: job.name, + inputSchema: job.inputSchema as string | ValidatorFn, + invalidJobExchange: job.invalidJobExchange, + }); + } + + // Register all consumers from config + for (const consumer of config.consumers) { + app.#consumerDefinitions.push({ + name: consumer.name, + queue: consumer.queue, + runner: consumer.runner as ConsumerDefinition['runner'], + }); + } + + return app; + } +} diff --git a/src/cli/commands/queues/base.ts b/src/cli/commands/queues/base.ts index 463aeb1..e92a79c 100644 --- a/src/cli/commands/queues/base.ts +++ b/src/cli/commands/queues/base.ts @@ -1,5 +1,5 @@ import { Command, Flags } from '@oclif/core'; -import { Config, default as Arnavon } from '../../..'; +import ArnavonApp from '../../../app'; import logger from '../../../logger'; import bunyan from 'bunyan'; @@ -17,25 +17,26 @@ export default abstract class BaseQueueCommand extends Command { }), } + protected app: ArnavonApp; + /** - * Initialize Arnavon from config file. + * Initialize ArnavonApp from config file. * Silences the logger to avoid noise in CLI output. */ - protected initArnavon(configPath: string) { + protected initApp(configPath: string) { logger.level(bunyan.FATAL + 1); - const config = Config.fromFile(configPath); - Arnavon.init(config); + this.app = ArnavonApp.fromYaml(configPath); } /** * Execute a function with queue connection, ensuring proper disconnect. */ protected async withQueue(fn: () => Promise): Promise { - await Arnavon.queue.connect(); + await this.app.queue.connect(); try { return await fn(); } finally { - await Arnavon.queue.disconnect(); + await this.app.queue.disconnect(); } } } diff --git a/src/cli/commands/queues/requeue.ts b/src/cli/commands/queues/requeue.ts index 9d8fe99..1284c06 100644 --- a/src/cli/commands/queues/requeue.ts +++ b/src/cli/commands/queues/requeue.ts @@ -1,5 +1,4 @@ import { Flags } from '@oclif/core'; -import { default as Arnavon } from '../../..'; import BaseQueueCommand from './base'; export default class RequeueCommand extends BaseQueueCommand { @@ -40,12 +39,12 @@ Examples: this.error('Count must be a positive integer'); } - this.initArnavon(flags.config); + this.initApp(flags.config); this.log('Connecting to queue...'); await this.withQueue(async () => { this.log(`Requeuing messages from ${queueName}${count ? ` (limit: ${count})` : ''}...`); - const result = await Arnavon.queue.requeue(queueName, { count }); + const result = await this.app.queue.requeue(queueName, { count }); this.log(''); this.log(`Status: ${result.status}`); diff --git a/src/cli/commands/queues/status.ts b/src/cli/commands/queues/status.ts index 0767c81..77914a7 100644 --- a/src/cli/commands/queues/status.ts +++ b/src/cli/commands/queues/status.ts @@ -1,5 +1,4 @@ import { CliUx } from '@oclif/core'; -import { default as Arnavon } from '../../..'; import BaseQueueCommand from './base'; export default class StatusCommand extends BaseQueueCommand { @@ -21,11 +20,11 @@ Examples: async run() { const { flags } = await this.parse(StatusCommand); - this.initArnavon(flags.config); + this.initApp(flags.config); await this.withQueue(async () => { // Get queue names from config topology - const queueConfig = Arnavon.config.queue.config as { topology?: { queues?: Array<{ name: string }> } }; + const queueConfig = this.app.queueDefinition.config as { topology?: { queues?: Array<{ name: string }> } }; const queueNames = queueConfig.topology?.queues?.map(q => q.name) || []; if (queueNames.length === 0) { @@ -33,7 +32,7 @@ Examples: return; } - const queues = await Arnavon.queue.getQueuesInfo(queueNames); + const queues = await this.app.queue.getQueuesInfo(queueNames); CliUx.ux.table(queues, { name: { diff --git a/src/cli/commands/start/api.ts b/src/cli/commands/start/api.ts index 7da28d0..9e856cd 100644 --- a/src/cli/commands/start/api.ts +++ b/src/cli/commands/start/api.ts @@ -1,5 +1,5 @@ import { Flags, Command } from '@oclif/core'; -import { Server, Config, default as Arnavon } from '../../..'; +import ArnavonApp from '../../../app'; export default class StartApiCommand extends Command { @@ -17,20 +17,18 @@ export default class StartApiCommand extends Command { async run() { const { flags } = await this.parse(StartApiCommand); const configPath = flags.config || 'config.yaml'; + const port = flags.port || 3000; - const config = Config.fromFile(configPath); - Arnavon.init(config); + const app = ArnavonApp.fromYaml(configPath); + await app.startApi({ port }); - const port = flags.port || 3000; - const server = new Server(Arnavon.config); - await server.start(port); // Quit properly on SIGINT (typically ctrl-c) process.on('SIGINT', async () => { - await server.stop(); + await app.stop(); }); // Quit properly on SIGTERM (typically kubernetes termination) process.on('SIGTERM', async () => { - await server.stop(); + await app.stop(); }); } } diff --git a/src/cli/commands/start/consumer.ts b/src/cli/commands/start/consumer.ts index 0a2101e..b55e4ec 100644 --- a/src/cli/commands/start/consumer.ts +++ b/src/cli/commands/start/consumer.ts @@ -1,6 +1,6 @@ import { Command, Flags } from '@oclif/core'; -import { Config, Consumer, default as Arnavon } from '../../../'; -import { JobDispatcher } from '../../../jobs'; +import ArnavonApp from '../../../app'; + export default class StartConsumerCommand extends Command { static summary = `Starts an Arnavon consumer @@ -41,59 +41,27 @@ Please note that the --all flag can be used to start all consumers at once, but async run() { const { args, flags } = await this.parse(StartConsumerCommand); const configPath = flags.config || 'config.yaml'; - - const config = Config.fromFile(configPath); - Arnavon.init(config); - const port = flags.port || 3000; - const dispatcher = new JobDispatcher(Arnavon.config); - const ensureConsumerExists = (name: string) => { - const consumerConfig = Arnavon.config.consumers.find(c => c.name === name); - if (!consumerConfig) { - throw new Error(`No consumer with name '${name}' found`); - } - return consumerConfig; - }; - - // Ensure -x/--except lists existing consumers - if (flags.except) { - flags.except.forEach((name: string) => { - ensureConsumerExists(name); - }); - } + const app = ArnavonApp.fromYaml(configPath); - let configs = []; if (flags.all) { - configs = [...Arnavon.config.consumers]; - if (flags.except) { - configs = configs.filter((c) => (flags.except || []).indexOf(c.name) < 0); - } + const except = flags.except || []; + await app.startAllConsumers({ port, except }); } else { if (!args.name) { throw new Error('The name of a consumer must be provided'); } - const consumerConfig = ensureConsumerExists(args.name); - configs.push(consumerConfig); - } - - if (!configs.length) { - throw new Error('Empty list of consumers'); + await app.startConsumer(args.name, { port }); } - // eslint-disable-next-line no-console - console.log('Starting consumers:', configs.map(c => c.name)); - const consumer = new Consumer(configs, dispatcher); - await consumer.start(port); - // Quit properly on SIGINT (typically ctrl-c) process.on('SIGINT', async () => { - await consumer.stop(); + await app.stop(); }); // Quit properly on SIGTERM (typically kubernetes termination) process.on('SIGTERM', async () => { - await consumer.stop(); + await app.stop(); }); - } } diff --git a/src/config/index.ts b/src/config/index.ts index 76575c7..c827a74 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -6,6 +6,42 @@ import Finitio from 'finitio'; import JobConfig from '../jobs/config'; import { QueueConfig } from '../queue'; import ConsumerConfig from '../consumer/config'; +import { ValidatorFn } from '../jobs/validator'; +import { HandlerFn } from '../jobs/runners/function'; +import { Mode } from '../jobs/runner'; + +/** + * TypeScript interfaces for programmatic config construction. + */ +export interface JobDefinition { + name: string + inputSchema: string | ValidatorFn + invalidJobExchange?: string +} + +export interface RunnerDefinition { + type: string + mode?: Mode | string + config?: Record +} + +export interface ConsumerDefinition { + name: string + queue: string + runner?: RunnerDefinition + handler?: HandlerFn +} + +export interface QueueDefinition { + driver: string + config: Record +} + +export interface ArnavonOptions { + queue: QueueDefinition + jobs: JobDefinition[] + consumers: ConsumerDefinition[] +} export default class ArnavonConfig { @@ -21,6 +57,9 @@ export default class ArnavonConfig { this.cwd = cwd; } + /** + * Create an ArnavonConfig from a YAML file (v1 path). + */ static fromFile(fname = 'config.yaml') { const fpath = path.join(process.cwd(), fname); try { @@ -54,4 +93,43 @@ export default class ArnavonConfig { return new ArnavonConfig(types.ArnavonConfig.dressFromFile(fpath, baseSystem), configFolder); } + + /** + * Create an ArnavonConfig from plain TypeScript objects (v2 path). + * + * This allows programmatic configuration without YAML or Finitio. + * Job input validation can use validator functions instead of Finitio schemas. + */ + static from(options: ArnavonOptions, cwd?: string): ArnavonConfig { + const jobs = options.jobs.map((jobDef) => { + return new JobConfig({ + name: jobDef.name, + inputSchema: jobDef.inputSchema, + invalidJobExchange: jobDef.invalidJobExchange, + } as unknown as JobConfig); + }); + + const consumers = options.consumers.map((consumerDef) => { + if (consumerDef.handler) { + return new ConsumerConfig({ + name: consumerDef.name, + queue: consumerDef.queue, + handler: consumerDef.handler, + }); + } + return new ConsumerConfig({ + name: consumerDef.name, + queue: consumerDef.queue, + runner: consumerDef.runner, + } as unknown as ConsumerConfig); + }); + + const data = { + queue: options.queue as QueueConfig, + jobs, + consumers, + }; + + return new ArnavonConfig(data as unknown as ArnavonConfig, cwd || process.cwd()); + } } diff --git a/src/consumer/config.ts b/src/consumer/config.ts index 94da2a2..cdfb71d 100644 --- a/src/consumer/config.ts +++ b/src/consumer/config.ts @@ -1,4 +1,5 @@ import { JobRunnerConfig } from '../jobs/runner'; +import { HandlerFn } from '../jobs/runners/function'; import { inspect } from '../robust'; export default class ConsumerConfig { @@ -6,10 +7,14 @@ export default class ConsumerConfig { public readonly name: string; public readonly queue: string; public readonly runner: JobRunnerConfig + /** - * Constructs a new ConsumerConfig object + * Constructs a new ConsumerConfig object. + * + * Accepts either a runner config ({ type, config }) or a handler function. + * When a handler function is provided, it is wrapped in a FunctionRunner config. */ - constructor(cfg: ConsumerConfig) { + constructor(cfg: ConsumerConfig | { name: string, queue: string, handler: HandlerFn }) { if (!cfg) { throw new Error(`Config object expected, got ${inspect(cfg)}`); } @@ -19,8 +24,19 @@ export default class ConsumerConfig { } this.name = cfg.name; - this.runner = cfg.runner; this.queue = cfg.queue; + + if ('handler' in cfg && typeof cfg.handler === 'function') { + // v2 style: wrap handler function as a FunctionRunner config + this.runner = { + type: 'function', + config: { handler: cfg.handler }, + } as JobRunnerConfig; + } else if ('runner' in cfg) { + this.runner = (cfg as ConsumerConfig).runner; + } else { + throw new Error(`Either runner or handler expected in consumer config '${cfg.name}'`); + } } /** diff --git a/src/consumer/index.ts b/src/consumer/index.ts index 95aa79d..de738fd 100644 --- a/src/consumer/index.ts +++ b/src/consumer/index.ts @@ -6,7 +6,8 @@ import Arnavon from '..'; import { Server } from 'http'; import { JobRunner, JobDispatcher } from '../jobs'; import { JobRunnerConfig, JobRunnerContext } from '../jobs/runner'; -import { Registry } from 'prom-client'; +import promClient, { Registry } from 'prom-client'; +import Queue from '../queue'; export type JobConsumerContext = JobRunnerContext & { dispatcher: JobDispatcher @@ -19,9 +20,12 @@ export default class Consumer { #server?: Server; #configs; #dispatcher; + #registry; + #queue: Queue; + #cwd: string; #processes; - constructor(configs: Array, dispatcher: JobDispatcher) { + constructor(configs: Array, dispatcher: JobDispatcher, registry?: promClient.Registry, queue?: Queue, cwd?: string) { configs = ([] as Array).concat(configs).flat(); configs.forEach((cfg) => { if (!(cfg instanceof ConsumerConfig)) { @@ -31,7 +35,10 @@ export default class Consumer { if (!(dispatcher instanceof JobDispatcher)) { throw new Error(`JobDispatcher expected, got ${inspect(dispatcher)}`); } - this.#api = createApi(); + this.#registry = registry || Arnavon.registry; + this.#queue = queue || Arnavon.queue; + this.#cwd = cwd || (Arnavon.config ? Arnavon.config.cwd : process.cwd()); + this.#api = createApi({ registry: this.#registry }); this.#dispatcher = dispatcher; this.#configs = configs; } @@ -52,19 +59,19 @@ export default class Consumer { } _connectQueue() { - Arnavon.queue.on('error', () => { + this.#queue.on('error', () => { logger.error('Queue errored, quitting'); process.exit(10); }); - Arnavon.queue.on('close', () => { + this.#queue.on('close', () => { logger.error('Queue disconnected, quitting'); process.exit(10); }); - return Arnavon.queue.connect(); + return this.#queue.connect(); } _disconnectQueue() { - return Arnavon.queue.disconnect(); + return this.#queue.disconnect(); } _startConsuming() { @@ -72,9 +79,10 @@ export default class Consumer { this.#processes = this.#configs.map((config: ConsumerConfig) => { const runner = JobRunner.factor(config.runner.type, { mode: config.runner.mode, + cwd: this.#cwd, ...config.runner.config as JobRunnerConfig, - }); - return Arnavon.queue.consume(config.queue, (_job, context) => { + }, this.#registry); + return this.#queue.consume(config.queue, (_job, context) => { // Dress the payload const validator = this.#dispatcher.getValidator(_job.meta); // @ts-expect-error we need to refactor this @@ -82,7 +90,7 @@ export default class Consumer { // Extend context to include dispatcher and prometheus registry const extendedContext: JobConsumerContext = Object.assign({}, context, { dispatcher: this.#dispatcher, - prometheusRegistry: Arnavon.registry, + prometheusRegistry: this.#registry, }); return runner.run(_job, extendedContext); }); diff --git a/src/index.ts b/src/index.ts index aa1a45a..8d99918 100644 --- a/src/index.ts +++ b/src/index.ts @@ -9,8 +9,14 @@ import { inspect } from './robust'; import ArnavonConfig from './config'; /** - * Arnavon uses a singleton pattern for the main "module" - * exposing the prometheus instance, the queue, the config etc + * Arnavon singleton - convenience facade for CLI usage. + * + * Internal components (Server, Consumer, JobDispatcher, JobRunner, createApi) + * no longer depend on this singleton. They accept registry, queue, and cwd + * as explicit parameters, falling back to Arnavon.* only when not provided. + * + * For library usage, pass dependencies explicitly instead of relying on + * Arnavon.init(). */ class Arnavon { static registry: promClient.Registry; @@ -26,15 +32,16 @@ class Arnavon { Arnavon.config = config; } + /** @deprecated Use config.cwd directly or pass cwd to components */ static cwd() { return Arnavon.config.cwd; } + /** @deprecated Pass module functions directly or use cwd config option */ static require(fname: string) { return require(path.join(Arnavon.cwd(), fname)); } - // for test purposes, shouldn't really be used static reset() { // Disconnect old queue to prevent memory leaks if (Arnavon.queue) { @@ -49,13 +56,17 @@ class Arnavon { export default Arnavon; +import ArnavonApp from './app'; + export { + ArnavonApp, Config, Queue, Server, Consumer, }; +export * from './app'; export * from './consumer'; export * from './server'; export * from './queue'; diff --git a/src/jobs/config.ts b/src/jobs/config.ts index 6bd85dd..f6a0702 100644 --- a/src/jobs/config.ts +++ b/src/jobs/config.ts @@ -1,9 +1,10 @@ import Finitio from 'finitio'; import { inspect } from '../robust'; +import { ValidatorFn } from './validator'; export default class JobConfig { - public readonly inputSchema: Finitio.System; + public readonly inputSchema: Finitio.System | ValidatorFn; public readonly name: string public readonly invalidJobExchange?: string @@ -12,7 +13,7 @@ export default class JobConfig { * @param {Object} cfg: a configuration object with valid `id` and `schema` * @param {Finitio.System} system: an existing finitio system the inputSchema should inherit from */ - constructor(cfg: JobConfig, system: Finitio.System) { + constructor(cfg: JobConfig, system?: Finitio.System) { if (!cfg) { throw new Error(`Config object expected, got ${inspect(cfg)}`); } @@ -40,7 +41,12 @@ export default class JobConfig { } // Private utils -function ensureSchema(schema: string, parentSystem: Finitio.System): Finitio.System { +function ensureSchema(schema: string | Finitio.System | ValidatorFn, parentSystem?: Finitio.System): Finitio.System | ValidatorFn { + // If it's already a validator function, return as-is + if (typeof schema === 'function') { + return schema; + } + if (!parentSystem) { parentSystem = Finitio.system('@import finitio/data'); } @@ -61,4 +67,3 @@ function ensureSchema(schema: string, parentSystem: Finitio.System): Finitio.Sys return system; } - diff --git a/src/jobs/dispatcher.ts b/src/jobs/dispatcher.ts index 7468663..7b401e8 100644 --- a/src/jobs/dispatcher.ts +++ b/src/jobs/dispatcher.ts @@ -1,6 +1,7 @@ import Arnavon from '../'; import promClient from 'prom-client'; import ArnavonConfig from '../config'; +import Queue from '../queue'; import JobValidator from './validator'; import { DataValidationError, inspect, UnknownJobError, InvalidBatch } from '../robust'; import Job, { JobMeta } from './job'; @@ -22,34 +23,40 @@ export default class JobDispatcher { * Private collection of counters (unknown/invalid/valid jobs) */ #counters; + #queue: Queue; /** * Constructs a new JobDispatcher * @param {ArnavonConfig} config a valid config object + * @param {promClient.Registry} registry optional prometheus registry (defaults to Arnavon.registry) + * @param {Queue} queue optional queue instance (defaults to Arnavon.queue) */ - constructor(config: ArnavonConfig) { + constructor(config: ArnavonConfig, registry?: promClient.Registry, queue?: Queue) { if (!(config instanceof ArnavonConfig)) { throw new Error(`ArnavonConfig expected, got ${inspect(config)}`); } + this.#queue = queue || Arnavon.queue; + const reg = registry || Arnavon.registry; + this.#counters = { valid: new promClient.Counter({ name: 'dispatcher_valid_jobs', help: 'number of valid jobs passing through the dispacther', labelNames: ['jobName'], - registers: [Arnavon.registry], + registers: [reg], }), invalid: new promClient.Counter({ name: 'dispatcher_invalid_jobs', help: 'number of invalid jobs passing through the dispacther', labelNames: ['jobName'], - registers: [Arnavon.registry], + registers: [reg], }), unknown: new promClient.Counter({ name: 'dispatcher_unknown_jobs', help: 'number of unknown jobs rejected by the dispacther', labelNames: ['jobName'], - registers: [Arnavon.registry], + registers: [reg], }), }; this.jobs = config.jobs.reduce((jobs: JobCollection, jobConfig: JobConfig) => { @@ -100,7 +107,7 @@ export default class JobDispatcher { const pushOptions = { ...options }; delete (pushOptions as { strict?: boolean })['strict']; - const promises = jobs.map(job => Arnavon.queue.push(jobName, job, pushOptions)); + const promises = jobs.map(job => this.#queue.push(jobName, job, pushOptions)); return Promise.all(promises) .then(() => jobs); @@ -127,7 +134,7 @@ export default class JobDispatcher { this.#counters.invalid.inc({ jobName }); if (jobConfig.invalidJobExchange) { // Await the push to ensure invalid job is queued before rejecting - await Arnavon.queue.push(jobName, data, { exchange: jobConfig.invalidJobExchange }); + await this.#queue.push(jobName, data, { exchange: jobConfig.invalidJobExchange }); } throw err; } @@ -138,7 +145,7 @@ export default class JobDispatcher { dispatched: new Date(), })); - await Arnavon.queue.push(jobName, job, extraOptions); + await this.#queue.push(jobName, job, extraOptions); return job; } diff --git a/src/jobs/runner.ts b/src/jobs/runner.ts index 2d22abd..9e1b876 100644 --- a/src/jobs/runner.ts +++ b/src/jobs/runner.ts @@ -9,14 +9,14 @@ import Logger from 'bunyan'; * The prometheus counters are shared amongst JobRunner classes * but use labels to distinguish the implementating-class/job */ -const ensureCounter = >(type: new(params: unknown) => T, name: string, help: string, extraLabels: string[] = []): T => { - let metric = Arnavon.registry.getSingleMetric(name); +const ensureCounter = >(registry: promClient.Registry, type: new(params: unknown) => T, name: string, help: string, extraLabels: string[] = []): T => { + let metric = registry.getSingleMetric(name); if (!metric) { metric = new type({ name, help, labelNames: ['jobName'].concat(extraLabels), - registers: [Arnavon.registry], + registers: [registry], }); } return metric as T; @@ -30,7 +30,8 @@ export enum Mode { export type JobRunnerConfig = { type: string mode?: Mode, - config: unknown + config: unknown, + cwd?: string } export type JobRunnerMetricCollection = { @@ -54,30 +55,35 @@ export default class JobRunner { protected static metrics: JobRunnerMetricCollection; public readonly mode: Mode; - constructor(config: Partial = {}) { + constructor(config: Partial = {}, registry?: promClient.Registry) { this.mode = config.mode || Mode.ARNAVON; - JobRunner.ensureMetrics(); + JobRunner.ensureMetrics(registry || Arnavon.registry); } - static ensureMetrics() { + static ensureMetrics(registry?: promClient.Registry) { + const reg = registry || Arnavon.registry; JobRunner.metrics = JobRunner.metrics || {}; JobRunner.metrics.success = ensureCounter( + reg, promClient.Counter, 'runner_successful_jobs', 'number of successful job runs', ); JobRunner.metrics.failures = ensureCounter( + reg, promClient.Counter, 'runner_failed_jobs', 'number of failed job runs', ); JobRunner.metrics.leadTime = ensureCounter( + reg, promClient.Histogram, 'runner_job_lead_time', 'time spent between queueing and end of job execution', ['success'], ); JobRunner.metrics.touchTime = ensureCounter( + reg, promClient.Histogram, 'runner_job_touch_time', 'time spent on job execution', @@ -87,7 +93,6 @@ export default class JobRunner { /** * Runs a job - * @param {Job} job */ run(message: unknown, context: Partial = {}) { context.logger = context.logger ? context.logger : mainLogger; @@ -152,29 +157,32 @@ export default class JobRunner { return result .then((result) => { context.logger.info({ result }, `${this.constructor.name} run succeeded`); - updateMetrics(); + if (updateMetrics) { + updateMetrics(); + } return result; }) .catch((err) => { context.logger.error(err, `${this.constructor.name} Runner failed`); - updateMetrics(err); + if (updateMetrics) { + updateMetrics(err); + } throw err; }); } /** * Implementation specific, should be implemented by subclasses - * @param {Job} job */ _run(_message: unknown, _context: JobRunnerContext) { throw new Error('#_run should be implemented by subclasses'); } - static factor(type: string, config: JobRunnerConfig) { + static factor(type: string, config: JobRunnerConfig, registry?: promClient.Registry) { // circular dependency... no choice :( // eslint-disable-next-line @typescript-eslint/no-var-requires const runners = require('./runners').default; - return runners.factor(type, config); + return runners.factor(type, config, registry); } } diff --git a/src/jobs/runners/binary.ts b/src/jobs/runners/binary.ts index 64c50ed..5d8fac9 100644 --- a/src/jobs/runners/binary.ts +++ b/src/jobs/runners/binary.ts @@ -4,6 +4,7 @@ import { inspect } from '../../robust'; import { spawn, ChildProcess } from 'child_process'; import { sync as commandExistsSync } from 'command-exists'; import Job from '../job'; +import promClient from 'prom-client'; // Default timeout: 5 minutes const DEFAULT_TIMEOUT_MS = 5 * 60 * 1000; @@ -22,8 +23,8 @@ export default class BinaryRunner extends JobRunner { protected args: string[]; protected timeout: number; - constructor(config: BinaryRunnerConfig) { - super(config); + constructor(config: BinaryRunnerConfig, registry?: promClient.Registry) { + super(config, registry); if (!config.path) { throw new Error(`Binary path expected, got ${inspect(config.path)}`); diff --git a/src/jobs/runners/function.ts b/src/jobs/runners/function.ts new file mode 100644 index 0000000..ed7fcbc --- /dev/null +++ b/src/jobs/runners/function.ts @@ -0,0 +1,23 @@ +import JobRunner, { JobRunnerConfig, JobRunnerContext } from '../runner'; +import Job from '../job'; +import promClient from 'prom-client'; + +export type HandlerFn = (job: Job, context: JobRunnerContext) => Promise + +export interface FunctionRunnerConfig extends JobRunnerConfig { + handler: HandlerFn +} + +export default class FunctionRunner extends JobRunner { + + private handler: HandlerFn; + + constructor(config: FunctionRunnerConfig, registry?: promClient.Registry) { + super(config, registry); + this.handler = config.handler; + } + + _run(job: Job, context: JobRunnerContext) { + return this.handler(job, context); + } +} diff --git a/src/jobs/runners/index.ts b/src/jobs/runners/index.ts index 473a407..07ea214 100644 --- a/src/jobs/runners/index.ts +++ b/src/jobs/runners/index.ts @@ -1,5 +1,6 @@ import NodeJSRunner from './nodejs'; import BinaryRunner from './binary'; +import FunctionRunner from './function'; import JobRunner, { JobRunnerConfig } from '../runner'; export type RunnerRegistry = {[key: string]: typeof JobRunner}; @@ -21,14 +22,15 @@ class RunnersFactory { return this.runners[type]; } - factor(type: string, config: JobRunnerConfig) { + factor(type: string, config: JobRunnerConfig, registry?: import('prom-client').Registry) { const clazz = this.get(type); - return new clazz(config); + return new clazz(config, registry); } } const factory = new RunnersFactory(); factory.register('nodejs', NodeJSRunner); factory.register('binary', BinaryRunner); +factory.register('function', FunctionRunner); export default factory; diff --git a/src/jobs/runners/nodejs.ts b/src/jobs/runners/nodejs.ts index 065f466..e910040 100644 --- a/src/jobs/runners/nodejs.ts +++ b/src/jobs/runners/nodejs.ts @@ -1,12 +1,15 @@ require('@babel/register'); +import path from 'path'; import Arnavon from '../../'; import JobRunner, { JobRunnerConfig, JobRunnerContext } from '../runner'; import { inspect } from '../../robust'; import logger from '../../logger'; import Job from '../job'; +import promClient from 'prom-client'; export interface NodeJSRunnerConfig extends JobRunnerConfig { module: string + cwd?: string } export type NodeJSRunnerModule = (job: Job, context: JobRunnerContext) => Promise @@ -14,15 +17,18 @@ export type NodeJSRunnerModule = (job: Job, context: JobRunnerContext) => Promis export default class NodeJSRunner extends JobRunner { private module: NodeJSRunnerModule; - constructor(private config: NodeJSRunnerConfig) { - super(config); + constructor(private config: NodeJSRunnerConfig, registry?: promClient.Registry) { + super(config, registry); if (!config.module) { throw new Error(`Module path expected, got ${inspect(config.module)}`); } + const cwd = config.cwd || Arnavon.cwd(); + try { - const module = Arnavon.require(config.module); + // eslint-disable-next-line @typescript-eslint/no-var-requires + const module = require(path.join(cwd, config.module)); this.module = module.default ? module.default : module; } catch (err) { logger.error(err); diff --git a/src/jobs/validator.ts b/src/jobs/validator.ts index 1a63d6a..7764728 100644 --- a/src/jobs/validator.ts +++ b/src/jobs/validator.ts @@ -2,39 +2,47 @@ import Finitio from 'finitio'; import { inspect, DataValidationError } from '../robust'; import logger from '../logger'; +/** + * A validator function takes unknown input data and returns the + * validated/dressed data. It should throw on invalid input. + */ +export type ValidatorFn = (data: unknown) => unknown; + export default class JobValidator { - #schema; + #validateFn: ValidatorFn; /** * Creates a new JobValidator * - * @param {Finitio.Schema} schema a finitio schema that will be used to ensure the job input data - * is valid + * Accepts either a Finitio.System or a plain validator function. */ - constructor(schema: Finitio.System) { - if (!(schema instanceof Finitio.System)) { - throw new Error(`Finitio system expected, got ${inspect(schema)}`); + constructor(schema: Finitio.System | ValidatorFn) { + if (typeof schema === 'function') { + this.#validateFn = schema; + } else if (schema instanceof Finitio.System) { + this.#validateFn = (inputData: unknown) => { + try { + return schema.dress(inputData); + } catch (err) { + if (err instanceof Finitio.TypeError) { + logger.error('Invalid job payload', err); + logger.debug({ inputData }, 'Input data does not respect the schema'); + throw DataValidationError.fromFinitioError('Invalid input data:', err); + } + throw err; + } + }; + } else { + throw new Error(`Finitio system or validator function expected, got ${inspect(schema)}`); } - this.#schema = schema; } /** * Validates the input data for a job - * - * @param {Object} inputData */ validate(inputData: unknown) { - try { - return this.#schema.dress(inputData); - } catch (err) { - if (err instanceof Finitio.TypeError) { - logger.error('Invalid job payload', err); - logger.debug({ inputData }, 'Input data does not respect the schema'); - throw DataValidationError.fromFinitioError('Invalid input data:', err); - } - throw err; - } + return this.#validateFn(inputData); } } diff --git a/src/server/index.ts b/src/server/index.ts index 17d589f..4de5d47 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -5,19 +5,24 @@ import { inspect } from '../robust'; import ArnavonConfig from '../config'; import Arnavon from '../'; import { Server as HttpServer } from 'http'; +import promClient from 'prom-client'; +import Queue from '../queue'; export default class Server { #api; #server?: HttpServer; #dispatcher; + #queue: Queue; - constructor(config: ArnavonConfig) { + constructor(config: ArnavonConfig, registry?: promClient.Registry, queue?: Queue) { if (!(config instanceof ArnavonConfig)) { throw new Error(`ArnavonConfig expected, got ${inspect(config)}`); } - this.#dispatcher = new JobDispatcher(config); - this.#api = createApi(this.#dispatcher); + const reg = registry || Arnavon.registry; + this.#queue = queue || Arnavon.queue; + this.#dispatcher = new JobDispatcher(config, reg, this.#queue); + this.#api = createApi(this.#dispatcher, { registry: reg, queue: this.#queue, config }); } _startApi(port: number) { @@ -36,19 +41,19 @@ export default class Server { } _connectQueue() { - Arnavon.queue.on('error', () => { + this.#queue.on('error', () => { logger.error('Queue errored, quitting'); process.exit(10); }); - Arnavon.queue.on('close', () => { + this.#queue.on('close', () => { logger.error('Queue disconnected, quitting'); process.exit(10); }); - return Arnavon.queue.connect(); + return this.#queue.connect(); } _disconnectQueue() { - return Arnavon.queue.disconnect(); + return this.#queue.disconnect(); } start(port = 3000) { diff --git a/src/server/rest/index.ts b/src/server/rest/index.ts index 5cab3d2..c89a730 100644 --- a/src/server/rest/index.ts +++ b/src/server/rest/index.ts @@ -4,6 +4,9 @@ import { JobDispatcher } from '../../jobs'; import { ArnavonError, UnknownJobError, DataValidationError } from '../../robust'; import logger from '../../logger'; import Arnavon from '../../index'; +import promClient from 'prom-client'; +import Queue from '../../queue'; +import ArnavonConfig from '../../config'; // Valid x-arnavon-push-modes const PUSH_MODES = ['SINGLE', 'BATCH']; @@ -11,8 +14,17 @@ const PUSH_MODES = ['SINGLE', 'BATCH']; // Valid x-arnavon-batch-input-validation const VALIDATION_MODES = ['ALL-OR-NOTHING', 'BEST-EFFORT']; -export default (dispatcher: JobDispatcher) => { - const api = createApi(); +type RestApiOptions = { + registry?: promClient.Registry, + queue?: Queue, + config?: ArnavonConfig, +} + +export default (dispatcher?: JobDispatcher, options: RestApiOptions = {}) => { + const { registry, queue, config } = options; + const q = queue || Arnavon.queue; + const cfg = config || Arnavon.config; + const api = createApi({ registry }); api.post('/jobs/:id', (req, res, next) => { // Check X-Arnavon-Push-Mode: @@ -87,10 +99,10 @@ export default (dispatcher: JobDispatcher) => { api.get('/queues', async (req, res, next) => { try { // Get queue names from config topology - const config = Arnavon.config.queue.config as { topology?: { queues?: Array<{ name: string }> } }; - const queueNames = config.topology?.queues?.map(q => q.name) || []; + const queueConfig = cfg.queue.config as { topology?: { queues?: Array<{ name: string }> } }; + const queueNames = queueConfig.topology?.queues?.map(q => q.name) || []; - const queues = await Arnavon.queue.getQueuesInfo(queueNames); + const queues = await q.getQueuesInfo(queueNames); return res.status(200).send({ queues }); } catch (err) { next(err); @@ -109,7 +121,7 @@ export default (dispatcher: JobDispatcher) => { } try { - const result = await Arnavon.queue.requeue(queueName, { count }); + const result = await q.requeue(queueName, { count }); return res.status(200).send(result); } catch (err) { next(err); diff --git a/tests/app.spec.ts b/tests/app.spec.ts new file mode 100644 index 0000000..0b14d33 --- /dev/null +++ b/tests/app.spec.ts @@ -0,0 +1,122 @@ +import { expect } from 'chai'; +import ArnavonApp from '../src/app'; +import MemoryQueue from '../src/queue/drivers/memory'; + +describe('ArnavonApp', () => { + + it('exports a class', () => { + expect(ArnavonApp).to.be.an.instanceof(Function); + expect(ArnavonApp.name).to.equal('ArnavonApp'); + }); + + describe('constructor', () => { + it('creates an instance with a queue definition', () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }); + expect(app).to.be.an.instanceof(ArnavonApp); + }); + + it('creates a queue from the definition', () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }); + expect(app.queue).to.be.an.instanceof(MemoryQueue); + }); + + it('creates its own prometheus registry', () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }); + expect(app.registry).to.exist; + }); + }); + + describe('#job', () => { + it('registers a job definition and returns self for chaining', () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }); + const result = app.job('send-email', { inputSchema: '.' }); + expect(result).to.equal(app); + }); + + it('accepts a validator function', () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }); + app.job('send-email', { validate: (data) => data }); + // No error thrown + }); + }); + + describe('#consumer', () => { + it('registers a consumer with a handler and returns self for chaining', () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }); + const result = app.consumer('mailer', { + queue: 'emails', + handler: async (job) => job, + }); + expect(result).to.equal(app); + }); + + it('registers a consumer with a runner config', () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }); + app.consumer('mailer', { + queue: 'emails', + runner: { type: 'nodejs', config: { module: './test' } }, + }); + // No error thrown + }); + }); + + describe('#startApi', () => { + it('starts a server and can be stopped', async () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }); + app.job('test-job', { inputSchema: '.' }); + await app.startApi({ port: 0 }); // port 0 = random available port + await app.stop(); + }); + }); + + describe('#startConsumer', () => { + it('throws if consumer name not found', async () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }); + app.job('test-job', { inputSchema: '.' }); + + try { + await app.startConsumer('nonexistent'); + throw new Error('should have thrown'); + } catch (err) { + expect(err.message).to.include('No consumer with name'); + } + }); + }); + + describe('#startAllConsumers', () => { + it('throws if no consumers defined', async () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }); + + try { + await app.startAllConsumers(); + throw new Error('should have thrown'); + } catch (err) { + expect(err.message).to.include('No consumers to start'); + } + }); + }); + + describe('.fromYaml', () => { + it('creates an ArnavonApp from a YAML config file', () => { + const app = ArnavonApp.fromYaml('example/config.yaml'); + expect(app).to.be.an.instanceof(ArnavonApp); + }); + + it('throws for non-existent config file', () => { + expect(() => ArnavonApp.fromYaml('nonexistent.yaml')).to.throw(/Config file not found/); + }); + }); + + describe('fluent API', () => { + it('allows chaining job and consumer definitions', () => { + const app = new ArnavonApp({ queue: { driver: 'memory', config: {} } }) + .job('send-email', { validate: (d) => d }) + .job('log-info', { inputSchema: '.' }) + .consumer('mailer', { queue: 'emails', handler: async (job) => job }) + .consumer('logger', { queue: 'logs', handler: async (job) => job }); + + expect(app).to.be.an.instanceof(ArnavonApp); + }); + }); + +}); diff --git a/tests/config/config.spec.ts b/tests/config/config.spec.ts index 3e9966d..84ce899 100644 --- a/tests/config/config.spec.ts +++ b/tests/config/config.spec.ts @@ -36,6 +36,22 @@ describe('ArnavonConfig', () => { const test = () => ArnavonConfig.fromFile(); expect(test).to.throw(/Config file not found/); }); + + it('complains on malformed YAML syntax', () => { + const test = () => ArnavonConfig.fromFile('tests/config/malformed.yaml'); + expect(test).to.throw(); + }); + + it('works without schema.fio and schema.world.js in config directory', () => { + const config = ArnavonConfig.fromFile('tests/config/minimal/config.yaml'); + expect(config).to.be.an.instanceOf(ArnavonConfig); + expect(config.queue.driver).to.equal('amqp'); + expect(config.jobs).to.be.an.instanceof(Array); + expect(config.jobs.length).to.equal(1); + expect(config.jobs[0].name).to.equal('test-job'); + expect(config.consumers).to.be.an.instanceof(Array); + expect(config.consumers.length).to.equal(1); + }); }); describe('constructor', () => { @@ -69,4 +85,74 @@ describe('ArnavonConfig', () => { }); }); + describe('.from', () => { + + it('creates a config from plain objects', () => { + const config = ArnavonConfig.from({ + queue: { driver: 'memory', config: {} }, + jobs: [ + { name: 'test-job', inputSchema: '.' }, + ], + consumers: [ + { name: 'test-consumer', queue: 'test-queue', runner: { type: 'nodejs', config: { module: './test' } } }, + ], + }); + expect(config).to.be.an.instanceOf(ArnavonConfig); + expect(config.jobs).to.have.length(1); + expect(config.jobs[0].name).to.equal('test-job'); + expect(config.consumers).to.have.length(1); + expect(config.consumers[0].name).to.equal('test-consumer'); + expect(config.queue.driver).to.equal('memory'); + }); + + it('accepts validator functions as inputSchema', () => { + const myValidator = (data) => { + if (!data.email) throw new Error('email required'); + return data; + }; + const config = ArnavonConfig.from({ + queue: { driver: 'memory', config: {} }, + jobs: [ + { name: 'send-email', inputSchema: myValidator }, + ], + consumers: [], + }); + expect(config.jobs[0].name).to.equal('send-email'); + expect(config.jobs[0].inputSchema).to.equal(myValidator); + }); + + it('accepts handler functions in consumer definitions', () => { + const handler = async (job) => { return job; }; + const config = ArnavonConfig.from({ + queue: { driver: 'memory', config: {} }, + jobs: [ + { name: 'test-job', inputSchema: '.' }, + ], + consumers: [ + { name: 'my-consumer', queue: 'test-queue', handler }, + ], + }); + expect(config.consumers[0].runner.type).to.equal('function'); + }); + + it('sets cwd to process.cwd() by default', () => { + const config = ArnavonConfig.from({ + queue: { driver: 'memory', config: {} }, + jobs: [], + consumers: [], + }); + expect(config.cwd).to.equal(process.cwd()); + }); + + it('accepts custom cwd', () => { + const config = ArnavonConfig.from({ + queue: { driver: 'memory', config: {} }, + jobs: [], + consumers: [], + }, '/custom/path'); + expect(config.cwd).to.equal('/custom/path'); + }); + + }); + }); diff --git a/tests/config/malformed.yaml b/tests/config/malformed.yaml new file mode 100644 index 0000000..5dda708 --- /dev/null +++ b/tests/config/malformed.yaml @@ -0,0 +1 @@ +{{{invalid yaml \ No newline at end of file diff --git a/tests/config/minimal/config.yaml b/tests/config/minimal/config.yaml new file mode 100644 index 0000000..efecb96 --- /dev/null +++ b/tests/config/minimal/config.yaml @@ -0,0 +1,28 @@ +queue: + driver: amqp + config: + url: amqp://localhost + topology: + exchanges: + - name: test-exchange + type: topic + default: true + options: + durable: true + queues: + - name: test-queue + options: + durable: true + bindings: + - routingKey: test-job + exchange: test-exchange +jobs: + - name: test-job + inputSchema: . +consumers: + - name: test-consumer + queue: test-queue + runner: + type: nodejs + config: + module: ./test-module diff --git a/tests/consumer/config.spec.ts b/tests/consumer/config.spec.ts new file mode 100644 index 0000000..3dacabd --- /dev/null +++ b/tests/consumer/config.spec.ts @@ -0,0 +1,82 @@ +import ConsumerConfig from '../../src/consumer/config'; +import { expect } from 'chai'; + +describe('ConsumerConfig', () => { + + it('exports a class', () => { + expect(ConsumerConfig).to.be.an.instanceof(Function); + expect(ConsumerConfig.name).to.equal('ConsumerConfig'); + }); + + describe('its constructor', () => { + + it('throws when given null', () => { + expect(() => new ConsumerConfig(null as any)).to.throw(/Config object expected, got null/); + }); + + it('throws when given undefined', () => { + expect(() => new ConsumerConfig(undefined as any)).to.throw(/Config object expected, got undefined/); + }); + + it('throws when name is missing', () => { + expect(() => new ConsumerConfig({} as any)).to.throw(/Valid job name expected/); + }); + + it('throws when name is not a string', () => { + expect(() => new ConsumerConfig({ name: 42 } as any)).to.throw(/Valid job name expected/); + }); + + it('throws when name is empty string', () => { + expect(() => new ConsumerConfig({ name: '' } as any)).to.throw(/Valid job name expected/); + }); + + it('works with a valid config', () => { + const cfg = { + name: 'test-consumer', + queue: 'test-queue', + runner: { type: 'nodejs', config: { module: './test' } }, + }; + expect(() => new ConsumerConfig(cfg as any)).to.not.throw(); + }); + + it('sets properties correctly', () => { + const runner = { type: 'nodejs', config: { module: './test' } }; + const cfg = { + name: 'my-consumer', + queue: 'my-queue', + runner, + }; + const config = new ConsumerConfig(cfg as any); + expect(config.name).to.equal('my-consumer'); + expect(config.queue).to.equal('my-queue'); + expect(config.runner).to.equal(runner); + }); + + }); + + describe('.json', () => { + + it('returns a ConsumerConfig instance', () => { + const cfg = { + name: 'test-consumer', + queue: 'test-queue', + runner: { type: 'nodejs', config: { module: './test' } }, + }; + const result = ConsumerConfig.json(cfg as any); + expect(result).to.be.an.instanceof(ConsumerConfig); + }); + + it('sets properties from the data', () => { + const cfg = { + name: 'json-consumer', + queue: 'json-queue', + runner: { type: 'binary', config: { path: '/bin/echo' } }, + }; + const result = ConsumerConfig.json(cfg as any); + expect(result.name).to.equal('json-consumer'); + expect(result.queue).to.equal('json-queue'); + }); + + }); + +}); diff --git a/tests/consumer/consumer.spec.ts b/tests/consumer/consumer.spec.ts index 6829405..148d8ea 100644 --- a/tests/consumer/consumer.spec.ts +++ b/tests/consumer/consumer.spec.ts @@ -1,37 +1,30 @@ import Arnavon from '../../src/'; import Config from '../../src/config'; +import Consumer from '../../src/consumer'; import { expect, default as chai } from 'chai'; import sinon from 'sinon'; import sinonChai from 'sinon-chai'; -import proxyquire from 'proxyquire'; -import { Job, JobDispatcher, JobRunner } from '../../src/jobs'; +import { JobDispatcher, JobRunner } from '../../src/jobs'; chai.should(); chai.use(sinonChai); describe('Consumer', () => { - let consumer, config, Consumer, listen, processExit, dispatcher; + let consumer, config, processExit, dispatcher; beforeEach(() => { const arncfg = Config.fromFile('example/config.yaml'); config = arncfg.consumers[0]; - listen = sinon.stub().yields(); processExit = sinon.stub(process, 'exit'); dispatcher = new JobDispatcher(arncfg); - Consumer = proxyquire('../../src/consumer', { - '../api': { - default: function() { - return { - listen, - }; - }, - }, - }).default; consumer = new Consumer(config, dispatcher); + // Stub _startApi to avoid actually starting an HTTP server + sinon.stub(consumer, '_startApi').resolves(consumer); }); afterEach(() => { processExit.restore(); + sinon.restore(); }); it('is a class', () => { @@ -52,7 +45,7 @@ describe('Consumer', () => { expect(test({})).to.throw(/JobDispatcher expected, got/); }); it('it works', () => { - expect(new Consumer(config, dispatcher)).to.be.an.instanceof(Consumer); + expect(consumer).to.be.an.instanceof(Consumer); }); }); @@ -72,13 +65,14 @@ describe('Consumer', () => { Arnavon.queue.connect = sinon.stub().resolves(true); return consumer.start() .then(() => { - expect(listen).to.be.calledOnce; + expect(consumer._startApi).to.be.calledOnce; }); }); it('quits if unable to start the api', () => { Arnavon.queue.connect = sinon.stub().resolves(true); - listen.yields(new Error('oops')); + consumer._startApi.restore(); + sinon.stub(consumer, '_startApi').rejects(new Error('oops')); return consumer.start() .finally(() => { expect(processExit).to.be.calledOnceWith(10); @@ -177,4 +171,63 @@ describe('Consumer', () => { }); + describe('#stop', () => { + + it('disconnects the queue', async () => { + Arnavon.queue.connect = sinon.stub().resolves(true); + const disconnectSpy = sinon.stub(Arnavon.queue, 'disconnect').resolves(Arnavon.queue); + await consumer.start(); + await consumer.stop(); + expect(disconnectSpy).to.be.calledOnce; + }); + + it('stops the internal API server', async () => { + Arnavon.queue.connect = sinon.stub().resolves(true); + sinon.stub(Arnavon.queue, 'disconnect').resolves(Arnavon.queue); + const stopApiSpy = sinon.spy(consumer, '_stopApi'); + await consumer.start(); + await consumer.stop(); + expect(stopApiSpy).to.be.calledOnce; + }); + }); + + describe('with multiple consumer configs', () => { + let multiConsumer; + beforeEach(() => { + // Reset registry to avoid metric registration conflicts from prior Consumer creation + Arnavon.reset(); + const arncfg = Config.fromFile('example/config.yaml'); + dispatcher = new JobDispatcher(arncfg); + const configs = arncfg.consumers; + multiConsumer = new Consumer(configs, dispatcher); + sinon.stub(multiConsumer, '_startApi').resolves(multiConsumer); + }); + + it('starts consuming all configured queues', () => { + Arnavon.queue.connect = sinon.stub().resolves(true); + sinon.stub(JobRunner, 'factor').returns({ run: sinon.stub().resolves() }); + const consumeSpy = sinon.stub(Arnavon.queue, 'consume').resolves(true); + return multiConsumer.start() + .then(() => { + expect(consumeSpy.callCount).to.equal(2); + const queueNames = consumeSpy.getCalls().map(c => c.args[0]); + expect(queueNames).to.include('send-email'); + expect(queueNames).to.include('send-email-via-binary'); + }); + }); + + it('factors a runner for each consumer config', () => { + Arnavon.queue.connect = sinon.stub().resolves(true); + sinon.stub(Arnavon.queue, 'consume').resolves(true); + const factorSpy = sinon.stub(JobRunner, 'factor').returns({ run: sinon.stub().resolves() }); + return multiConsumer.start() + .then(() => { + expect(factorSpy.callCount).to.equal(2); + const types = factorSpy.getCalls().map(c => c.args[0]); + expect(types).to.include('nodejs'); + expect(types).to.include('binary'); + }); + }); + }); + }); diff --git a/tests/jobs/dispatcher.spec.ts b/tests/jobs/dispatcher.spec.ts index ccfdf10..10c4b09 100644 --- a/tests/jobs/dispatcher.spec.ts +++ b/tests/jobs/dispatcher.spec.ts @@ -181,6 +181,16 @@ describe('JobDispatcher', () => { }); }); + it('propagates queue push failure as a rejection', () => { + sinon.stub(Arnavon.queue, 'push').rejects(new Error('queue down')); + const payload = [validPayload, validPayload]; + return dispatcher.dispatchBatch('send-email', payload) + .then(() => { throw new Error('should have rejected'); }) + .catch((err) => { + expect(err.message).to.equal('queue down'); + }); + }); + }); describe('#dispatch', () => { @@ -272,6 +282,71 @@ describe('JobDispatcher', () => { }); }); + describe('when invalidJobExchange is configured', () => { + // send-email has invalidJobExchange: 'invalid-jobs' in example config + + it('pushes invalid payload to the invalidJobExchange', () => { + const spy = sinon.spy(Arnavon.queue, 'push'); + const invalidPayload = { bad: 'data' }; + return dispatcher.dispatch('send-email', invalidPayload) + .then(() => { throw new Error('should have rejected'); }) + .catch((err) => { + expect(err).to.be.an.instanceof(DataValidationError); + expect(spy).to.be.calledOnce; + const { args } = spy.getCall(0); + expect(args[0]).to.equal('send-email'); + expect(args[1]).to.equal(invalidPayload); + expect(args[2]).to.eql({ exchange: 'invalid-jobs' }); + }); + }); + + it('still rejects with DataValidationError after pushing to invalid exchange', () => { + sinon.stub(Arnavon.queue, 'push').resolves(); + const invalidPayload = { bad: 'data' }; + return dispatcher.dispatch('send-email', invalidPayload) + .then(() => { throw new Error('should have rejected'); }) + .catch((err) => { + expect(err).to.be.an.instanceof(DataValidationError); + }); + }); + }); + + describe('when queue push fails', () => { + + it('propagates the push error as a rejection', () => { + sinon.stub(Arnavon.queue, 'push').rejects(new Error('queue down')); + return dispatcher.dispatch('send-email', validPayload) + .then(() => { throw new Error('should have rejected'); }) + .catch((err) => { + expect(err.message).to.equal('queue down'); + }); + }); + }); + + describe('when invalidJobExchange is NOT configured', () => { + // log-info has inputSchema: '.' (accepts anything) and no invalidJobExchange + // Use a job that has no invalidJobExchange but will fail validation + // Actually log-info accepts everything, so let's verify no push happens + // by checking that for a valid dispatch, push is called exactly once (for the job itself) + + it('does not push to any exchange on invalid payload', () => { + const spy = sinon.spy(Arnavon.queue, 'push'); + // send-email-via-binary also has invalidJobExchange, log-info accepts everything + // We need a job without invalidJobExchange - but all jobs in example config + // that have strict schemas also have invalidJobExchange. + // log-info has inputSchema: '.' which accepts everything. + // So we test the negative: log-info accepts any payload, no invalid push needed. + return dispatcher.dispatch('log-info', { anything: 'goes' }) + .then(() => { + expect(spy).to.be.calledOnce; + // The single push is for the actual job, not an invalid exchange push + const { args } = spy.getCall(0); + expect(args[0]).to.equal('log-info'); + expect(args[1]).to.be.an.instanceof(Job); + }); + }); + }); + }); }); diff --git a/tests/jobs/result.spec.ts b/tests/jobs/result.spec.ts new file mode 100644 index 0000000..63b8dfc --- /dev/null +++ b/tests/jobs/result.spec.ts @@ -0,0 +1,100 @@ +import JobResult from '../../src/jobs/result'; +import { expect } from 'chai'; + +describe('JobResult', () => { + + it('exports a class', () => { + expect(JobResult).to.be.an.instanceof(Function); + expect(JobResult.name).to.equal('JobResult'); + }); + + describe('its constructor', () => { + + it('sets success, error, and result properties', () => { + const err = new Error('test error'); + const result = new JobResult({ success: false, error: err, result: 'some data' }); + expect((result as any).success).to.equal(false); + expect((result as any).error).to.equal(err); + expect((result as any).result).to.equal('some data'); + }); + + it('makes properties immutable', () => { + const result = new JobResult({ success: true, result: 'data' }); + + expect(() => { + (result as any).success = false; + }).to.throw(TypeError); + + expect(() => { + (result as any).result = 'changed'; + }).to.throw(TypeError); + + expect(() => { + (result as any).error = new Error('nope'); + }).to.throw(TypeError); + + expect((result as any).success).to.equal(true); + expect((result as any).result).to.equal('data'); + }); + + it('defaults error to undefined when not provided', () => { + const result = new JobResult({ success: true, result: 42 }); + expect((result as any).error).to.be.undefined; + }); + + }); + + describe('.fail', () => { + + it('returns a JobResult instance', () => { + const err = new Error('failure'); + const result = JobResult.fail(err, 'fail data'); + expect(result).to.be.an.instanceof(JobResult); + }); + + it('sets success to false', () => { + const err = new Error('failure'); + const result = JobResult.fail(err, 'fail data'); + expect((result as any).success).to.equal(false); + }); + + it('stores the error', () => { + const err = new Error('test failure'); + const result = JobResult.fail(err, null); + expect((result as any).error).to.equal(err); + expect((result as any).error.message).to.equal('test failure'); + }); + + it('stores the result', () => { + const err = new Error('failure'); + const result = JobResult.fail(err, { partial: 'data' }); + expect((result as any).result).to.deep.equal({ partial: 'data' }); + }); + + }); + + describe('.success', () => { + + it('returns a JobResult instance', () => { + const result = JobResult.success('ok'); + expect(result).to.be.an.instanceof(JobResult); + }); + + it('sets success to true', () => { + const result = JobResult.success('ok'); + expect((result as any).success).to.equal(true); + }); + + it('stores the result', () => { + const result = JobResult.success({ answer: 42 }); + expect((result as any).result).to.deep.equal({ answer: 42 }); + }); + + it('does not set an error', () => { + const result = JobResult.success('ok'); + expect((result as any).error).to.be.undefined; + }); + + }); + +}); diff --git a/tests/jobs/runner.spec.ts b/tests/jobs/runner.spec.ts index 5ca7905..5398535 100644 --- a/tests/jobs/runner.spec.ts +++ b/tests/jobs/runner.spec.ts @@ -2,10 +2,13 @@ import Arnavon from '../../src'; import sinon from 'sinon'; import Runners from '../../src/jobs/runners'; import JobRunner from '../../src/jobs/runner'; -import { expect } from 'chai'; +import { expect, default as chai } from 'chai'; +import sinonChai from 'sinon-chai'; import Job from '../../src/jobs/job'; import { InvalidRunError } from '../../src/robust'; +chai.use(sinonChai); + describe('JobRunner', () => { let testJob; @@ -22,8 +25,8 @@ describe('JobRunner', () => { }); class TestRunner extends JobRunner { - constructor(config) { - super(); + constructor(config = {}) { + super(config); this.config = config; } @@ -189,6 +192,84 @@ describe('JobRunner', () => { }); + describe('#run in raw mode', () => { + let rawRunner; + beforeEach(() => { + rawRunner = new TestRunner({ mode: 'raw' }); + }); + + it('passes the message as-is to _run', () => { + const rawMessage = { some: 'data' }; + rawRunner._run = sinon.stub().returns(Promise.resolve(42)); + rawRunner.run(rawMessage); + expect(rawRunner._run).to.be.calledOnceWith(rawMessage); + }); + + it('does not deserialize the message as a Job', () => { + const rawMessage = { some: 'data' }; + rawRunner._run = sinon.stub().returns(Promise.resolve(42)); + return rawRunner.run(rawMessage) + .then(() => { + const passedMessage = rawRunner._run.getCall(0).args[0]; + // In raw mode, the message is NOT converted to a Job instance + expect(passedMessage).to.equal(rawMessage); + expect(passedMessage).to.not.be.an.instanceof(Job); + }); + }); + + it('does not track prometheus metrics', () => { + rawRunner._run = sinon.stub().returns(Promise.resolve(42)); + const successMetric = Arnavon.registry.getSingleMetric('runner_successful_jobs'); + const failureMetric = Arnavon.registry.getSingleMetric('runner_failed_jobs'); + const spySuccess = sinon.spy(successMetric, 'inc'); + const spyFailure = sinon.spy(failureMetric, 'inc'); + + return rawRunner.run({ some: 'data' }) + .then(() => { + expect(spySuccess).to.not.have.been.called; + expect(spyFailure).to.not.have.been.called; + }); + }); + + it('does not track prometheus metrics on failure either', () => { + rawRunner._run = sinon.stub().returns(Promise.reject(new Error('fail'))); + const successMetric = Arnavon.registry.getSingleMetric('runner_successful_jobs'); + const failureMetric = Arnavon.registry.getSingleMetric('runner_failed_jobs'); + const spySuccess = sinon.spy(successMetric, 'inc'); + const spyFailure = sinon.spy(failureMetric, 'inc'); + + return rawRunner.run({ some: 'data' }) + .catch(() => { + expect(spySuccess).to.not.have.been.called; + expect(spyFailure).to.not.have.been.called; + }); + }); + + it('still returns a Promise', () => { + rawRunner._run = sinon.stub().returns(Promise.resolve(42)); + const res = rawRunner.run({ some: 'data' }); + expect(res).to.be.an.instanceof(Promise); + }); + + it('still rejects when _run throws', () => { + rawRunner._run = sinon.stub().throws(new Error('oops')); + return rawRunner.run({ some: 'data' }) + .then(() => { throw new Error('should not resolve'); }) + .catch(err => { + expect(err.message).to.equal('oops'); + }); + }); + + it('still rejects for non-promise results', () => { + rawRunner._run = sinon.stub().returns('not a promise'); + return rawRunner.run({ some: 'data' }) + .then(() => { throw new Error('should not resolve'); }) + .catch(err => { + expect(err).to.be.an.instanceof(InvalidRunError); + }); + }); + }); + describe('.factor', () => { beforeEach(() => { diff --git a/tests/jobs/runners/binary.spec.ts b/tests/jobs/runners/binary.spec.ts index 546b8f0..1488dae 100644 --- a/tests/jobs/runners/binary.spec.ts +++ b/tests/jobs/runners/binary.spec.ts @@ -147,8 +147,8 @@ describe('BinaryRunner', () => { }); }); - it.skip('gets every line printed on stderr and forwards to logger'); - it.skip('parses every line printed on stderr when possible and forwards to logger'); + it('gets every line printed on stderr and forwards to logger'); + it('parses every line printed on stderr when possible and forwards to logger'); }); }); diff --git a/tests/jobs/runners/function.spec.ts b/tests/jobs/runners/function.spec.ts new file mode 100644 index 0000000..6f82e6b --- /dev/null +++ b/tests/jobs/runners/function.spec.ts @@ -0,0 +1,55 @@ +import FunctionRunner from '../../../src/jobs/runners/function'; +import { expect, default as chai } from 'chai'; +import sinonChai from 'sinon-chai'; +import chaiAsPromised from 'chai-as-promised'; +import Job from '../../../src/jobs/job'; + +chai.use(sinonChai); +chai.use(chaiAsPromised); + +describe('FunctionRunner', () => { + + let testJob; + beforeEach(() => { + testJob = new Job({ email: 'test@test.com' }, { + dispatched: new Date(), + dequeued: new Date(), + }); + }); + + it('exports a class', () => { + expect(FunctionRunner).to.be.an.instanceof(Function); + expect(FunctionRunner.name).to.equal('FunctionRunner'); + }); + + describe('its constructor', () => { + it('accepts a handler function in config', () => { + const handler = async (job) => job; + const runner = new FunctionRunner({ type: 'function', handler, config: {} }); + expect(runner).to.be.an.instanceof(FunctionRunner); + }); + }); + + describe('#run', () => { + it('calls the handler with the job and context', () => { + let receivedJob, receivedContext; + const handler = async (job, context) => { + receivedJob = job; + receivedContext = context; + return 'done'; + }; + const runner = new FunctionRunner({ type: 'function', handler, config: {} }); + return runner.run(testJob).then((result) => { + expect(result).to.equal('done'); + expect(receivedJob).to.be.an.instanceof(Job); + }); + }); + + it('propagates handler rejections', () => { + const handler = async () => { throw new Error('handler failed'); }; + const runner = new FunctionRunner({ type: 'function', handler, config: {} }); + return expect(runner.run(testJob)).to.be.rejectedWith('handler failed'); + }); + }); + +}); diff --git a/tests/jobs/runners/nodejs.spec.ts b/tests/jobs/runners/nodejs.spec.ts index 0fe365d..8de9e61 100644 --- a/tests/jobs/runners/nodejs.spec.ts +++ b/tests/jobs/runners/nodejs.spec.ts @@ -50,11 +50,18 @@ describe('NodeJSRunner', () => { describe('#run', () => { - describe.skip('when used in raw mode', () => { - it('should pass the raw message to the node module', () => { - runner.run(testJob); + describe('when used in raw mode', () => { + let rawRunner; + beforeEach(() => { + rawRunner = new NodeJSRunner({ module: './dummy.runner', mode: 'raw' }); + dummy.runner.reset(); + }); + + it('should pass the raw message to the node module unchanged', () => { + const rawMessage = { some: 'raw data' }; + rawRunner.run(rawMessage); expect(dummy.runner.calls).to.have.length(1); - expect(dummy.runner.calls[0]).to.equal(testJob); + expect(dummy.runner.calls[0]).to.equal(rawMessage); }); }); diff --git a/tests/jobs/validator.spec.ts b/tests/jobs/validator.spec.ts index 24871e0..c9c37dc 100644 --- a/tests/jobs/validator.spec.ts +++ b/tests/jobs/validator.spec.ts @@ -1,6 +1,7 @@ import Finitio from 'finitio'; import JobValidator from '../../src/jobs/validator'; import { expect } from 'chai'; +import { DataValidationError } from '../../src/robust'; describe('JobValidator', () => { @@ -24,13 +25,15 @@ describe('JobValidator', () => { }); describe('its constructor', () => { - it('expects a finitio type as first argument', () => { + it('expects a finitio type or validator function as first argument', () => { const test = (t) => () => new JobValidator(t); - expect(test()).to.throw(/Finitio system expected, got undefined/); - expect(test(null)).to.throw(/Finitio system expected, got null/); - expect(test({})).to.throw(/Finitio system expected, got/); - // correct + expect(test()).to.throw(/Finitio system or validator function expected, got undefined/); + expect(test(null)).to.throw(/Finitio system or validator function expected, got null/); + expect(test({})).to.throw(/Finitio system or validator function expected, got/); + // correct: Finitio system expect(test(jobSchema)).to.not.throw(); + // correct: validator function + expect(test((data) => data)).to.not.throw(); }); }); @@ -51,6 +54,76 @@ describe('JobValidator', () => { expect(payload.date).to.be.an.instanceOf(Date); }); + it('validates with optional fields present', () => { + const payload = validator.validate({ id: 1, name: 'test', date: '2021-06-15 12:00:00' }); + expect(payload.id).to.equal(1); + expect(payload.name).to.equal('test'); + expect(payload.date).to.be.an.instanceOf(Date); + }); + + it('validates with optional fields absent', () => { + const payload = validator.validate({ id: 1, name: 'test' }); + expect(payload.id).to.equal(1); + expect(payload.name).to.equal('test'); + expect(payload.date).to.be.undefined; + }); + + it('validates nested object schemas', () => { + const nestedSchema = Finitio.system(` + @import finitio/data + { + user: { + name: String + age: Integer + } + } + `); + const nestedValidator = new JobValidator(nestedSchema); + const payload = nestedValidator.validate({ user: { name: 'Alice', age: 30 } }); + expect(payload.user.name).to.equal('Alice'); + expect(payload.user.age).to.equal(30); + }); + + it('coerces values during dressing (date string becomes Date object)', () => { + const input = { id: 5, name: 'coerce-test', date: '2023-03-15 08:30:00' }; + const payload = validator.validate(input); + // The original input date is a string + expect(input.date).to.be.a('string'); + // The dressed output date is a Date object + expect(payload.date).to.be.an.instanceOf(Date); + }); + + it('throws a DataValidationError on invalid input', () => { + expect(() => validator.validate({ id: 'not-a-number', name: 'test' })).to.throw(DataValidationError); + }); + + }); + + describe('with a validator function', () => { + + it('uses the function for validation', () => { + const fn = (data: any) => { + if (!data || !data.email) throw new Error('email required'); + return { email: data.email.toLowerCase() }; + }; + const v = new JobValidator(fn); + const result = v.validate({ email: 'FOO@BAR.COM' }); + expect(result.email).to.equal('foo@bar.com'); + }); + + it('throws when the validator function throws', () => { + const fn = () => { throw new Error('invalid'); }; + const v = new JobValidator(fn); + expect(() => v.validate({})).to.throw('invalid'); + }); + + it('returns the transformed data from the function', () => { + const fn = (data: any) => ({ ...data, validated: true }); + const v = new JobValidator(fn); + const result = v.validate({ foo: 'bar' }); + expect(result).to.eql({ foo: 'bar', validated: true }); + }); + }); }); diff --git a/tests/queue/drivers/amqp.spec.ts b/tests/queue/drivers/amqp.spec.ts index 84ab548..62a57c4 100644 --- a/tests/queue/drivers/amqp.spec.ts +++ b/tests/queue/drivers/amqp.spec.ts @@ -1,18 +1,52 @@ -import { expect } from 'chai'; -import chai from 'chai'; +import { expect, default as chai } from 'chai'; +import sinon from 'sinon'; import sinonChai from 'sinon-chai'; import AmqpQueue from '../../../src/queue/drivers/amqp'; import Config from '../../../src/config'; +import amqplib from 'amqplib'; chai.should(); chai.use(sinonChai); +// Helpers to build a mock AMQP connection and channel +function createMockChannel(overrides: Record = {}) { + return { + prefetch: sinon.stub(), + assertExchange: sinon.stub().resolves(), + assertQueue: sinon.stub().resolves(), + bindQueue: sinon.stub().resolves(), + publish: sinon.stub().callsFake((exchange, key, content, options, cb) => cb && cb(null)), + consume: sinon.stub().resolves(), + ack: sinon.stub(), + nack: sinon.stub(), + reject: sinon.stub(), + on: sinon.stub(), + ...overrides, + }; +} + +function createMockConnection(channel) { + return { + createConfirmChannel: sinon.stub().resolves(channel), + on: sinon.stub(), + close: sinon.stub().resolves(), + }; +} + +async function connectQueue(config, sandbox) { + const channel = createMockChannel(); + const conn = createMockConnection(channel); + sandbox.stub(amqplib, 'connect').resolves(conn); + const queue = new AmqpQueue(config); + await queue._connect(); + return { queue, channel, conn }; +} + describe('AmqpQueue', () => { let config; beforeEach(() => { config = Config.fromFile('example/config.yaml'); - // Clear AMQP_URL env var to test url parameter delete process.env.AMQP_URL; }); @@ -36,46 +70,58 @@ describe('AmqpQueue', () => { }); it('complains if more than one exchange is defined as default', () => { - config.queue.config.topology.exchanges.push({ - name: 'second', - default: true, - type: 'topic', - }); + config.queue.config.topology.exchanges.push({ name: 'second', default: true, type: 'topic' }); expect(() => new AmqpQueue(config.queue.config)).to.throw(/only one exchange can be set as default/); }); it('complains if no url is provided and AMQP_URL env var is not set', () => { - const configWithoutUrl = { ...config.queue.config }; - delete configWithoutUrl.url; - expect(() => new AmqpQueue(configWithoutUrl)).to.throw(/AMQP: url parameter required/); + const cfg = { ...config.queue.config }; + delete cfg.url; + expect(() => new AmqpQueue(cfg)).to.throw(/AMQP: url parameter required/); }); it('uses AMQP_URL env var if url parameter is not provided', () => { process.env.AMQP_URL = 'amqp://env-host:5672'; - const configWithoutUrl = { ...config.queue.config }; - delete configWithoutUrl.url; - const queue = new AmqpQueue(configWithoutUrl); - expect(queue).to.be.an.instanceof(AmqpQueue); - // The URL is stored in a protected property + const cfg = { ...config.queue.config }; + delete cfg.url; + const queue = new AmqpQueue(cfg); expect((queue as any).url).to.equal('amqp://env-host:5672'); }); it('uses AMQP_URL env var over url parameter when both are present', () => { process.env.AMQP_URL = 'amqp://env-host:5672'; const queue = new AmqpQueue(config.queue.config); - expect(queue).to.be.an.instanceof(AmqpQueue); - // Env var takes precedence over config url expect((queue as any).url).to.equal('amqp://env-host:5672'); }); it('uses default values for connectRetries and prefetchCount', () => { - const configWithDefaults = { - url: 'amqp://localhost', - topology: config.queue.config.topology, - }; - const queue = new AmqpQueue(configWithDefaults); + const queue = new AmqpQueue({ url: 'amqp://localhost', topology: config.queue.config.topology }); expect(queue).to.be.an.instanceof(AmqpQueue); - // Default values should be applied internally + }); + }); + + describe('.deriveManagementUrl', () => { + it('derives management URL from AMQP URL', () => { + const result = AmqpQueue.deriveManagementUrl('amqp://user:pass@myhost:5672/myvhost'); + expect(result.url).to.equal('http://myhost:15672'); + expect(result.vhost).to.equal('myvhost'); + expect(result.auth).to.equal(Buffer.from('user:pass').toString('base64')); + }); + + it('uses default vhost when none specified', () => { + const result = AmqpQueue.deriveManagementUrl('amqp://user:pass@myhost:5672'); + expect(result.vhost).to.equal('/'); + }); + + it('builds AMQP URI for shovel with encoded vhost', () => { + const result = AmqpQueue.deriveManagementUrl('amqp://user:pass@myhost:5672/my%2Fvhost'); + expect(result.amqpUri).to.include('user:pass@myhost:5672'); + }); + + it('handles URL with query string', () => { + const result = AmqpQueue.deriveManagementUrl('amqp://user:pass@myhost:5672?heartbeat=30'); + expect(result.url).to.equal('http://myhost:15672'); + expect(result.vhost).to.equal('/'); }); }); @@ -86,18 +132,419 @@ describe('AmqpQueue', () => { }); }); + describe('#_connect', () => { + const sandbox = sinon.createSandbox(); + afterEach(() => sandbox.restore()); + + it('connects and installs topology', async () => { + const channel = createMockChannel(); + const conn = createMockConnection(channel); + sandbox.stub(amqplib, 'connect').resolves(conn); + + const queue = new AmqpQueue(config.queue.config); + await queue._connect(); + + expect(conn.createConfirmChannel).to.be.calledOnce; + expect(channel.prefetch).to.be.calledOnceWith(5); + expect(channel.assertExchange.callCount).to.equal(3); + expect(channel.assertQueue.callCount).to.equal(4); + }); + + it('retries on connection failure then succeeds', async () => { + const channel = createMockChannel(); + const conn = createMockConnection(channel); + let callCount = 0; + sandbox.stub(amqplib, 'connect').callsFake(() => { + callCount++; + if (callCount === 1) return Promise.reject(new Error('ECONNREFUSED')); + return Promise.resolve(conn); + }); + + const queue = new AmqpQueue({ + url: 'amqp://localhost', + connectRetries: 3, + topology: config.queue.config.topology, + }); + await queue._connect(); + + expect(callCount).to.be.at.least(2); + }).timeout(5000); + + it('propagates connection close events', async () => { + const channel = createMockChannel(); + const conn = createMockConnection(channel); + sandbox.stub(amqplib, 'connect').resolves(conn); + + const queue = new AmqpQueue(config.queue.config); + const closeSpy = sinon.spy(); + queue.on('close', closeSpy); + await queue._connect(); + + const closeHandler = conn.on.getCalls().find(c => c.args[0] === 'close'); + closeHandler.args[1](new Error('connection closed')); + expect(closeSpy).to.be.calledOnce; + }); + + it('propagates connection error events', async () => { + const channel = createMockChannel(); + const conn = createMockConnection(channel); + sandbox.stub(amqplib, 'connect').resolves(conn); + + const queue = new AmqpQueue(config.queue.config); + const errorSpy = sinon.spy(); + queue.on('error', errorSpy); + await queue._connect(); + + const errorHandler = conn.on.getCalls().find(c => c.args[0] === 'error'); + errorHandler.args[1](new Error('connection error')); + expect(errorSpy).to.be.calledOnce; + }); + }); + describe('#_push', () => { + const sandbox = sinon.createSandbox(); + afterEach(() => sandbox.restore()); + it('throws an error if no channel is available', () => { const queue = new AmqpQueue(config.queue.config); expect(() => queue._push('key', { data: 'test' }, { exchange: 'test' })).to.throw(/Cannot push, no channel found/); }); + + it('publishes JSON-serialized data to the default exchange', async () => { + const { queue, channel } = await connectQueue(config.queue.config, sandbox); + const data = { foo: 'bar' }; + await queue._push('my-key', data, {}); + + expect(channel.publish).to.be.calledOnce; + const [exchange, key, payload, options] = channel.publish.getCall(0).args; + expect(exchange).to.equal('example-arnavon'); + expect(key).to.equal('my-key'); + expect(JSON.parse(payload.toString())).to.eql(data); + expect(options.persistent).to.be.true; + }); + + it('publishes to a custom exchange when specified', async () => { + const { queue, channel } = await connectQueue(config.queue.config, sandbox); + await queue._push('my-key', { foo: 'bar' }, { exchange: 'custom-exchange' }); + expect(channel.publish.getCall(0).args[0]).to.equal('custom-exchange'); + }); + + it('forwards headers in publish options', async () => { + const { queue, channel } = await connectQueue(config.queue.config, sandbox); + const headers = { 'x-delay': '5000' }; + await queue._push('my-key', {}, { exchange: undefined, headers }); + expect(channel.publish.getCall(0).args[3].headers).to.eql(headers); + }); + + it('rejects when publish reports an error', async () => { + const channel = createMockChannel({ + publish: sinon.stub().callsFake((ex, key, content, opts, cb) => cb(new Error('publish failed'))), + }); + const conn = createMockConnection(channel); + sandbox.stub(amqplib, 'connect').resolves(conn); + const queue = new AmqpQueue(config.queue.config); + await queue._connect(); + + try { + await queue._push('key', {}, {}); + throw new Error('should have rejected'); + } catch (err) { + expect(err.message).to.equal('publish failed'); + } + }); }); describe('#_consume', () => { + const sandbox = sinon.createSandbox(); + afterEach(() => sandbox.restore()); + it('throws an error if no channel is available', () => { const queue = new AmqpQueue(config.queue.config); expect(() => queue._consume('queue-name', () => Promise.resolve())).to.throw(/Cannot consume, no channel found/); }); + + it('starts consuming from the specified queue', async () => { + const { queue, channel } = await connectQueue(config.queue.config, sandbox); + await queue._consume('my-queue', sinon.stub().resolves()); + expect(channel.consume).to.be.calledOnce; + expect(channel.consume.getCall(0).args[0]).to.equal('my-queue'); + }); + + it('ACKs on successful processing', async () => { + const channel = createMockChannel({ + consume: sinon.stub().callsFake((queueName, handler) => { + handler({ + content: Buffer.from(JSON.stringify({ payload: 'test', meta: {} })), + fields: { routingKey: 'test' }, + }); + return Promise.resolve(); + }), + }); + const conn = createMockConnection(channel); + sandbox.stub(amqplib, 'connect').resolves(conn); + const queue = new AmqpQueue(config.queue.config); + await queue._connect(); + + const processor = sinon.stub().resolves(); + await queue._consume('my-queue', processor); + await new Promise(resolve => setTimeout(resolve, 10)); + + expect(processor).to.be.calledOnce; + expect(channel.ack).to.be.calledOnce; + }); + + it('NACKs (rejects) on processor failure', async () => { + const channel = createMockChannel({ + consume: sinon.stub().callsFake((queueName, handler) => { + handler({ + content: Buffer.from(JSON.stringify({ payload: 'test', meta: {} })), + fields: { routingKey: 'test' }, + }); + return Promise.resolve(); + }), + }); + const conn = createMockConnection(channel); + sandbox.stub(amqplib, 'connect').resolves(conn); + const queue = new AmqpQueue(config.queue.config); + await queue._connect(); + + const processor = sinon.stub().rejects(new Error('processing failed')); + await queue._consume('my-queue', processor); + await new Promise(resolve => setTimeout(resolve, 10)); + + expect(channel.reject).to.be.calledOnce; + expect(channel.reject.getCall(0).args[1]).to.be.false; + }); + + it('NACKs on invalid JSON payload', async () => { + const channel = createMockChannel({ + consume: sinon.stub().callsFake((queueName, handler) => { + handler({ content: Buffer.from('not valid json{{{'), fields: {} }); + return Promise.resolve(); + }), + }); + const conn = createMockConnection(channel); + sandbox.stub(amqplib, 'connect').resolves(conn); + const queue = new AmqpQueue(config.queue.config); + await queue._connect(); + + const processor = sinon.stub().resolves(); + await queue._consume('my-queue', processor); + await new Promise(resolve => setTimeout(resolve, 10)); + + expect(processor).to.not.have.been.called; + expect(channel.nack).to.be.calledOnce; + expect(channel.nack.getCall(0).args[1]).to.be.false; + expect(channel.nack.getCall(0).args[2]).to.be.false; + }); + + it('handles null message (consumption cancelled)', async () => { + const channel = createMockChannel({ + consume: sinon.stub().callsFake((queueName, handler) => { + handler(null); + return Promise.resolve(); + }), + }); + const conn = createMockConnection(channel); + sandbox.stub(amqplib, 'connect').resolves(conn); + const queue = new AmqpQueue(config.queue.config); + await queue._connect(); + + const processor = sinon.stub().resolves(); + await queue._consume('my-queue', processor); + + expect(processor).to.not.have.been.called; + expect(channel.ack).to.not.have.been.called; + expect(channel.nack).to.not.have.been.called; + }); + + it('passes parsed payload and metadata to processor', async () => { + const jobData = { payload: { email: 'test@test.com' }, meta: { id: '123' } }; + const channel = createMockChannel({ + consume: sinon.stub().callsFake((queueName, handler) => { + handler({ + content: Buffer.from(JSON.stringify(jobData)), + fields: { routingKey: 'send-email', deliveryTag: 42 }, + }); + return Promise.resolve(); + }), + }); + const conn = createMockConnection(channel); + sandbox.stub(amqplib, 'connect').resolves(conn); + const queue = new AmqpQueue(config.queue.config); + await queue._connect(); + + const processor = sinon.stub().resolves(); + await queue._consume('my-queue', processor); + await new Promise(resolve => setTimeout(resolve, 10)); + + expect(processor).to.be.calledOnce; + const [payload, metadata] = processor.getCall(0).args; + expect(payload).to.eql(jobData); + expect(metadata.routingKey).to.equal('send-email'); + }); + }); + + describe('#_disconnect', () => { + const sandbox = sinon.createSandbox(); + afterEach(() => sandbox.restore()); + + it('closes the connection', async () => { + const { queue, conn } = await connectQueue(config.queue.config, sandbox); + await queue._disconnect(); + expect(conn.close).to.be.calledOnce; + }); + + it('handles disconnect when not connected', async () => { + const queue = new AmqpQueue(config.queue.config); + await queue._disconnect(); + }); + }); + + describe('#_requeue', () => { + const sandbox = sinon.createSandbox(); + let queue, fetchStub; + + beforeEach(async () => { + const result = await connectQueue(config.queue.config, sandbox); + queue = result.queue; + fetchStub = sandbox.stub(globalThis, 'fetch'); + }); + + afterEach(() => sandbox.restore()); + + it('creates a shovel to requeue messages', async () => { + fetchStub.onFirstCall().resolves({ ok: false, status: 404 }); + fetchStub.onSecondCall().resolves({ + ok: true, + json: () => Promise.resolve({ messages: 5 }), + }); + fetchStub.onThirdCall().resolves({ ok: true }); + + const result = await queue._requeue('dead-letters', {}); + + expect(result.status).to.equal('initiated'); + expect(result.requeued).to.equal(5); + expect(fetchStub.callCount).to.equal(3); + + const createCall = fetchStub.getCall(2); + expect(createCall.args[1].method).to.equal('PUT'); + const body = JSON.parse(createCall.args[1].body); + expect(body.value['src-queue']).to.equal('dead-letters'); + expect(body.value['dest-exchange']).to.equal('example-arnavon'); + }); + + it('limits requeue count when specified', async () => { + fetchStub.onFirstCall().resolves({ ok: false, status: 404 }); + fetchStub.onSecondCall().resolves({ + ok: true, + json: () => Promise.resolve({ messages: 100 }), + }); + fetchStub.onThirdCall().resolves({ ok: true }); + + const result = await queue._requeue('dead-letters', { count: 10 }); + expect(result.requeued).to.equal(10); + const body = JSON.parse(fetchStub.getCall(2).args[1].body); + expect(body.value['src-delete-after']).to.equal(10); + }); + + it('throws when a requeue is already in progress', async () => { + fetchStub.onFirstCall().resolves({ ok: true }); + + try { + await queue._requeue('dead-letters', {}); + throw new Error('should have thrown'); + } catch (err) { + expect(err.message).to.include('already in progress'); + } + }); + + it('throws when shovel plugin is not available', async () => { + fetchStub.onFirstCall().resolves({ ok: false, status: 404 }); + fetchStub.onSecondCall().resolves({ + ok: true, + json: () => Promise.resolve({ messages: 5 }), + }); + fetchStub.onThirdCall().resolves({ + ok: false, + status: 404, + text: () => Promise.resolve('Not Found'), + }); + + try { + await queue._requeue('dead-letters', {}); + throw new Error('should have thrown'); + } catch (err) { + expect(err.message).to.include('Shovel plugin not available'); + } + }); + + it('throws on authentication failure', async () => { + fetchStub.onFirstCall().resolves({ ok: false, status: 404 }); + fetchStub.onSecondCall().resolves({ + ok: true, + json: () => Promise.resolve({ messages: 5 }), + }); + fetchStub.onThirdCall().resolves({ + ok: false, + status: 401, + text: () => Promise.resolve('Unauthorized'), + }); + + try { + await queue._requeue('dead-letters', {}); + throw new Error('should have thrown'); + } catch (err) { + expect(err.message).to.include('authentication failed'); + } + }); + }); + + describe('#_getQueuesInfo', () => { + const sandbox = sinon.createSandbox(); + let queue, fetchStub; + + beforeEach(async () => { + const result = await connectQueue(config.queue.config, sandbox); + queue = result.queue; + fetchStub = sandbox.stub(globalThis, 'fetch'); + }); + + afterEach(() => sandbox.restore()); + + it('fetches info for each queue from management API', async () => { + fetchStub.onFirstCall().resolves({ + ok: true, + json: () => Promise.resolve({ name: 'q1', messages: 10, consumers: 2, state: 'running' }), + }); + fetchStub.onSecondCall().resolves({ + ok: true, + json: () => Promise.resolve({ name: 'q2', messages: 0, consumers: 0, state: 'idle' }), + }); + + const result = await queue._getQueuesInfo(['q1', 'q2']); + expect(result).to.have.length(2); + expect(result[0]).to.eql({ name: 'q1', messages: 10, consumers: 2, state: 'running' }); + expect(result[1]).to.eql({ name: 'q2', messages: 0, consumers: 0, state: 'idle' }); + }); + + it('returns unknown state for non-existent queues (404)', async () => { + fetchStub.resolves({ ok: false, status: 404 }); + const result = await queue._getQueuesInfo(['missing-queue']); + expect(result[0]).to.eql({ name: 'missing-queue', messages: 0, consumers: 0, state: 'unknown' }); + }); + + it('returns unknown state on network errors', async () => { + fetchStub.rejects(new Error('ECONNREFUSED')); + const result = await queue._getQueuesInfo(['some-queue']); + expect(result[0]).to.eql({ name: 'some-queue', messages: 0, consumers: 0, state: 'unknown' }); + }); + + it('returns unknown state for non-200/404 responses', async () => { + fetchStub.resolves({ ok: false, status: 500 }); + const result = await queue._getQueuesInfo(['some-queue']); + expect(result[0].state).to.equal('unknown'); + }); }); }); diff --git a/tests/server/rest/api.spec.ts b/tests/server/rest/api.spec.ts index 26c423c..ba5d168 100644 --- a/tests/server/rest/api.spec.ts +++ b/tests/server/rest/api.spec.ts @@ -1,12 +1,12 @@ -import proxyquire from 'proxyquire'; import { expect, default as chai } from 'chai'; import chaiHttp from 'chai-http'; -import createApiHelper from '../../../src/api'; +import createApi from '../../../src/server/rest'; import { UnknownJobError, DataValidationError } from '../../../src/robust'; import sinon from 'sinon'; import sinonChai from 'sinon-chai'; import { JobDispatcher } from '../../../src/jobs'; import Arnavon from '../../../src/index'; +import { validate as uuidValidate } from 'uuid'; chai.should(); chai.use(sinonChai); @@ -14,31 +14,9 @@ chai.use(chaiHttp); describe('server/createApi', () => { - let createApi: (params?: any) => Express.Application, helperCalled: boolean, staticUuid: string; - beforeEach(() => { - staticUuid = 'ea47ac69-c0ca-4960-b905-6f14f2029744'; - helperCalled = false; - createApi = proxyquire('../../../src/server/rest', { - '../../api': { - default: function() { - helperCalled = true; - // eslint-disable-next-line prefer-rest-params - const api = createApiHelper(arguments as any as { agent: string}); - // make sure all request ids use our static uuid - api.use((req, res, next) => { - req.id = staticUuid; - next(); - }); - return api; - }, - }, - }).default; - }); - it('calls the createApi helper (src/api)', () => { - expect(helperCalled).to.equal(false); - createApi(); - expect(helperCalled).to.equal(true); + const api = createApi(); + expect(api).to.be.a('function'); }); describe('its POST /jobs/:id endpoint', () => { @@ -163,7 +141,11 @@ describe('server/createApi', () => { .send(jobPayload) .end((err, res) => { res.should.have.status(201); - expect(dispatcher.dispatch).to.be.calledOnceWith('foo-bar', jobPayload, { id: staticUuid }); + expect(dispatcher.dispatch).to.be.calledOnce; + const { args } = dispatcher.dispatch.getCall(0); + expect(args[0]).to.equal('foo-bar'); + expect(args[1]).to.eql(jobPayload); + expect(uuidValidate(args[2].id)).to.be.true; done(); }); }); @@ -180,7 +162,11 @@ describe('server/createApi', () => { .send(jobPayload) .end((err, res) => { res.should.have.status(201); - expect(dispatcher.dispatchBatch).to.be.calledOnceWith('foo-bar', jobPayload, { id: staticUuid }); + expect(dispatcher.dispatchBatch).to.be.calledOnce; + const { args } = dispatcher.dispatchBatch.getCall(0); + expect(args[0]).to.equal('foo-bar'); + expect(args[1]).to.eql(jobPayload); + expect(uuidValidate(args[2].id)).to.be.true; done(); }); }); diff --git a/tests/server/server.spec.ts b/tests/server/server.spec.ts index 865a11d..693eddd 100644 --- a/tests/server/server.spec.ts +++ b/tests/server/server.spec.ts @@ -1,31 +1,25 @@ -import Arnavon, { Server } from '../../src'; +import Arnavon from '../../src'; import Config from '../../src/config'; +import Server from '../../src/server'; import { expect, default as chai } from 'chai'; import sinon from 'sinon'; import sinonChai from 'sinon-chai'; -import proxyquire from 'proxyquire'; -import ArnavonConfig from '../../src/config'; chai.should(); chai.use(sinonChai); describe('Server', () => { - let config, server: Server, Server: new(config: ArnavonConfig) => Server, listen: sinon.SinonStub; + let config, server: Server; beforeEach(() => { config = Config.fromFile('example/config.yaml'); - listen = sinon.stub().yields(); - Server = proxyquire('../../src/server', { - './rest': { - default: function() { - return { - listen, - }; - }, - }, - }).default; + server = new Server(config); + // Stub _startApi to avoid actually starting an HTTP server + sinon.stub(server, '_startApi').resolves(server); + }); - server = new Server(config as ArnavonConfig); + afterEach(() => { + sinon.restore(); }); it('is a class', () => { @@ -58,7 +52,7 @@ describe('Server', () => { Arnavon.queue.connect = sinon.stub().resolves(true); return server.start() .then(() => { - expect(listen).to.be.calledOnce; + expect(server._startApi).to.be.calledOnce; }); }); @@ -72,4 +66,24 @@ describe('Server', () => { }); }); }); + + describe('#stop', () => { + + it('disconnects the queue', async () => { + Arnavon.queue.connect = sinon.stub().resolves(true); + const disconnectSpy = sinon.stub(Arnavon.queue, 'disconnect').resolves(Arnavon.queue); + await server.start(); + await server.stop(); + expect(disconnectSpy).to.be.calledOnce; + }); + + it('stops the internal API server', async () => { + Arnavon.queue.connect = sinon.stub().resolves(true); + sinon.stub(Arnavon.queue, 'disconnect').resolves(Arnavon.queue); + const stopApiSpy = sinon.spy(server, '_stopApi'); + await server.start(); + await server.stop(); + expect(stopApiSpy).to.be.calledOnce; + }); + }); });