Compare commits

..

30 Commits

Author SHA1 Message Date
e7707d7edb Add abort execution feature for stuck/running processes 2026-01-08 22:11:59 -05:00
ea7a2d82e6 Fix execution details endpoint - remove reference to deleted activeExecutions 2026-01-08 22:06:25 -05:00
8224f3b6a4 Add missing helper functions for workflow execution (addExecutionLog, updateExecutionStatus) 2026-01-08 22:03:00 -05:00
9b31b7619f Add WebSocket error handling with stack traces
Wrapped ws.onmessage in try-catch to capture full stack trace
when errors occur during message handling. This will help identify
where the 'Cannot read properties of undefined' error is coming from.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 23:33:07 -05:00
06eb2d2593 Add detailed logging to workflow creation endpoint
This will help diagnose the 'Cannot read properties of undefined' error
by logging each step of the workflow creation process.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 23:30:50 -05:00
aaeb59a8e2 Improve error handling in workflow creation and data loading
- Separated JSON validation from API call error handling
- Changed refreshData() to async with individual try-catch blocks
- Better error messages: "Invalid JSON" vs "Error creating workflow"
- Console.error logging for each data loading function
- Changed success alert to terminal notification
- This will help identify which specific function is failing

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 23:27:54 -05:00
b85bd58c4b Fix: Remove duplicate old workflow execution code
Removed old/obsolete workflow execution system that was conflicting
with the new executeWorkflowSteps() engine:

Removed:
- activeExecutions Map (old tracking system)
- executeWorkflow() - old workflow executor
- executeNextStep() - old step processor
- executeCommandStep() - old command executor (duplicate)
- handleUserInput() - unimplemented prompt handler
- Duplicate app.post('/api/executions') endpoint
- app.post('/api/executions/:id/respond') endpoint

This was causing "Cannot read properties of undefined (reading 'target')"
error because the old code was being called instead of the new engine.

The new executeWorkflowSteps() engine is now the only workflow system.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 23:24:42 -05:00
018752813a Implement Complete Workflow Execution Engine
Added full workflow execution engine that actually runs workflow steps:

Server-Side (server.js):
- executeWorkflowSteps() - Main workflow orchestration function
- executeCommandStep() - Executes commands on target workers
- waitForCommandResult() - Polls for command completion
- Support for step types: execute, wait, prompt (prompt skipped for now)
- Sequential step execution with failure handling
- Worker targeting: "all" or specific worker IDs/names
- Automatic status updates (running -> completed/failed)
- Real-time WebSocket broadcasts for step progress
- Command result tracking with command_id for workflows
- Only updates status for non-workflow quick commands

Client-Side (index.html):
- Enhanced formatLogEntry() with workflow-specific log types
- step_started - Shows step number and name with amber color
- step_completed - Shows completion with green checkmark
- waiting - Displays wait duration
- no_workers - Error when no workers available
- worker_offline - Warning for offline workers
- workflow_error - Critical workflow errors
- Better visual feedback for workflow progress

Workflow Definition Format:
{
  "steps": [
    {
      "name": "Step Name",
      "type": "execute",
      "targets": ["all"] or ["worker-name"],
      "command": "your command here"
    },
    {
      "type": "wait",
      "duration": 5
    }
  ]
}

Features:
- Executes steps sequentially
- Stops on first failure
- Supports multiple workers per step
- Real-time progress updates
- Comprehensive logging
- Terminal-themed workflow logs

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 23:19:12 -05:00
4511fac486 Phase 10: Command Scheduler
Added comprehensive command scheduling system:

Backend:
- New scheduled_commands database table
- Scheduler processor runs every minute
- Support for three schedule types: interval, hourly, daily
- calculateNextRun() function for intelligent scheduling
- API endpoints: GET, POST, PUT (toggle), DELETE
- Executions automatically created and tracked
- Enable/disable schedules without deleting

Frontend:
- New Scheduler tab in navigation
- Create Schedule modal with worker selection
- Dynamic schedule input based on type
- Schedule list showing status, next/last run times
- Enable/Disable toggle for each schedule
- Delete schedule functionality
- Terminal-themed scheduler UI
- Integration with existing worker and execution systems

Schedule Types:
- Interval: Every X minutes (e.g., 30 for every 30 min)
- Hourly: Every X hours (e.g., 2 for every 2 hours)
- Daily: At specific time (e.g., 03:00 for 3 AM daily)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 23:13:27 -05:00
4e17fdbf8c Phase 9: Execution Diff View
Added powerful execution comparison and diff view:

- Compare Mode toggle button in executions tab
- Multi-select up to 5 executions for comparison
- Visual selection indicators with checkmarks
- Comparison modal with summary table (status, duration, timestamps)
- Side-by-side output view for all selected executions
- Line-by-line diff analysis for 2-execution comparisons
- Highlights identical vs. different lines
- Shows identical/different line counts
- Color-coded diff (green for exec 1, amber for exec 2)
- Perfect for comparing same command across workers
- Terminal-themed comparison UI

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 23:08:55 -05:00
5c41afed85 Phase 8: Execution Search & Filtering
Added comprehensive search and filtering for execution history:

- Search bar to filter by command text, execution ID, or workflow name
- Status filter dropdown (All, Running, Completed, Failed, Waiting)
- Real-time client-side filtering as user types
- Filter statistics showing X of Y executions
- Clear Filters button to reset all filters
- Extracts command text from logs for quick command searches
- Maintains all executions in memory for instant filtering
- Terminal-themed filter UI matching existing aesthetic

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 23:06:43 -05:00
4baecc54d3 Phase 7: Multi-Worker Command Execution
Added ability to execute commands on multiple workers simultaneously:

- Added execution mode selector (Single/Multiple Workers)
- Multi-worker mode with checkbox list for worker selection
- Helper buttons: Select All, Online Only, Clear All
- Sequential execution across selected workers
- Results summary showing success/fail count per worker
- Updated command history to track multi-worker executions
- Terminal beep feedback based on overall success/failure
- Maintained backward compatibility with single worker mode

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 23:03:45 -05:00
661c83a578 Fix clearCompletedExecutions for new pagination API format
Changes:
- Handle new pagination response format (data.executions vs data)
- Request up to 1000 executions to ensure all are checked
- Track successful deletions count
- Use terminal notification instead of alert
- Better error handling for individual delete failures

Fixes regression from Phase 5 pagination changes.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:55:13 -05:00
c6e3e5704e Phase 6: Terminal aesthetic refinements and notifications
Changes:
- Added blinking terminal cursor animation
- Smooth hover effects for execution/worker/workflow items
- Hover animation: background highlight + border expand + slide
- Loading pulse animation for loading states
- Slide-in animation for log entries
- Terminal beep sound using Web Audio API (different tones for success/error)
- Real-time terminal notifications for command completion
- Toast-style notifications with green glow effects
- Auto-dismiss after 3 seconds with fade-out
- Visual and audio feedback for user actions

Sound features:
- 800Hz tone for success (higher pitch)
- 200Hz tone for errors (lower pitch)
- 440Hz tone for info (standard A note)
- 100ms duration, exponential fade-out
- Graceful fallback if Web Audio API not supported

