Messages Queue

Auxiliary message queue for n8n.

Lightweight service written in Go (Fiber) that acts as an intermediate message queue between n8n and your workflows. It allows decoupling message production from processing, automatically retrying failed messages, and isolating traffic by tenant (client/instance).

Why does it exist?

n8n executes workflows asynchronously, but in high-volume scenarios or with external rate limits, it is advisable to have a temporary storage and retry layer before consuming the data. This API:

• Receives incoming messages and queues them by tenant.
• Automatically prioritizes failed messages (re-queued to the front).
• Allows confirming (ack) or reporting errors (error) per message.
• Exposes query and cleanup endpoints by tenant or global.

Architecture

flowchart LR
    subgraph n8n["n8n"]
        WF1[Entry workflow]
        WF2[Processing workflow]
    end

    subgraph MQ["Messages Queue (Go + Fiber)"]
        API[HTTP API]
        Q[(In-memory queue\nper tenant)]
    end

    subgraph Ext["External services"]
        APIEXT[API / LLM / DB]
    end

    WF1 -- "POST /:tenant_id" --> API
    API --> Q
    WF2 -- "GET /:tenant_id" --> API
    API --> Q
    WF2 -- "processes message" --> APIEXT
    WF2 -- "POST /error/:tenant_id" --> API
    WF2 -- "POST /ack/:id" --> API
    API --> Q

    Q -- "order: error > date" --> WF2

Loading

Message lifecycle

stateDiagram-v2
    [*] --> Pending: POST /:tenant_id
    Pending --> Processing: taken by n8n
    Processing --> Completed: POST /ack/:id
    Processing --> Error: POST /error/:tenant_id
    Error --> Pending: re-queued with high priority
    Completed --> [*]
    Error --> [*]: ack after retry

Loading

Data model

Each message (internal/models/message.go) contains:

Field	Type	Description
id	string (uuid)	Unique message identifier.
tenant_id	int	Owner client/instance.
priority	int	1 high (error), 2 normal.
status	string	pending / processing / completed / error.
body	json.RawMessage	Original message payload.
error_code	int	Error code (if applicable).
error_message	string	Error description (if applicable).
retry_count	int	Number of retries.
created_at	time.Time	Creation timestamp.
updated_at	time.Time	Last update.
locked	bool	Indicates if it is being processed.
locked_until	time.Time	Lock expiration.

The queue (internal/queue/queue.go) maintains a tenant_id -> []message map and sorts each tenant by priority: errors first, then by creation date (oldest first).

classDiagram
    class Queue {
        - messages map[int][]*Message
        + Enqueue(msg)
        + Ack(id) error
        + MarkError(id, status, message) error
        + Release(id) error
        + GetAll() map
        + GetByTenant(id) []*Message
        + Clear()
        + ClearTenant(id) error
    }
    class Message {
        + string ID
        + int TenantID
        + Priority Priority
        + MessageStatus Status
        + json.RawMessage Body
        + int ErrorCode
        + string ErrorMessage
        + int RetryCount
    }
    Queue "1" o-- "*" Message

Loading

API

Base URL: http://localhost:8080

Method	Route	Description
POST	/:tenant_id	Creates and queues a message (body = JSON).
GET	/	Returns all messages by tenant.
GET	/:tenant_id	Returns messages for a tenant.
POST	/error/:tenant_id	Marks a message as error (re-queued to front).
POST	/ack/:id	Confirms and removes a message from the queue.
DELETE	/	Clears the entire queue.
DELETE	/:tenant_id	Clears messages for a tenant.
GET	/health	Health check (200 OK).

Examples

Create a message:

curl -X POST http://localhost:8080/1 \
  -H "Content-Type: application/json" \
  -d '{"conversation_id": 123, "message": "Hello"}'

Mark error:

curl -X POST http://localhost:8080/error/1 \
  -H "Content-Type: application/json" \
  -d '{"id": "<uuid>", "status": 429, "message": "Rate limit"}'

Acknowledge (ack):

curl -X POST http://localhost:8080/ack/<uuid>

Note: the error payload is {"id": string, "status": int, "message": string} (see internal/api/handlers.go → ErrorMessage).

You can also use requests.http with the VS Code REST Client extension.

Running

Local (Go)

PORT=8080 go run ./cmd/server

Docker

docker build -t messages-queue .
docker run -p 8080:8080 -e PORT=8080 messages-queue

Considerations

• The queue is in-memory: messages are lost when the process restarts. Use it as an auxiliary buffer for n8n, not as a persistence system.
• It is thread-safe (uses sync.RWMutex per operation).