🧩 @trap_stevo/cynq

Automation Evolved.

The next-generation, event-driven, real-time CI/CD and pipeline orchestration engine.
Unifying pipelines, triggers, deployments, metrics, and real-time observability into a single, composable core — powering ultra-dynamic continuous integration and delivery workflows across any environment.

🚀 Features

• 🧠 Pipeline-as-Code – Define fully declarative, composable pipelines that run anywhere
• ⚡ Real-Time Engine – React instantly to events, webhooks, and remote triggers
• 🕹️ Dynamic Drivers – Register sources, runners, triggers, and deployers with simple interfaces
• 🌐 Cynq Route Engine – Secure HTTP layer for remote enqueues and external control
• 📡 Remote Execution Ready – Dispatch builds or deployments to other instances
• 🔁 Queue & Retry Logic – Resilient queue system with per-tenant isolation and idempotency
• 🔒 Security-First Architecture – HMAC / JWT auth, replay protection, allowlists, and TTL enforcement
• 📊 Integrated Metrics – Seamless telemetry for build time, deploy time, and success ratios
• 🧩 Vault Integration – Secret and credential management for pipelines
• 🛰️ Multi-Target Sync Loops – Orchestrate multiple pipelines in parallel
• 🧹 Graceful Shutdowns – Unified close() method to stop routes, triggers, and sync loops safely

⚙️ System Requirements

🧩 Built-In Drivers

Register a custom driver:

engine.registerDriver("runner", "python", (ctx) => ({
     exec : async ({ with : w }) => {
          const { spawn } = require("child_process");
          
          await new Promise((resolve) =>
               spawn("python", [w.script], { stdio : "inherit" })
               .on("exit", resolve)
          );
     }
}));

🧠 Driver Authoring

Cynq enables fully modular extensions through drivers — small, isolated units that define custom logic for different stages of the pipeline lifecycle.
Drivers transform how builds run, deploy, trigger, and source artifacts across environments.

⚙️ Driver Fundamentals

🧰 Shared Context Reference

🏃 Runner Drivers

Runner drivers execute actual work — builds, tests, packaging, or orchestration steps.

Example

cynq.registerDriver("runner", "shell-task", (ctx) => ({
     start : async ({ with : w }) => {
          const { exec } = require("child_process");
          if (!w?.cmd) { throw new Error("Missing cmd parameter"); }

          await new Promise((resolve, reject) => {
               exec(w.cmd, { cwd : w.cwd || process.cwd(), env : { ...process.env, ...w.env } }, (err, stdout, stderr) => {
                    if (err) reject(stderr || err);
                    else { ctx.logger?.(`stdout: ${stdout}`); resolve(); }
               });
          });
     }
}));

Common with Parameters

🚀 Deployer Drivers

Deployer drivers publish or distribute artifacts — pushing to registries, cloud storage, or remote systems.

Example

cynq.registerDriver("deployer", "artifact-uploader", (ctx) => ({
     activate : async ({ with : w }) => {
          const fs = require("fs");
          const axios = require("axios");

          if (!w?.url || !w?.filePath) {
               throw new Error("artifact-uploader requires url and filePath");
          }

          const data = fs.createReadStream(w.filePath);
          await axios.post(w.url, data, {
               headers : {
                    "Content-Type" : "application/octet-stream",
                    Authorization : w.auth || ""
               }
          });

          ctx.logger?.(`Uploaded ${w.filePath} to ${w.url}`);
     }
}));

🌐 Trigger Drivers

Trigger drivers listen for external or scheduled events and enqueue new jobs dynamically.

Example

cynq.registerDriver("trigger", "webhook", (ctx) => ({
     start : async ({ with : w }, enqueue) => {
          const express = require("express");
          const app = express();
          app.use(express.json());

          const route = w.path || "/hook";
          const port = w.port || 3100;

          app.post(route, async (req, res) => {
               await enqueue({ event : "webhook", payload : req.body });
               res.json({ ok : true });
          });

          app.listen(port, () => ctx.logger?.(`Webhook listening on :${port}${route}`));
     }
}));