Notification features:
- Fixed position top-right
- Terminal-themed styling with glow
- Color-coded: green for success, red for errors
- Icons: ✓ success, ✗ error, ℹ info
- Smooth animations (slide-in, fade-out)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:52:51 -05:00
9f972182b2 Phase 5: Auto-cleanup and pagination for executions
Changes:
Server-side:
- Added automatic cleanup of old executions (runs daily)
- Configurable retention period via EXECUTION_RETENTION_DAYS env var (default: 30 days)
- Cleanup runs on server startup and every 24 hours
- Only cleans completed/failed executions, keeps running ones
- Added pagination support to /api/executions endpoint
- Returns total count, limit, offset, and hasMore flag

Client-side:
- Implemented "Load More" button for execution pagination
- Loads 50 executions at a time
- Appends additional executions when "Load More" clicked
- Shows total execution count info
- Backward compatible with old API format

Benefits:
- Automatic database maintenance
- Prevents execution table from growing indefinitely
- Better performance with large execution histories
- User can browse all executions via pagination
- Configurable retention policy per deployment

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:50:39 -05:00
fff50f19da Phase 4: Execution detail enhancements with re-run and download
Changes:
- Added "Re-run Command" button to execution details modal
- Added "Download Logs" button to export execution data as JSON
- Re-run automatically switches to Quick Command tab and pre-fills form
- Download includes all execution metadata and logs
- Buttons only show for applicable execution types
- Terminal-themed button styling

Features:
- Re-run: Quickly repeat a previous command on same worker
- Download: Export execution logs for auditing/debugging
- JSON format includes: execution_id, status, timestamps, logs
- Filename includes execution ID and date for easy organization

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:49:20 -05:00
8152a827e6 Phase 3: Quick command enhancements with templates and history
Changes:
- Added command templates modal with 12 common system commands
- Added command history tracking (stored in localStorage)
- History saves last 50 commands with timestamp and worker name
- Template categories: system info, disk/memory, network, Docker, logs
- Click templates to auto-fill command field
- Click history items to reuse previous commands
- Terminal-themed modals with green/amber styling
- History persists across browser sessions

Templates included:
- System: uname, uptime, CPU info, processes
- Resources: df -h, free -h, memory usage
- Network: ip addr, active connections
- Docker: container list
- Logs: syslog tail, who is logged in, last logins

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:45:40 -05:00
bc3524e163 Phase 2: Enhanced worker status display with metadata
Changes:
- Show worker system metrics in dashboard and worker list
- Display CPU cores, memory usage, load average, uptime
- Added formatBytes() to display memory in human-readable format
- Added formatUptime() to show uptime as days/hours/minutes
- Added getTimeAgo() to show relative last-seen time
- Improved worker list with detailed metadata panel
- Show active tasks vs max concurrent tasks
- Terminal-themed styling for metadata display
- Amber labels for metadata fields

Benefits:
- See worker health at a glance
- Monitor resource usage (CPU, RAM, load)
- Track worker activity (active tasks)
- Better operational visibility

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:43:13 -05:00
f8ec651e73 Phase 1: Improve log display formatting
Changes:
- Added formatLogEntry() function to parse and format log entries
- Replaced raw JSON display with readable formatted logs
- Added specific formatting for command_sent and command_result logs
- Show timestamp, status, duration, stdout/stderr in organized layout
- Color-coded success (green) and failure (red) states
- Added scrollable output sections with max-height
- Syntax highlighting for command code blocks
- Terminal-themed styling with green/amber colors

Benefits:
- Much easier to read execution logs
- Clear visual distinction between sent/result logs
- Professional terminal aesthetic maintained
- Better UX for debugging command execution

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:41:29 -05:00
7656f4a151 Add execution cleanup functionality
Changes:
- Added DELETE /api/executions/:id endpoint
- Added "Clear Completed" button to Executions tab
- Deletes all completed and failed executions
- Broadcasts execution_deleted event to update all clients
- Shows count of deleted executions

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:36:51 -05:00
e13fe9d22f Fix worker ID mapping - use database ID for command routing
Problem:
- Workers generate random UUID on startup (runtime ID)
- Database stores workers with persistent IDs (database ID)
- UI sends commands using database ID
- Server couldn't find worker connection (stored by runtime ID)
- Result: 400 Bad Request "Worker not connected"

Solution:
- When worker connects, look up database ID by worker name
- Store WebSocket connection in Map using BOTH IDs:
  * Runtime ID (from worker_connect message)
  * Database ID (from database lookup by name)
- Commands from UI use database ID → finds correct WebSocket
- Cleanup both IDs when worker disconnects

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:29:23 -05:00
8bda9672d6 Fix worker command execution and execution status updates
Changes:
- Removed duplicate /api/executions/:id endpoint that didn't parse logs
- Added workers Map to track worker_id -> WebSocket connection
- Store worker connections when they send worker_connect message
- Send commands to specific worker instead of broadcasting to all clients
- Clean up workers Map when worker disconnects
- Update execution status to completed/failed when command results arrive
- Add proper error handling when worker is not connected

Fixes:
- execution.logs.forEach is not a function (logs now properly parsed)
- Commands stuck in "running" status (now update to completed/failed)
- Commands not reaching workers (now sent to specific worker WebSocket)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:23:02 -05:00
4974730dc8 Remove database migrations after direct schema fixes
Changes:
- Removed all migration code from server.js
- Database schema fixed directly via MySQL:
  * Dropped users.role column (SSO only)
  * Dropped users.password column (SSO only)
  * Added executions.started_by column
  * Added workflows.created_by column
  * All tables now match expected schema
- Server startup will be faster without migrations

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:11:07 -05:00
df581e85a8 Remove password column from users table
Changes:
- Drop password column from users table (SSO authentication only)
- PULSE uses Authelia SSO, not password-based authentication
- Fixes 500 error: Field 'password' doesn't have a default value

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 22:01:58 -05:00
e4574627f1 Add last_login column to users table migration
Changes:
- Add last_login TIMESTAMP column to existing users table
- Complete the users table migration with all required columns
- Fixes 500 error: Unknown column 'last_login' in 'INSERT INTO'

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 20:33:04 -05:00
1d994dc8d6 Add migration to update users table schema
Changes:
- Add display_name, email, and groups columns to existing users table
- Handle MariaDB lack of IF NOT EXISTS in ALTER TABLE
- Gracefully skip columns that already exist
- Fixes 500 error when authenticating users

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 20:28:47 -05:00
02ed71e3e0 Allow NULL workflow_id in executions table for quick commands
Changes:
- Modified executions table schema to allow NULL workflow_id
- Removed foreign key constraint that prevented NULL values
- Added migration to update existing table structure
- Quick commands can now be stored without a workflow reference

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 20:27:02 -05:00
cff058818e Fix quick command executions not appearing in execution tab
Changes:
- Create execution record in database when quick command is sent
- Store initial log entry with command details
- Broadcast execution_started event to update UI
- Display quick commands as "[Quick Command]" in execution list
- Fix worker communication to properly track all executions

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-07 20:24:11 -05:00
05c304f2ed Updated websocket handler 2026-01-07 20:20:18 -05:00
20cff59cee updates aesthetic 2026-01-07 20:12:16 -05:00
4 changed files with 2043 additions and 1941 deletions

1372
Claude.md

File diff suppressed because it is too large Load Diff

477
README.md
View File

