Skip to content
BranchesTags

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

4 Commits
grafana
grafana
 
 
init-scripts
init-scripts
 
 
n8n
n8n
 
 
prometheus
prometheus
 
 
scripts
scripts
 
 
traefik
traefik
 
 
.DS_Store
.DS_Store
 
 
.env
.env
 
 
.gitignore
.gitignore
 
 
README.md
README.md
 
 
docker-compose.yml
docker-compose.yml
 
 

Repository files navigation

CRM Workflow Engine

Multi-tenant CRM automation built on n8n, Postgres, and Mistral AI. Telegram and HTTP API entry points, workflow-as-code build system, tenant-isolated data.

Architecture

                 ┌─────────────┐     ┌─────────────┐
                 │  Telegram    │     │  HTTP POST   │
                 │  Bot Input   │     │ /webhook/    │
                 │  (text/voice)│     │  crm-api     │
                 └──────┬───────┘     └──────┬───────┘
                        │                    │
                        └────────┬───────────┘
                                 ▼
                   ┌─────────────────────────┐
                   │  Master: Ingest & Route  │ classify intent,
                   │  (mAst3rIngestCRM01)     │ resolve tenant,
                   │  Mistral Small → Large   │ dynamic dispatch
                   └────────────┬────────────┘
                    ┌───────────┼───────────┐
                    ▼           ▼           ▼
              ┌──────────┐ ┌────────┐ ┌──────────┐
              │ contacts  │ │ todos  │ │  data    │
              │ (nl2sql)  │ │(nl2sql)│ │ (export) │
              └──────────┘ └────────┘ └──────────┘
                    │           │           │
                    └───────────┼───────────┘
                                ▼
                   ┌─────────────────────────┐
                   │  Sub: Finalize & Reply   │
                   │  + Error Handler         │
                   └─────────────────────────┘

---
config:
  layout: elk
---
graph TB
    subgraph containers["Docker Containers"]
        n8n["n8n<br/>Workflow Automation"]
        postgres["PostgreSQL<br/>Database"]
        prometheus["Prometheus<br/>Metrics"]
        grafana["Grafana<br/>Dashboards"]
        whisper["Whisper ASR<br/>Speech Recognition"]
        ollama["Ollama<br/>LLM Engine"]
        traefik["Traefik<br/>Reverse Proxy"]
    end
    
    traefik --> n8n
    traefik --> grafana
    n8n --> postgres
    n8n --> whisper
    n8n --> ollama
    prometheus --> n8n
    grafana --> prometheus
    
    classDef container stroke #818cf8,fill #eef2ff
    class n8n,postgres,prometheus,grafana,whisper,ollama,traefik container
Loading

Workflows (7 active)

ID Name Role
mAst3rIngestCRM01 Master: Ingest & Route Entry point: Telegram + HTTP webhook, classification, dispatch
cNtRoutrNL2SQL01 contact-router-nl2sql Contacts module: CRUD + NL2SQL search/count
t0d0R0ut3rCRM7q1 todo-router Todos module: create, list (NL2SQL), complete, update, delete
dAtAExP0rtCRM7q1 data-router Data export module
f1nAndRep1yCRM7q Sub: Finalize & Reply Shared subworkflow: log result + send Telegram reply
eH4nd1erCRM7x2Qp Error Handler Error workflow: log error to DB + notify user
adm1nPr0vCRM7q01 Admin Provision Tenant provisioning + activate/deactivate (bearer auth)

Model stack

Layer Model Purpose
Classification (primary) Mistral Small (mistral-small-latest) Intent extraction, action routing
Classification (fallback) Mistral Large (open-mistral-nemo) Escalation when Small returns null
NL2SQL generation Codestral (codestral-latest) SQL generation for search/count/list
Help responses Mistral Small Channel-aware (markdown for HTTP, plain text for Telegram)
Audio transcription Whisper (small, local CPU) Voice message → text
Local LLM (optional) Ollama qwen2.5:7b Used by contact-router for extraction prompt

Project structure

