Add GET /queues API endpoint for queue status

Add endpoint to retrieve information about all configured queues from
RabbitMQ Management API, including message count and consumer count.

API:
  GET /queues
  Response: { "queues": [{ "name", "messages", "consumers", "state" }] }

Implementation:
- src/queue/index.ts: Add QueueInfo type and getQueuesInfo method
- src/queue/drivers/amqp.ts: Query /api/queues/{vhost}/{name} for each queue
- src/queue/drivers/memory.ts: Return queue lengths from internal state
- src/server/rest/index.ts: Add /queues endpoint using config topology

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: llambeau

example/webspicy/arnavon/queues/get.yml

@@ -0,0 +1,41 @@
+---
+name: |-
+  Get queues status
+
+url: |-
+  /queues
+
+services:
+- method: |-
+    GET
+
+  description: |-
+    Retrieve status information for all queues defined in the Arnavon
+    configuration, including message count and consumer count.
+
+  preconditions:
+    - RabbitMQ Management plugin must be enabled
+    - Arnavon must be configured with queue topology
+
+  postconditions:
+    - Returns status for all configured queues
+    - Each queue includes name, messages, consumers, and state
+
+  input_schema: |-
+    .
+
+  output_schema: |-
+    Queue.ListResponse
+
+  error_schema: |-
+    .
+
+  examples:
+
+    - description: |-
+        FOR: Returns status for all configured queues
+      expected:
+        content_type: application/json; charset=utf-8
+        status: 200
+      assert:
+        - 'size("queues", 4)'

example/webspicy/schema.fio

@@ -60,6 +60,23 @@ Dlq.Error = {
   error : String
 }
 
+###
+### Queues Info
+###
+
+Queue.State = String
+
+Queue.Info = {
+  name      : String
+  messages  : Numeric
+  consumers : Numeric
+  state     : Queue.State
+}
+
+Queue.ListResponse = {
+  queues : [Queue.Info]
+}
+
 ###
 ### Example
 ###

src/queue/drivers/amqp.ts

@@ -1,4 +1,4 @@
-import Queue, { QueueInternalProcessor, RequeueOptions, RequeueResult } from '../index';
+import Queue, { QueueInternalProcessor, QueueInfo, RequeueOptions, RequeueResult } from '../index';
 import amqplib, { ConfirmChannel } from 'amqplib';
 import logger from '../../logger';
 import { JobMeta } from '../../jobs/job';
@@ -392,6 +392,48 @@ class AmqpQueue extends Queue {
     }
   }
 
+  async _getQueuesInfo(queueNames: string[]): Promise<QueueInfo[]> {
+    const vhostEncoded = encodeURIComponent(this.#vhost);
+
+    const fetchQueueInfo = async (queueName: string): Promise<QueueInfo> => {
+      const queueNameEncoded = encodeURIComponent(queueName);
+      const queueInfoUrl = `${this.#managementUrl}/api/queues/${vhostEncoded}/${queueNameEncoded}`;
+
+      try {
+        const response = await fetch(queueInfoUrl, {
+          headers: { 'Authorization': `Basic ${this.#managementAuth}` },
+        });
+
+        if (response.ok) {
+          const info = await response.json() as {
+            name: string;
+            messages?: number;
+            consumers?: number;
+            state?: string;
+          };
+
+          return {
+            name: info.name,
+            messages: info.messages ?? 0,
+            consumers: info.consumers ?? 0,
+            state: info.state === 'running' ? 'running' : info.state === 'idle' ? 'idle' : 'unknown',
+          };
+        } else if (response.status === 404) {
+          // Queue doesn't exist in RabbitMQ
+          return { name: queueName, messages: 0, consumers: 0, state: 'unknown' };
+        } else {
+          logger.warn(`${this.constructor.name}: Failed to get info for queue ${queueName}: ${response.status}`);
+          return { name: queueName, messages: 0, consumers: 0, state: 'unknown' };
+        }
+      } catch (err) {
+        logger.error(err, `${this.constructor.name}: Error getting info for queue ${queueName}`);
+        return { name: queueName, messages: 0, consumers: 0, state: 'unknown' };
+      }
+    };
+
+    return Promise.all(queueNames.map(fetchQueueInfo));
+  }
+
 }
 
 export default AmqpQueue;

src/queue/drivers/memory.ts

