site-publish/scripts/deploy.py

"""Deploy phase — S3 sync, manifest rendering, alias reconcile."""

import fnmatch
import hashlib
import json
import mimetypes
import os
import re
import shutil
import subprocess
from pathlib import Path, PurePosixPath
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

from utils import (
    NAMESPACE,
    clone_apps,
    commit_and_push,
    die,
    env,
    k8s_name,
    parse_site_yaml,
    render_templates,
    run,
    validate_artifact_inputs,
)

GARAGE_ADMIN_ENDPOINT = os.environ.get(
    "GARAGE_ADMIN_ENDPOINT", "http://garage.storage.svc:3903"
)
HISTORY_FILE = "site-publish-history.json"


def validate_artifact_output(site_dir, artifact):
    """Prove every artifact and declared cache prefix exists before publishing any."""
    html_dir = site_dir / artifact["build_dir"]
    if not html_dir.is_dir() or not any(path.is_file() for path in html_dir.rglob("*")):
        die(f"artifact {artifact['name']} build output is absent or empty: {html_dir}")
    for rule in artifact["cache_rules"]:
        if not rule["path"]:
            continue
        cache_root = html_dir / rule["path"]
        if not cache_root.exists() or not any(path.is_file() for path in cache_root.rglob("*")):
            die(f"artifact {artifact['name']} cache path /{rule['path']} has no built files")


def validate_publication_environment(cfg):
    """Resolve every declared credential before the first bucket is changed."""
    for artifact in cfg["artifacts"]:
        env(artifact["credentials"]["access_key_env"])
        env(artifact["credentials"]["secret_key_env"])
    if cfg["compatibility"] and cfg["aliases"] and not os.environ.get("GARAGE_ADMIN_TOKEN"):
        die("GARAGE_ADMIN_TOKEN is required when aliases are declared")


def _is_immutable(rule):
    return "immutable" in {
        part.strip().lower().split("=", 1)[0]
        for part in rule["cache_control"].split(",")
    }


def _aws_capture(args, aws_env):
    """Run a non-streaming AWS request without exposing environment credentials."""
    print(f"  $ {' '.join(str(part) for part in args)}")
    return subprocess.run(args, env=aws_env, text=True, capture_output=True, check=False)


def _immutable_head(endpoint, bucket, key, aws_env):
    args = ["aws", "--endpoint-url", endpoint, "s3api", "head-object",
            "--bucket", bucket, "--key", key, "--output", "json"]
    result = _aws_capture(args, aws_env)
    if result.returncode == 0:
        return json.loads(result.stdout)
    error = f"{result.stdout}\n{result.stderr}"
    if any(marker in error for marker in ("404", "Not Found", "NoSuchKey")):
        return None
    raise RuntimeError(f"head-object failed for s3://{bucket}/{key}: {error.strip()}")


def _immutable_digests(source, cache_control, content_type):
    with source.open("rb") as stream:
        content_digest = hashlib.file_digest(stream, "sha256").hexdigest()
    publication = hashlib.sha256()
    publication.update(cache_control.encode())
    publication.update(b"\0")
    publication.update(content_type.encode())
    publication.update(b"\0")
    with source.open("rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            publication.update(block)
    return content_digest, publication.hexdigest()


def _same_immutable_object(info, content_digest, publication_digest, cache_control, content_type):
    metadata = {key.lower(): value for key, value in (info.get("Metadata") or {}).items()}
    return (
        metadata.get("sha256") == content_digest
        and metadata.get("publication-sha256") == publication_digest
        and info.get("CacheControl") == cache_control
        and info.get("ContentType") == content_type
    )


def publish_immutable_file(endpoint, bucket, key, source, cache_control, aws_env):
    """Publish a content-addressed key; identical retries converge."""
    content_type = mimetypes.guess_type(source.name)[0] or "application/octet-stream"
    content_digest, publication_digest = _immutable_digests(source, cache_control, content_type)
    address_digests = re.findall(r"(?<![0-9a-f])([0-9a-f]{64})(?![0-9a-f])", key.lower())
    if address_digests != [publication_digest]:
        raise RuntimeError(
            f"immutable key must contain its one publication SHA-256 {publication_digest}: "
            f"s3://{bucket}/{key}"
        )
    existing = _immutable_head(endpoint, bucket, key, aws_env)
    if existing is not None:
        if _same_immutable_object(
            existing, content_digest, publication_digest, cache_control, content_type,
        ):
            print(f"  Immutable object already matches: s3://{bucket}/{key}")
            return False
        raise RuntimeError(f"immutable object differs or lacks publisher digest: s3://{bucket}/{key}")
    args = ["aws", "--endpoint-url", endpoint, "s3api", "put-object",
            "--bucket", bucket, "--key", key, "--body", str(source),
            "--content-type", content_type, "--cache-control", cache_control,
            "--metadata", f"sha256={content_digest},publication-sha256={publication_digest}"]
    result = _aws_capture(args, aws_env)
    if result.returncode == 0:
        return True
    error = f"{result.stdout}\n{result.stderr}"
    raise RuntimeError(f"put-object failed for s3://{bucket}/{key}: {error.strip()}")


def publish_immutable_rule(artifact, route, rule, html_dir, aws_env):
    """Publish one immutable cache partition without overwrite or deletion."""
    rule_root = html_dir / rule["path"]
    child_paths = [candidate["path"] for candidate in artifact["cache_rules"]
                   if candidate["path"].startswith(f"{rule['path'].rstrip('/')}/")]
    object_prefix = route["path"].strip("/")
    for source in sorted(path for path in rule_root.rglob("*") if path.is_file()):
        relative = source.relative_to(html_dir).as_posix()
        if any(relative == child or relative.startswith(f"{child}/") for child in child_paths):
            continue
        if any(fnmatch.fnmatch(relative, pattern) for pattern in artifact["excludes"]):
            continue
        key = "/".join(part for part in (object_prefix, relative) if part)
        publish_immutable_file(
            artifact["s3_endpoint"], artifact["bucket"], key, source,
            rule["cache_control"], aws_env,
        )


def publication_aws_env(artifact, credential_env_names=None):
    """Build the route-scoped AWS environment without leaking other credentials."""
    access_key = env(artifact["credentials"]["access_key_env"])
    secret_key = env(artifact["credentials"]["secret_key_env"])
    aws_env = os.environ.copy()
    for name in credential_env_names or artifact["credentials"].values():
        aws_env.pop(name, None)
    for name in ("CI_BOT_TOKEN", "GARAGE_ADMIN_TOKEN", "AWS_PROFILE",
                 "AWS_SHARED_CREDENTIALS_FILE", "AWS_SESSION_TOKEN"):
        aws_env.pop(name, None)
    aws_env.update({
        "AWS_ACCESS_KEY_ID": access_key,
        "AWS_SECRET_ACCESS_KEY": secret_key,
        "AWS_DEFAULT_REGION": os.environ.get("AWS_DEFAULT_REGION", "sjc001"),
    })
    return aws_env


def publish_route_immutables(artifact, route, site_dir, credential_env_names=None):
    """Publish one route's immutable partitions during the global preflight."""
    html_dir = site_dir / artifact["build_dir"]
    aws_env = publication_aws_env(artifact, credential_env_names)
    for rule in artifact["cache_rules"]:
        if _is_immutable(rule):
            publish_immutable_rule(artifact, route, rule, html_dir, aws_env)


def retired_immutable_filters(artifact, route, html_dir, previous_contract):
    """Protect every historical immutable prefix inside the current sync scope."""
    if not previous_contract:
        return []
    current_prefix = route["path"].strip("/")
    filters = []
    current_immutable = set(immutable_prefixes(artifact, route))
    for immutable_prefix in previous_contract["immutable_prefixes"]:
        if current_prefix:
            marker = f"{current_prefix}/"
            if not immutable_prefix.startswith(marker):
                continue
            relative_path = immutable_prefix[len(marker):]
        else:
            relative_path = immutable_prefix
        collision = html_dir / relative_path
        if (immutable_prefix not in current_immutable and collision.exists()
                and any(path.is_file() for path in collision.rglob("*"))):
            raise RuntimeError(
                f"current artifact collides with retired immutable partition: {relative_path}"
            )
        filters.extend(("--exclude", f"{relative_path}/*"))
    return filters


def immutable_prefixes(artifact, route):
    route_prefix = route["path"].strip("/")
    return sorted({
        "/".join(part for part in (route_prefix, rule["path"]) if part)
        for rule in artifact["cache_rules"] if _is_immutable(rule)
    })


def s3_sync(artifact, route, site_dir, credential_env_names=None, previous_contract=None):
    endpoint = artifact["s3_endpoint"]
    html_dir = site_dir / artifact["build_dir"]
    aws_env = publication_aws_env(artifact, credential_env_names)
    bucket = artifact["bucket"]
    object_prefix = route["path"].strip("/")
    destination = f"s3://{bucket}/{object_prefix + '/' if object_prefix else ''}"
    default_cache = next(rule["cache_control"] for rule in artifact["cache_rules"] if not rule["path"])
    immutable_paths = [rule["path"] for rule in artifact["cache_rules"] if _is_immutable(rule)]
    # `excludes` are patterns (site.yaml `excludes:` list) that should never
    # be uploaded *and* should never be deleted from the bucket — escape hatch
    # for assets managed out-of-band (e.g. large PDFs uploaded via aws-cli).
    exclude_args = [arg for pattern in artifact["excludes"] for arg in ("--exclude", pattern)]
    if artifact["excludes"]:
        print(f"Excluding patterns: {artifact['excludes']}")
    print(f"Syncing artifact {artifact['name']} → {destination} via {endpoint}")
    # Upload with the final cache policy before cleanup. Sync and deletion are
    # scoped to the same current route prefix and cache partition. A route move
    # leaves its old bucket partition intact but unreachable after the old
    # Ingress disappears, while stale mutable keys on the serving prefix are
    # deleted. Immutable subtrees are structurally excluded. `cp --recursive`
    # refreshes metadata atomically per object before `sync --delete` removes
    # stale keys without ever exposing new bytes under a provisional policy.
    # A no-op deploy therefore transfers the artifact bytes once.
    # AWS CLI guesses Content-Type from file extension on local→S3 uploads,
    # so a fresh upload always carries the right MIME type.
    specific_paths = [rule["path"] for rule in artifact["cache_rules"] if rule["path"]]
    default_filters = [arg for path in specific_paths for arg in ("--exclude", f"{path}/*")]
    retired_filters = retired_immutable_filters(artifact, route, html_dir, previous_contract)
    run(["aws", "--endpoint-url", endpoint, "s3", "cp", f"{html_dir}/", destination,
         "--recursive", "--only-show-errors", "--cache-control", default_cache,
         *default_filters, *retired_filters, *exclude_args], env=aws_env)
    run(["aws", "--endpoint-url", endpoint, "s3", "sync", f"{html_dir}/", destination,
         "--delete", "--only-show-errors", "--cache-control", default_cache,
         *default_filters, *retired_filters, *exclude_args], env=aws_env)
    for rule in artifact["cache_rules"]:
        if not rule["path"]:
            continue
        if _is_immutable(rule):
            continue
        include = f"{rule['path'].rstrip('/')}/*"
        child_filters = [arg for path in specific_paths
                         if path.startswith(f"{rule['path'].rstrip('/')}/")
                         for arg in ("--exclude", f"{path}/*")]
        # Apply rules from the artifact root so artifact-level exclusions keep
        # their original meaning under every cache override.
        run(["aws", "--endpoint-url", endpoint, "s3", "cp", f"{html_dir}/", destination,
             "--recursive", "--only-show-errors", "--cache-control", rule["cache_control"],
             "--exclude", "*", "--include", include, *child_filters, *exclude_args], env=aws_env)
        run(["aws", "--endpoint-url", endpoint, "s3", "sync", f"{html_dir}/", destination,
             "--delete", "--only-show-errors", "--cache-control", rule["cache_control"],
             "--exclude", "*", "--include", include, *child_filters, *exclude_args], env=aws_env)


def garage_admin(method, path, token, body=None):
    url = f"{GARAGE_ADMIN_ENDPOINT}{path}"
    data = json.dumps(body).encode() if body is not None else None
    headers = {"Authorization": f"Bearer {token}"}
    if data is not None:
        headers["Content-Type"] = "application/json"
    req = Request(url, data=data, method=method, headers=headers)
    with urlopen(req) as resp:
        raw = resp.read()
        return json.loads(raw) if raw else {}


def ensure_bucket_aliases(site_name, aliases, admin_token):
    """Add cfg['aliases'] as Garage globalAliases on the site bucket.

    Idempotent: skips aliases already present. Never removes aliases not in
    the desired set (safety — orphan removal is manual).
    """
    if not aliases:
        return
    if not admin_token:
        die("GARAGE_ADMIN_TOKEN is required when aliases are declared")

    try:
        info = garage_admin("GET", f"/v2/GetBucketInfo?globalAlias={site_name}",
                            admin_token)
    except (HTTPError, URLError) as e:
        raise RuntimeError(f"bucket lookup failed for {site_name}: {e}") from e

    bucket_id = info.get("id")
    existing = set(info.get("globalAliases") or [])
    print(f"  Bucket {site_name} ({bucket_id[:12]}…) currently aliases: {sorted(existing)}")

    for alias in aliases:
        if alias in existing:
            continue
        print(f"  Adding globalAlias: {alias}")
        try:
            garage_admin("POST", "/v2/AddBucketAlias", admin_token,
                         {"bucketId": bucket_id, "globalAlias": alias})
        except HTTPError as e:
            body = e.read().decode(errors="replace") if hasattr(e, "read") else ""
            print(f"  ERROR adding alias {alias}: {e} {body}")
            raise


def render_site_manifests(site_name, action_dir, app_dir, manifests_dir, cfg):
    """Always re-render manifests from current site.yaml. Templates own
    domain + aliases, so changes propagate without manual edits."""
    manifests_dir.mkdir(parents=True, exist_ok=True)
    artifact_by_name = {artifact["name"]: artifact for artifact in cfg["artifacts"]}
    routes = []
    for route in cfg["routes"]:
        artifact = artifact_by_name[route["artifact"]]
        resource_name = k8s_name(site_name) if cfg["compatibility"] else f"{k8s_name(site_name)}-{route['name']}"
        routes.append({**route, "resource_name": resource_name, "artifact_config": artifact})
    template_vars = {
        "site": site_name,
        "site_k8s": k8s_name(site_name),
        "domain": cfg["domain"],
        "aliases": cfg["aliases"],
        "namespace": NAMESPACE,
        "compatibility": cfg["compatibility"],
        "routes": routes,
    }
    render_templates(action_dir, template_vars, app_dir, manifests_dir)


def previous_route_contracts(app_dir):
    """Read the append-only bucket history kept beside generated manifests."""
    path = app_dir / HISTORY_FILE
    if not path.exists():
        return {}
    try:
        document = json.loads(path.read_text())
    except (OSError, json.JSONDecodeError) as exc:
        raise RuntimeError(f"invalid site-publish route history in {path}") from exc
    if (not isinstance(document, dict) or set(document) != {"schemaVersion", "buckets"}
            or document["schemaVersion"] != 1 or not isinstance(document["buckets"], dict)):
        raise RuntimeError(f"invalid site-publish route history in {path}")
    for bucket, contract in document["buckets"].items():
        if (not isinstance(bucket, str) or not isinstance(contract, dict)
                or set(contract) != {"access", "artifact", "immutablePrefixes", "routePath"}
                or contract["access"] not in {"legacy", "protected", "public"}
                or not isinstance(contract["artifact"], str)
                or not isinstance(contract["routePath"], str)
                or not contract["routePath"].startswith("/")
                or not isinstance(contract["immutablePrefixes"], list)
                or any(not _valid_immutable_prefix(value)
                       for value in contract["immutablePrefixes"])
                or len(contract["immutablePrefixes"]) != len(set(contract["immutablePrefixes"]))):
            raise RuntimeError(f"invalid site-publish route history in {path}")
    return {
        bucket: {
            "access": contract["access"],
            "artifact": contract["artifact"],
            "immutable_prefixes": sorted(contract["immutablePrefixes"]),
            "path": contract["routePath"],
        }
        for bucket, contract in document["buckets"].items()
    }


def next_route_contracts(cfg, previous_contracts):
    """Carry protected access and immutable prefixes forward for every known bucket."""
    contracts = json.loads(json.dumps(previous_contracts))
    artifacts = {artifact["name"]: artifact for artifact in cfg["artifacts"]}
    for route in cfg["routes"]:
        artifact = artifacts[route["artifact"]]
        bucket = artifact["bucket"]
        previous = previous_contracts.get(bucket)
        access = "protected" if (
            route["access"] == "protected" or previous and previous["access"] == "protected"
        ) else route["access"]
        contracts[bucket] = {
            "access": access,
            "artifact": route["artifact"],
            "immutable_prefixes": sorted(set(
                (previous or {}).get("immutable_prefixes", []) + immutable_prefixes(artifact, route)
            )),
            "path": route["path"],
        }
    return contracts


def write_route_contracts(app_dir, contracts):
    app_dir.mkdir(parents=True, exist_ok=True)
    document = {
        "schemaVersion": 1,
        "buckets": {
            bucket: {
                "access": contract["access"],
                "artifact": contract["artifact"],
                "immutablePrefixes": contract["immutable_prefixes"],
                "routePath": contract["path"],
            }
            for bucket, contract in sorted(contracts.items())
        },
    }
    (app_dir / HISTORY_FILE).write_text(f"{json.dumps(document, indent=2, sort_keys=True)}\n")


def _valid_immutable_prefix(value):
    return (isinstance(value, str) and value and not value.startswith("/")
            and not value.endswith("/") and ".." not in PurePosixPath(value).parts)


def validate_route_migrations(cfg, previous_contracts):
    artifacts = {artifact["name"]: artifact for artifact in cfg["artifacts"]}
    for route in cfg["routes"]:
        artifact = artifacts[route["artifact"]]
        previous = previous_contracts.get(artifact["bucket"])
        if (previous and previous["access"] == "protected"
                and route["access"] in {"public", "legacy"}
        ):
            raise RuntimeError(
                f"artifact {route['artifact']} cannot become public while reusing protected "
                f"bucket {artifact['bucket']}"
            )


def deploy_static(site_name, site_dir, action_dir, token, cfg):
    artifact_by_name = {artifact["name"]: artifact for artifact in cfg["artifacts"]}
    credential_env_names = {
        name for artifact in cfg["artifacts"] for name in artifact["credentials"].values()
    }
    validate_publication_environment(cfg)
    for artifact in cfg["artifacts"]:
        validate_artifact_output(site_dir, artifact)
    apps_dir = clone_apps(token)
    app_dir = apps_dir / "sjc001" / "websites" / site_name
    manifests_dir = app_dir / "manifests"
    previous_contracts = previous_route_contracts(app_dir)
    validate_route_migrations(cfg, previous_contracts)
    # Complete immutable work across the whole publication before any route's
    # mutable pointers can change. Partial immutable success is safe; mixing a
    # new route with an old route after a later immutable failure is not.
    for route in cfg["routes"]:
        publish_route_immutables(
            artifact_by_name[route["artifact"]], route, site_dir, credential_env_names,
        )
    for route in cfg["routes"]:
        s3_sync(
            artifact_by_name[route["artifact"]], route, site_dir, credential_env_names,
            previous_contracts.get(artifact_by_name[route["artifact"]]["bucket"]),
        )
    if cfg["compatibility"]:
        ensure_bucket_aliases(site_name, cfg["aliases"], os.environ.get("GARAGE_ADMIN_TOKEN"))

    write_route_contracts(app_dir, next_route_contracts(cfg, previous_contracts))
    render_site_manifests(site_name, action_dir, app_dir, manifests_dir, cfg)

    commit_and_push(apps_dir, f"Deploy {site_name}", token)


def decommission(site_name, token, buckets=None):
    """Remove manifests from apps repo."""
    apps_dir = clone_apps(token)
    site_path = apps_dir / "sjc001" / "websites" / site_name
    if not site_path.exists():
        print(f"No manifests for {site_name} — nothing to remove")
        return
    history_path = site_path / HISTORY_FILE
    history = history_path.read_bytes() if history_path.exists() else None
    shutil.rmtree(site_path)
    if history is not None:
        site_path.mkdir(parents=True)
        (site_path / HISTORY_FILE).write_bytes(history)
    commit_and_push(apps_dir, f"Decommission {site_name}", token)
    for bucket in buckets or [site_name]:
        print(f"Bucket {bucket} and its objects are NOT purged automatically.")
        print(f"  garage bucket delete {bucket} --yes")


def cmd_deploy():
    site_repo = env("SITE_REPO")
    site_dir = Path(env("SITE_DIR"))
    action_dir = Path(env("ACTION_DIR"))
    token = env("CI_BOT_TOKEN")
    site_name = site_repo.split("/", 1)[1]

    cfg = parse_site_yaml(site_dir)

    if not cfg["enabled"]:
        print("Site disabled — running decommission...")
        decommission(site_name, token, [artifact["bucket"] for artifact in cfg["artifacts"]])
        return

    validate_artifact_inputs(site_dir, cfg)
    deploy_static(site_name, site_dir, action_dir, token, cfg)