.
├── docker-compose.yml              # traefik, postgres, redis, n8n, whisper-asr, prometheus, grafana
├── .env                            # credentials, model config
├── init-scripts/
│   ├── init.sql                    # global_registry, roles, provisioning functions
│   ├── contacts.sql                # contacts table + provisioning function
│   ├── todos.sql                   # todos table + provisioning function
│   ├── invoices.sql                # invoices table (provisioning function, no workflow yet)
│   ├── mig_add_module.sql          # migration: module column in execution_logs
│   └── mig_add_routing_path.sql    # migration: routing_path column in execution_logs
├── n8n/
│   ├── prompts/                    # prompt templates injected at build time
│   │   ├── master-classification.md
│   │   ├── todo-extraction.md
│   │   └── todo-sql-generation.md
│   ├── scripts/
│   │   ├── lib/workflow-builder.js # builder DSL: node(), conn(), write(), uid(), src()
│   │   ├── build-master-router.js  # → master-router.json
│   │   ├── build-contact-router.js # → contact-router-nl2sql.json
│   │   ├── build-todo-router.js    # → todo-router.json
│   │   ├── build-data-router.js    # → data-router.json
│   │   ├── build-admin.js          # → admin-provision.json
│   │   ├── build-error-handler.js  # → error-handler.json
│   │   ├── check-contracts.js      # registry↔builder field validation
│   │   └── deploy.sh               # build → import → publish → restart → activate
│   └── workflows/                  # generated JSON (do not edit by hand)
├── scripts/
│   ├── refresh-tunnel.sh           # localtunnel + Telegram webhook re-registration
│   └── test-api.sh                 # smoke tests against the HTTP gateway
├── traefik/
│   └── traefik.yml                 # reverse proxy config (TLS, routing rules)
├── prometheus/
│   └── prometheus.yml              # scrape config for n8n /metrics
└── grafana/
    └── dashboards/
        └── n8n-dashboard.json      # exported Grafana dashboard (8 panels)

Getting started

# 1. Start services
docker compose up -d

# 2. Deploy all workflows
./n8n/scripts/deploy.sh

# 3. Provision a tenant
curl -X POST http://localhost:5678/webhook/admin-provision \
  -H 'Content-Type: application/json' \
  -H 'Authorization: bearer MyAuthToken' \
  -d '{
    "tenant_id": "acme_corp",
    "company_name": "Acme Corporation",
    "telegram_chat_id": "7363939976"
  }'

# 4. Start the HTTPS tunnel (needed for Telegram webhook)
./scripts/refresh-tunnel.sh

Admin API

Provision a tenant

curl -X POST http://localhost:5678/webhook/admin-provision \
  -H 'Content-Type: application/json' \
  -H 'Authorization: bearer <ADMIN_TOKEN>' \
  -d '{
    "tenant_id": "acme_corp",
    "company_name": "Acme Corporation",
    "telegram_chat_id": "7363939976"
  }'

Creates the tenant schema, tables (contacts, todos, invoices, execution_logs), grants, and maps the user. Returns {"success": true, "schema_name": "customer_acme_corp", ...}.

Activate / deactivate a tenant

# Deactivate (tenant is locked out immediately)
curl -X POST http://localhost:5678/webhook/admin-tenant-toggle \
  -H 'Content-Type: application/json' \
  -H 'Authorization: bearer <ADMIN_TOKEN>' \
  -d '{"tenant_id": "acme_corp", "active": false}'

# Reactivate
curl -X POST http://localhost:5678/webhook/admin-tenant-toggle \
  -H 'Content-Type: application/json' \
  -H 'Authorization: bearer <ADMIN_TOKEN>' \
  -d '{"tenant_id": "acme_corp", "active": true}'

is_active = false blocks the tenant at the first gate in master-router — all three entry paths (Telegram, HTTP, callback) check WHERE t.is_active = TRUE.

Build & deploy

Deploy all workflows

./n8n/scripts/deploy.sh

Deploy a single workflow

./n8n/scripts/deploy.sh build-master-router.js
./n8n/scripts/deploy.sh build-contact-router.js
./n8n/scripts/deploy.sh build-todo-router.js
./n8n/scripts/deploy.sh build-data-router.js
./n8n/scripts/deploy.sh build-admin.js
./n8n/scripts/deploy.sh build-error-handler.js

What deploy.sh does

  1. Contract check — check-contracts.js queries global_registry.module_registry.params_schema for declared field names, verifies each appears in the builder file. Blocks deploy on mismatch.
  2. Build — runs builder scripts inside n8n container → generates JSON in /workflows/
  3. Import — n8n import:workflow upserts into n8n's internal DB (deactivates the workflow)
  4. Publish — n8n publish:workflow pins the version
  5. Restart — docker compose restart n8n
  6. Activate — waits for /healthz, then n8n update:workflow --active=true to re-register triggers

