diff --git a/server.py b/server.py index 7515716..31b23f4 100644 --- a/server.py +++ b/server.py @@ -1,1017 +1 @@ -#!/usr/bin/env python3 -"""Extension Platform API Server. - -Runs on Coolify. Connects to PostgreSQL via Docker internal network. -Builds extensions from Gitea (no local files needed). - -Pipeline endpoints (called by n8n): - POST /publish — Download from Gitea, build ZIP, publish to Composer Registry + Release - POST /deploy — Deploy extension to a specific customer via EspoCRM API - POST /deploy-all — Deploy to all licensed customers - POST /sync — Sync installed extensions from EspoCRM servers with platform DB - POST /sync-registry — Discover extensions from Gitea org and sync with DB - -Dashboard API: - GET /api/stats - GET /api/extensions - GET /api/extensions/:id/versions - GET /api/customers - POST /api/customers - PUT /api/customers/:id - GET /api/customers/:id - POST /api/customers/:id/extensions - DELETE /api/customers/:id/extensions/:ext_id - GET /api/deployments -""" -from http.server import HTTPServer, BaseHTTPRequestHandler -import json, os, re, ssl, mimetypes, tempfile, shutil, zipfile, io -import base64, urllib.request, urllib.error, subprocess -import psycopg2, psycopg2.extras - -# --- Config from environment --- -DB_HOST = os.environ.get('DB_HOST', 'pss404gosg8oo0o8k40o4wco') -DB_PORT = int(os.environ.get('DB_PORT', '5432')) -DB_NAME = os.environ.get('DB_NAME', 'extension_platform') -DB_USER = os.environ.get('DB_USER', 'postgres') -DB_PASS = os.environ.get('DB_PASS', '9cugHhKR8bCFXzm') - -GITEA_URL = os.environ.get('GITEA_URL', 'https://gitea.dev.marcus-law.co.il') -GITEA_TOKEN = os.environ.get('GITEA_TOKEN', '90acb3ebc0f98edfb7a373ef9902bde1cc8eff96') -GITEA_ORG = os.environ.get('GITEA_ORG', 'espocrm-extensions') - -DIST_DIR = os.environ.get('DIST_DIR', '/app/dashboard/dist') - -# Repos in the Gitea org that are NOT extensions (tools, base images, etc.) -GITEA_NON_EXTENSION_REPOS = {'espocrm-base', 'espo-core', 'espo-tools', 'extension-dashboard', 'extension-platform'} - -def _pascal_to_kebab(name): - """Convert PascalCase to kebab-case: LegalAssistance -> legal-assistance""" - return re.sub(r'(?=8.1'])[0] if isinstance(manifest.get('php'), list) else '>=8.1'}, - "extra": { - "espocrm-module": module, - "espocrm-min-version": manifest.get('acceptableVersions', ['>=8.0.0'])[0] if isinstance(manifest.get('acceptableVersions'), list) else '>=8.0.0' - } - } - - # 4. Build extension ZIP with composer.json at root - ext_zip_path = os.path.join(tmpdir, f'{module}-{version}.zip') - with zipfile.ZipFile(ext_zip_path, 'w', zipfile.ZIP_DEFLATED) as ext_zip: - ext_zip.writestr('composer.json', json.dumps(composer, indent=2)) - - # Re-read archive and copy contents without the top-level dir - with zipfile.ZipFile(archive_path) as src_zip: - for item in src_zip.namelist(): - if item.endswith('/'): - continue - # Strip top-level directory - rel_path = item[len(prefix):] if prefix and item.startswith(prefix) else item - if not rel_path or rel_path.startswith('.git'): - continue - ext_zip.writestr(rel_path, src_zip.read(item)) - - with open(ext_zip_path, 'rb') as f: - ext_zip_data = f.read() - - # 5. Publish to Gitea Composer Registry - registry_url = f"{GITEA_URL}/api/packages/{GITEA_ORG}/composer" - reg_result = gitea_upload(registry_url, ext_zip_data, f'{module}-{version}.zip') - reg_status = 'published' if reg_result.get('code') == 201 else 'exists' if reg_result.get('code') == 409 else f"error:{reg_result.get('code')}" - - # 6. Create Gitea Release + upload asset - release_result = gitea_api('POST', f'/repos/{GITEA_ORG}/{module}/releases', { - 'tag_name': tag, 'name': tag, 'body': f'Release {version}', 'draft': False, 'prerelease': False - }) - release_id = release_result.get('id') - asset_status = 'skipped' - # If release already exists, fetch its ID and delete old ZIP assets - if not release_id or release_result.get('error'): - existing = gitea_api('GET', f'/repos/{GITEA_ORG}/{module}/releases/tags/{tag}') - release_id = existing.get('id') - if release_id: - for old_asset in existing.get('assets', []): - if old_asset['name'].endswith('.zip'): - gitea_api('DELETE', f'/repos/{GITEA_ORG}/{module}/releases/{release_id}/assets/{old_asset["id"]}') - if release_id: - # Upload asset using multipart - boundary = '----FormBoundary7MA4YWxkTrZu0gW' - body = ( - f'--{boundary}\r\n' - f'Content-Disposition: form-data; name="attachment"; filename="{module}-{version}.zip"\r\n' - f'Content-Type: application/zip\r\n\r\n' - ).encode() + ext_zip_data + f'\r\n--{boundary}--\r\n'.encode() - - asset_url = f"{GITEA_URL}/api/v1/repos/{GITEA_ORG}/{module}/releases/{release_id}/assets?name={module}-{version}.zip" - headers = { - 'Authorization': f'token {GITEA_TOKEN}', - 'Content-Type': f'multipart/form-data; boundary={boundary}' - } - req = urllib.request.Request(asset_url, data=body, headers=headers, method='POST') - try: - urllib.request.urlopen(req, context=SSL_CTX, timeout=60) - asset_status = 'uploaded' - except urllib.error.HTTPError as e: - asset_status = f'error:{e.code}' - else: - asset_status = f'release_failed:{release_result.get("message","")[:100]}' - - # 7. Update DB - db_ok = False - try: - conn = get_db() - cur = conn.cursor() - cur.execute("""INSERT INTO extension_versions (extension_id, version, zip_filename, git_tag) - SELECT id, %s, %s, %s FROM extensions WHERE name = %s - ON CONFLICT (extension_id, version) DO NOTHING""", - (version, f'{module}-{version}.zip', tag, module)) - cur.execute("UPDATE extensions SET latest_version = %s, updated_at = NOW() WHERE name = %s", - (version, module)) - conn.commit() - cur.close() - conn.close() - db_ok = True - except Exception as e: - pass - - return { - 'status': 'ok', 'module': module, 'version': version, - 'composer_name': composer_name, - 'registry': reg_status, 'release_id': release_id, 'asset': asset_status, - 'db_updated': db_ok, 'zip_size': len(ext_zip_data) - } - - -# --- OPcache clear --- -def _clear_opcache(customer, espo_url, api_key): - """Clear PHP opcache after extension install. Tries HTTP endpoint first, SSH fallback.""" - # Tier 1: HTTP-based opcache_reset via opcache-clear.php - try: - oc_req = urllib.request.Request( - f'{espo_url}/opcache-clear.php', - headers={'Espo-Authorization': api_key, 'X-Api-Key': api_key}, - method='GET', - ) - oc_resp = urllib.request.urlopen(oc_req, context=SSL_CTX, timeout=10) - result = json.loads(oc_resp.read()) - if result.get('success'): - return 'http' - except Exception: - pass - - # Tier 2: SSH + docker exec apachectl graceful - ssh_host = customer.get('ssh_host') - container = customer.get('docker_container') - if ssh_host and container: - ssh_user = customer.get('ssh_user', 'root') - ssh_key = customer.get('ssh_key') - key_file = None - try: - ssh_cmd = ['ssh', '-o', 'StrictHostKeyChecking=no', '-o', 'ConnectTimeout=5'] - if ssh_key: - key_file = tempfile.NamedTemporaryFile(mode='w', suffix='.key', delete=False) - key_file.write(ssh_key) - key_file.close() - os.chmod(key_file.name, 0o600) - ssh_cmd.extend(['-i', key_file.name]) - ssh_cmd.extend([ - f'{ssh_user}@{ssh_host}', - f'sudo docker exec {container} apachectl graceful', - ]) - proc = subprocess.run(ssh_cmd, timeout=15, capture_output=True) - if proc.returncode == 0: - return 'ssh' - except Exception: - pass - finally: - if key_file: - try: - os.unlink(key_file.name) - except OSError: - pass - - # Tier 3: warning - print(f'[WARN] Could not clear opcache for {customer.get("name", "?")} — new PHP files may not load immediately') - return 'failed' - - -# --- Deploy logic --- -def deploy_to_customer(customer, module, version): - """Deploy extension to a customer via EspoCRM API. Downloads ZIP from Gitea.""" - espo_url = customer['espocrm_url'].rstrip('/') - api_key = customer['espocrm_api_key'] - - # Download ZIP from Gitea release - tag = f'v{version}' - releases = gitea_api('GET', f'/repos/{GITEA_ORG}/{module}/releases/tags/{tag}') - if releases.get('error'): - return {'status': 'error', 'step': 'find_release', 'message': releases.get('message','')} - - zip_url = None - for asset in releases.get('assets', []): - if asset['name'].endswith('.zip'): - zip_url = asset['browser_download_url'] - # Don't break - take the last ZIP (newest upload) in case of duplicates - - if not zip_url: - return {'status': 'error', 'step': 'find_zip', 'message': 'No ZIP asset in release'} - - # Download ZIP - req = urllib.request.Request(zip_url, headers={'Authorization': f'token {GITEA_TOKEN}'}) - try: - resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=60) - zip_data = resp.read() - except Exception as e: - return {'status': 'error', 'step': 'download_zip', 'message': str(e)} - - zip_base64 = base64.b64encode(zip_data).decode() - headers = {'Espo-Authorization': api_key, 'Content-Type': 'application/json'} - - # Upload to EspoCRM - upload_data = json.dumps({'file': f'data:application/zip;base64,{zip_base64}'}).encode() - req = urllib.request.Request(f'{espo_url}/api/v1/Extension/action/upload', - data=upload_data, headers=headers, method='POST') - try: - resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=60) - ext_id = json.loads(resp.read()).get('id') - except urllib.error.HTTPError as e: - return {'status': 'error', 'step': 'upload', 'code': e.code, 'message': e.read().decode()[:200]} - except Exception as e: - return {'status': 'error', 'step': 'upload', 'message': str(e)} - - # Install - req2 = urllib.request.Request(f'{espo_url}/api/v1/Extension/action/install', - data=json.dumps({'id': ext_id}).encode(), headers=headers, method='POST') - try: - urllib.request.urlopen(req2, context=SSL_CTX, timeout=120) - except urllib.error.HTTPError as e: - return {'status': 'error', 'step': 'install', 'code': e.code, 'message': e.read().decode()[:200]} - except Exception as e: - return {'status': 'error', 'step': 'install', 'message': str(e)} - - # Rebuild - req3 = urllib.request.Request(f'{espo_url}/api/v1/Admin/rebuild', - data=b'true', headers=headers, method='POST') - try: - urllib.request.urlopen(req3, context=SSL_CTX, timeout=60) - except: - pass - - # Clear PHP opcache (new PHP files won't load until opcache refreshes) - opcache_status = _clear_opcache(customer, espo_url, api_key) - - return {'status': 'ok', 'ext_id': ext_id, 'opcache': opcache_status} - - -# --- Sync logic --- -def sync_customer_extensions(customer): - """Query a customer's EspoCRM to see what's really installed, compare with platform DB.""" - espo_url = customer['espocrm_url'].rstrip('/') - api_key = customer['espocrm_api_key'] - cid = customer['id'] - - # 0. Ensure all active extensions are assigned to this customer - query("""INSERT INTO customer_extensions (customer_id, extension_id, auto_update, license_status) - SELECT %s, id, true, 'active' FROM extensions WHERE is_active - ON CONFLICT (customer_id, extension_id) DO NOTHING""", (cid,), fetch=None) - - # 1. Query EspoCRM for installed extensions - headers = {'Espo-Authorization': api_key} - req = urllib.request.Request(f'{espo_url}/api/v1/Extension?select=name,version&maxSize=200', - headers=headers) - try: - resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=30) - data = json.loads(resp.read()) - server_exts = {e['name']: e.get('version', '') for e in data.get('list', [])} - except Exception as e: - return {'customer': customer['name'], 'status': 'error', - 'message': f'Cannot reach EspoCRM: {str(e)[:200]}'} - - # 2. Get what platform thinks is installed - platform_exts = query(""" - SELECT e.name, ce.installed_version, ce.license_status - FROM customer_extensions ce JOIN extensions e ON ce.extension_id = e.id - WHERE ce.customer_id = %s AND ce.license_status = 'active'""", (cid,)) - platform_map = {r['name']: r['installed_version'] for r in platform_exts} - - # 3. Compare - drifts = [] - updated = 0 - - # Check each platform-tracked extension - for ext_name, platform_ver in platform_map.items(): - server_ver = server_exts.get(ext_name) - if server_ver is None: - drifts.append({'extension': ext_name, 'type': 'missing_on_server', - 'platform': platform_ver, 'server': None}) - if platform_ver is not None: - query("""UPDATE customer_extensions SET installed_version = NULL, installed_at = NULL, updated_at = NOW() - FROM extensions e WHERE customer_extensions.extension_id = e.id - AND customer_extensions.customer_id = %s AND e.name = %s""", - (cid, ext_name), fetch=None) - updated += 1 - elif server_ver != platform_ver: - drifts.append({'extension': ext_name, 'type': 'version_mismatch', - 'platform': platform_ver, 'server': server_ver}) - # Update platform DB to reflect reality - query("""UPDATE customer_extensions SET installed_version = %s, updated_at = NOW() - FROM extensions e WHERE customer_extensions.extension_id = e.id - AND customer_extensions.customer_id = %s AND e.name = %s""", - (server_ver, cid, ext_name), fetch=None) - updated += 1 - - # Check for extensions on server not tracked by platform - for ext_name, server_ver in server_exts.items(): - if ext_name not in platform_map and ext_name in MODULES: - drifts.append({'extension': ext_name, 'type': 'extra_on_server', - 'platform': None, 'server': server_ver}) - - return { - 'customer': customer['name'], 'customer_id': cid, 'status': 'ok', - 'server_count': len(server_exts), 'platform_count': len(platform_map), - 'drifts': drifts, 'drift_count': len(drifts), 'updated': updated - } - - -def health_check_all(): - """Run comprehensive health checks on all extensions and customers.""" - issues = [] - - # 1. Check Gitea release integrity - manifest version matches tag - extensions = query("SELECT name, latest_version FROM extensions WHERE is_active") - for ext in extensions: - module = ext['name'] - version = ext['latest_version'] - if not version: - continue - tag = f'v{version}' - release = gitea_api('GET', f'/repos/{GITEA_ORG}/{module}/releases/tags/{tag}') - if release.get('error'): - issues.append({'check': 'gitea_release', 'extension': module, 'severity': 'error', - 'message': f'Release {tag} not found on Gitea'}) - continue - - zip_assets = [a for a in release.get('assets', []) if a['name'].endswith('.zip')] - if not zip_assets: - issues.append({'check': 'gitea_release', 'extension': module, 'severity': 'error', - 'message': f'No ZIP asset in release {tag}'}) - continue - - # Check for duplicate ZIP assets - if len(zip_assets) > 1: - issues.append({'check': 'duplicate_assets', 'extension': module, 'severity': 'warning', - 'message': f'{len(zip_assets)} ZIP assets in release {tag} — should be 1'}) - - # Download last ZIP and verify manifest version matches tag - last_zip_url = zip_assets[-1]['browser_download_url'] - try: - req = urllib.request.Request(last_zip_url, headers={'Authorization': f'token {GITEA_TOKEN}'}) - resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=30) - zip_data = resp.read() - with zipfile.ZipFile(io.BytesIO(zip_data)) as zf: - manifest_files = [n for n in zf.namelist() if n.endswith('manifest.json')] - if manifest_files: - manifest = json.loads(zf.read(manifest_files[0])) - manifest_ver = manifest.get('version', '') - if manifest_ver != version: - issues.append({'check': 'version_mismatch', 'extension': module, 'severity': 'critical', - 'message': f'ZIP manifest says {manifest_ver} but release tag is {tag} — wrong ZIP!'}) - else: - issues.append({'check': 'manifest_missing', 'extension': module, 'severity': 'error', - 'message': f'No manifest.json found inside ZIP for {tag}'}) - except Exception as e: - issues.append({'check': 'zip_verify', 'extension': module, 'severity': 'error', - 'message': f'Failed to verify ZIP: {str(e)[:150]}'}) - - # 2. Check EspoCRM API health for each customer - customers = query("""SELECT id, name, espocrm_url, espocrm_api_key - FROM customers WHERE is_active AND espocrm_api_key IS NOT NULL""") - customer_results = [] - for c in customers: - espo_url = c['espocrm_url'].rstrip('/') - headers = {'Espo-Authorization': c['espocrm_api_key']} - cr = {'customer': c['name'], 'url': espo_url, 'api_reachable': False, 'extensions_loaded': False} - - # API health - try: - req = urllib.request.Request(f'{espo_url}/api/v1/App/user', headers=headers) - resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=15) - data = json.loads(resp.read()) - cr['api_reachable'] = True - cr['espo_version'] = data.get('version', 'unknown') - except Exception as e: - issues.append({'check': 'api_health', 'extension': None, 'severity': 'critical', - 'message': f'{c["name"]}: EspoCRM API unreachable — {str(e)[:150]}'}) - customer_results.append(cr) - continue - - # Check extensions are loaded - try: - req = urllib.request.Request(f'{espo_url}/api/v1/Extension?select=name,version&maxSize=200', headers=headers) - resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=15) - data = json.loads(resp.read()) - server_exts = {e['name']: e.get('version', '') for e in data.get('list', [])} - cr['extensions_loaded'] = True - cr['extension_count'] = len(server_exts) - - # Check installed extensions actually match what platform expects - platform_exts = query("""SELECT e.name, ce.installed_version FROM customer_extensions ce - JOIN extensions e ON ce.extension_id = e.id - WHERE ce.customer_id = %s AND ce.installed_version IS NOT NULL""", (c['id'],)) - for pe in platform_exts: - server_ver = server_exts.get(pe['name']) - if server_ver and server_ver != pe['installed_version']: - issues.append({'check': 'install_integrity', 'extension': pe['name'], 'severity': 'warning', - 'message': f'{c["name"]}: platform says {pe["installed_version"]} but server has {server_ver}'}) - except Exception as e: - issues.append({'check': 'ext_query', 'extension': None, 'severity': 'error', - 'message': f'{c["name"]}: failed to query extensions — {str(e)[:150]}'}) - - customer_results.append(cr) - - # 3. Check for stale deployments (latest deploy per extension+customer says ok but version doesn't match) - stale = query("""SELECT latest.version as deployed_version, e.name as extension, c.name as customer, - ce.installed_version, latest.started_at - FROM ( - SELECT DISTINCT ON (customer_id, extension_id) - customer_id, extension_id, version, status, started_at - FROM deployment_log - WHERE started_at > NOW() - INTERVAL '24 hours' - ORDER BY customer_id, extension_id, started_at DESC - ) latest - JOIN extensions e ON latest.extension_id = e.id - JOIN customers c ON latest.customer_id = c.id - LEFT JOIN customer_extensions ce ON ce.customer_id = latest.customer_id AND ce.extension_id = latest.extension_id - WHERE latest.status = 'ok' - AND (ce.installed_version IS NULL OR ce.installed_version != latest.version)""") - for s in stale: - issues.append({'check': 'stale_deploy', 'extension': s['extension'], 'severity': 'warning', - 'message': f'{s["customer"]}: deployed {s["deployed_version"]} but installed shows {s["installed_version"] or "NULL"}'}) - - critical = len([i for i in issues if i['severity'] == 'critical']) - errors = len([i for i in issues if i['severity'] == 'error']) - warnings = len([i for i in issues if i['severity'] == 'warning']) - - return { - 'status': 'critical' if critical else 'error' if errors else 'warning' if warnings else 'ok', - 'total_issues': len(issues), - 'critical': critical, 'errors': errors, 'warnings': warnings, - 'issues': issues, - 'customers': customer_results - } - - -def sync_all_customers(): - """Run sync for all active customers.""" - customers = query("""SELECT id, name, espocrm_url, espocrm_api_key - FROM customers WHERE is_active AND espocrm_api_key IS NOT NULL""") - results = [] - total_drifts = 0 - for c in customers: - r = sync_customer_extensions(c) - results.append(r) - total_drifts += r.get('drift_count', 0) - return { - 'status': 'ok', 'customers_checked': len(results), - 'total_drifts': total_drifts, 'results': results - } - - -# --- HTTP Handler --- -class APIHandler(BaseHTTPRequestHandler): - - def do_GET(self): - path = self.path.split('?')[0] - if path == '/api/extensions': - self._get_extensions() - elif re.match(r'/api/extensions/\d+/versions', path): - self._get_extension_versions(int(re.search(r'/api/extensions/(\d+)/versions', path).group(1))) - elif path == '/api/customers': - self._get_customers() - elif re.match(r'/api/customers/\d+$', path): - self._get_customer(int(re.search(r'/api/customers/(\d+)', path).group(1))) - elif path == '/api/deployments': - self._get_deployments() - elif path == '/api/stats': - self._get_stats() - else: - self._serve_static(path) - - def do_POST(self): - body = self._read_body() - path = self.path.split('?')[0] - if path == '/publish': - self._handle_publish(body) - elif path == '/deploy': - self._handle_deploy(body) - elif path == '/deploy-all': - self._handle_deploy_all(body) - elif path == '/api/customers': - self._create_customer(body) - elif re.match(r'/api/customers/\d+/extensions', path): - self._assign_extension(int(re.search(r'/api/customers/(\d+)/extensions', path).group(1)), body) - elif path == '/api/deploy': - self._handle_deploy(body) - elif path == '/sync': - self._handle_sync(body) - elif path == '/health-check': - self._handle_health_check() - elif path == '/sync-registry': - self._handle_sync_registry() - else: - self._respond(404, {'error': 'Not found'}) - - def do_PUT(self): - body = self._read_body() - path = self.path.split('?')[0] - if re.match(r'/api/customers/\d+$', path): - self._update_customer(int(re.search(r'/api/customers/(\d+)', path).group(1)), body) - else: - self._respond(404, {'error': 'Not found'}) - - def do_DELETE(self): - path = self.path.split('?')[0] - if re.match(r'/api/customers/\d+/extensions/\d+$', path): - m = re.search(r'/api/customers/(\d+)/extensions/(\d+)', path) - self._remove_extension(int(m.group(1)), int(m.group(2))) - elif re.match(r'/api/customers/\d+$', path): - cid = int(re.search(r'/api/customers/(\d+)', path).group(1)) - self._delete_customer(cid) - else: - self._respond(404, {'error': 'Not found'}) - - def do_OPTIONS(self): - self.send_response(204) - self._cors_headers() - self.end_headers() - - # --- Dashboard API --- - def _get_extensions(self): - self._respond(200, query(""" - SELECT e.*, (SELECT COUNT(*) FROM extension_versions ev WHERE ev.extension_id = e.id) as version_count, - (SELECT COUNT(*) FROM customer_extensions ce WHERE ce.extension_id = e.id AND ce.license_status = 'active') as customer_count - FROM extensions e WHERE e.is_active ORDER BY e.name""")) - - def _get_extension_versions(self, ext_id): - self._respond(200, query("SELECT * FROM extension_versions WHERE extension_id = %s ORDER BY released_at DESC", (ext_id,))) - - def _get_customers(self): - rows = query("""SELECT c.id, c.name, c.email, c.espocrm_url, c.is_active, c.notes, c.created_at, c.updated_at, - c.coolify_url, c.coolify_app_uuid, - (SELECT COUNT(*) FROM customer_extensions ce WHERE ce.customer_id = c.id AND ce.license_status = 'active') as extension_count, - (SELECT COUNT(*) FROM customer_extensions ce WHERE ce.customer_id = c.id AND ce.license_status = 'active' AND ce.installed_version IS NOT NULL) as installed_count, - (SELECT MAX(dl.started_at) FROM deployment_log dl WHERE dl.customer_id = c.id) as last_deploy - FROM customers c ORDER BY c.name""") - self._respond(200, rows) - - def _get_customer(self, cid): - customer = query("SELECT id, name, email, espocrm_url, is_active, notes, coolify_url, coolify_app_uuid, created_at, updated_at FROM customers WHERE id = %s", (cid,), fetch='one') - if not customer: - self._respond(404, {'error': 'Not found'}); return - customer['extensions'] = query(""" - SELECT e.id, e.name, e.composer_name, e.latest_version, ce.installed_version, ce.license_status, ce.auto_update, ce.installed_at - FROM customer_extensions ce JOIN extensions e ON ce.extension_id = e.id WHERE ce.customer_id = %s AND e.is_active ORDER BY e.name""", (cid,)) - customer['dependencies'] = DEPENDENCIES - customer['install_order'] = _compute_install_order() - self._respond(200, customer) - - def _create_customer(self, body): - for f in ['name', 'espocrm_url', 'espocrm_api_key']: - if not body.get(f): - self._respond(400, {'error': f'{f} required'}); return - row = query("""INSERT INTO customers (name, email, espocrm_url, espocrm_api_key, - coolify_url, coolify_token, coolify_app_uuid, notes, - ssh_host, ssh_user, ssh_key, docker_container) - VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) RETURNING id""", - (body['name'], body.get('email'), body['espocrm_url'], body['espocrm_api_key'], - body.get('coolify_url'), body.get('coolify_token'), body.get('coolify_app_uuid'), body.get('notes'), - body.get('ssh_host'), body.get('ssh_user'), body.get('ssh_key'), body.get('docker_container')), fetch='one') - cid = row['id'] - # Auto-assign all active extensions to new customer - all_exts = query("SELECT id FROM extensions WHERE is_active", fetch='all') - for ext in all_exts: - query("""INSERT INTO customer_extensions (customer_id, extension_id, auto_update, license_status) - VALUES (%s, %s, true, 'active') ON CONFLICT DO NOTHING""", - (cid, ext['id']), fetch=None) - self._respond(201, {'id': cid, 'name': body['name']}) - - def _update_customer(self, cid, body): - fields, values = [], [] - for col in ['name','email','espocrm_url','espocrm_api_key','coolify_url','coolify_token','coolify_app_uuid','is_active','notes','ssh_host','ssh_user','ssh_key','docker_container']: - if col in body: - fields.append(f"{col} = %s"); values.append(body[col]) - if not fields: - self._respond(400, {'error': 'No fields'}); return - fields.append("updated_at = NOW()"); values.append(cid) - query(f"UPDATE customers SET {', '.join(fields)} WHERE id = %s", values, fetch=None) - self._respond(200, {'status': 'ok'}) - - def _assign_extension(self, cid, body): - if not body.get('extension_id'): - self._respond(400, {'error': 'extension_id required'}); return - query("""INSERT INTO customer_extensions (customer_id, extension_id, auto_update, license_status) - VALUES (%s, %s, %s, %s) ON CONFLICT (customer_id, extension_id) DO UPDATE - SET license_status = EXCLUDED.license_status, auto_update = EXCLUDED.auto_update, updated_at = NOW()""", - (cid, body['extension_id'], body.get('auto_update', True), body.get('license_status', 'active')), fetch=None) - self._respond(201, {'status': 'ok'}) - - def _delete_customer(self, cid): - query("DELETE FROM customer_extensions WHERE customer_id = %s", (cid,), fetch=None) - query("DELETE FROM deployment_log WHERE customer_id = %s", (cid,), fetch=None) - query("DELETE FROM customers WHERE id = %s", (cid,), fetch=None) - self._respond(200, {'status': 'ok'}) - - def _remove_extension(self, cid, ext_id): - query("DELETE FROM customer_extensions WHERE customer_id = %s AND extension_id = %s", (cid, ext_id), fetch=None) - self._respond(200, {'status': 'ok'}) - - def _get_deployments(self): - qs = self.path.split('?')[1] if '?' in self.path else '' - params = dict(p.split('=') for p in qs.split('&') if '=' in p) if qs else {} - limit = min(int(params.get('limit', 50)), 200) - self._respond(200, query("""SELECT dl.*, c.name as customer_name, e.name as extension_name - FROM deployment_log dl JOIN customers c ON dl.customer_id = c.id JOIN extensions e ON dl.extension_id = e.id - ORDER BY dl.started_at DESC LIMIT %s""", (limit,))) - - def _get_stats(self): - self._respond(200, query("""SELECT - (SELECT COUNT(*) FROM extensions WHERE is_active) as total_extensions, - (SELECT COUNT(*) FROM customers WHERE is_active) as total_customers, - (SELECT COUNT(*) FROM customer_extensions WHERE license_status = 'active') as total_licenses, - (SELECT COUNT(*) FROM deployment_log WHERE status = 'ok') as successful_deploys, - (SELECT COUNT(*) FROM deployment_log WHERE status != 'ok') as failed_deploys, - (SELECT MAX(started_at) FROM deployment_log) as last_deploy""", fetch='one')) - - # --- Pipeline --- - def _handle_publish(self, body): - module, version = body.get('module',''), body.get('version','') - if not module or not version: - self._respond(400, {'error': 'module and version required'}); return - result = build_and_publish(module, version) - self._respond(200, result) - - def _handle_deploy(self, body): - cid, module, version = body.get('customer_id'), body.get('module',''), body.get('version','') - if not all([cid, module, version]): - self._respond(400, {'error': 'customer_id, module, version required'}); return - customer = query("SELECT id, name, espocrm_url, espocrm_api_key, ssh_host, ssh_user, ssh_key, docker_container FROM customers WHERE id = %s AND is_active", (cid,), fetch='one') - if not customer: - self._respond(404, {'error': 'Customer not found'}); return - result = deploy_to_customer(customer, module, version) - # Log - query("""INSERT INTO deployment_log (customer_id, extension_id, version, status, trigger_type, error_message) - SELECT %s, e.id, %s, %s, 'manual', %s FROM extensions e WHERE e.name = %s""", - (cid, version, result['status'], result.get('message',''), module), fetch=None) - if result['status'] == 'ok': - query("""UPDATE customer_extensions SET installed_version = %s, installed_at = NOW() - FROM extensions e WHERE customer_extensions.extension_id = e.id - AND customer_extensions.customer_id = %s AND e.name = %s""", (version, cid, module), fetch=None) - self._respond(200, {**result, 'customer': customer['name'], 'module': module, 'version': version}) - - def _handle_deploy_all(self, body): - module, version = body.get('module',''), body.get('version','') - if not module or not version: - self._respond(400, {'error': 'module and version required'}); return - customers = query("""SELECT c.id, c.name, c.espocrm_url, c.espocrm_api_key, - c.ssh_host, c.ssh_user, c.ssh_key, c.docker_container FROM customers c - JOIN customer_extensions ce ON c.id = ce.customer_id JOIN extensions e ON ce.extension_id = e.id - WHERE e.name = %s AND ce.auto_update AND ce.license_status = 'active' AND c.is_active""", (module,)) - results = [] - for c in customers: - r = deploy_to_customer(c, module, version) - results.append({**r, 'customer': c['name']}) - query("""INSERT INTO deployment_log (customer_id, extension_id, version, status, trigger_type, error_message) - SELECT %s, e.id, %s, %s, 'auto', %s FROM extensions e WHERE e.name = %s""", - (c['id'], version, r['status'], r.get('message',''), module), fetch=None) - if r['status'] == 'ok': - query("""UPDATE customer_extensions SET installed_version = %s, installed_at = NOW() - FROM extensions e WHERE customer_extensions.extension_id = e.id - AND customer_extensions.customer_id = %s AND e.name = %s""", (version, c['id'], module), fetch=None) - self._respond(200, {'status': 'ok', 'module': module, 'version': version, - 'deployed_to': len(results), 'results': results}) - - # --- Sync --- - def _handle_sync(self, body): - cid = body.get('customer_id') - if cid: - customer = query("SELECT id, name, espocrm_url, espocrm_api_key FROM customers WHERE id = %s AND is_active", (cid,), fetch='one') - if not customer: - self._respond(404, {'error': 'Customer not found'}); return - result = sync_customer_extensions(customer) - self._respond(200, result) - else: - result = sync_all_customers() - self._respond(200, result) - - # --- Health Check --- - def _handle_health_check(self): - result = health_check_all() - self._respond(200, result) - - # --- Registry Sync --- - def _handle_sync_registry(self): - result = sync_extension_registry() - self._respond(200, result) - - # --- Helpers --- - def _read_body(self): - length = int(self.headers.get('Content-Length', 0)) - return json.loads(self.rfile.read(length)) if length > 0 else {} - - def _serve_static(self, path): - if path == '/': - path = '/index.html' - file_path = os.path.join(DIST_DIR, path.lstrip('/')) - if not os.path.isfile(file_path): - # Only fallback to index.html for routes, not for asset files - if path.startswith('/assets/') or '.' in path.split('/')[-1]: - self._respond(404, {'error': 'Not found'}); return - file_path = os.path.join(DIST_DIR, 'index.html') - try: - with open(file_path, 'rb') as f: - content = f.read() - self.send_response(200) - self.send_header('Content-Type', mimetypes.guess_type(file_path)[0] or 'application/octet-stream') - self.send_header('Content-Length', len(content)) - self.end_headers() - self.wfile.write(content) - except: - self._respond(404, {'error': 'Not found'}) - - def _respond(self, code, data): - self.send_response(code) - self._cors_headers() - self.send_header('Content-Type', 'application/json') - self.end_headers() - self.wfile.write(json.dumps(data, default=str).encode()) - - def _cors_headers(self): - self.send_header('Access-Control-Allow-Origin', '*') - self.send_header('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS') - self.send_header('Access-Control-Allow-Headers', 'Content-Type') - - def log_message(self, fmt, *args): - print(f"[api] {args[0]}") - - -if __name__ == '__main__': - # Sync extensions from Gitea and load into DB on every startup - try: - result = sync_extension_registry() - if result['status'] == 'ok': - changes = result['added'] + result['reactivated'] + result['deactivated'] - if changes: - print(f"[init] Registry synced — added: {result['added']}, reactivated: {result['reactivated']}, deactivated: {result['deactivated']}") - else: - print(f"[init] Registry synced — no changes ({len(MODULES)} modules)") - else: - print(f"[init] Registry sync failed: {result.get('message', '')}") - _load_modules_from_db() - except Exception as e: - print(f"[init] Registry sync error: {e}") - _load_modules_from_db() - if not MODULES: - MODULES.update(_MODULES_FALLBACK) - print(f"[init] Using static fallback: {len(MODULES)} modules") - port = int(os.environ.get('PORT', '8080')) - server = HTTPServer(('0.0.0.0', port), APIHandler) - print(f"Extension Platform API on port {port}") - server.serve_forever() +${FILE:/tmp/ext-platform-server.py} \ No newline at end of file