@@ -1,240 +1,375 @@
# PULSE - Pipelined Unified Logic & Server Engine
A distributed workflow orchestration platform for managing and executing complex multi-step operations across server clusters through an intuitive web interface.
A distributed workflow orchestration platform for managing and executing complex multi-step operations across server clusters through a retro terminal-themed web interface.
## Overview
PULSE is a centralized workflow execution system designed to orchestrate operations across distributed infrastructure. It provides a powerful web-based interface for defining, managing, and executing workflows that can span multiple servers, require human interaction, and perform complex automation tasks at scale.
PULSE is a centralized workflow execution system designed to orchestrate operations across distributed infrastructure. It provides a powerful web-based interface with a vintage CRT terminal aesthetic for defining, managing, and executing workflows that can span multiple servers, require human interaction, and perform complex automation tasks at scale.
### Key Features
- **Interactive Workflow Management**: Define and execute multi-step workflows with conditional logic, user prompts, and decision points
- **Distributed Execution**: Run commands and scripts across multiple worker nodes simultaneously
- **High Availability Architecture**: Deploy redundant worker nodes in LXC containers with Ceph storage for fault tolerance
- **Web-Based Control Center**: Intuitive interface for workflow selection, monitoring, and interactive input
- **Flexible Worker Pool**: Scale horizontally by adding worker nodes as needed
- **Real-Time Monitoring**: Track workflow progress, view logs, and receive notifications
- **🎨 Retro Terminal Interface**: Phosphor green CRT-style interface with scanlines, glow effects, and ASCII art
- **⚡ Quick Command Execution**: Instantly execute commands on any worker with built-in templates and command history
- **📊 Real-Time Worker Monitoring**: Live system metrics including CPU, memory, load average, and active tasks
- **🔄 Interactive Workflow Management**: Define and execute multi-step workflows with conditional logic and user prompts
- **🌐 Distributed Execution**: Run commands across multiple worker nodes simultaneously via WebSocket
- **📈 Execution Tracking**: Comprehensive logging with formatted output, re-run capabilities, and JSON export
- **🔐 SSO Authentication**: Seamless integration with Authelia for enterprise authentication
- **🧹 Auto-Cleanup**: Automatic removal of old executions with configurable retention policies
- **🔔 Terminal Notifications**: Audio beeps and visual toasts for command completion events
## Architecture
PULSE consists of two core components:
### PULSE Server
**Location:** `10.10.10.65` (LXC Container ID: 122)
**Directory:** `/opt/pulse-server`
The central orchestration hub that:
- Hosts the web interface for workflow management
- Hosts the retro terminal web interface
- Manages workflow definitions and execution state
- Coordinates task distribution to worker nodes
- Handles user interactions and input collection
- Coordinates task distribution to worker nodes via WebSocket
- Handles user interactions through Authelia SSO
- Provides real-time status updates and logging
- Stores all data in MariaDB database
**Technology Stack:**
- Node.js 20.x
- Express.js (web framework)
- WebSocket (ws package) for real-time bidirectional communication
- MySQL2 (MariaDB driver)
- Authelia SSO integration
### PULSE Worker
**Example:** `10.10.10.151` (LXC Container ID: 153, hostname: pulse-worker-01)
**Directory:** `/opt/pulse-worker`
Lightweight execution agents that:
- Connect to the PULSE server and await task assignments
- Execute commands, scripts, and code on target infrastructure
- Report execution status and results back to the server
- Support multiple concurrent workflow executions
- Automatically reconnect and resume on failure
- Connect to PULSE server via WebSocket with heartbeat monitoring
- Execute shell commands and report results in real-time
- Provide system metrics (CPU, memory, load, uptime)
- Support concurrent task execution with configurable limits
- Automatically reconnect on connection loss
**Technology Stack:**
- Node.js 20.x
- WebSocket client
- Child process execution
- System metrics collection
```
┌─────────────────────┐
│ PULSE Server
(Web Interface)
────────────────────
┌──────┴───────┬──────────────┬──────────────┐
│ │ │
───────┐ ┌───────┐ ┌───────┐ ┌───────
│ Worker │ │ Worker │ │ Worker │ │ Worker │
│ Node 1 │ │ Node 2 │ │ Node 3 │ │ Node N │
└────────┘ └────────┘ └────────┘ └────────┘
LXC Containers in Proxmox with Ceph
┌─────────────────────────────────
│ PULSE Server (10.10.10.65)
Terminal Web Interface + API
│ ┌───────────┐ ┌──────────┐ │
│ MariaDB │ │ Authelia │
│ Database │ │ SSO │ │
│ └───────────┘ └──────────┘
────────────────────────────────
│ WebSocket
┌────────┴────────┬───────────┐
│ │ │
┌───▼────────┐ ┌───▼────┐ ┌──▼─────┐
│ Worker 1 │ │Worker 2│ │Worker N│
│10.10.10.151│ │ ... │ │ ... │
└────────────┘ └────────┘ └────────┘
LXC Containers in Proxmox with Ceph
```
## Deployment
## Installation
### Prerequisites
- **Proxmox VE Cluster**: Hypervisor environment for container deployment
- **Ceph Storage**: Distributed storage backend for high availability
- **LXC Support**: Container runtime for worker node deployment
- **Network Connectivity**: Communication between server and workers
- **Node.js 20.x** or higher
- **MariaDB 10.x** or higher
- **Authelia** configured for SSO (optional but recommended)
- **Network Connectivity** between server and workers
### Installation
### PULSE Server Setup
#### PULSE Server
```bash
# Clone the repository
git clone https://github.com/yourusername/pulse.git
cd pulse
# Clone repository
cd /opt
git clone <your-repo-url> pulse-server
cd pulse-server
# Install dependencies
npm install # or pip install -r requirements.txt
npm install
# Configure server settings
cp config.example.yml config.yml
nano config.yml
# Create .env file with configuration
cat > .env << EOF
# Server Configuration
PORT=8080
SECRET_KEY=your-secret-key-here
# Start the PULSE server
npm start # or python server.py
# MariaDB Configuration
DB_HOST=10.10.10.50
DB_PORT=3306
DB_NAME=pulse
DB_USER=pulse_user
DB_PASSWORD=your-db-password
# Worker API Key (for worker authentication)
WORKER_API_KEY=your-worker-api-key
# Auto-cleanup configuration (optional)
EXECUTION_RETENTION_DAYS=30
EOF
# Create systemd service
cat > /etc/systemd/system/pulse.service << EOF
[Unit]
Description=PULSE Workflow Orchestration Server
After=network.target
[Service]
Type=simple
User=root
WorkingDirectory=/opt/pulse-server
ExecStart=/usr/bin/node server.js
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
EOF
# Start service
systemctl daemon-reload
systemctl enable pulse.service
systemctl start pulse.service
```
#### PULSE Worker
### PULSE Worker Setup
```bash
# On each worker node (LXC container)
# On each worker node
cd /opt
git clone <your-repo-url> pulse-worker
cd pulse-worker
# Install dependencies
npm install # or pip install -r requirements.txt
npm install
# Configure worker connection
cp worker-config.example.yml worker-config.yml
nano worker-config.yml
# Create .env file
cat > .env << EOF
# Worker Configuration
WORKER_NAME=pulse-worker-01
PULSE_SERVER=http://10.10.10.65:8080
PULSE_WS=ws://10.10.10.65:8080
WORKER_API_KEY=your-worker-api-key
# Start the worker daemon
npm start # or python worker.py
```
# Performance Settings
HEARTBEAT_INTERVAL=30
MAX_CONCURRENT_TASKS=5
EOF
### High Availability Setup
# Create systemd service
cat > /etc/systemd/system/pulse-worker.service << EOF
[Unit]
Description=PULSE Worker Node
After=network.target
Deploy multiple worker nodes across Proxmox hosts:
```bash
# Create LXC template
pct create 1000 local:vztmpl/ubuntu-22.04-standard_amd64.tar.zst \
--rootfs ceph-pool:8 \
--memory 2048 \
--cores 2 \
--net0 name=eth0,bridge=vmbr0,ip=dhcp
[Service]
Type=simple
User=root
WorkingDirectory=/opt/pulse-worker
ExecStart=/usr/bin/node worker.js
Restart=always
RestartSec=10
# Clone for additional workers
pct clone 1000 1001 --full --storage ceph-pool
pct clone 1000 1002 --full --storage ceph-pool
pct clone 1000 1003 --full --storage ceph-pool
[Install]
WantedBy=multi-user.target
EOF
# Start all workers
for i in {1000..1003}; do pct start $i; done
# Start service
systemctl daemon-reload
systemctl enable pulse-worker.service
systemctl start pulse-worker.service
```
## Usage
### Creating a Workflow
### Quick Command Execution
1. Access the PULSE web interface at `http://your-server:8080`
2. Navigate to **Workflows****Create New**
3. Define workflow steps using the visual editor or YAML syntax
4. Specify execution targets (specific nodes, groups, or all workers)
5. Add interactive prompts where user input is required
6. Save and activate the workflow
1. Access PULSE at `http://your-server:8080`
2. Navigate to **⚡ Quick Command** tab
3. Select a worker from the dropdown
4. Use **Templates** for pre-built commands or **History** for recent commands
5. Enter your command and click **Execute**
6. View results in the **Executions** tab
### Example Workflow
```yaml
name: "System Update and Reboot"
description: "Update all servers in the cluster with user confirmation"
steps:
- name: "Check Current Versions"
type: "execute"
targets: ["all"]
command: "apt list --upgradable"
- name: "User Approval"
type: "prompt"
message: "Review available updates. Proceed with installation?"
options: ["Yes", "No", "Cancel"]
- name: "Install Updates"
type: "execute"
targets: ["all"]
command: "apt-get update && apt-get upgrade -y"
condition: "prompt_response == 'Yes'"
- name: "Reboot Confirmation"
type: "prompt"
message: "Updates complete. Reboot all servers?"
options: ["Yes", "No"]
- name: "Rolling Reboot"
type: "execute"
targets: ["all"]
command: "reboot"
strategy: "rolling"
condition: "prompt_response == 'Yes'"
```
**Built-in Command Templates:**
- System Info: `uname -a`
- Disk Usage: `df -h`
- Memory Usage: `free -h`
- CPU Info: `lscpu`
- Running Processes: `ps aux --sort=-%mem | head -20`
- Network Interfaces: `ip addr show`
- Docker Containers: `docker ps -a`
- System Logs: `tail -n 50 /var/log/syslog`
### Running a Workflow
### Worker Monitoring
1. Select a workflow from the dashboard
2. Click **Execute**
3. Monitor progress in real-time
4. Respond to interactive prompts as they appear
5. View detailed logs for each execution step
The **Workers** tab displays real-time metrics for each worker:
- System information (OS, architecture, CPU cores)
- Memory usage (used/total with percentage)
- Load averages (1m, 5m, 15m)
- System uptime
- Active tasks vs. maximum concurrent capacity
## Configuration
### Execution Management
### Server Configuration (`config.yml`)
```yaml
server:
host: "0.0.0.0"
port: 8080
secret_key: "your-secret-key"
- **View Details**: Click any execution to see formatted logs with timestamps, status, and output
- **Re-run Command**: Click "Re-run" button in execution details to repeat a command
- **Download Logs**: Export execution data as JSON for auditing
- **Clear Completed**: Bulk delete finished executions
- **Auto-Cleanup**: Executions older than 30 days are automatically removed
database:
type: "postgresql"
host: "localhost"
port: 5432
name: "pulse"
### Workflow Creation (Future Feature)
workers:
heartbeat_interval: 30
timeout: 300
max_concurrent_tasks: 10
security:
enable_authentication: true
require_approval: true
```
### Worker Configuration (`worker-config.yml`)
```yaml
worker:
name: "worker-01"
server_url: "http://pulse-server:8080"
api_key: "worker-api-key"
resources:
max_cpu_percent: 80
max_memory_mb: 1024
executor:
shell: "/bin/bash"
working_directory: "/tmp/pulse"
timeout: 3600
```
1. Navigate to **Workflows****Create New**
2. Define workflow steps using JSON syntax
3. Specify target workers
4. Add interactive prompts where needed
5. Save and execute
## Features in Detail
### Interactive Workflows
- Pause execution to collect user input via web forms
- Display intermediate results for review
- Conditional branching based on user decisions
- Multi-choice prompts with validation
### Terminal Aesthetic
- Phosphor green (#00ff41) on black (#0a0a0a) color scheme
- CRT scanline animation effect
- Text glow and shadow effects
- ASCII box-drawing characters for borders
- Boot sequence animation on first load
- Hover effects with smooth transitions
### Mass Execution
- Run commands across all workers simultaneously
- Target specific node groups or individual servers
- Rolling execution for zero-downtime updates
- Parallel and sequential execution strategies
### Real-Time Communication
- WebSocket-based bidirectional communication
- Instant command result notifications
- Live worker status updates
- Terminal beep sounds for events
- Toast notifications with visual feedback
### Monitoring & Logging
- Real-time workflow execution dashboard
- Detailed per-step logging and output capture
- Historical execution records and analytics
- Alert notifications for failures or completion
### Execution Tracking
- Formatted log display (not raw JSON)
- Color-coded success/failure indicators
- Timestamp and duration for each step
- Scrollable output with syntax highlighting
- Persistent history with pagination
- Load More button for large execution lists
### Security
- Role-based access control (RBAC)
- Authelia SSO integration for user authentication
- API key authentication for workers
- Workflow approval requirements
- Audit logging for all actions
- User session management
- Admin-only operations (worker deletion, workflow management)
- Audit logging for all executions
### Performance
- Automatic cleanup of old executions (configurable retention)
- Pagination for large execution lists (50 at a time)
- Efficient WebSocket connection pooling
- Worker heartbeat monitoring
- Database connection pooling
## Configuration
### Environment Variables
**Server (.env):**
```bash
PORT=8080 # Server port
SECRET_KEY=<random-string> # Session secret
DB_HOST=10.10.10.50 # MariaDB host
DB_PORT=3306 # MariaDB port
DB_NAME=pulse # Database name
DB_USER=pulse_user # Database user
DB_PASSWORD=<password> # Database password
WORKER_API_KEY=<api-key> # Worker authentication key
EXECUTION_RETENTION_DAYS=30 # Auto-cleanup retention (default: 30)
```
**Worker (.env):**
```bash
WORKER_NAME=pulse-worker-01 # Unique worker name
PULSE_SERVER=http://10.10.10.65:8080 # Server HTTP URL
PULSE_WS=ws://10.10.10.65:8080 # Server WebSocket URL
WORKER_API_KEY=<api-key> # Must match server key
HEARTBEAT_INTERVAL=30 # Heartbeat seconds (default: 30)
MAX_CONCURRENT_TASKS=5 # Max parallel tasks (default: 5)
```
## Database Schema
PULSE uses MariaDB with the following tables:
- **users**: User accounts from Authelia SSO
- **workers**: Worker node registry with metadata
- **workflows**: Workflow definitions (JSON)
- **executions**: Execution history with logs
See [Claude.md](Claude.md) for complete schema details.
## Troubleshooting
### Worker Not Connecting
```bash
# Check worker service status
systemctl status pulse-worker
# Check worker logs
journalctl -u pulse-worker -n 50 -f
# Verify API key matches server
grep WORKER_API_KEY /opt/pulse-worker/.env
```
### Commands Stuck in "Running"
- This was fixed in recent updates - restart the server:
```bash
systemctl restart pulse.service
```
### Clear All Executions
Use the database directly if needed:
```bash
mysql -h 10.10.10.50 -u pulse_user -p pulse
> DELETE FROM executions WHERE status IN ('completed', 'failed');
```
## Development
### Recent Updates
**Phase 1-6 Improvements:**
- Formatted log display with color-coding
- Worker system metrics monitoring
- Command templates and history
- Re-run and download execution features
- Auto-cleanup and pagination
- Terminal aesthetic refinements
- Audio notifications and visual toasts
See git history for detailed changelog.
### Future Enhancements
- Full workflow system implementation
- Multi-worker command execution
- Scheduled/cron job support
- Execution search and filtering
- Dark/light theme toggle
- Mobile-responsive design
- REST API documentation
- Webhook integrations
## License
MIT License - See LICENSE file for details
---
**PULSE** - Orchestrating your infrastructure, one heartbeat at a time.
**PULSE** - Orchestrating your infrastructure, one heartbeat at a time.
Built with retro terminal aesthetics 🖥️ | Powered by WebSockets 🔌 | Secured by Authelia 🔐

File diff suppressed because it is too large Load Diff

866
server.js
View File

@@ -89,6 +89,24 @@ async function initDatabase() {
)
`);
await connection.query(`
CREATE TABLE IF NOT EXISTS scheduled_commands (
id VARCHAR(36) PRIMARY KEY,
name VARCHAR(255) NOT NULL,
command TEXT NOT NULL,
worker_ids JSON NOT NULL,
schedule_type VARCHAR(50) NOT NULL,
schedule_value VARCHAR(255) NOT NULL,
enabled BOOLEAN DEFAULT TRUE,
created_by VARCHAR(255),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_run TIMESTAMP NULL,
next_run TIMESTAMP NULL,
INDEX idx_enabled (enabled),
INDEX idx_next_run (next_run)
)
`);
console.log('Database tables initialized successfully');
} catch (error) {
console.error('Database initialization error:', error);
@@ -98,6 +116,116 @@ async function initDatabase() {
}
}
// Auto-cleanup old executions (runs daily)
async function cleanupOldExecutions() {
try {
const retentionDays = parseInt(process.env.EXECUTION_RETENTION_DAYS) || 30;
const [result] = await pool.query(
`DELETE FROM executions
WHERE status IN ('completed', 'failed')
AND started_at < DATE_SUB(NOW(), INTERVAL ? DAY)`,
[retentionDays]
);
console.log(`[Cleanup] Removed ${result.affectedRows} executions older than ${retentionDays} days`);
} catch (error) {
console.error('[Cleanup] Error removing old executions:', error);
}
}
// Run cleanup daily at 3 AM
setInterval(cleanupOldExecutions, 24 * 60 * 60 * 1000);
// Run cleanup on startup
cleanupOldExecutions();
// Scheduled Commands Processor
async function processScheduledCommands() {
try {
const [schedules] = await pool.query(
`SELECT * FROM scheduled_commands
WHERE enabled = TRUE
AND (next_run IS NULL OR next_run <= NOW())`
);
for (const schedule of schedules) {
console.log(`[Scheduler] Running scheduled command: ${schedule.name}`);
const workerIds = JSON.parse(schedule.worker_ids);
// Execute command on each worker
for (const workerId of workerIds) {
const workerWs = workers.get(workerId);
if (workerWs && workerWs.readyState === WebSocket.OPEN) {
const executionId = generateUUID();
// Create execution record
await pool.query(
'INSERT INTO executions (id, workflow_id, status, started_by, started_at, logs) VALUES (?, ?, ?, ?, NOW(), ?)',
[executionId, null, 'running', `scheduler:${schedule.name}`, JSON.stringify([{
step: 'scheduled_command',
action: 'command_sent',
worker_id: workerId,
command: schedule.command,
timestamp: new Date().toISOString()
}])]
);
// Send command to worker
workerWs.send(JSON.stringify({
type: 'execute_command',
execution_id: executionId,
command: schedule.command,
worker_id: workerId,
timeout: 300000 // 5 minute timeout for scheduled commands
}));
broadcast({ type: 'execution_started', execution_id: executionId, workflow_id: null });
}
}
// Update last_run and calculate next_run
const nextRun = calculateNextRun(schedule.schedule_type, schedule.schedule_value);
await pool.query(
'UPDATE scheduled_commands SET last_run = NOW(), next_run = ? WHERE id = ?',
[nextRun, schedule.id]
);
}
} catch (error) {
console.error('[Scheduler] Error processing scheduled commands:', error);
}
}
function calculateNextRun(scheduleType, scheduleValue) {
const now = new Date();
if (scheduleType === 'interval') {
// Interval in minutes
const minutes = parseInt(scheduleValue);
return new Date(now.getTime() + minutes * 60000);
} else if (scheduleType === 'daily') {
// Daily at HH:MM
const [hours, minutes] = scheduleValue.split(':').map(Number);
const next = new Date(now);
next.setHours(hours, minutes, 0, 0);
// If time has passed today, schedule for tomorrow
if (next <= now) {
next.setDate(next.getDate() + 1);
}
return next;
} else if (scheduleType === 'hourly') {
// Every N hours
const hours = parseInt(scheduleValue);
return new Date(now.getTime() + hours * 3600000);
}
return null;
}
// Run scheduler every minute
setInterval(processScheduledCommands, 60 * 1000);
// Initial run on startup
setTimeout(processScheduledCommands, 5000);
// WebSocket connections
const clients = new Set();
const workers = new Map(); // Map worker_id -> WebSocket connection
@@ -113,12 +241,13 @@ wss.on('connection', (ws) => {
if (message.type === 'command_result') {
// Handle command result from worker
const { execution_id, worker_id, success, stdout, stderr, duration, timestamp } = message;
const { execution_id, command_id, worker_id, success, stdout, stderr, duration, timestamp } = message;
// Add result to execution logs
await addExecutionLog(execution_id, {
step: 'command_execution',
action: 'command_result',
command_id: command_id, // Include command_id for workflow tracking
worker_id: worker_id,
success: success,
stdout: stdout,
@@ -127,9 +256,14 @@ wss.on('connection', (ws) => {
timestamp: timestamp || new Date().toISOString()
});
// Update execution status to completed or failed
const finalStatus = success ? 'completed' : 'failed';
await updateExecutionStatus(execution_id, finalStatus);
// For non-workflow executions, update status immediately
// For workflow executions, the workflow engine will update status
const [execution] = await pool.query('SELECT workflow_id FROM executions WHERE id = ?', [execution_id]);
if (execution.length > 0 && !execution[0].workflow_id) {
// Only update status for quick commands (no workflow_id)
const finalStatus = success ? 'completed' : 'failed';
await updateExecutionStatus(execution_id, finalStatus);
}
// Broadcast result to all connected clients
broadcast({
@@ -141,7 +275,7 @@ wss.on('connection', (ws) => {
stderr: stderr
});
console.log(`Command result received for execution ${execution_id}: ${finalStatus}`);
console.log(`Command result received for execution ${execution_id}: ${success ? 'success' : 'failed'}`);
}
if (message.type === 'workflow_result') {
@@ -252,6 +386,45 @@ function broadcast(data) {
});
}
// Helper function to add log entry to execution
async function addExecutionLog(executionId, logEntry) {
try {
const [execution] = await pool.query('SELECT logs FROM executions WHERE id = ?', [executionId]);
if (execution.length > 0) {
const logs = typeof execution[0].logs === 'string' ? JSON.parse(execution[0].logs) : execution[0].logs;
logs.push(logEntry);
await pool.query(
'UPDATE executions SET logs = ? WHERE id = ?',
[JSON.stringify(logs), executionId]
);
}
} catch (error) {
console.error(`[Workflow] Error adding execution log:`, error);
}
}
// Helper function to update execution status
async function updateExecutionStatus(executionId, status) {
try {
await pool.query(
'UPDATE executions SET status = ?, completed_at = NOW() WHERE id = ?',
[status, executionId]
);
broadcast({
type: 'execution_status',
execution_id: executionId,
status: status
});
console.log(`[Workflow] Execution ${executionId} status updated to: ${status}`);
} catch (error) {
console.error(`[Workflow] Error updating execution status:`, error);
}
}
// Authelia SSO Middleware
async function authenticateSSO(req, res, next) {
// Check for Authelia headers
@@ -308,6 +481,237 @@ async function authenticateSSO(req, res, next) {
next();
}
// Workflow Execution Engine
async function executeWorkflowSteps(executionId, workflowId, definition, username) {
try {
console.log(`[Workflow] Starting execution ${executionId} for workflow ${workflowId}`);
const steps = definition.steps || [];
let allStepsSucceeded = true;
for (let i = 0; i < steps.length; i++) {
const step = steps[i];
console.log(`[Workflow] Execution ${executionId} - Step ${i + 1}/${steps.length}: ${step.name}`);
// Add step start log
await addExecutionLog(executionId, {
step: i + 1,
step_name: step.name,
action: 'step_started',
timestamp: new Date().toISOString()
});
broadcast({
type: 'workflow_step_started',
execution_id: executionId,
step: i + 1,
step_name: step.name
});
if (step.type === 'execute') {
// Execute command step
const success = await executeCommandStep(executionId, step, i + 1);
if (!success) {
allStepsSucceeded = false;
break; // Stop workflow on failure
}
} else if (step.type === 'wait') {
// Wait step (delay in seconds)
const seconds = step.duration || 5;
await addExecutionLog(executionId, {
step: i + 1,
action: 'waiting',
duration: seconds,
timestamp: new Date().toISOString()
});
await new Promise(resolve => setTimeout(resolve, seconds * 1000));
} else if (step.type === 'prompt') {
// Interactive prompt step (not fully implemented, would need user interaction)
await addExecutionLog(executionId, {
step: i + 1,
action: 'prompt_skipped',
message: 'Interactive prompts not yet supported',
timestamp: new Date().toISOString()
});
}
// Add step completion log
await addExecutionLog(executionId, {
step: i + 1,
step_name: step.name,
action: 'step_completed',
timestamp: new Date().toISOString()
});
broadcast({
type: 'workflow_step_completed',
execution_id: executionId,
step: i + 1,
step_name: step.name
});
}
// Mark execution as completed or failed
const finalStatus = allStepsSucceeded ? 'completed' : 'failed';
await updateExecutionStatus(executionId, finalStatus);
console.log(`[Workflow] Execution ${executionId} finished with status: ${finalStatus}`);
} catch (error) {
console.error(`[Workflow] Execution ${executionId} error:`, error);
await addExecutionLog(executionId, {
action: 'workflow_error',
error: error.message,
timestamp: new Date().toISOString()
});
await updateExecutionStatus(executionId, 'failed');
}
}
async function executeCommandStep(executionId, step, stepNumber) {
try {
const command = step.command;
const targets = step.targets || ['all'];
// Determine which workers to target
let targetWorkerIds = [];
if (targets.includes('all')) {
// Get all online workers
const [onlineWorkers] = await pool.query('SELECT id FROM workers WHERE status = ?', ['online']);
targetWorkerIds = onlineWorkers.map(w => w.id);
} else {
// Specific worker IDs or names
for (const target of targets) {
// Try to find by ID first, then by name
const [workerById] = await pool.query('SELECT id FROM workers WHERE id = ?', [target]);
if (workerById.length > 0) {
targetWorkerIds.push(workerById[0].id);
} else {
const [workerByName] = await pool.query('SELECT id FROM workers WHERE name = ?', [target]);
if (workerByName.length > 0) {
targetWorkerIds.push(workerByName[0].id);
}
}
}
}
if (targetWorkerIds.length === 0) {
await addExecutionLog(executionId, {
step: stepNumber,
action: 'no_workers',
message: 'No workers available for this step',
timestamp: new Date().toISOString()
});
return false;
}
// Execute command on each target worker and wait for results
const results = [];
for (const workerId of targetWorkerIds) {
const workerWs = workers.get(workerId);
if (!workerWs || workerWs.readyState !== WebSocket.OPEN) {
await addExecutionLog(executionId, {
step: stepNumber,
action: 'worker_offline',
worker_id: workerId,
timestamp: new Date().toISOString()
});
continue;
}
// Send command to worker
const commandId = generateUUID();
await addExecutionLog(executionId, {
step: stepNumber,
action: 'command_sent',
worker_id: workerId,
command: command,
command_id: commandId,
timestamp: new Date().toISOString()
});
workerWs.send(JSON.stringify({
type: 'execute_command',
execution_id: executionId,
command_id: commandId,
command: command,
worker_id: workerId,
timeout: 120000 // 2 minute timeout
}));
// Wait for command result (with timeout)
const result = await waitForCommandResult(executionId, commandId, 120000);
results.push(result);
if (!result.success) {
// Command failed, workflow should stop
return false;
}
}
// All commands succeeded
return results.every(r => r.success);
} catch (error) {
console.error(`[Workflow] Error executing command step:`, error);
await addExecutionLog(executionId, {
step: stepNumber,
action: 'step_error',
error: error.message,
timestamp: new Date().toISOString()
});
return false;
}
}
async function waitForCommandResult(executionId, commandId, timeout) {
return new Promise((resolve) => {
const startTime = Date.now();
const checkInterval = setInterval(async () => {
try {
// Check if we've received the command result in logs
const [execution] = await pool.query('SELECT logs FROM executions WHERE id = ?', [executionId]);
if (execution.length > 0) {
const logs = typeof execution[0].logs === 'string' ? JSON.parse(execution[0].logs) : execution[0].logs;
const resultLog = logs.find(log => log.command_id === commandId && log.action === 'command_result');
if (resultLog) {
clearInterval(checkInterval);
resolve({
success: resultLog.success,
stdout: resultLog.stdout,
stderr: resultLog.stderr
});
return;
}
}
// Check timeout
if (Date.now() - startTime > timeout) {
clearInterval(checkInterval);
resolve({
success: false,
error: 'Command timeout'
});
}
} catch (error) {
console.error('[Workflow] Error checking command result:', error);
clearInterval(checkInterval);
resolve({
success: false,
error: error.message
});
}
}, 500); // Check every 500ms
});
}
// Routes - All protected by SSO
app.get('/api/user', authenticateSSO, (req, res) => {
res.json(req.user);
@@ -326,15 +730,24 @@ app.post('/api/workflows', authenticateSSO, async (req, res) => {
try {
const { name, description, definition } = req.body;
const id = generateUUID();
console.log('[Workflow] Creating workflow:', name);
console.log('[Workflow] Definition:', JSON.stringify(definition, null, 2));
await pool.query(
'INSERT INTO workflows (id, name, description, definition, created_by) VALUES (?, ?, ?, ?, ?)',
[id, name, description, JSON.stringify(definition), req.user.username]
);
console.log('[Workflow] Successfully inserted workflow:', id);
res.json({ id, name, description, definition });
console.log('[Workflow] Broadcasting workflow_created');
broadcast({ type: 'workflow_created', workflow_id: id });
console.log('[Workflow] Broadcast complete');
} catch (error) {
console.error('[Workflow] Error creating workflow:', error);
res.status(500).json({ error: error.message });
}
});
@@ -394,13 +807,29 @@ app.post('/api/executions', authenticateSSO, async (req, res) => {
try {
const { workflow_id } = req.body;
const id = generateUUID();
// Get workflow definition
const [workflows] = await pool.query('SELECT * FROM workflows WHERE id = ?', [workflow_id]);
if (workflows.length === 0) {
return res.status(404).json({ error: 'Workflow not found' });
}
const workflow = workflows[0];
const definition = typeof workflow.definition === 'string' ? JSON.parse(workflow.definition) : workflow.definition;
// Create execution record
await pool.query(
'INSERT INTO executions (id, workflow_id, status, started_by, started_at, logs) VALUES (?, ?, ?, ?, NOW(), ?)',
[id, workflow_id, 'running', req.user.username, JSON.stringify([])]
);
broadcast({ type: 'execution_started', execution_id: id, workflow_id });
// Start workflow execution asynchronously
executeWorkflowSteps(id, workflow_id, definition, req.user.username).catch(err => {
console.error(`[Workflow] Execution ${id} failed:`, err);
});
res.json({ id, workflow_id, status: 'running' });
} catch (error) {
res.status(500).json({ error: error.message });
@@ -409,10 +838,25 @@ app.post('/api/executions', authenticateSSO, async (req, res) => {
app.get('/api/executions', authenticateSSO, async (req, res) => {
try {
const limit = parseInt(req.query.limit) || 50;
const offset = parseInt(req.query.offset) || 0;
const [rows] = await pool.query(
'SELECT e.*, w.name as workflow_name FROM executions e LEFT JOIN workflows w ON e.workflow_id = w.id ORDER BY e.started_at DESC LIMIT 50'
'SELECT e.*, w.name as workflow_name FROM executions e LEFT JOIN workflows w ON e.workflow_id = w.id ORDER BY e.started_at DESC LIMIT ? OFFSET ?',
[limit, offset]
);
res.json(rows);
// Get total count
const [countRows] = await pool.query('SELECT COUNT(*) as total FROM executions');
const total = countRows[0].total;
res.json({
executions: rows,
total: total,
limit: limit,
offset: offset,
hasMore: offset + rows.length < total
});
} catch (error) {
res.status(500).json({ error: error.message });
}
@@ -428,6 +872,104 @@ app.delete('/api/executions/:id', authenticateSSO, async (req, res) => {
}
});
app.post('/api/executions/:id/abort', authenticateSSO, async (req, res) => {
try {
const executionId = req.params.id;
// Check if execution exists and is running
const [execution] = await pool.query('SELECT status FROM executions WHERE id = ?', [executionId]);
if (execution.length === 0) {
return res.status(404).json({ error: 'Execution not found' });
}
if (execution[0].status !== 'running') {
return res.status(400).json({ error: 'Execution is not running' });
}
// Add abort log entry
await addExecutionLog(executionId, {
action: 'execution_aborted',
aborted_by: req.user.username,
timestamp: new Date().toISOString()
});
// Update execution status to failed
await updateExecutionStatus(executionId, 'failed');
console.log(`[Execution] Execution ${executionId} aborted by ${req.user.username}`);
res.json({ success: true });
} catch (error) {
console.error('[Execution] Error aborting execution:', error);
res.status(500).json({ error: error.message });
}
});
// Scheduled Commands API
app.get('/api/scheduled-commands', authenticateSSO, async (req, res) => {
try {
const [schedules] = await pool.query(
'SELECT * FROM scheduled_commands ORDER BY created_at DESC'
);
res.json(schedules);
} catch (error) {
res.status(500).json({ error: error.message });
}
});
app.post('/api/scheduled-commands', authenticateSSO, async (req, res) => {
try {
const { name, command, worker_ids, schedule_type, schedule_value } = req.body;
if (!name || !command || !worker_ids || !schedule_type || !schedule_value) {
return res.status(400).json({ error: 'Missing required fields' });
}
const id = generateUUID();
const nextRun = calculateNextRun(schedule_type, schedule_value);
await pool.query(
`INSERT INTO scheduled_commands
(id, name, command, worker_ids, schedule_type, schedule_value, created_by, next_run)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
[id, name, command, JSON.stringify(worker_ids), schedule_type, schedule_value, req.user.username, nextRun]
);
res.json({ success: true, id });
} catch (error) {
res.status(500).json({ error: error.message });
}
});
app.put('/api/scheduled-commands/:id/toggle', authenticateSSO, async (req, res) => {
try {
const { id } = req.params;
const [current] = await pool.query('SELECT enabled FROM scheduled_commands WHERE id = ?', [id]);
if (current.length === 0) {
return res.status(404).json({ error: 'Schedule not found' });
}
const newEnabled = !current[0].enabled;
await pool.query('UPDATE scheduled_commands SET enabled = ? WHERE id = ?', [newEnabled, id]);
res.json({ success: true, enabled: newEnabled });
} catch (error) {
res.status(500).json({ error: error.message });
}
});
app.delete('/api/scheduled-commands/:id', authenticateSSO, async (req, res) => {
try {
const { id } = req.params;
await pool.query('DELETE FROM scheduled_commands WHERE id = ?', [id]);
res.json({ success: true });
} catch (error) {
res.status(500).json({ error: error.message });
}
});
// Health check (no auth required)
app.get('/health', async (req, res) => {
try {
@@ -467,299 +1009,6 @@ initDatabase().then(() => {
// ============================================
// WORKFLOW EXECUTION ENGINE
// ============================================
// Store active workflow executions in memory
const activeExecutions = new Map();
// Execute workflow step by step
async function executeWorkflow(workflowId, executionId, userId, targetWorkers = 'all') {
try {
// Get workflow definition
const [workflows] = await pool.query('SELECT * FROM workflows WHERE id = ?', [workflowId]);
if (workflows.length === 0) {
throw new Error('Workflow not found');
}
const workflow = workflows[0];
const definition = JSON.parse(workflow.definition);
// Initialize execution state
const executionState = {
id: executionId,
workflowId: workflowId,
currentStep: 0,
steps: definition.steps || [],
results: [],
status: 'running',
waitingForInput: false,
targetWorkers: targetWorkers,
userId: userId
};
activeExecutions.set(executionId, executionState);
// Start executing steps
await executeNextStep(executionId);
} catch (error) {
console.error('Workflow execution error:', error);
await updateExecutionStatus(executionId, 'failed', error.message);
}
}
// Execute the next step in a workflow
async function executeNextStep(executionId) {
const state = activeExecutions.get(executionId);
if (!state) return;
// Check if we've completed all steps
if (state.currentStep >= state.steps.length) {
await updateExecutionStatus(executionId, 'completed');
activeExecutions.delete(executionId);
return;
}
const step = state.steps[state.currentStep];
// Check if step has a condition
if (step.condition && !evaluateCondition(step.condition, state)) {
console.log(`[Workflow] Skipping step ${state.currentStep}: condition not met`);
state.currentStep++;
return executeNextStep(executionId);
}
console.log(`[Workflow] Executing step ${state.currentStep}: ${step.name}`);
try {
switch (step.type) {
case 'execute':
await executeCommandStep(executionId, step);
break;
case 'prompt':
await executePromptStep(executionId, step);
break;
case 'wait':
await executeWaitStep(executionId, step);
break;
default:
throw new Error(`Unknown step type: ${step.type}`);
}
} catch (error) {
await addExecutionLog(executionId, {
step: state.currentStep,
error: error.message,
timestamp: new Date().toISOString()
});
await updateExecutionStatus(executionId, 'failed', error.message);
activeExecutions.delete(executionId);
}
}
// Execute a command on workers
async function executeCommandStep(executionId, step) {
const state = activeExecutions.get(executionId);
// Get target workers
const [workers] = await pool.query(
'SELECT * FROM workers WHERE status = "online"'
);
if (workers.length === 0) {
throw new Error('No online workers available');
}
// Filter workers based on target
let targetWorkers = workers;
if (step.targets && step.targets[0] !== 'all') {
targetWorkers = workers.filter(w => step.targets.includes(w.name));
}
// Send command to workers via WebSocket
const commandMessage = {
type: 'execute_command',
execution_id: executionId,
step_index: state.currentStep,
command: step.command,
timeout: step.timeout || 300000
};
// Broadcast to target workers
clients.forEach(client => {
if (client.readyState === WebSocket.OPEN) {
client.send(JSON.stringify(commandMessage));
}
});
await addExecutionLog(executionId, {
step: state.currentStep,
action: 'command_sent',
command: step.command,
workers: targetWorkers.map(w => w.name),
timestamp: new Date().toISOString()
});
// For now, move to next step immediately
// In production, we'd wait for worker responses
state.currentStep++;
// Small delay to allow command to execute
setTimeout(() => executeNextStep(executionId), 1000);
}
// Execute a user prompt step
async function executePromptStep(executionId, step) {
const state = activeExecutions.get(executionId);
state.waitingForInput = true;
state.promptData = {
message: step.message,
options: step.options,
step: state.currentStep
};
await addExecutionLog(executionId, {
step: state.currentStep,
action: 'waiting_for_input',
prompt: step.message,
timestamp: new Date().toISOString()
});
// Notify frontend that input is needed
broadcast({
type: 'execution_prompt',
execution_id: executionId,
prompt: state.promptData
});
}
// Execute a wait/delay step
async function executeWaitStep(executionId, step) {
const state = activeExecutions.get(executionId);
const delay = step.duration || 1000;
await addExecutionLog(executionId, {
step: state.currentStep,
action: 'waiting',
duration: delay,
timestamp: new Date().toISOString()
});
state.currentStep++;
setTimeout(() => executeNextStep(executionId), delay);
}
// Handle user input for prompts
function handleUserInput(executionId, response) {
const state = activeExecutions.get(executionId);
if (!state || !state.waitingForInput) {
return false;
}
state.promptResponse = response;
state.waitingForInput = false;
state.currentStep++;
addExecutionLog(executionId, {
step: state.currentStep - 1,
action: 'user_response',
response: response,
timestamp: new Date().toISOString()
});
executeNextStep(executionId);
return true;
}
// Evaluate conditions
function evaluateCondition(condition, state) {
try {
// Simple condition evaluation
// In production, use a proper expression evaluator
const promptResponse = state.promptResponse;
return eval(condition);
} catch (error) {
console.error('Condition evaluation error:', error);
return false;
}
}
// Helper functions
async function updateExecutionStatus(executionId, status, error = null) {
const updates = { status };
if (status === 'completed' || status === 'failed') {
updates.completed_at = new Date();
}
if (error) {
// Add error to logs
await addExecutionLog(executionId, { error, timestamp: new Date().toISOString() });
}
await pool.query(
'UPDATE executions SET status = ?, completed_at = ? WHERE id = ?',
[status, updates.completed_at || null, executionId]
);
broadcast({
type: 'execution_status',
execution_id: executionId,
status: status
});
}
async function addExecutionLog(executionId, logEntry) {
const [rows] = await pool.query('SELECT logs FROM executions WHERE id = ?', [executionId]);
if (rows.length === 0) return;
const logs = JSON.parse(rows[0].logs || '[]');
logs.push(logEntry);
await pool.query('UPDATE executions SET logs = ? WHERE id = ?', [
JSON.stringify(logs),
executionId
]);
}
// ============================================
// API ROUTES - Add these to your server.js
// ============================================
// Start workflow execution
app.post('/api/executions', authenticateSSO, async (req, res) => {
try {
const { workflow_id, target_workers } = req.body;
const id = generateUUID();
await pool.query(
'INSERT INTO executions (id, workflow_id, status, started_by, started_at, logs) VALUES (?, ?, ?, ?, NOW(), ?)',
[id, workflow_id, 'running', req.user.username, JSON.stringify([])]
);
// Start execution
executeWorkflow(workflow_id, id, req.user.username, target_workers || 'all');
broadcast({ type: 'execution_started', execution_id: id, workflow_id });
res.json({ id, workflow_id, status: 'running' });
} catch (error) {
res.status(500).json({ error: error.message });
}
});
// Respond to workflow prompt
app.post('/api/executions/:id/respond', authenticateSSO, async (req, res) => {
try {
const { response } = req.body;
const success = handleUserInput(req.params.id, response);
if (success) {
res.json({ success: true });
} else {
res.status(400).json({ error: 'Execution not waiting for input' });
}
} catch (error) {
res.status(500).json({ error: error.message });
}
});
// Get execution details with logs
app.get('/api/executions/:id', authenticateSSO, async (req, res) => {
@@ -768,15 +1017,12 @@ app.get('/api/executions/:id', authenticateSSO, async (req, res) => {
if (rows.length === 0) {
return res.status(404).json({ error: 'Not found' });
}
const execution = rows[0];
const state = activeExecutions.get(req.params.id);
res.json({
...execution,
logs: JSON.parse(execution.logs || '[]'),
waiting_for_input: state?.waitingForInput || false,
prompt: state?.promptData || null
logs: JSON.parse(execution.logs || '[]')
});
} catch (error) {
res.status(500).json({ error: error.message });