Testing

# Run all smoke tests (17 CRM + 5 admin tests)
./scripts/test-api.sh

# Verbose mode (show full response on failure)
./scripts/test-api.sh --verbose

# Filter by name
./scripts/test-api.sh --filter help
./scripts/test-api.sh --filter admin

The test suite covers: contacts (search, count, mutation blocking), todos (create, list, date filtering, update blocking, complete), help (3 variants), routing (none, data export), auth (unknown user rejection), and admin (provision, deactivate, lockout verification, reactivate).

Adding a new workflow module

1. Create the database table

Create init-scripts/<module>.sql with a provisioning function, following the pattern in contacts.sql or todos.sql:

CREATE OR REPLACE FUNCTION global_registry.provision_<module>_table(schema_name TEXT)
RETURNS void AS $$ ...

Apply to existing tenants:

docker compose exec -T postgres psql -U postgres_admin -d app_workspace < ./init-scripts/<module>.sql
docker compose exec -T postgres psql -U postgres_admin -d app_workspace -c "
  SELECT global_registry.provision_<module>_table(schema_name)
  FROM global_registry.tenants;
"

2. Register actions in module_registry

INSERT INTO global_registry.module_registry
  (workflow_name, workflow_id, action, params_schema)
VALUES
  ('<module>', '<WORKFLOW_ID>', 'search',
   '{"required":[{"name":"search_field","hint":"value to search for"}],"optional":[]}');

The params_schema drives both the LLM extraction prompt (master-router reads it dynamically) and the contract check validation.

3. Create the builder script

Create n8n/scripts/build-<module>-router.js. Required structure:

#!/usr/bin/env node
'use strict';
const { createBuilder, src } = require('./lib/workflow-builder');
const { node, conn, write, uid } = createBuilder();

const WORKFLOW_ID = '<unique_id>';
const PG_ROUTER = { postgres: { id: 'JAwz8BcESuFCZEfv', name: 'n8n router' } };

// Execute Workflow Trigger with standard inputs
node('Execute Workflow Trigger', 'n8n-nodes-base.executeWorkflowTrigger', 1.1, {
  workflowInputs: { values: [
    { name: 'mode' }, { name: 'tenant_schema' }, { name: 'chat_id' },
    { name: 'log_id' }, { name: 'processed_text' }, { name: 'proposed_params' },
    { name: 'pending_log_id' }, { name: 'callback_query_id' },
    { name: 'workflow' }, { name: 'action' }, { name: 'params' }, { name: 'channel' },
  ]},
}, [400, 400]);

// ... your nodes ...

// Final output node must produce: { output, action, routing_path, log_id, skip_finalize }
write('<WORKFLOW_ID>', '<module>-router', '/workflows/<module>-router.json');

4. Wire into deploy.sh

Add to WORKFLOW_MAP and TARGETS:

WORKFLOW_MAP=(
  ...
  "build-<module>-router.js:<WORKFLOW_ID>"
)
TARGETS=("..." "build-<module>-router.js")

5. Add to check-contracts.js

Add to BUILDER_MAP:

const BUILDER_MAP = {
  contacts: 'build-contact-router.js',
  todos:    'build-todo-router.js',
  data:     'build-data-router.js',
  '<module>': 'build-<module>-router.js',
};

6. Deploy and test

./n8n/scripts/deploy.sh build-<module>-router.js

The master-router picks up new modules automatically via the registry query — no changes needed in build-master-router.js.

HTTP API

# Search
curl -s -X POST http://localhost:5678/webhook/crm-api \
  -H 'Content-Type: application/json' \
  -d '{"user_id":"7363939976","text":"cherche Marie"}' | jq .

# Count
curl -s -X POST http://localhost:5678/webhook/crm-api \
  -H 'Content-Type: application/json' \
  -d '{"user_id":"7363939976","text":"combien de contacts"}' | jq .

# Help
curl -s -X POST http://localhost:5678/webhook/crm-api \
  -H 'Content-Type: application/json' \
  -d '{"user_id":"7363939976","text":"aide"}' | jq .