@@ -1,4 +1,4 @@
-import Queue, { QueueInternalProcessor, RequeueOptions, RequeueResult } from '../index';
+import Queue, { QueueInternalProcessor, QueueInfo, RequeueOptions, RequeueResult } from '../index';
 import Job from '../../jobs/job';
 
 export type MemoryQueueConfig = unknown;
@@ -75,6 +75,15 @@ class MemoryQueue extends Queue {
     return this.#queue.length;
   }
 
+  async _getQueuesInfo(queueNames: string[]): Promise<QueueInfo[]> {
+    return queueNames.map(name => ({
+      name,
+      messages: this.#queues.has(name) ? this.#queues.get(name)!.length : 0,
+      consumers: 0, // MemoryQueue doesn't track consumers
+      state: 'running' as const,
+    }));
+  }
+
 }
 
 export default MemoryQueue;

src/queue/index.ts

@@ -22,6 +22,13 @@ export type RequeueResult = {
   errors: Array<{ error: string }>;
 }
 
+export type QueueInfo = {
+  name: string;
+  messages: number;
+  consumers: number;
+  state: 'running' | 'idle' | 'unknown';
+}
+
 export type QueueProcessor = (job: Job, context: JobRunnerContext) => Promise<unknown>
 export type QueueInternalProcessor = (job: Job, metadata: JobMeta) => Promise<unknown>
 
@@ -105,6 +112,15 @@ class Queue extends EventEmitter {
     });
   }
 
+  // subclasses should implement _getQueuesInfo(queueNames)
+  getQueuesInfo(queueNames: string[]): Promise<QueueInfo[]> {
+    logger.info(`${this.constructor.name} - Getting info for queues: ${queueNames.join(', ')}`);
+    return this._getQueuesInfo(queueNames).then((result) => {
+      logger.info(`${this.constructor.name} - Got info for ${result.length} queues`);
+      return result;
+    });
+  }
+
   // To be implemented by subclasses
   _consume(_queueName: string, _processor: QueueInternalProcessor) {
     throw new Error('NotImplemented');
@@ -126,6 +142,10 @@ class Queue extends EventEmitter {
     throw new Error('NotImplemented');
   }
 
+  _getQueuesInfo(_queueNames: string[]): Promise<QueueInfo[]> {
+    throw new Error('NotImplemented');
+  }
+
 }
 
 export default Queue;

src/server/rest/index.ts

@@ -84,6 +84,19 @@ 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 queues = await Arnavon.queue.getQueuesInfo(queueNames);
+      return res.status(200).send({ queues });
+    } catch (err) {
+      next(err);
+    }
+  });
+
   api.post('/dlq/:queueName/requeue', async (req, res, next) => {
     const { queueName } = req.params;
     const count = req.query.count ? parseInt(req.query.count as string, 10) : undefined;

tests/queue/drivers/memory.spec.ts

@@ -196,4 +196,30 @@ describe('MemoryQueue', () => {
     });
   });
 
+  describe('#_getQueuesInfo', () => {
+    it('returns info for all requested queues', async () => {
+      const queue = new MemoryQueue();
+
+      queue.pushToQueue('queue-1', 'key', { data: 'item1' });
+      queue.pushToQueue('queue-1', 'key', { data: 'item2' });
+      queue.pushToQueue('queue-2', 'key', { data: 'item3' });
+
+      const result = await queue._getQueuesInfo(['queue-1', 'queue-2', 'queue-3']);
+
+      expect(result).to.have.length(3);
+      expect(result[0]).to.eql({ name: 'queue-1', messages: 2, consumers: 0, state: 'running' });
+      expect(result[1]).to.eql({ name: 'queue-2', messages: 1, consumers: 0, state: 'running' });
+      expect(result[2]).to.eql({ name: 'queue-3', messages: 0, consumers: 0, state: 'running' });
+    });
+
+    it('returns zero messages for non-existent queues', async () => {
+      const queue = new MemoryQueue();
+
+      const result = await queue._getQueuesInfo(['non-existent']);
+
+      expect(result).to.have.length(1);
+      expect(result[0]).to.eql({ name: 'non-existent', messages: 0, consumers: 0, state: 'running' });
+    });
+  });
+
 });

tests/queue/index.spec.ts