📦 Source Drivers

Source drivers fetch repositories, packages, or other input materials before a build or deployment begins.

Example

cynq.registerDriver("source", "git-clone", (ctx) => ({
     fetch : async ({ with : w }) => {
          const { execSync } = require("child_process");
          if (!w?.repo) throw new Error("Missing repo URL");
          const dir = w.dir || `./workspace-${Date.now()}`;
          execSync(`git clone ${w.repo} ${dir}`, { stdio : "inherit" });
          ctx.logger?.(`Cloned ${w.repo} into ${dir}`);
          return dir;
     }
}));

🧩 Driver Safety Guidelines

🧠 Advanced Driver Patterns

1. Composable Drivers

Use drivers that delegate to other registered drivers internally.

cynq.registerDriver("runner", "composite", (ctx) => ({
     start : async ({ with : w }) => {
          for (const step of w.steps) {
               await ctx.enqueue(step.target, step.payload);
          }
     }
}));

2. Stateful Deployers

Maintain incremental state via ctx.storage.kvFacade().

const kv = ctx.storage.kvFacade("deploy", "artifact");
await kv.put("lastVersion", w.version);

🧩 Pipeline Spec Reference

The Pipeline Spec defines how a project builds, tests, deploys, and reacts to events.
Every spec describes sources, steps, environments, and follow-up actions under a unified JSON structure.

🧱 Top-Level Schema

Example

{
     "name": "backend-prod",
     "triggers": [
          { "driver": "webhook", "with": { "path": "/github", "port": 3100, "secret": "shared" } }
     ],
     "pipeline": {
          "source": { "driver": "git-clone", "with": { "repo": "https://github.com/org/app.git", "branch": "main" } },
          "steps": [
               { "kind": "run", "name": "build", "runner": "shell-task", "with": { "cmd": "npm ci && npm run build" } },
               { "kind": "deploy", "name": "publish", "deployer": "artifact-uploader", "with": { "url": "https://cdn.example.com/upload", "filePath": "./dist.zip" } }
          ],
          "on": {
               "success": [
                    { "kind": "deploy", "deployer": "cynq-enqueue", "with": { "project": "web", "target": "smoke-tests" } }
               ],
               "failure": [
                    { "kind": "run", "runner": "shell-task", "with": { "cmd": "bash scripts/rollback.sh" } }
               ]
          }
     },
     "env": {
          "NODE_ENV": "production",
          "REGION": "us-east-1"
     },
     "matrix": {
          "node": ["18", "20"],
          "region": ["us-east-1", "eu-west-1"]
     },
     "secrets": {
          "GITHUB_TOKEN": "vault:deploy.github"
     }
}

⚙️ triggers[]

Defines what initiates the pipeline.
Each trigger uses a registered driver and optional configuration.

Example

{ "driver": "webhook", "with": { "path": "/hook", "secret": "abc123" } }

🔗 pipeline.source

Defines how to retrieve or prepare the source materials before the build starts.

Example

{ "driver": "git-clone", "with": { "repo": "https://github.com/org/app.git", "branch": "main" } }

🧩 pipeline.steps[]

Describes ordered tasks inside the pipeline.
Each step specifies what to run, deploy, or trigger next.

Example

{ "kind": "run", "name": "build", "runner": "shell-task", "with": { "cmd": "npm run build" } }

🔁 pipeline.on

Defines follow-up actions based on the result of the main pipeline execution.

Example

"on": {
     "success": [
          { "kind": "deploy", "deployer": "cynq-enqueue", "with": { "project": "web", "target": "smoke-tests" } }
     ],
     "failure": [
          { "kind": "run", "runner": "shell-task", "with": { "cmd": "bash rollback.sh" } }
     ]
}

🌍 env

Defines environment variables that apply globally to all steps.
Can be overridden per step via its own with.env.

Example

"env": {
     "NODE_ENV": "production",
     "LOG_LEVEL": "debug"
}

🧮 matrix

Generates multiple parallel pipeline runs for parameter combinations.
Each key defines an axis with possible values.

