chaim
740866eb1c
manifest.description is for composer / dev consumption (often a long English text). The admin catalog needs short Hebrew labels. Introduces optional manifest.displayLabel field — only that field is synced into extensions.description (the column the admin UI reads). When displayLabel is missing the DB row is left untouched, so curated labels never get clobbered by long English manifest descriptions. Adds POST /admin/set-labels for bulk backfilling labels in one call. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1130 lines
54 KiB
Python
1130 lines
54 KiB
Python
#!/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 /release-deleted — Remove a deleted release from DB; deactivate extension if no versions remain
|
|
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'(?<!^)(?=[A-Z])', '-', name).lower()
|
|
|
|
# Module name mapping — loaded from DB on startup, synced from Gitea via /sync-registry
|
|
MODULES = {}
|
|
|
|
# Extension dependency tree (domain knowledge, not auto-discoverable)
|
|
DEPENDENCIES = {
|
|
'HebrewLanguage': [],
|
|
'KlearBranding': ['HebrewLanguage'],
|
|
'EmailDefaults': ['HebrewLanguage'],
|
|
'LocalDocuments': ['HebrewLanguage'],
|
|
'GoogleIntegration': ['HebrewLanguage'],
|
|
'NextCloudIntegration': ['HebrewLanguage'],
|
|
'NetworkStorageIntegration': ['HebrewLanguage'],
|
|
'SmsIntegration': ['HebrewLanguage'],
|
|
'WhatsAppIntegration': ['HebrewLanguage'],
|
|
'GreenInvoiceBilling': ['HebrewLanguage'],
|
|
'LegalCrm': ['HebrewLanguage'],
|
|
'NetHamishpat': ['HebrewLanguage', 'LegalCrm'],
|
|
'DigitalSignature': ['HebrewLanguage', 'LegalCrm'],
|
|
'SmartAssistant': ['HebrewLanguage', 'LegalCrm'],
|
|
'LegalAssistance': ['HebrewLanguage', 'LegalCrm'],
|
|
'DataMigration': [],
|
|
}
|
|
|
|
def _compute_install_order():
|
|
"""Compute install order from DEPENDENCIES via topological sort. Extensions not in DEPENDENCIES are appended at the end."""
|
|
order = []
|
|
visited = set()
|
|
def visit(name):
|
|
if name in visited:
|
|
return
|
|
visited.add(name)
|
|
for dep in DEPENDENCIES.get(name, []):
|
|
visit(dep)
|
|
order.append(name)
|
|
for name in DEPENDENCIES:
|
|
visit(name)
|
|
# Append any MODULES not in DEPENDENCIES (no deps assumed)
|
|
for name in sorted(MODULES):
|
|
if name not in visited:
|
|
order.append(name)
|
|
return order
|
|
|
|
def _load_modules_from_db():
|
|
"""Load MODULES dict from the extensions table."""
|
|
global MODULES
|
|
try:
|
|
rows = query("SELECT name, composer_name FROM extensions WHERE is_active")
|
|
if rows:
|
|
MODULES = {
|
|
r['name']: r['composer_name'].split('/')[-1] if r.get('composer_name') else _pascal_to_kebab(r['name'])
|
|
for r in rows
|
|
}
|
|
print(f"[init] Loaded {len(MODULES)} modules from database")
|
|
except Exception as e:
|
|
print(f"[init] Could not load modules from DB: {e}")
|
|
|
|
# Static fallback used only if DB is not available on startup
|
|
_MODULES_FALLBACK = {
|
|
"DataMigration": "data-migration", "DigitalSignature": "digital-signature",
|
|
"EmailDefaults": "email-defaults", "GoogleIntegration": "google-integration",
|
|
"GreenInvoiceBilling": "green-invoice-billing", "HebrewLanguage": "hebrew-language",
|
|
"KlearBranding": "klear-branding", "LegalAssistance": "legal-assistance",
|
|
"LegalCrm": "legal-crm", "LocalDocuments": "local-documents",
|
|
"NetHamishpat": "net-hamishpat", "NetworkStorageIntegration": "network-storage-integration",
|
|
"NextCloudIntegration": "nextcloud-integration", "SmartAssistant": "smart-assistant",
|
|
"SmsIntegration": "sms-integration", "WhatsAppIntegration": "whatsapp-integration",
|
|
}
|
|
|
|
|
|
def sync_extension_registry():
|
|
"""Discover extensions from Gitea org and sync with database.
|
|
Adds new extensions, reactivates returned ones, deactivates removed ones."""
|
|
repos = gitea_api('GET', f'/orgs/{GITEA_ORG}/repos?limit=100')
|
|
if isinstance(repos, dict) and repos.get('error'):
|
|
return {'status': 'error', 'message': f'Failed to list Gitea repos: {repos.get("message", "")}'}
|
|
|
|
# Filter to extension repos: PascalCase names not in the skip list
|
|
gitea_extensions = {}
|
|
for repo in repos:
|
|
name = repo['name']
|
|
if name in GITEA_NON_EXTENSION_REPOS or not name[0].isupper():
|
|
continue
|
|
gitea_extensions[name] = _pascal_to_kebab(name)
|
|
|
|
# Get current DB state
|
|
db_extensions = query("SELECT name, is_active FROM extensions")
|
|
db_names = {r['name'] for r in db_extensions}
|
|
db_active = {r['name'] for r in db_extensions if r['is_active']}
|
|
|
|
added, reactivated, deactivated = [], [], []
|
|
|
|
# Extensions deactivated via /release-deleted (zero versions) must stay deactivated even if the repo still exists.
|
|
versions_count = {r['name']: r['cnt'] for r in query(
|
|
"SELECT e.name, COUNT(ev.id) AS cnt FROM extensions e LEFT JOIN extension_versions ev ON ev.extension_id = e.id GROUP BY e.name"
|
|
)}
|
|
|
|
for name, kebab in gitea_extensions.items():
|
|
composer_name = f'{GITEA_ORG}/{kebab}'
|
|
if name not in db_names:
|
|
query("INSERT INTO extensions (name, composer_name, is_active) VALUES (%s, %s, true)",
|
|
(name, composer_name), fetch=None)
|
|
added.append(name)
|
|
elif name not in db_active and versions_count.get(name, 0) > 0:
|
|
query("UPDATE extensions SET is_active = true, updated_at = NOW() WHERE name = %s",
|
|
(name,), fetch=None)
|
|
reactivated.append(name)
|
|
|
|
for name in db_active:
|
|
if name not in gitea_extensions:
|
|
query("UPDATE extensions SET is_active = false, updated_at = NOW() WHERE name = %s",
|
|
(name,), fetch=None)
|
|
deactivated.append(name)
|
|
|
|
# Fix latest_version: set to the highest version from extension_versions
|
|
query("""UPDATE extensions e SET latest_version = sub.max_ver, updated_at = NOW()
|
|
FROM (SELECT extension_id, version as max_ver FROM extension_versions ev1
|
|
WHERE NOT EXISTS (SELECT 1 FROM extension_versions ev2
|
|
WHERE ev2.extension_id = ev1.extension_id
|
|
AND string_to_array(ev2.version, '.')::int[] > string_to_array(ev1.version, '.')::int[])
|
|
) sub WHERE e.id = sub.extension_id AND (e.latest_version IS NULL OR e.latest_version != sub.max_ver)""",
|
|
fetch=None)
|
|
|
|
# Sync catalog display label from each extension's manifest.json (main branch).
|
|
# Source: optional `displayLabel` field. `manifest.description` is intentionally NOT used —
|
|
# it's for composer/dev consumption and is often long English; the catalog needs short Hebrew.
|
|
description_updates = []
|
|
for name in gitea_extensions:
|
|
if _sync_display_label_from_manifest(name):
|
|
description_updates.append(name)
|
|
|
|
# Reload MODULES from DB
|
|
_load_modules_from_db()
|
|
|
|
return {
|
|
'status': 'ok',
|
|
'gitea_extensions': len(gitea_extensions),
|
|
'added': added, 'reactivated': reactivated, 'deactivated': deactivated,
|
|
'description_updates': description_updates,
|
|
'active_modules': sorted(MODULES.keys())
|
|
}
|
|
|
|
|
|
def _sync_display_label_from_manifest(module):
|
|
"""Fetch manifest.json from the extension's Gitea main branch and update extensions.description
|
|
from manifest.displayLabel (the short Hebrew catalog label). Leaves DB untouched if displayLabel
|
|
is missing — `manifest.description` is intentionally NOT consulted (it's for composer/dev use
|
|
and is often a long English text unsuitable for the admin UI).
|
|
Returns True if the row was changed."""
|
|
resp = gitea_api('GET', f'/repos/{GITEA_ORG}/{module}/contents/manifest.json')
|
|
if not isinstance(resp, dict) or not resp.get('content'):
|
|
return False
|
|
try:
|
|
manifest = json.loads(base64.b64decode(resp['content']))
|
|
except Exception:
|
|
return False
|
|
label = (manifest.get('displayLabel') or '').strip()
|
|
if not label:
|
|
return False
|
|
current = query("SELECT description FROM extensions WHERE name = %s", (module,), fetch='one')
|
|
if current and (current.get('description') or '').strip() == label:
|
|
return False
|
|
query("UPDATE extensions SET description = %s, updated_at = NOW() WHERE name = %s",
|
|
(label, module), fetch=None)
|
|
return True
|
|
|
|
SSL_CTX = ssl.create_default_context()
|
|
SSL_CTX.check_hostname = False
|
|
SSL_CTX.verify_mode = ssl.CERT_NONE
|
|
|
|
|
|
# --- DB helpers ---
|
|
def get_db():
|
|
return psycopg2.connect(host=DB_HOST, port=DB_PORT, dbname=DB_NAME, user=DB_USER, password=DB_PASS)
|
|
|
|
def query(sql, params=None, fetch='all'):
|
|
conn = get_db()
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
|
cur.execute(sql, params)
|
|
if fetch == 'all':
|
|
result = [dict(r) for r in cur.fetchall()]
|
|
elif fetch == 'one':
|
|
row = cur.fetchone()
|
|
result = dict(row) if row else None
|
|
else:
|
|
result = None
|
|
conn.commit()
|
|
cur.close()
|
|
conn.close()
|
|
return result
|
|
|
|
|
|
# --- Gitea helpers ---
|
|
def gitea_api(method, path, data=None, raw=False):
|
|
url = f"{GITEA_URL}/api/v1{path}"
|
|
headers = {'Authorization': f'token {GITEA_TOKEN}'}
|
|
if data and not raw:
|
|
headers['Content-Type'] = 'application/json'
|
|
data = json.dumps(data).encode()
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=60)
|
|
body = resp.read()
|
|
if raw:
|
|
return body
|
|
return json.loads(body) if body else {}
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode()
|
|
return {'error': True, 'code': e.code, 'message': body[:300]}
|
|
|
|
|
|
def gitea_upload(url, file_data, filename):
|
|
"""Upload file to Gitea (Composer registry or release asset)."""
|
|
boundary = '----FormBoundary' + os.urandom(8).hex()
|
|
body = (
|
|
f'--{boundary}\r\n'
|
|
f'Content-Disposition: form-data; name="package"; filename="{filename}"\r\n'
|
|
f'Content-Type: application/zip\r\n\r\n'
|
|
).encode() + file_data + f'\r\n--{boundary}--\r\n'.encode()
|
|
headers = {
|
|
'Authorization': f'token {GITEA_TOKEN}',
|
|
'Content-Type': f'multipart/form-data; boundary={boundary}'
|
|
}
|
|
req = urllib.request.Request(url, data=body, headers=headers, method='PUT')
|
|
try:
|
|
resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=60)
|
|
return {'status': 'ok', 'code': resp.getcode()}
|
|
except urllib.error.HTTPError as e:
|
|
code = e.code
|
|
msg = e.read().decode()[:200]
|
|
if code == 409:
|
|
return {'status': 'exists', 'code': 409, 'message': 'already exists'}
|
|
return {'status': 'error', 'code': code, 'message': msg}
|
|
|
|
|
|
# --- Build logic (from Gitea, no local files) ---
|
|
def build_and_publish(module, version):
|
|
"""Download extension from Gitea, inject composer.json, publish to registry + release."""
|
|
if module not in MODULES:
|
|
return {'status': 'error', 'message': f'Unknown module: {module}'}
|
|
|
|
kebab = MODULES[module]
|
|
tag = f'v{version}'
|
|
composer_name = f'{GITEA_ORG}/{kebab}'
|
|
|
|
# 1. Download tag archive from Gitea
|
|
archive_url = f"{GITEA_URL}/api/v1/repos/{GITEA_ORG}/{module}/archive/{tag}.zip"
|
|
headers = {'Authorization': f'token {GITEA_TOKEN}'}
|
|
req = urllib.request.Request(archive_url, headers=headers)
|
|
try:
|
|
resp = urllib.request.urlopen(req, context=SSL_CTX, timeout=60)
|
|
archive_data = resp.read()
|
|
except Exception as e:
|
|
return {'status': 'error', 'step': 'download', 'message': str(e)}
|
|
|
|
# 2. Extract and read manifest.json
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
archive_path = os.path.join(tmpdir, 'archive.zip')
|
|
with open(archive_path, 'wb') as f:
|
|
f.write(archive_data)
|
|
|
|
with zipfile.ZipFile(archive_path) as zf:
|
|
# Gitea archives have a top-level directory
|
|
top_dirs = set(n.split('/')[0] for n in zf.namelist() if '/' in n)
|
|
prefix = list(top_dirs)[0] + '/' if len(top_dirs) == 1 else ''
|
|
|
|
# Read manifest
|
|
manifest_path = f'{prefix}manifest.json'
|
|
try:
|
|
manifest = json.loads(zf.read(manifest_path))
|
|
except KeyError:
|
|
return {'status': 'error', 'step': 'manifest', 'message': f'manifest.json not found (tried {manifest_path})'}
|
|
|
|
# Validate version
|
|
if manifest.get('version') != version:
|
|
return {'status': 'error', 'step': 'version',
|
|
'message': f"Mismatch: tag={version}, manifest={manifest.get('version')}"}
|
|
|
|
# 3. Generate composer.json
|
|
composer = {
|
|
"name": composer_name,
|
|
"description": manifest.get('description', ''),
|
|
"version": version,
|
|
"type": "espocrm-extension",
|
|
"license": "proprietary",
|
|
"authors": [{"name": manifest.get('author', 'klear')}],
|
|
"require": {"php": manifest.get('php', ['>=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 AND (latest_version IS NULL OR string_to_array(latest_version, '.')::int[] < string_to_array(%s, '.')::int[])""",
|
|
(version, module, version))
|
|
# Sync catalog display label from manifest.displayLabel (short Hebrew). Skip manifest.description
|
|
# — it's for composer/dev use, often long English, and unsuitable for the admin UI.
|
|
manifest_label = (manifest.get('displayLabel') or '').strip()
|
|
if manifest_label:
|
|
cur.execute("UPDATE extensions SET description = %s, updated_at = NOW() WHERE name = %s",
|
|
(manifest_label, 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 == '/release-deleted':
|
|
self._handle_release_deleted(body)
|
|
elif path == '/admin/set-labels':
|
|
self._handle_admin_set_labels(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_admin_set_labels(self, body):
|
|
"""Bulk-set extensions.description (the catalog display label) for one or many extensions.
|
|
Body: { "labels": { "ModuleName": "תווית קצרה", ... } }"""
|
|
labels = body.get('labels', {})
|
|
if not isinstance(labels, dict) or not labels:
|
|
self._respond(400, {'error': 'labels dict required'}); return
|
|
updated, skipped = [], []
|
|
for name, label in labels.items():
|
|
if not isinstance(label, str):
|
|
skipped.append({'name': name, 'reason': 'label must be string'}); continue
|
|
ext = query("SELECT id FROM extensions WHERE name = %s", (name,), fetch='one')
|
|
if not ext:
|
|
skipped.append({'name': name, 'reason': 'not found'}); continue
|
|
query("UPDATE extensions SET description = %s, updated_at = NOW() WHERE id = %s",
|
|
(label.strip(), ext['id']), fetch=None)
|
|
updated.append(name)
|
|
self._respond(200, {'status': 'ok', 'updated': updated, 'skipped': skipped})
|
|
|
|
def _handle_release_deleted(self, body):
|
|
"""Triggered by Gitea release-deleted webhook (via n8n).
|
|
Removes the version row, recomputes latest_version, and deactivates the extension if no versions remain.
|
|
Body: { module: str, version: str (optional — if omitted, only recomputes/deactivates) }"""
|
|
module = body.get('module', '')
|
|
version = body.get('version', '')
|
|
if not module:
|
|
self._respond(400, {'error': 'module required'}); return
|
|
ext = query("SELECT id FROM extensions WHERE name = %s", (module,), fetch='one')
|
|
if not ext:
|
|
self._respond(404, {'error': f'Extension {module} not found'}); return
|
|
ext_id = ext['id']
|
|
if version:
|
|
query("DELETE FROM extension_versions WHERE extension_id = %s AND version = %s",
|
|
(ext_id, version), fetch=None)
|
|
remaining = query("SELECT COUNT(*) AS cnt FROM extension_versions WHERE extension_id = %s",
|
|
(ext_id,), fetch='one')['cnt']
|
|
if remaining == 0:
|
|
query("UPDATE extensions SET is_active = false, latest_version = NULL, updated_at = NOW() WHERE id = %s",
|
|
(ext_id,), fetch=None)
|
|
self._respond(200, {'status': 'ok', 'module': module, 'version': version,
|
|
'action': 'deactivated', 'remaining_versions': 0})
|
|
return
|
|
query("""UPDATE extensions e SET latest_version = sub.max_ver, updated_at = NOW()
|
|
FROM (SELECT extension_id, version as max_ver FROM extension_versions ev1
|
|
WHERE ev1.extension_id = %s
|
|
AND NOT EXISTS (SELECT 1 FROM extension_versions ev2
|
|
WHERE ev2.extension_id = ev1.extension_id
|
|
AND string_to_array(ev2.version, '.')::int[] > string_to_array(ev1.version, '.')::int[])
|
|
) sub WHERE e.id = sub.extension_id""", (ext_id,), fetch=None)
|
|
self._respond(200, {'status': 'ok', 'module': module, 'version': version,
|
|
'action': 'version_removed', 'remaining_versions': remaining})
|
|
|
|
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()
|