@@ -45,6 +45,9 @@ describe('Queue', () => {
     _requeue(_sourceQueue, _options) {
       return Promise.resolve({ status: 'completed', requeued: 1, failed: 0, errors: [] });
     }
+    _getQueuesInfo(queueNames) {
+      return Promise.resolve(queueNames.map(name => ({ name, messages: 0, consumers: 0, state: 'running' })));
+    }
   }
 
   let queue;
@@ -144,4 +147,20 @@ describe('Queue', () => {
     });
   });
 
+  describe('#getQueuesInfo', () => {
+    it('is a function', () => {
+      expect(queue.getQueuesInfo).to.be.an.instanceof(Function);
+    });
+    it('returns a promise', () => {
+      sinon.stub(queue, '_getQueuesInfo').returns(Promise.resolve([]));
+      expect(queue.getQueuesInfo(['queue-1'])).to.be.an.instanceof(Promise);
+    });
+    it('calls the subclass _getQueuesInfo implementation', () => {
+      const spy = sinon.stub(queue, '_getQueuesInfo')
+        .returns(Promise.resolve([{ name: 'q1', messages: 5, consumers: 1, state: 'running' }]));
+      queue.getQueuesInfo(['q1', 'q2']);
+      expect(spy).to.be.calledOnceWith(['q1', 'q2']);
+    });
+  });
+
 });

tests/server/rest/api.spec.ts

@@ -601,4 +601,95 @@ describe('server/createApi', () => {
     });
   });
 
+  describe('its GET /queues endpoint', () => {
+    let api: Express.Application, dispatcher: Partial<JobDispatcher>, getQueuesInfoStub: sinon.SinonStub;
+    let originalQueue: any, originalConfig: any;
+
+    beforeEach(() => {
+      dispatcher = {
+        dispatch: sinon.stub().returns(Promise.resolve()),
+        dispatchBatch: sinon.stub().returns(Promise.resolve()),
+      };
+      getQueuesInfoStub = sinon.stub().returns(Promise.resolve([
+        { name: 'queue-1', messages: 10, consumers: 2, state: 'running' },
+        { name: 'queue-2', messages: 0, consumers: 0, state: 'idle' },
+      ]));
+      // Save and replace the queue with a mock
+      originalQueue = Arnavon.queue;
+      originalConfig = Arnavon.config;
+      Arnavon.queue = {
+        getQueuesInfo: getQueuesInfoStub,
+        disconnect: sinon.stub().returns(Promise.resolve()),
+      } as any;
+      Arnavon.config = {
+        queue: {
+          config: {
+            topology: {
+              queues: [
+                { name: 'queue-1' },
+                { name: 'queue-2' },
+              ],
+            },
+          },
+        },
+      } as any;
+      api = createApi(dispatcher);
+    });
+
+    afterEach(() => {
+      Arnavon.queue = originalQueue;
+      Arnavon.config = originalConfig;
+      sinon.restore();
+    });
+
+    it('returns 200 with queue info from config', (done) => {
+      chai.request(api)
+        .get('/queues')
+        .end((err, res) => {
+          res.should.have.status(200);
+          expect(getQueuesInfoStub).to.be.calledOnceWith(['queue-1', 'queue-2']);
+          res.body.should.eql({
+            queues: [
+              { name: 'queue-1', messages: 10, consumers: 2, state: 'running' },
+              { name: 'queue-2', messages: 0, consumers: 0, state: 'idle' },
+            ],
+          });
+          done();
+        });
+    });
+
+    it('returns empty array when no queues in config', (done) => {
+      Arnavon.config = {
+        queue: {
+          config: {
+            topology: {
+              queues: [],
+            },
+          },
+        },
+      } as any;
+      getQueuesInfoStub.returns(Promise.resolve([]));
+
+      chai.request(api)
+        .get('/queues')
+        .end((err, res) => {
+          res.should.have.status(200);
+          res.body.should.eql({ queues: [] });
+          done();
+        });
+    });
+
+    it('returns 500 on queue info errors', (done) => {
+      getQueuesInfoStub.returns(Promise.reject(new Error('Connection failed')));
+
+      chai.request(api)
+        .get('/queues')
+        .end((err, res) => {
+          res.should.have.status(500);
+          res.body.should.eql({ error: 'Connection failed' });
+          done();
+        });
+    });
+  });
+
 });