Example

"matrix": {
     "node": ["18", "20"],
     "region": ["us-east-1", "eu-west-1"]
}

This expands into four runs: (18, us-east-1), (18, eu-west-1), (20, us-east-1), (20, eu-west-1).

🔐 secrets

Lists vault references used in the pipeline.
Each secret resolves securely at runtime through the configured vault backend.

Example

"secrets": {
     "GITHUB_TOKEN": "vault:deploy.github",
     "DOCKER_PASSWORD": "vault:docker.pass"
}

When the pipeline runs, these values resolve securely at runtime.

🧠 Execution Flow Summary

1️⃣ Trigger fires (webhook, manual, cron, etc.) 2️⃣ Source driver fetches code or assets 3️⃣ Steps execute sequentially or in matrix form 4️⃣ Environment and secrets inject automatically 5️⃣ On success/failure handlers run 6️⃣ Metrics, logs, and states persist securely through the configured storage backend

Every source, step, and secret becomes structured, reproducible, and composable.

🧭 Cynq Core Methods

🧹 Graceful Shutdown

process.on("SIGINT", async () => {
     await engine.close();
     
     process.exit(0);
});

Stops all routes, trigger listeners, and active synchronization loops cleanly.

🌐 Cynq Route Engine

Overview

Cynq Route Engine exposes a minimal, secure HTTP interface for remote interaction and job enqueueing.

Configuration Example

engine : {
     routes : {
          enabled : true,
          autoStart : true,
          port : 3333,
          hmacSecret : "super-secret",
          ipAllowlist : ["10.0.0.0/24"],
          maxBytes : "5mb",
          rate : { capacity : 100, refillPerSec : 5 }
     }
}

Security Checklist

✅ HMAC or JWT authentication
✅ Replay protection (X-Timestamp, nonce, short TTL)
✅ IP allowlist and CIDR support
✅ Content-type and size limits
✅ Rate limiting and quotas
✅ Idempotency via Idempotency-Key header

🔄 Remote Communication

Remote Enqueue Example

{
     "steps": [
          {
               "kind": "deploy",
               "name": "trigger-next",
               "deployer": "remote-cynq-enqueue",
               "with": {
                    "url": "https://cynq-node-b.example.com",
                    "project": "frontend",
                    "target": "production",
                    "auth": "Bearer xyz123",
                    "payload": { "version": "1.2.0" }
               }
          }
     ]
}

🛰️ Cynq Event Reference

Cynq emits structured events for real-time dashboards, external integrations, or telemetry pipelines.
Events follow a consistent envelope:

{
     "event": "cynq:step:ok",
     "project": "frontend",
     "target": "production",
     "timestamp": 1734819100000,
     "data": { /* event-specific payload */ }
}

🔔 Core Lifecycle Events

⚙️ Step-Level Events

🚀 Deployment Events

🧩 Diagnostic & Audit Events

📊 Metrics Integration

All metric events flow through the same channel.
Metrics describe runtime timing, counts, and performance distribution across pipelines.

Example Metric Stream

{
     "event": "metric",
     "metric": "step.time.ms",
     "labels": { "project": "frontend", "step": "build" },
     "value": 5234
}

🧠 Realtime Consumption

Consume events from the runtime emitter or integrate IoTide for networked streams:

engine.on("event", (evt, data) => {
     console.log("[event]", evt, data);
});

// or with IoTide
engine.realtime.emit("cynq:step:ok", { step: "build", durationMs: 5123 });

All events preserve deterministic naming (cynq:*) for consistent filtering across observability systems such as Grafana, Prometheus, or custom dashboards.

🪝 Realtime Hook Patterns

Cynq exposes an event stream for live monitoring, logging, and notification systems.
Hooks can run inline, forward to third-party services, or store in custom observability backends.

🔧 Basic Listener

Attach directly to the realtime emitter to react to all events.

const { Cynq } = require("@trap_stevo/cynq");

const engine = new Cynq({
     realtime : { emit : (evt, data) => console.log("[event]", evt, data) }
});

