enterprise_digital-platform/portal/app/services/module_manager.py

"""
module_manager.py — start/stop/status helpers for EDP sub-services.

Used exclusively in development mode (start-dev.sh launcher).  Each service
writes its PID to   .dev-logs/<name>.pid  so the portal can signal it.
Module state (enabled/disabled) is persisted in the portal's SQLite DB
(ModuleConfig table) and mirrored to  .dev-module-state.json  so that
start-dev.sh can respect it on the next launch.
"""
import json
import logging
import os
import signal
import subprocess
from typing import Optional

log = logging.getLogger(__name__)


# ── Directory helpers ─────────────────────────────────────────────────────────

def _root() -> str:
    """Return the Enterprise_digital-platform root directory at runtime."""
    # This file lives at:  portal/app/services/module_manager.py
    # → services/  → app/  → portal/  → ROOT
    return os.path.dirname(
        os.path.dirname(
            os.path.dirname(
                os.path.dirname(os.path.abspath(__file__))
            )
        )
    )


def _log_dir() -> str:
    return os.path.join(_root(), '.dev-logs')


def _pid_path(app_id: str) -> str:
    return os.path.join(_log_dir(), f'{app_id}.pid')


def _state_path() -> str:
    return os.path.join(_root(), '.dev-module-state.json')


# ── PID file helpers ──────────────────────────────────────────────────────────

def _read_pid(app_id: str) -> Optional[int]:
    path = _pid_path(app_id)
    if not os.path.exists(path):
        return None
    try:
        with open(path) as fp:
            return int(fp.read().strip())
    except (ValueError, OSError):
        return None


def _write_pid(app_id: str, pid: int) -> None:
    os.makedirs(_log_dir(), exist_ok=True)
    with open(_pid_path(app_id), 'w') as fp:
        fp.write(str(pid))


def _clear_pid(app_id: str) -> None:
    path = _pid_path(app_id)
    try:
        os.unlink(path)
    except OSError:
        pass


# ── Module state file ─────────────────────────────────────────────────────────

def write_module_state(app_id: str, enabled: bool) -> None:
    """Mirror enabled/disabled state to .dev-module-state.json for start-dev.sh."""
    path = _state_path()
    state: dict = {}
    if os.path.exists(path):
        try:
            with open(path) as fp:
                state = json.load(fp)
        except (json.JSONDecodeError, OSError):
            state = {}
    state[app_id] = enabled
    with open(path, 'w') as fp:
        json.dump(state, fp, indent=2)


# ── Public API ────────────────────────────────────────────────────────────────

def is_running(app_id: str) -> bool:
    """Return True if the service process is alive."""
    pid = _read_pid(app_id)
    if pid is None:
        return False
    try:
        os.kill(pid, 0)
        return True
    except ProcessLookupError:
        _clear_pid(app_id)
        return False
    except PermissionError:
        return True  # process exists, owned by another uid


def stop_service(app_id: str) -> bool:
    """Send SIGTERM to the service. Returns True if signal was delivered."""
    pid = _read_pid(app_id)
    if pid is None:
        return False
    try:
        os.kill(pid, signal.SIGTERM)
        _clear_pid(app_id)
        log.info('Sent SIGTERM to %s (PID %d)', app_id, pid)
        return True
    except ProcessLookupError:
        _clear_pid(app_id)
        return False


def start_service(app_id: str) -> bool:
    """Spawn the service as a detached background process. Returns True on success."""
    root = _root()
    spec = _get_spec(app_id, root)
    if spec is None:
        log.warning('No service spec for app_id=%s', app_id)
        return False

    log_file = os.path.join(_log_dir(), f'{app_id}.log')
    os.makedirs(_log_dir(), exist_ok=True)

    env = {**os.environ, **spec.get('env', {})}
    with open(log_file, 'a') as lf:
        proc = subprocess.Popen(
            spec['cmd'],
            env=env,
            cwd=spec.get('cwd'),
            stdout=lf,
            stderr=lf,
            start_new_session=True,   # detach from the Flask process group
        )
    _write_pid(app_id, proc.pid)
    log.info('Started %s as PID %d', app_id, proc.pid)
    return True


# ── Service command specs ─────────────────────────────────────────────────────

def _get_spec(app_id: str, root: str) -> Optional[dict]:
    """Return the launch spec for a given app_id, or None if unknown."""
    jwt = os.environ.get('PORTAL_JWT_SECRET', 'change-this-jwt-secret-in-production')

    specs: dict = {
        'digiserver': {
            'cmd': [
                f'{root}/digiserver-v2/.venv/bin/gunicorn',
                '--bind', '0.0.0.0:5002',
                '--workers', '2',
                '--timeout', '120',
                '--chdir', f'{root}/digiserver-v2',
                'wsgi:application',
            ],
            'env': {
                'FLASK_ENV': 'development',
                'ADMIN_USERNAME': os.environ.get('ADMIN_USERNAME', 'admin'),
                'ADMIN_PASSWORD': os.environ.get('ADMIN_PASSWORD', 'admin123'),
                'PORTAL_JWT_SECRET': jwt,
            },
            'cwd': f'{root}/digiserver-v2',
        },
        'itassets': {
            'cmd': [
                f'{root}/IT_asset_management/.venv/bin/python',
                f'{root}/IT_asset_management/run.py',
            ],
            'env': {
                'FLASK_ENV': 'development',
                'SQLALCHEMY_DATABASE_URI': f'sqlite:///{root}/IT_asset_management/data/itassets.db',
                'PORTAL_JWT_SECRET': jwt,
                'FLASK_APP': 'run.py',
            },
            'cwd': f'{root}/IT_asset_management',
        },
        'networkview': {
            'cmd': ['node', f'{root}/NetworkView/backend/src/index.js'],
            'env': {
                'PORT': '3001',
                'PORTAL_JWT_SECRET': jwt,
                'NODE_ENV': 'development',
                'DB_PATH': f'{root}/NetworkView/data/networkview.db',
            },
            'cwd': f'{root}/NetworkView/backend',
        },
    }
    return specs.get(app_id)