Mutations (2-step confirmation)

Mutations (update/delete) require a confirmation step on both HTTP and Telegram channels.

# Step 1: request the mutation → returns pending_id
RESP=$(curl -s -X POST http://localhost:5678/webhook/crm-api \
  -H 'Content-Type: application/json' \
  -d '{"user_id":"7363939976","text":"modifie le téléphone de Marie à 0611111111"}')
# → {"output":"Confirmez-vous...","pending_id":"uuid","action":"pending"}

# Step 2: confirm (or reject)
PENDING_ID=$(echo "$RESP" | jq -r .pending_id)
curl -s -X POST http://localhost:5678/webhook/crm-confirm \
  -H 'Content-Type: application/json' \
  -d "{\"user_id\":\"7363939976\",\"pending_id\":\"$PENDING_ID\",\"decision\":\"confirm\"}" | jq .
# → {"output":"Contact modifié...","action":"update"}

On Telegram, the same flow uses inline keyboard buttons instead of /crm-confirm.

Optional auth: set HTTP_API_KEY in .env to a long random string, then pass X-API-Key: <value> header.

Telegram

Bot setup

Bot: @yWorkflowTestBot — created via @BotFather.

HTTPS tunnel (required for webhook)

./scripts/refresh-tunnel.sh

Starts localtunnel, updates .env WEBHOOK_URL, re-registers the Telegram webhook with setWebhook. Safe to re-run when the bot stops responding.

Manual webhook registration

curl -s "https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/setWebhook" \
  -d "url=https://nasty-toys-swim.loca.lt/webhook/845bc70a-937e-42e5-a626-51d3efe365e0/webhook" \
  -d 'allowed_updates=["message","callback_query"]' \
  -d "secret_token=mAst3rIngestCRM01_218bae93-869c-4e18-9d22-73f4c96c8d28"

Check webhook status

curl -s "https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/getWebhookInfo" | jq .

Observability

Stack

Service Port Purpose
Grafana localhost:3000 Dashboards (login: admin / ADMIN_TOKEN)
Prometheus localhost:9090 Scrapes n8n /metrics every 15s
n8n /metrics localhost:5678/metrics Prometheus-format workflow metrics

Data sources in Grafana

Source What it provides
Postgres Business metrics: cost per hour, requests per module, error rate, generated SQL
Prometheus Infrastructure metrics: execution rate, average duration, failed executions, CPU usage

Dashboard (8 panels)

Exported to grafana/dashboards/n8n-dashboard.json. Import via Grafana UI: Dashboards → New → Import.

Panel Source Query
Cost per hour Postgres date_trunc('hour', logged_at) + token pricing
Requests per module Postgres date_trunc('hour', logged_at) grouped by module
Error rate Postgres COUNT(*) FILTER (WHERE status = 'error')
Execution by module Postgres Bar chart, DATE(logged_at) grouped by module
Execution rate Prometheus sum(rate(n8n_workflow_execution_duration_seconds_sum[$__rate_interval]))
Average duration Prometheus sum(rate(..._sum)) / sum(rate(..._count))
Failed executions Prometheus rate(..._count{status="failed"}[$__rate_interval])
CPU usage Prometheus rate(n8n_process_cpu_seconds_total[$__rate_interval])

execution_logs (job state table)

Every request gets a row at ingestion. Columns updated throughout the pipeline:

Column When Source
raw_input, channel, triggered_by Ingestion SQL: Log & Load Registry
router_model, prompt_tokens, total_tokens After classification SQL: Update Log Tokens
output, action, routing_path, completed_at After module execution SQL: Finalize Log
error_message, error_node On workflow error Error Handler → SQL: Log Error
generated_sql NL2SQL queries SQL: Log Generated SQL
status, proposed_params Mutation confirmation flow SQL: Stage Pending Action

Cost report

-- Cost per tenant per day (last 30 days)
SELECT * FROM global_registry.cost_report(30);

-- Total spend
SELECT tenant_id, SUM(estimated_cost_usd) FROM global_registry.cost_report(30) GROUP BY tenant_id;

Grafana SQL tips

Use Grafana time range variables to sync with the dashboard time picker:

WHERE logged_at BETWEEN $__timeFrom() AND $__timeTo()

Use date_trunc('hour', ...) for time series panels (not DATE() which pins to midnight UTC).

