mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-03 10:36:45 +09:00
Juan DenisandClaude Opus 4.8
8428326a26
Generalize the Notification Bus pattern (thin {id} message -> fat DB row
-> single consumer) into a Job/WorkUnit layer on top of the Queue Bus, so
background work flows through one pipe instead of many ad-hoc threads.
- app/jobs: Job + ScheduledJob models, handler registry, JobService +
ScheduledJobService, one generic JobConsumer, one JobScheduler ticker,
and a plugin SDK surface (app.plugins_sdk.jobs).
- Migrate the 8 per-domain daemon scheduler threads (auto-sync,
snapshot-retention, workflow, health-check, wp-update, api-background,
pairing-prune, registrar-expiry) to ScheduledJob rows + built-in
handlers that reuse the exact prior check functions: cadence and
per-task de-dup unchanged, only the trigger moved.
- Retire the orphaned in-memory BackgroundJobService; repoint
/api/v1/performance/jobs at JobService.
- Add /api/v1/jobs (list/get/cancel/retry/stats; scheduled list/run/
toggle) as a single pane over all background work.
- tests/test_jobs.py: enqueue->run->succeed, dead-letter->failed,
unknown-kind, cancel, retry, scheduler tick/advance, built-ins.
No Alembic migration (tables created via create_all), avoiding a
multi-head clash with the in-flight telemetry work. Plan lives in
docs/ORCHESTRATION_ROADMAP.md (local/git-ignored).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
from flask import Blueprint, request, jsonify
|
|
from flask_jwt_extended import jwt_required
|
|
from app.services.cache_service import CacheService
|
|
from app.jobs.service import JobService
|
|
|
|
performance_bp = Blueprint('performance', __name__)
|
|
|
|
|
|
def get_current_user():
|
|
from flask_jwt_extended import get_jwt_identity
|
|
from app.models.user import User
|
|
return User.query.get(get_jwt_identity())
|
|
|
|
|
|
@performance_bp.route('/cache/stats', methods=['GET'])
|
|
@jwt_required()
|
|
def cache_stats():
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
return jsonify(CacheService.get_stats())
|
|
|
|
|
|
@performance_bp.route('/cache/flush', methods=['POST'])
|
|
@jwt_required()
|
|
def cache_flush():
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
CacheService.flush()
|
|
return jsonify({'message': 'Cache flushed'})
|
|
|
|
|
|
@performance_bp.route('/jobs', methods=['GET'])
|
|
@jwt_required()
|
|
def list_jobs():
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
# Keyed by id to preserve the prior {job_id: info} response shape.
|
|
return jsonify({'jobs': {j.id: j.to_dict() for j in JobService.list(limit=200)}})
|
|
|
|
|
|
@performance_bp.route('/jobs/<job_id>', methods=['GET'])
|
|
@jwt_required()
|
|
def get_job(job_id):
|
|
job = JobService.get(job_id)
|
|
if not job:
|
|
return jsonify({'error': 'Job not found'}), 404
|
|
return jsonify({'job_id': job_id, **job.to_dict(include_payload=True)})
|
|
|
|
|
|
@performance_bp.route('/jobs/stats', methods=['GET'])
|
|
@jwt_required()
|
|
def job_stats():
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
stats = JobService.stats()
|
|
return jsonify({
|
|
'total_jobs': stats['total'],
|
|
'by_status': stats['by_status'],
|
|
'by_kind': stats['by_kind'],
|
|
'running': stats['by_status'].get('running', 0),
|
|
'queue_size': stats['by_status'].get('pending', 0),
|
|
})
|
|
|
|
|
|
@performance_bp.route('/jobs/cleanup', methods=['POST'])
|
|
@jwt_required()
|
|
def cleanup_jobs():
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
deleted = JobService.cleanup_old()
|
|
return jsonify({'message': f'Cleaned up {deleted} old job(s)'})
|