ServerKit/backend/app/services/deploy_settings.py

"""Per-app deploy settings.

Stored as one JSON object on ``Application.deploy_settings``; NULL or a
missing key means the default below. ``effective(app)`` is what every reader
uses, so a default changes in one place.
"""
import json

DEFAULTS = {
    # Health gate: how long a new release may take to answer, whether a
    # 4xx on the health path still counts as up, and how many answers in a row
    # it takes.
    'healthcheck_timeout': 120,
    'healthcheck_allow_4xx': False,
    'healthcheck_consecutive': 3,
    # How many deployments' images stay on disk for rollback.
    'keep_images': 3,
    # Slot deploys: how long to watch a release after the switch, how
    # long the old slot stays warm, and the operator's word that two copies
    # sharing the app's volumes for a moment is safe.
    'watch_seconds': 60,
    'standby_warm_minutes': 10,
    'slot_volumes_confirmed': False,
    # Release phase: a command to run once before the switch (overrides
    # the Procfile's release: line), whether to dump the app's databases
    # first, and the announced-downtime fallback for schema changes that
    # cannot stay backward-compatible for one deploy.
    'release_command': None,
    'snapshot_databases': True,
    'stop_old_before_release': False,
    # Compose slots, written by the panel, never by the settings API:
    # the web service, the port it published before slots, and whether the
    # stateful services have moved to the shared data project.
    'compose_web': None,
    'compose_web_port': None,
    'compose_data_split': False,
}

# key -> (type, min, max). Anything outside is rejected, not clamped: a typo'd
# timeout of 0 would make every deploy fail.
_RULES = {
    'healthcheck_timeout': (int, 5, 1800),
    'healthcheck_allow_4xx': (bool, None, None),
    'healthcheck_consecutive': (int, 1, 20),
    'keep_images': (int, 1, 20),
    'watch_seconds': (int, 0, 900),
    'standby_warm_minutes': (int, 0, 1440),
    'slot_volumes_confirmed': (bool, None, None),
    'release_command': (str, None, 500),
    'snapshot_databases': (bool, None, None),
    'stop_old_before_release': (bool, None, None),
}


def stored(app) -> dict:
    raw = getattr(app, 'deploy_settings', None)
    if not raw:
        return {}
    try:
        value = json.loads(raw)
    except (TypeError, ValueError):
        return {}
    return value if isinstance(value, dict) else {}


def effective(app) -> dict:
    merged = dict(DEFAULTS)
    merged.update({k: v for k, v in stored(app).items() if k in DEFAULTS})
    return merged


def get(app, key):
    return effective(app)[key]


def validate(changes) -> tuple:
    """``(clean, error)``: the recognised keys coerced to their types."""
    if not isinstance(changes, dict):
        return None, 'deploy_settings must be an object'
    clean = {}
    for key, value in changes.items():
        if key not in _RULES:
            return None, f'Unknown deploy setting: {key}'
        kind, low, high = _RULES[key]
        if value is None:
            clean[key] = None          # back to the default
            continue
        if kind is bool:
            if not isinstance(value, bool):
                return None, f'{key} must be true or false'
            clean[key] = value
            continue
        if kind is str:
            if not isinstance(value, str):
                return None, f'{key} must be text'
            if len(value) > high:
                return None, f'{key} must be at most {high} characters'
            clean[key] = value.strip() or None
            continue
        if isinstance(value, bool):
            return None, f'{key} must be a number'
        try:
            number = kind(value)
        except (TypeError, ValueError):
            return None, f'{key} must be a number'
        if (low is not None and number < low) or (high is not None and number > high):
            return None, f'{key} must be between {low} and {high}'
        clean[key] = number
    return clean, None


def set_internal(app, key, value) -> None:
    """Record a panel-owned setting (no validation, no commit)."""
    if key not in DEFAULTS or key in _RULES:
        raise KeyError(key)
    current = stored(app)
    current[key] = value
    app.deploy_settings = json.dumps(current)


def update(app, changes) -> tuple:
    """Merge validated ``changes`` into the app's stored settings (no commit)."""
    clean, error = validate(changes)
    if error:
        return None, error
    current = stored(app)
    for key, value in clean.items():
        if value is None:
            current.pop(key, None)
        else:
            current[key] = value
    app.deploy_settings = json.dumps(current) if current else None
    return effective(app), None