Dev inspection commands

Postgres

# Interactive psql prompt
docker exec -it app-postgres psql -U postgres_admin -d app_workspace

# List tenants
docker compose exec postgres psql -U postgres_admin -d app_workspace -c \
  "SELECT tenant_id, company_name, schema_name, is_active FROM global_registry.tenants;"

# List contacts for a tenant
docker compose exec postgres psql -U postgres_admin -d app_workspace -c \
  "SELECT id, first_name, last_name, phone, email, company_name FROM customer_acme_corp.contacts;"

# List todos for a tenant
docker compose exec postgres psql -U postgres_admin -d app_workspace -c \
  "SELECT id, subject, due_date, done, created_by FROM customer_acme_corp.todos;"

# Execution logs (recent, with new columns)
docker compose exec postgres psql -U postgres_admin -d app_workspace -c \
  "SELECT id, action, routing_path, router_model, output, error_message, generated_sql, completed_at
   FROM customer_acme_corp.execution_logs ORDER BY logged_at DESC LIMIT 10;"

# Module registry (what master-router dispatches)
docker compose exec postgres psql -U postgres_admin -d app_workspace -c \
  "SELECT workflow_name, action, workflow_id FROM global_registry.module_registry ORDER BY workflow_name, action;"

# User mappings
docker compose exec postgres psql -U postgres_admin -d app_workspace -c \
  "SELECT * FROM global_registry.user_mappings;"

# Verify least-privilege role (should fail)
docker compose exec postgres psql -U n8n_router -d app_workspace -c \
  "INSERT INTO global_registry.tenants (tenant_id, company_name, schema_name) VALUES ('x','x','x');"

n8n

# List active workflows
docker compose exec -T n8n n8n export:workflow --all 2>/dev/null | \
  python3 -c "import json,sys; [print(f\"{w['id']} | active={w.get('active',False)} | {w['name']}\") for w in json.load(sys.stdin)]"

# Check n8n health
docker compose exec -T n8n wget -q -O - http://localhost:5678/healthz

# n8n Prometheus metrics
curl -s http://localhost:5678/metrics | head -20

# n8n logs (live)
docker compose logs -f n8n

Ollama

# Test n8n → Ollama connectivity
docker compose exec -it n8n node -e \
  "fetch('http://host.docker.internal:11434').then(r => console.log('Status:', r.status)).catch(e => console.log('Error:', e))"

# List available models
docker compose exec -it n8n node -e \
  "fetch('http://host.docker.internal:11434/api/tags').then(r => r.json().then(j => j.models.forEach(m => console.log(m.name)))).catch(console.error)"

# Host setup (macOS, Metal GPU)
brew install --cask ollama
ollama pull qwen2.5:7b
launchctl setenv OLLAMA_HOST "0.0.0.0"
brew services start ollama

Whisper

# Health check
curl -s http://localhost:8000/health | jq .

# Test transcription (from host)
curl -s http://localhost:8000/v1/audio/transcriptions \
  -F "file=@test.ogg" -F "language=fr" | jq .text

Migrations

Apply a migration to all existing tenants:

docker compose exec -T postgres psql -U postgres_admin -d app_workspace < ./init-scripts/mig_<name>.sql

Security notes

  • Port binding: all services bound to 127.0.0.1 — not exposed to the internet
  • Tenant isolation: each tenant gets its own Postgres schema; n8n_router role has CRUD only within tenant schemas, SELECT-only on global_registry
  • Todo ownership: created_by column ensures users only see/modify their own todos within a shared tenant
  • NL2SQL sandboxing: generated SQL is validated (SELECT-only, single-statement, ownership filter required) before execution
  • Admin API: bearer token auth via ADMIN_TOKEN env var
  • HTTP API: mutations require 2-step confirmation (pending → confirm/reject); optional X-API-Key auth via HTTP_API_KEY env var
  • Traefik: reverse proxy routes only /webhook/* and /grafana/* — n8n UI blocked from external access. Self-signed TLS for dev, Let's Encrypt for production (DOMAIN in .env)

About

multi-tenant n8n workflow router with Cloud LLM (Mistral) transcription assistance for contacts, todos, with telemetry and job queueing

Topics

grafana-dashboardmistral-smallmulti-tenantn8nprometheus

Resources

Readme
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages