diff --git a/app_server.py b/app_server.py index 31b23f4..c1f9316 100644 --- a/app_server.py +++ b/app_server.py @@ -1 +1,1019 @@ -${FILE:/tmp/ext-platform-server.py} \ No newline at end of file +#!/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() + elif path == '/api/health': + self._respond(200, {'status': 'ok'}) + 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()