admin/ServerKit
1
0 Fork 0
mirror of https://github.com/jhd3197/ServerKit.git synced 2026-10-31 00:57:22 +09:00
Code Issues Packages Projects Releases Wiki Activity

refactor(automations): retire Workflow Builder engine, port emitters (Phase 4 backend)

Browse Source 
The React-Flow Workflow Builder is replaced by the Automations extension
(tramo). Remove its core Python engine and port its four event emitters to
the panel event bus so tramo workflows can trigger on them via the events
bridge.

- event_service.py: new EVENT_CATALOG entries health.check_failed, git.push,
  monitor.high_cpu, monitor.high_memory.
- docker_service.py: drop the app_stopped emit (app.stopped already fires via
  the audit trail -- no double-emit).
- environment_health_service/git_service/monitoring_service: emit the new
  panel events via EventService instead of WorkflowEventBus.
- Delete app/api/workflows.py, app/services/workflow_engine.py,
  app/services/workflow_service.py; drop the blueprint + startup job
  registration; neutralize the cron workflow-scheduler handler to a no-op.
- KEEP app/models/workflow.py + its tables (read-only legacy export lives in
  the tramo blueprint; table drop is a future cleanup).
- Remove the two workflow-engine test classes from test_jobs.py.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Juan DenisandClaude Opus 4.8 committed 2026-07-11 22:15:10 -04:00
1 parent 675543541b
commit 531fdf3d42
11 files changed
+51 -2171
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/app/__init__.py
+3 -5
View File
@@ -332,9 +332,8 @@ def create_app(config_name=None):
from app.api.metrics import metrics_bp
app.register_blueprint(metrics_bp, url_prefix='/api/v1/metrics')
# Register blueprints - Workflows
from app.api.workflows import workflows_bp
app.register_blueprint(workflows_bp, url_prefix='/api/v1/workflows')
# The /api/v1/workflows blueprint (React-Flow Workflow Builder) was retired
# in plan 45 Phase 4 -- the Automations extension (tramo) replaces it.
# Register blueprints - Servers (Multi-server management)
from app.api.servers import servers_bp
@@ -620,8 +619,7 @@ def create_app(config_name=None):
# scheduled backups).
from app.services.deployment_job_service import DeploymentJobService
DeploymentJobService.register_jobs()
from app.services.workflow_engine import WorkflowEngine
WorkflowEngine.register_jobs()
# WorkflowEngine.register_jobs() removed in plan 45 Phase 4 (engine retired).
from app.services.backup_service import BackupService
BackupService.register_jobs()
from app.services.backup_policy_service import BackupPolicyService
backend/app/api/workflows.py
-373
View File
@@ -1,373 +0,0 @@
import json
import secrets
from datetime import datetime
from flask import Blueprint, request, jsonify
from flask_jwt_extended import jwt_required, get_jwt_identity
from app import db
from app.models import Workflow, User, WorkflowExecution, WorkflowLog
workflows_bp = Blueprint('workflows', __name__)
def _validate_workflow_graph(data):
"""Validate workflow graph for cycles if nodes/edges are present."""
nodes = data.get('nodes', [])
edges = data.get('edges', [])
if nodes and edges:
from app.services.workflow_engine import WorkflowEngine
err = WorkflowEngine.validate_graph(nodes, edges)
if err:
return err
return None
@workflows_bp.route('', methods=['GET'])
@jwt_required()
def get_workflows():
"""Get all workflows for the current user."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
if user.role == 'admin':
workflows = Workflow.query.all()
else:
workflows = Workflow.query.filter_by(user_id=current_user_id).all()
return jsonify({
'workflows': [w.to_dict() for w in workflows]
})
@workflows_bp.route('/<int:workflow_id>', methods=['GET'])
@jwt_required()
def get_workflow(workflow_id):
"""Get a single workflow by ID."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
workflow = Workflow.query.get(workflow_id)
if not workflow:
return jsonify({'error': 'Workflow not found'}), 404
if user.role != 'admin' and workflow.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
return jsonify(workflow.to_dict())
@workflows_bp.route('', methods=['POST'])
@jwt_required()
def create_workflow():
"""Create a new workflow."""
current_user_id = get_jwt_identity()
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
name = data.get('name')
if not name:
return jsonify({'error': 'Name is required'}), 400
# Validate graph for cycles
cycle_err = _validate_workflow_graph(data)
if cycle_err:
return jsonify({'error': cycle_err}), 400
# Auto-generate webhook_id if trigger is webhook
trigger_config = data.get('trigger_config', {})
if data.get('trigger_type') == 'webhook' and not trigger_config.get('webhook_id'):
trigger_config['webhook_id'] = secrets.token_urlsafe(24)
workflow = Workflow(
name=name,
description=data.get('description', ''),
nodes=json.dumps(data.get('nodes', [])),
edges=json.dumps(data.get('edges', [])),
viewport=json.dumps(data.get('viewport')) if data.get('viewport') else None,
is_active=data.get('is_active', False),
trigger_type=data.get('trigger_type', 'manual'),
trigger_config=json.dumps(trigger_config),
user_id=current_user_id
)
db.session.add(workflow)
db.session.commit()
return jsonify({
'message': 'Workflow created successfully',
'workflow': workflow.to_dict()
}), 201
@workflows_bp.route('/<int:workflow_id>', methods=['PUT'])
@jwt_required()
def update_workflow(workflow_id):
"""Update an existing workflow."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
workflow = Workflow.query.get(workflow_id)
if not workflow:
return jsonify({'error': 'Workflow not found'}), 404
if user.role != 'admin' and workflow.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
# Validate graph for cycles
cycle_err = _validate_workflow_graph(data)
if cycle_err:
return jsonify({'error': cycle_err}), 400
if 'name' in data:
workflow.name = data['name']
if 'description' in data:
workflow.description = data['description']
if 'nodes' in data:
workflow.nodes = json.dumps(data['nodes'])
if 'edges' in data:
workflow.edges = json.dumps(data['edges'])
if 'viewport' in data:
workflow.viewport = json.dumps(data['viewport']) if data['viewport'] else None
# Automation fields
if 'is_active' in data:
workflow.is_active = data['is_active']
if 'trigger_type' in data:
workflow.trigger_type = data['trigger_type']
if 'trigger_config' in data:
trigger_config = data['trigger_config']
# Auto-generate webhook_id if switching to webhook trigger
if data.get('trigger_type') == 'webhook' and not trigger_config.get('webhook_id'):
existing = json.loads(workflow.trigger_config) if workflow.trigger_config else {}
trigger_config['webhook_id'] = existing.get('webhook_id') or secrets.token_urlsafe(24)
workflow.trigger_config = json.dumps(trigger_config)
db.session.commit()
return jsonify({
'message': 'Workflow updated successfully',
'workflow': workflow.to_dict()
})
@workflows_bp.route('/<int:workflow_id>', methods=['DELETE'])
@jwt_required()
def delete_workflow(workflow_id):
"""Delete a workflow."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
workflow = Workflow.query.get(workflow_id)
if not workflow:
return jsonify({'error': 'Workflow not found'}), 404
if user.role != 'admin' and workflow.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
db.session.delete(workflow)
db.session.commit()
return jsonify({'message': 'Workflow deleted successfully'})
@workflows_bp.route('/<int:workflow_id>/execute', methods=['POST'])
@jwt_required()
def execute_workflow(workflow_id):
"""Trigger a workflow execution manually."""
from app.services.workflow_engine import WorkflowEngine
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
workflow = Workflow.query.get(workflow_id)
if not workflow:
return jsonify({'error': 'Workflow not found'}), 404
if user.role != 'admin' and workflow.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
data = request.get_json() or {}
context = data.get('context', {})
execution_id = WorkflowEngine.enqueue_execution(
workflow_id=workflow_id,
trigger_type='manual',
context=context
)
return jsonify({
'message': 'Workflow execution started',
'execution_id': execution_id
})
@workflows_bp.route('/<int:workflow_id>/executions', methods=['GET'])
@jwt_required()
def get_workflow_executions(workflow_id):
"""Get execution history for a workflow."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
workflow = Workflow.query.get(workflow_id)
if not workflow:
return jsonify({'error': 'Workflow not found'}), 404
if user.role != 'admin' and workflow.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
executions = WorkflowExecution.query.filter_by(workflow_id=workflow_id).order_by(WorkflowExecution.started_at.desc()).all()
return jsonify({
'executions': [e.to_dict() for e in executions]
})
@workflows_bp.route('/executions/<int:execution_id>', methods=['GET'])
@jwt_required()
def get_execution_details(execution_id):
"""Get details for a specific execution."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
execution = WorkflowExecution.query.get(execution_id)
if not execution:
return jsonify({'error': 'Execution not found'}), 404
workflow = execution.workflow
if user.role != 'admin' and workflow.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
return jsonify(execution.to_dict())
@workflows_bp.route('/executions/<int:execution_id>/logs', methods=['GET'])
@jwt_required()
def get_execution_logs(execution_id):
"""Get logs for a specific execution."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
execution = WorkflowExecution.query.get(execution_id)
if not execution:
return jsonify({'error': 'Execution not found'}), 404
workflow = execution.workflow
if user.role != 'admin' and workflow.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
logs = WorkflowLog.query.filter_by(execution_id=execution_id).order_by(WorkflowLog.timestamp.asc()).all()
return jsonify({
'logs': [l.to_dict() for l in logs]
})
@workflows_bp.route('/<int:workflow_id>/deploy', methods=['POST'])
@jwt_required()
def deploy_workflow(workflow_id):
"""
Deploy all resources from a workflow.
"""
from app.services.workflow_service import WorkflowService
current_user_id = get_jwt_identity()
result = WorkflowService.deploy_workflow(workflow_id, current_user_id)
if result.get('success'):
return jsonify(result), 200
elif result.get('error') == 'Workflow not found':
return jsonify(result), 404
elif result.get('error') == 'Access denied':
return jsonify(result), 403
else:
# Partial success or errors - return 200 with error details
return jsonify(result), 200
@workflows_bp.route('/hooks/<webhook_id>', methods=['POST'])
def webhook_trigger(webhook_id):
"""
Public webhook endpoint to trigger a workflow.
No JWT required — the webhook_id acts as the authentication token.
"""
from app.services.workflow_engine import WorkflowEngine, CycleDetectedError
# Find the workflow with this webhook_id
workflows = Workflow.query.filter_by(
is_active=True,
trigger_type='webhook'
).all()
target_workflow = None
for wf in workflows:
try:
config = json.loads(wf.trigger_config) if wf.trigger_config else {}
if config.get('webhook_id') == webhook_id:
target_workflow = wf
break
except (json.JSONDecodeError, TypeError):
continue
if not target_workflow:
return jsonify({'error': 'Webhook not found'}), 404
# Build context from the incoming request
context = {
'webhook_id': webhook_id,
'method': request.method,
'headers': dict(request.headers),
'triggered_at': datetime.utcnow().isoformat()
}
# Include request body
if request.is_json:
context['body'] = request.get_json(silent=True) or {}
else:
body = request.get_data(as_text=True)
context['body'] = body[:8192] if body else ''
# Include query parameters
if request.args:
context['query'] = dict(request.args)
try:
execution_id = WorkflowEngine.enqueue_execution(
workflow_id=target_workflow.id,
trigger_type='webhook',
context=context
)
return jsonify({
'message': 'Workflow triggered',
'execution_id': execution_id,
'workflow': target_workflow.name
}), 200
except CycleDetectedError as e:
return jsonify({'error': str(e)}), 400
except Exception as e:
return jsonify({'error': f'Execution failed: {str(e)}'}), 500
@workflows_bp.route('/validate', methods=['POST'])
@jwt_required()
def validate_workflow():
"""Validate a workflow graph for cycles and other issues."""
from app.services.workflow_engine import WorkflowEngine
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
nodes = data.get('nodes', [])
edges = data.get('edges', [])
err = WorkflowEngine.validate_graph(nodes, edges)
if err:
return jsonify({'valid': False, 'error': err}), 200
return jsonify({'valid': True}), 200
backend/app/jobs/builtin_handlers.py
+9 -40
View File
@@ -61,47 +61,16 @@ def check_auto_sync_schedules():
def check_workflow_schedules():
"""Check all active workflows with cron triggers and run those that are due."""
from app.models.workflow import Workflow
from app.services.workflow_engine import WorkflowEngine
from datetime import datetime
import json
"""Retired no-op (plan 45 Phase 4).
try:
from croniter import croniter
except ImportError:
logger.debug('croniter not installed, skipping workflow schedule check')
return
workflows = Workflow.query.filter_by(is_active=True, trigger_type='cron').all()
if not workflows:
return
now = datetime.utcnow()
for workflow in workflows:
try:
config = json.loads(workflow.trigger_config) if workflow.trigger_config else {}
cron_expr = config.get('cron')
if not cron_expr or not croniter.is_valid(cron_expr):
continue
cron = croniter(cron_expr, now)
prev_run = cron.get_prev(datetime)
seconds_since_due = (now - prev_run).total_seconds()
# Don't run multiple times for the same slot.
if 0 < seconds_since_due <= 90:
if workflow.last_run_at:
seconds_since_last_run = (now - workflow.last_run_at).total_seconds()
if seconds_since_last_run < 110:
continue
logger.info(f'Scheduled workflow triggered: {workflow.name} (ID: {workflow.id})')
WorkflowEngine.enqueue_execution(
workflow_id=workflow.id,
trigger_type='cron',
context={'scheduled_at': prev_run.isoformat()},
)
except Exception as e:
logger.error(f'Workflow schedule check failed for workflow {workflow.id}: {e}')
The React-Flow Workflow Builder and its execution engine were removed in
favour of the Automations extension (tramo), whose own in-process scheduler
drives ``cron-trigger`` nodes inside the managed engine. The legacy
``workflows`` tables are kept for read-only export, so this scheduled handler
is retained as a no-op (avoids an orphaned-handler error for any persisted
``workflow-schedules`` schedule row) but no longer executes anything.
"""
return
# How often the per-site WordPress health poller runs (seconds).
backend/app/services/docker_service.py
+4 -7
View File
@@ -263,13 +263,10 @@ class DockerService:
capture_output=True, text=True
)
if result.returncode == 0:
try:
from app.services.workflow_engine import WorkflowEventBus
WorkflowEventBus.emit('app_stopped', {
'container_id': container_id
})
except Exception as e:
logger.error(f"Failed to emit app_stopped event: {e}")
# The panel emits the `app.stopped` event via the audit trail
# (AuditService.log('app.stop') -> EventService.emit_for_audit),
# so this low-level helper no longer emits its own event -- doing
# so would double-fire it for Automations triggers (plan 45 Ph4).
return {'success': True}
return {'success': False, 'error': result.stderr}
except Exception as e:
backend/app/services/environment_health_service.py
+7 -5
View File
@@ -100,15 +100,17 @@ class EnvironmentHealthService:
site.last_health_check = datetime.utcnow()
db.session.commit()
# Emit event for workflow triggers on every failing poll (back-compat for
# existing event-trigger workflows keyed on 'health_check_failed').
# Emit a panel event on every failing poll so Automations (tramo)
# workflows can trigger on it through the events bridge (plan 45 Ph4,
# ported from the retired Workflow Builder's 'health_check_failed').
if overall in ('unhealthy', 'degraded'):
try:
from app.services.workflow_engine import WorkflowEventBus
WorkflowEventBus.emit('health_check_failed', {
from app.services.event_service import EventService
EventService.emit('health.check_failed', {
'event': 'health.check_failed',
'site_id': site_id,
'overall_status': overall,
'checks': checks
'checks': checks,
})
except Exception:
pass
backend/app/services/event_service.py
+6
View File
@@ -57,6 +57,12 @@ EVENT_CATALOG = [
{'type': 'wordpress.update_rolled_back', 'category': 'WordPress', 'description': 'A WordPress update was auto-rolled-back'},
{'type': 'wordpress.deployed', 'category': 'WordPress', 'description': 'A WordPress git deploy completed'},
{'type': 'wordpress.deploy_failed', 'category': 'WordPress', 'description': 'A WordPress git deploy failed'},
# Ported from the retired Workflow Builder event bus (plan 45 Phase 4) so
# Automations (tramo) workflows can trigger on them via the events bridge.
{'type': 'health.check_failed', 'category': 'Monitoring', 'description': 'An environment health check failed'},
{'type': 'git.push', 'category': 'Deployments', 'description': 'A git push was received for an app'},
{'type': 'monitor.high_cpu', 'category': 'Monitoring', 'description': 'A server crossed its high-CPU alert threshold'},
{'type': 'monitor.high_memory', 'category': 'Monitoring', 'description': 'A server crossed its high-memory alert threshold'},
]
backend/app/services/git_service.py
+7 -4
View File
@@ -436,13 +436,16 @@ class GitService:
'message': f'Ignoring push to {ref}, configured branch is {branch}'
}
# Emit event for workflow triggers
# Emit a panel event so Automations (tramo) workflows can trigger on a
# git push through the events bridge (plan 45 Ph4, ported from the
# retired Workflow Builder's 'git_push').
try:
from app.services.workflow_engine import WorkflowEventBus
WorkflowEventBus.emit('git_push', {
from app.services.event_service import EventService
EventService.emit('git.push', {
'event': 'git.push',
'app_id': app_id,
'branch': branch,
'ref': ref
'ref': ref,
})
except Exception:
pass
backend/app/services/monitoring_service.py
+11 -7
View File
@@ -303,24 +303,28 @@ class MonitoringService:
commit=True,
)
# Emit events for workflow triggers
# Emit panel events so Automations (tramo) workflows can trigger on a
# high-CPU/memory alert through the events bridge (plan 45 Ph4, ported
# from the retired Workflow Builder's 'high_cpu'/'high_memory').
try:
from app.services.workflow_engine import WorkflowEventBus
from app.services.event_service import EventService
for alert in alerts_to_send:
if alert['type'] == 'cpu':
WorkflowEventBus.emit('high_cpu', {
EventService.emit('monitor.high_cpu', {
'event': 'monitor.high_cpu',
'percent': alert.get('value'),
'threshold': alert.get('threshold'),
'severity': alert.get('severity')
'severity': alert.get('severity'),
})
elif alert['type'] == 'memory':
WorkflowEventBus.emit('high_memory', {
EventService.emit('monitor.high_memory', {
'event': 'monitor.high_memory',
'percent': alert.get('value'),
'threshold': alert.get('threshold'),
'severity': alert.get('severity')
'severity': alert.get('severity'),
})
except Exception:
logger.exception("Error emitting workflow events for alerts")
logger.exception("Error emitting monitor events for alerts")
@classmethod
def log_alert(cls, alerts: List[Dict]) -> None:
backend/app/services/workflow_engine.py
-1001
View File
File diff suppressed because it is too large. Load diff
backend/app/services/workflow_service.py
-621
View File
@@ -1,621 +0,0 @@
"""
Workflow Deployment Service
Orchestrates deployment of workflows by converting visual nodes and edges
into actual infrastructure resources (Docker apps, databases, domains).
"""
import json
import os
from app import db
from app.models import Workflow, Application, Domain, User
from app.services.docker_service import DockerService
from app.services.database_service import DatabaseService
from app import paths
class WorkflowService:
"""Service for deploying workflows to infrastructure."""
# Base path for app deployments
APP_BASE_PATH = paths.APPS_DIR
@staticmethod
def deploy_workflow(workflow_id, user_id):
"""
Deploy all resources from a workflow.
Args:
workflow_id: ID of the workflow to deploy
user_id: ID of the user deploying
Returns:
dict with results, errors, and updated workflow
"""
workflow = Workflow.query.get(workflow_id)
if not workflow:
return {'success': False, 'error': 'Workflow not found'}
# Verify user access
user = User.query.get(user_id)
if not user:
return {'success': False, 'error': 'User not found'}
if user.role != 'admin' and workflow.user_id != user_id:
return {'success': False, 'error': 'Access denied'}
# Parse workflow data
nodes = json.loads(workflow.nodes) if workflow.nodes else []
edges = json.loads(workflow.edges) if workflow.edges else []
if not nodes:
return {'success': False, 'error': 'Workflow has no nodes to deploy'}
# Get deployment order
ordered_nodes = WorkflowService.get_deployment_order(nodes, edges)
# Track results and created resources
results = []
errors = []
created_resources = {} # nodeId -> resourceId mapping
# Deploy each node
for node in ordered_nodes:
result = WorkflowService.deploy_node(
node, edges, user_id, created_resources
)
results.append(result)
if result['success']:
created_resources[node['id']] = {
'type': node['type'],
'resourceId': result.get('resourceId'),
'resourceName': result.get('resourceName'),
'credentials': result.get('credentials'), # Store DB credentials for apps
'rootPath': result.get('rootPath')
}
else:
errors.append({
'nodeId': node['id'],
'type': node['type'],
'error': result.get('error')
})
# Update workflow nodes with created resource IDs
updated_nodes = WorkflowService.update_nodes_with_ids(nodes, created_resources)
workflow.nodes = json.dumps(updated_nodes)
db.session.commit()
return {
'success': len(errors) == 0,
'message': 'Deployment completed' if len(errors) == 0 else 'Deployment completed with errors',
'results': results,
'errors': errors,
'workflow': workflow.to_dict()
}
@staticmethod
def get_deployment_order(nodes, edges):
"""
Determine deployment order using topological sort.
Order: databases -> apps -> domains
Args:
nodes: List of workflow nodes
edges: List of workflow edges
Returns:
List of nodes in deployment order
"""
# Separate by type
databases = [n for n in nodes if n['type'] == 'database']
apps = [n for n in nodes if n['type'] in ('dockerApp', 'service')]
domains = [n for n in nodes if n['type'] == 'domain']
# Order: databases first, then apps, then domains
return databases + apps + domains
@staticmethod
def deploy_node(node, edges, user_id, created_resources):
"""
Deploy a single node based on its type.
Args:
node: Node to deploy
edges: All workflow edges
user_id: User ID for ownership
created_resources: Already created resources mapping
Returns:
dict with success, resourceId, etc.
"""
node_type = node.get('type')
node_data = node.get('data', {})
try:
if node_type == 'database':
return WorkflowService.deploy_database(node, user_id)
elif node_type == 'dockerApp':
return WorkflowService.deploy_docker_app(node, edges, user_id, created_resources)
elif node_type == 'service':
return WorkflowService.deploy_service(node, edges, user_id, created_resources)
elif node_type == 'domain':
return WorkflowService.deploy_domain(node, edges, user_id, created_resources)
else:
return {
'nodeId': node['id'],
'type': node_type,
'success': False,
'error': f'Unknown node type: {node_type}'
}
except Exception as e:
return {
'nodeId': node['id'],
'type': node_type,
'success': False,
'error': str(e)
}
@staticmethod
def deploy_database(node, user_id):
"""
Deploy a database node.
For MVP: Creates database using existing database service.
Note: Databases don't have a model in the current system,
so we track by name rather than ID.
Args:
node: Database node
user_id: User ID
Returns:
dict with deployment result
"""
node_data = node.get('data', {})
db_name = node_data.get('name', '').replace(' ', '_').replace('-', '_').lower()
db_type = node_data.get('type', 'mysql')
if not db_name:
return {
'nodeId': node['id'],
'type': 'database',
'success': False,
'error': 'Database name is required'
}
# Check if already deployed (has dbName set)
if node_data.get('deployed'):
return {
'nodeId': node['id'],
'type': 'database',
'success': True,
'resourceId': None,
'resourceName': db_name,
'message': 'Database already deployed'
}
try:
if db_type == 'mysql':
# Create MySQL database with user
result = DatabaseService.mysql_create_database(
db_name,
charset='utf8mb4',
collation='utf8mb4_unicode_ci'
)
if result.get('success'):
# Also create a user for the database
password = DatabaseService.generate_password()
user_result = DatabaseService.mysql_create_user(
username=db_name,
password=password,
host='%'
)
if user_result.get('success'):
# Grant privileges
DatabaseService.mysql_grant_privileges(
username=db_name,
database=db_name,
privileges='ALL',
host='%'
)
return {
'nodeId': node['id'],
'type': 'database',
'success': True,
'resourceId': None,
'resourceName': db_name,
'credentials': {
'database': db_name,
'username': db_name,
'password': password,
'host': 'localhost',
'port': 3306
}
}
else:
return {
'nodeId': node['id'],
'type': 'database',
'success': False,
'error': result.get('error', 'Failed to create MySQL database')
}
elif db_type == 'postgresql':
result = DatabaseService.pg_create_database(db_name)
if result.get('success'):
password = DatabaseService.generate_password()
user_result = DatabaseService.pg_create_user(
username=db_name,
password=password
)
if user_result.get('success'):
DatabaseService.pg_grant_privileges(
username=db_name,
database=db_name,
privileges='ALL'
)
return {
'nodeId': node['id'],
'type': 'database',
'success': True,
'resourceId': None,
'resourceName': db_name,
'credentials': {
'database': db_name,
'username': db_name,
'password': password,
'host': 'localhost',
'port': 5432
}
}
else:
return {
'nodeId': node['id'],
'type': 'database',
'success': False,
'error': result.get('error', 'Failed to create PostgreSQL database')
}
else:
# MongoDB and Redis would be handled as Docker containers
return {
'nodeId': node['id'],
'type': 'database',
'success': False,
'error': f'{db_type} databases should be deployed as Docker containers'
}
except Exception as e:
return {
'nodeId': node['id'],
'type': 'database',
'success': False,
'error': str(e)
}
@staticmethod
def deploy_docker_app(node, edges, user_id, created_resources):
"""
Deploy a Docker app node.
Args:
node: Docker app node
edges: Workflow edges (to find database connections)
user_id: User ID
created_resources: Already created resources
Returns:
dict with deployment result
"""
node_data = node.get('data', {})
app_name = node_data.get('name', '').replace(' ', '-').lower()
image = node_data.get('image', 'nginx:latest')
ports = node_data.get('ports', [])
memory = node_data.get('memory')
if not app_name:
return {
'nodeId': node['id'],
'type': 'dockerApp',
'success': False,
'error': 'App name is required'
}
# Check if already deployed
existing_app_id = node_data.get('appId')
if existing_app_id:
app = Application.query.get(existing_app_id)
if app:
# Update existing app
if image:
app.docker_image = image
if ports:
# Extract host port from first port mapping
first_port = ports[0] if ports else None
if first_port and ':' in first_port:
app.port = int(first_port.split(':')[0])
db.session.commit()
return {
'nodeId': node['id'],
'type': 'dockerApp',
'success': True,
'resourceId': app.id,
'resourceName': app.name,
'message': 'App updated'
}
# Find connected database for environment variables
env_vars = {}
for edge in edges:
if edge.get('source') == node['id']:
# This app connects to something
target_id = edge.get('target')
target_resource = created_resources.get(target_id)
if target_resource and target_resource['type'] == 'database':
# Get database credentials from the deployment result
credentials = target_resource.get('credentials', {})
if credentials:
env_vars.update({
'DB_HOST': credentials.get('host', 'localhost'),
'DB_NAME': credentials.get('database', ''),
'DB_USER': credentials.get('username', ''),
'DB_PASSWORD': credentials.get('password', ''),
'DB_PORT': str(credentials.get('port', 3306))
})
else:
# Fallback to naming convention
db_name = target_resource.get('resourceName', '')
env_vars.update({
'DB_HOST': 'localhost',
'DB_NAME': db_name,
'DB_USER': db_name,
'DB_PORT': '3306'
})
# Create the application record
app = Application(
name=app_name,
app_type='docker',
status='stopped',
docker_image=image,
user_id=user_id
)
# Extract host port from first port mapping
if ports:
first_port = ports[0]
if ':' in first_port:
app.port = int(first_port.split(':')[0])
else:
app.port = int(first_port)
# Set root path
app.root_path = os.path.join(WorkflowService.APP_BASE_PATH, app_name)
db.session.add(app)
db.session.commit()
# Create Docker compose files
docker_result = DockerService.create_docker_app(
app_path=app.root_path,
app_name=app_name,
image=image,
ports=ports if ports else None,
env=env_vars if env_vars else None
)
if not docker_result.get('success'):
# Rollback application creation
db.session.delete(app)
db.session.commit()
return {
'nodeId': node['id'],
'type': 'dockerApp',
'success': False,
'error': docker_result.get('error', 'Failed to create Docker app')
}
# Start the containers
start_result = DockerService.compose_up(app.root_path, detach=True)
if start_result.get('success'):
app.status = 'running'
db.session.commit()
else:
# App created but containers didn't start - still return success
# but include warning
return {
'nodeId': node['id'],
'type': 'dockerApp',
'success': True,
'resourceId': app.id,
'resourceName': app.name,
'port': app.port,
'rootPath': app.root_path,
'warning': f'App created but containers failed to start: {start_result.get("error", "Unknown error")}'
}
return {
'nodeId': node['id'],
'type': 'dockerApp',
'success': True,
'resourceId': app.id,
'resourceName': app.name,
'port': app.port,
'rootPath': app.root_path,
'status': 'running'
}
@staticmethod
def deploy_service(node, edges, user_id, created_resources):
"""
Deploy a service node.
Services are similar to Docker apps but may have specialized handling.
For now, treat them as Docker apps.
"""
return WorkflowService.deploy_docker_app(node, edges, user_id, created_resources)
@staticmethod
def deploy_domain(node, edges, user_id, created_resources):
"""
Deploy a domain node.
Domains must be connected to an app. Find the connected app
from edges and create the domain linked to that app.
Args:
node: Domain node
edges: Workflow edges
user_id: User ID
created_resources: Already created resources
Returns:
dict with deployment result
"""
node_data = node.get('data', {})
domain_name = node_data.get('name', '')
ssl_status = node_data.get('ssl', 'none')
if not domain_name:
return {
'nodeId': node['id'],
'type': 'domain',
'success': False,
'error': 'Domain name is required'
}
# Check if already deployed
existing_domain_id = node_data.get('domainId')
if existing_domain_id:
domain = Domain.query.get(existing_domain_id)
if domain:
return {
'nodeId': node['id'],
'type': 'domain',
'success': True,
'resourceId': domain.id,
'resourceName': domain.name,
'message': 'Domain already exists'
}
# Find the app this domain connects to
target_app_id = None
for edge in edges:
if edge.get('source') == node['id']:
target_id = edge.get('target')
target_resource = created_resources.get(target_id)
if target_resource and target_resource['type'] in ('dockerApp', 'service'):
target_app_id = target_resource.get('resourceId')
break
if not target_app_id:
return {
'nodeId': node['id'],
'type': 'domain',
'success': False,
'error': 'Domain must be connected to an app'
}
# Verify the app exists
app = Application.query.get(target_app_id)
if not app:
return {
'nodeId': node['id'],
'type': 'domain',
'success': False,
'error': 'Connected app not found'
}
# Check if domain already exists
existing = Domain.query.filter_by(name=domain_name).first()
if existing:
return {
'nodeId': node['id'],
'type': 'domain',
'success': False,
'error': f'Domain {domain_name} already exists'
}
# Create the domain
domain = Domain(
name=domain_name,
is_primary=True,
ssl_enabled=ssl_status == 'valid',
ssl_auto_renew=True,
application_id=target_app_id
)
db.session.add(domain)
db.session.commit()
# Create nginx config
from app.services.nginx_service import NginxService
if app.port:
all_domains = [d.name for d in Domain.query.filter_by(application_id=target_app_id).all()]
nginx_result = NginxService.create_site(
name=app.name,
app_type='docker',
domains=all_domains,
root_path=app.root_path or '',
port=app.port
)
if nginx_result.get('success'):
NginxService.enable_site(app.name)
return {
'nodeId': node['id'],
'type': 'domain',
'success': True,
'resourceId': domain.id,
'resourceName': domain.name,
'applicationId': target_app_id
}
@staticmethod
def update_nodes_with_ids(nodes, created_resources):
"""
Update node data with created resource IDs.
Args:
nodes: Original nodes list
created_resources: Mapping of nodeId to resource info
Returns:
Updated nodes list
"""
updated = []
for node in nodes:
node_copy = dict(node)
resource = created_resources.get(node['id'])
if resource:
data = dict(node_copy.get('data', {}))
if resource['type'] == 'dockerApp' or resource['type'] == 'service':
data['appId'] = resource.get('resourceId')
data['deployed'] = True
elif resource['type'] == 'domain':
data['domainId'] = resource.get('resourceId')
data['deployed'] = True
elif resource['type'] == 'database':
data['deployed'] = True
data['dbName'] = resource.get('resourceName')
node_copy['data'] = data
updated.append(node_copy)
return updated
backend/tests/test_jobs.py
+4 -108
View File
@@ -392,71 +392,6 @@ class TestDeploymentInstallHandler:
assert unified.correlation_id == dep.correlation_id
class TestWorkflowExecutionHandler:
"""Phase 6 — workflow executions run as a 'workflow.execute' unified job
instead of inline/threaded. Empty-node workflows run for real to success;
the failure path patches _run_execution to blow up."""
def _make_workflow(self, nodes='[]', edges='[]'):
from app.models.workflow import Workflow
wf = Workflow(name='wf-test', user_id=1, nodes=nodes, edges=edges,
is_active=True, trigger_type='manual')
db.session.add(wf)
db.session.commit()
return wf
def test_register_jobs_adds_handler(self, app):
from app.services.workflow_engine import WorkflowEngine, WORKFLOW_JOB_KIND
WorkflowEngine.register_jobs()
assert WORKFLOW_JOB_KIND == 'workflow.execute'
assert registry.is_registered('workflow.execute')
def test_enqueue_creates_running_row_and_job(self, app):
from app.services.workflow_engine import WorkflowEngine
from app.models.workflow import WorkflowExecution
wf = self._make_workflow()
execution_id = WorkflowEngine.enqueue_execution(wf.id, trigger_type='manual')
ex = WorkflowExecution.query.get(execution_id)
assert ex is not None and ex.status == 'running'
unified = Job.query.filter_by(kind='workflow.execute', owner_id=str(wf.id)).first()
assert unified is not None
assert unified.get_payload() == {'execution_id': execution_id}
assert unified.max_attempts == 1
def test_handler_runs_empty_workflow_to_success(self, app):
from app.services.workflow_engine import WorkflowEngine
from app.models.workflow import WorkflowExecution
WorkflowEngine.register_jobs()
wf = self._make_workflow(nodes='[]', edges='[]')
execution_id = WorkflowEngine.enqueue_execution(wf.id)
_drain_once()
assert WorkflowExecution.query.get(execution_id).status == 'success'
unified = Job.query.filter_by(kind='workflow.execute').first()
assert unified.status == Job.STATUS_SUCCEEDED
def test_handler_failure_marks_job_failed(self, app, monkeypatch):
from app.services.workflow_engine import WorkflowEngine
from app.models.workflow import WorkflowExecution
def boom(execution_id, nodes, edges):
raise RuntimeError('node blew up')
monkeypatch.setattr(WorkflowEngine, '_run_execution', staticmethod(boom))
WorkflowEngine.register_jobs()
wf = self._make_workflow(nodes='[{"id":"a","type":"trigger","data":{}}]', edges='[]')
execution_id = WorkflowEngine.enqueue_execution(wf.id)
_drain_once()
assert WorkflowExecution.query.get(execution_id).status == 'failed'
unified = Job.query.filter_by(kind='workflow.execute').first()
assert unified.status == Job.STATUS_FAILED
class TestBackupScheduleHandler:
"""Phase 7 — scheduled backups run as 'backup.run' unified jobs, enqueued by
the builtin.backup_scheduler tick (replacing the orphaned daemon loop).
@@ -522,46 +457,7 @@ class TestBackupScheduleHandler:
assert Job.query.get(bad.id).status == Job.STATUS_FAILED
class TestWorkflowEventDispatch:
"""Event-bus cleanup — WorkflowEventBus.emit enqueues a 'workflow.dispatch'
job instead of spawning a per-event thread; the handler fans out to matching
event-subscribed workflows."""
def _make_event_workflow(self, event_type):
import json as _json
from app.models.workflow import Workflow
wf = Workflow(name=f'evt-{event_type}', user_id=1, nodes='[]', edges='[]',
is_active=True, trigger_type='event',
trigger_config=_json.dumps({'eventType': event_type}))
db.session.add(wf)
db.session.commit()
return wf
def test_register_adds_dispatch_handler(self, app):
from app.services.workflow_engine import WorkflowEngine
WorkflowEngine.register_jobs()
assert registry.is_registered('workflow.dispatch')
def test_emit_enqueues_dispatch_job(self, app):
from app.services.workflow_engine import WorkflowEventBus
WorkflowEventBus.emit('high_cpu', {'value': 95})
evt_jobs = Job.query.filter_by(kind='workflow.dispatch').all()
assert len(evt_jobs) == 1
assert evt_jobs[0].get_payload() == {'event_type': 'high_cpu', 'data': {'value': 95}}
assert evt_jobs[0].owner_id == 'high_cpu'
def test_dispatch_fans_out_to_matching_workflow_only(self, app):
from app.services.workflow_engine import WorkflowEngine, WorkflowEventBus
from app.models.workflow import WorkflowExecution
WorkflowEngine.register_jobs()
match = self._make_event_workflow('high_cpu')
self._make_event_workflow('git_push') # different event — must be ignored
WorkflowEventBus.emit('high_cpu', {'value': 95})
_drain_once() # process workflow.dispatch -> enqueues workflow.execute
execs = WorkflowExecution.query.all()
assert len(execs) == 1
assert execs[0].workflow_id == match.id
assert execs[0].trigger_type == 'event'
assert Job.query.filter_by(kind='workflow.execute').count() == 1
# TestWorkflowEventDispatch removed in plan 45 Phase 4: the React-Flow Workflow
# Builder engine (WorkflowEngine / WorkflowEventBus) was retired. Its four event
# emitters were ported to EventService (health.check_failed, git.push,
# monitor.high_cpu/high_memory); those are covered in test_tramo_extension.py.