For targeted subscriptions, filter by prefix:

engine.on("cynq:step:ok", (data) => {
     console.log(`[ok] ${data.step} → ${data.durationMs}ms`);
});

🧠 Scoped Hook Helpers

Define scoped handlers to simplify observability integration.

engine.onPipelineStart = (fn) => engine.on("cynq:pipeline:start", fn);
engine.onPipelineEnd = (fn) => engine.on("cynq:pipeline:ok", fn);
engine.onPipelineFail = (fn) => engine.on("cynq:pipeline:fail", fn);

engine.onStepStart = (fn) => engine.on("cynq:step:start", fn);
engine.onStepEnd = (fn) => engine.on("cynq:step:ok", fn);
engine.onStepFail = (fn) => engine.on("cynq:step:fail", fn);

engine.onDeployStart = (fn) => engine.on("cynq:deploy:start", fn);
engine.onDeployEnd = (fn) => engine.on("cynq:deploy:ok", fn);
engine.onDeployFail = (fn) => engine.on("cynq:deploy:fail", fn);

Example

engine.onStepEnd((data) => {
     console.log(`✅ Step "${data.step}" completed in ${data.durationMs}ms`);
});

💬 Slack and Discord Hooks

Integrate with messaging platforms by posting from realtime events.

const axios = require("axios");

engine.on("cynq:pipeline:ok", async (data) => {
     await axios.post(process.env.SLACK_WEBHOOK_URL, {
          text : `✅ Pipeline "${data.name}" succeeded in ${data.durationMs}ms`
     });
});

engine.on("cynq:pipeline:fail", async (data) => {
     await axios.post(process.env.DISCORD_WEBHOOK_URL, {
          content : `❌ Pipeline "${data.name}" failed — reason: ${data.reason}`
     });
});

📊 Custom Telemetry Collector

Aggregate step times or pipeline durations for dashboards.

const metrics = [];

engine.on("cynq:step:ok", (data) => {
     metrics.push({
          step : data.step,
          duration : data.durationMs,
          ts : Date.now()
     });
});

Send data periodically:

setInterval(() => {
     if (metrics.length === 0) { return; }
     console.table(metrics);
     metrics.length = 0;
}, 5000);

🧩 Chained Reactivity

Forward events into another Cynq instance or any remote listener.

engine.on("cynq:deploy:ok", async (data) => {
     await axios.post("https://remote-node.example.com/enqueue", {
          project : "mirror",
          target : "sync",
          payload : data
     }, { headers : { Authorization : "Bearer xyz" } });
});

⚙️ Pattern Reference

🛰️ Example: Unified Realtime Dashboard Feed

engine.on("*", (evt, data) => {
     console.log(`[${new Date().toISOString()}] ${evt}`, data);
});

Combined with IoTide, events propagate instantly between distributed nodes for visualization, alerting, and streaming dashboards.

🔐 Production Tip

Always sanitize payloads before broadcasting externally.
Avoid exposing secret paths, credentials, or internal error stacks in event handlers.
Forward only relevant metadata for telemetry or logs.

💡 Example: End-to-End Flow

1️⃣ Push event triggers webhook
2️⃣ Job enters queue and runs build
3️⃣ Deploy step copies files to target path
4️⃣ Remote instance receives follow-up trigger
5️⃣ Metrics update in real time
6️⃣ Route engine awaits next event

📡 Distributed Mesh Mode

Cynq operates in mesh topology when IoTide connects multiple instances together.
Each node participates in a shared event fabric that synchronizes jobs, telemetry, and triggers across regions or data centers.

🧩 Mesh Overview

All nodes authenticate using IoTide peer configuration and auto-discover each other through shared topics.

⚙️ Configuration Example

const { Cynq } = require("@trap_stevo/cynq");

const IoTide   = require("@trap_stevo/iotide");

// Launch IoTide on each node
const tide = new IoTide(
     3101,
     { useCors : true, useHTTPS : false, tidalCoreOptions : { namespace : "cynq-mesh" } },
     true,
     (socket) =>  {
          console.log("[mesh] node connected:", socket.tideID);

          // Automatically join the shared mesh channel
          tide.joinChannel(socket.tideID, { roomName : "cynq-mesh", userID : socket.tideID });
     }
);

