Improvements

src/api/index.ts

@@ -4,14 +4,16 @@ 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 '../';
 
 /**
  * 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);
 

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<string, unknown>
+  }
+}
+
+/**
+ * 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;
+  }
+}

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<T>(fn: () => Promise<T>): Promise<T> {
-    await Arnavon.queue.connect();
+    await this.app.queue.connect();
     try {
       return await fn();
     } finally {
-      await Arnavon.queue.disconnect();
+      await this.app.queue.disconnect();
     }
   }
 }

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}`);

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,19 +20,19 @@ 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) {
         this.log('No queues configured in topology.');
         return;
       }
 
-      const queues = await Arnavon.queue.getQueuesInfo(queueNames);
+      const queues = await this.app.queue.getQueuesInfo(queueNames);
 
       CliUx.ux.table(queues, {
         name: {

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();
     });
   }
 }