const engine = new Cynq({
     realtime : {
          emit : (event, data) => tide.emitOnChannel("cynq-mesh", event, data)
     },
     logger : (msg) => console.log("[mesh]", msg)
});

await engine.sync("backend", "prod", spec);

Each Cynq instance emits and listens on the same IoTide channel.
When one node triggers a job, others react instantly through IoTide propagation.

🔄 Mesh Synchronization Events

🧠 Use Cases

🔐 Security & Isolation

✅ Peer authentication via signed tokens
✅ Namespace-scoped topic isolation
✅ HMAC validation for message integrity
✅ Configurable mesh-level ACLs
✅ Built-in replay and tamper protection

🔁 Example: Mirrored Event Relay

tide.on("cynq:deploy:ok", (data) =>  {
     console.log("[replica] received deploy confirmation", data);
     // Optionally enqueue follow-up job on another node
});

🛰️ Combined Architecture

┌────────────────────────────┐
│        Cynq Node A         │
│  • Receives webhook        │
│  • Runs build & deploy     │
│  • Emits events via IoTide │
└───────────▲────────────────┘
            │
        Mesh Fabric (IoTide)
            │
┌───────────▼────────────────┐
│        Cynq Node B         │
│  • Receives event mirror   │
│  • Performs post-deploy    │
│  • Sends metrics upstream  │
└────────────────────────────┘

🧩 Cluster-Wide Behavior

• Shared event namespace (cynq:*)
• Job delegation through mesh forwarding
• Metrics and locks synchronized across nodes
• Vault data kept local per node (no secret broadcast)
• RouteEngine remains independent per instance

⚡ Example: Cross-Node Job Flow

1️⃣ Node A receives webhook → triggers build
2️⃣ Build completes → emits cynq:deploy:ok
3️⃣ Node B listens → forwards remote-cynq-enqueue job
4️⃣ Node B deploys to its environment
5️⃣ Cluster updates metrics in real time

🛡️ Production Tip

Keep mesh namespaces separate for staging vs production.
Always rotate mesh authentication tokens periodically and monitor node join events.

🌍 Outcome

Distributed Mesh Mode turns independent Cynq instances into a single cooperative CI/CD fabric — enabling horizontally scalable pipelines, real-time telemetry, and autonomous recovery across global environments.

🔐 Secret Management

const secrets = engine.storage.secretsFacade("project", "target", { tenantId : "demo" });

await secrets.put("deploy.token", "ghp_ABC123");
const token = await secrets.get("deploy.token");

Use vault references inside pipeline specs:

{
     "with": {
          "repo": "https://github.com/org/repo.git",
          "token": "vault:deploy.token"
     }
}

🛡️ Security Principles

• Immutable deployment logs
• HMAC-protected enqueue endpoints
• Optional JWT-based authentication
• Replay-safe timestamp verification
• Strict content-type enforcement
• Tenant and project isolation

📦 Installation

npm install @trap_stevo/cynq

⚡ Quick Start

Minimal Example

const { Cynq } = require("@trap_stevo/cynq");

const engine = new Cynq({
     engine : {
          routes : { enabled : true, autoStart : true, port : 3333 }
     }
});

const spec = {
     name : "web-deploy",
     pipeline : {
          source : { driver : "git", with : { repo : "https://github.com/user/app" } },
          steps : [
               { kind : "run", name : "build", runner : "shell", with : { script : "npm run build" } },
               { kind : "deploy", name : "copy", deployer : "copy-folder", with : { from : "./dist", to : "/srv/app" } }
          ]
     }
};

await engine.deploy("myProject", "production", spec, { clientAuth : { token : "xyz" } });

📜 License

See License in LICENSE.md

⚡ Automation Evolved.

From commits to clouds, from triggers to telemetry — one intelligent engine unites it all.

Automate intelligently. Deploy infinitely.