#!/usr/bin/python3
#
# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""
Manage temporary VMs to discuss UI prototypes.

Setup:

1. apt install awscli python3-gitlab python3-boto3 python3-rich

2. edit ~/.config/freexian.ini:

   [tokens]
   debusine_playground_password = <password-for-the-playground-user>

3. aws configure sso --profile debusine

   SSO session name (Recommended): debusine
   SSO start URL [None]: https://freexian.awsapps.com/start/
   SSO region [None]: eu-west-3
   SSO registration scopes [sso:account:access]: (leave as they are)

4. aws --profile debusine sso login \
       --endpoint-url https://freexian.awsapps.com/start/

   Test if successful with: ``aws sts get-caller-identity --profile debusine``

5. aws --profile debusine ec2 import-key-pair --key-name $USER \
       --public-key-material fileb://~/.ssh/id_ed25519.pub

Usage:

* `playground-vm list`: lists open MRs and corresponding instances (if any)
* `playground-vm create`: nnn create an instance for the given MR
* `playground-vm provision nnn`: provision the instance
* `playground-vm delete nnn`: remove the instance
* `playground-vm login nnn`: root login on the given instance

To redeploy the MR branch after iterating on changes:

* `playground-vm provision nnn`
"""

import abc
import argparse
import contextlib
import logging
import os
import subprocess
import sys
import tempfile
import time as tm
from collections import defaultdict
from collections.abc import Generator
from configparser import ConfigParser
from functools import cached_property
from getpass import getuser
from pathlib import Path
from typing import Literal, NamedTuple, Self, TYPE_CHECKING, TypeAlias, cast

import boto3
import gitlab
import gitlab.v4.objects
import rich
import yaml
from botocore.exceptions import TokenRetrievalError
from rich import box
from rich.table import Table

if TYPE_CHECKING:
    from argparse import _SubParsersAction

    from gitlab.v4.objects import Project, ProjectMergeRequest
    from mypy_boto3_ec2 import EC2Client
    from mypy_boto3_ec2.literals import InstanceStateNameType, InstanceTypeType
    from mypy_boto3_ec2.type_defs import FilterTypeDef, InstanceTypeDef
    from mypy_boto3_route53 import Route53Client
    from mypy_boto3_route53.literals import ChangeActionType
    from mypy_boto3_route53.type_defs import ChangeBatchTypeDef, ChangeTypeDef

    SubParsers = _SubParsersAction[argparse.ArgumentParser]

log = logging.getLogger("playground-vm")

# Source: https://wiki.debian.org/Cloud/AmazonEC2Image/Trixie
AMI = "ami-0e0cf09f194b94a22"
SUBNET = "subnet-08cc8d03ce80f870e"

SECURITY_GROUPS = [
    "sg-08b4080e3dd2f0b9a",  # ssh
    "sg-0582fe2faa9363dfd",  # http
    "sg-0bcafaee56c57590f",  # https,
    "sg-0c0a99a9fa81cc520",  # egress-all
]

DNSType: TypeAlias = Literal["A", "AAAA", "CNAME"]


class Fail(Exception):
    """There was an error in playground VM management."""


DEPLOY_PLAYBOOK = r"""
- name: Provision a playground system
  hosts: all
  vars:
    debusine_packages:
     - python3-debusine
     - python3-debusine-signing
     - python3-debusine-server
     - debusine-server
  handlers:
   - name: Restart nginx
     ansible.builtin.service:
       name: nginx
       state: restarted
  tasks:
   - name: Set hostname
     ansible.builtin.hostname:
       name: "{{hostname}}"
   - name: Enable backports
     copy:
        owner: root
        group: root
        mode: 0644
        dest: /etc/apt/sources.list.d/debian-trixie-backports.list
        content: |
           deb [arch=amd64] https://deb.debian.org/debian/ trixie-backports main contrib
   - name: Update after enabling backports
     apt:
       update_cache: yes
   - name: Install git
     ansible.builtin.apt:
       name: [git, eatmydata, dpkg-dev, nginx, certbot, python3-certbot-nginx, fail2ban]
       state: present
       update_cache: true
       cache_valid_time: 3600
   - name: "Fetch debusine branch {{source_repository_path}}:{{source_branch}}"
     ansible.builtin.git:
       dest: "/srv/sources/debusine"
       repo: "https://salsa.debian.org/{{source_repository_path}}.git"
       version: "{{source_branch}}"
       force: true
     register: fetch_sources
   - name: Install debusine build-deps
     ansible.builtin.apt:
       name: "/srv/sources/debusine"
       state: build-dep
       default_release: trixie-backports
   - name: Rebuild debusine source
     ansible.builtin.shell:
       cmd: "DEB_BUILD_OPTIONS='nocheck' eatmydata dpkg-buildpackage -us -uc"
       chdir: "/srv/sources/debusine"
     when: fetch_sources.changed
   - name: Find debusine version
     ansible.builtin.shell:
       cmd: "dpkg-parsechangelog -SVersion"
       chdir: "/srv/sources/debusine"
     register: deb
   - name: "Remove old versions of built debs"
     ansible.builtin.apt:
       name: "{{item}}"
       state: absent
     loop: "{{debusine_packages|reverse}}"
   - name: "Install built debs"
     ansible.builtin.apt:
       deb: "/srv/sources/{{item}}_{{deb.stdout}}_all.deb"
       default_release: trixie-backports
     loop: "{{debusine_packages}}"
   - name: "Create debusine-server postgres user"
     become: yes
     become_user: postgres
     community.postgresql.postgresql_user:
       name: debusine-server
   - name: "Create debusine postgres db"
     become: yes
     become_user: postgres
     community.postgresql.postgresql_db:
       name: debusine
       owner: debusine-server
   - name: Initialize database
     become: yes
     become_user: debusine-server
     ansible.builtin.command:
       argv: ["debusine-admin", "migrate"]
   - name: Populate database
     become: yes
     become_user: debusine-server
     ansible.builtin.command:
       argv:
         - /srv/sources/debusine/bin/playground-populate
         - "--password"
         - "{{ playground_password }}"
   - name: "Get https certificate for {{hostname}} and deb.{{hostname}}"
     ansible.builtin.command:
       argv: [certbot, run, "--nginx", "--domain", "{{hostname}}",
              "--domain", "deb.{{hostname}}",
              "--noninteractive", "--agree-tos",
              "--register-unsafely-without-email",
              "--cert-name", "playground"]
       creates: /etc/letsencrypt/live/playground/fullchain.pem
   - name: Remove default nginx configuration
     ansible.builtin.file:
       state: absent
       path: /etc/nginx/sites-enabled/default
   - name: Configure nginx (copy template file)
     ansible.builtin.copy:
       remote_src: true
       src: /usr/share/doc/debusine-server/examples/nginx-vhost.conf
       dest: /etc/nginx/sites-enabled/debusine
   - name: Configure nginx (copy template file)
     ansible.builtin.copy:
       remote_src: true
       src: /usr/share/doc/debusine-server/examples/nginx-vhost-deb.conf
       dest: /etc/nginx/sites-enabled/debusine-deb
     notify: Restart nginx
   - name: Configure nginx (edit template file)
     ansible.builtin.lineinfile:
       path: /etc/nginx/sites-enabled/debusine
       line: "{{item.name}} {{item.value}};"
       regexp: "^\\s*{{item.name}} "
     loop:
      - { name: server_name, value: "{{hostname}}" }
      - { name: ssl_certificate, value: "/etc/letsencrypt/live/playground/fullchain.pem" }
      - { name: ssl_certificate_key, value: "/etc/letsencrypt/live/playground/privkey.pem" }
     notify: Restart nginx
   - name: Configure nginx (edit template file)
     ansible.builtin.lineinfile:
       path: /etc/nginx/sites-enabled/debusine-deb
       line: "{{item.name}} {{item.value}};"
       regexp: "^\\s*{{item.name}} "
     loop:
      - { name: server_name, value: "deb.{{hostname}}" }
      - { name: ssl_certificate, value: "/etc/letsencrypt/live/playground/fullchain.pem" }
      - { name: ssl_certificate_key, value: "/etc/letsencrypt/live/playground/privkey.pem" }
     notify: Restart nginx
"""  # noqa: E501


def load_config() -> ConfigParser:
    """Load configuration from freexian.ini."""
    config_home = Path(
        os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config"))
    )
    config_file = config_home / "freexian.ini"

    config = ConfigParser()
    config.read([config_file])

    config.add_section("META")
    config.set("META", "source", str(config_file))
    return config


class DNSRecord(NamedTuple):
    """Represents a simple DNS record."""

    instance_name: str
    name: str
    type: DNSType
    ttl: int
    value: str

    def __str__(self) -> str:
        """Return a string representation of the DNS record."""
        return f"{self.name:50s} {self.ttl:5d} IN {self.type:4s} {self.value}"

    @classmethod
    def from_dns(
        cls, name: str, dns_type: DNSType, ttl: int, value: str
    ) -> Self:
        """Instantiate a DNSRecord from an AWS ResourceRecordSet."""
        match dns_type:
            case "A" | "AAAA":
                return DNSRecord(
                    instance_name=name,
                    name=name,
                    ttl=ttl,
                    type=dns_type,
                    value=value,
                )
            case "CNAME":
                return DNSRecord(
                    instance_name=name.rsplit(".", 1)[-1],
                    name=name,
                    ttl=ttl,
                    type=dns_type,
                    value=value,
                )


class EC2Instance(NamedTuple):
    """Represents an AWS EC2 Instance."""

    instance_id: str
    name: str
    state: "InstanceStateNameType"
    public_ipv4: str | None
    public_ipv6: str

    def dns_records(self, domain: str, ttl: int) -> Generator[DNSRecord]:
        """Yield the expected DNS records with supplied ttl."""
        if self.public_ipv4:
            yield DNSRecord(
                instance_name=self.name,
                name=self.name,
                type="A",
                ttl=ttl,
                value=self.public_ipv4,
            )
        yield DNSRecord(
            instance_name=self.name,
            name=self.name,
            type="AAAA",
            ttl=ttl,
            value=self.public_ipv6,
        )
        yield DNSRecord(
            instance_name=self.name,
            name=f"deb.{self.name}",
            type="CNAME",
            ttl=ttl,
            value=f"{self.name}.{domain}.",
        )


class Route53Zone(NamedTuple):
    """Represents an AWS Route53 Zone."""

    zone_id: str
    name: str


class InstanceDoesNotExist(Exception):
    """An EC2 instance does not exist."""


class AWSClient:
    """Wrapper around boto3, simplifying interfaces."""

    session: boto3.Session

    def __init__(self, profile: str, account_id: str) -> None:
        """Initialize the AWS client."""
        self.session = self.boto3_session(profile, account_id)

    def boto3_session(self, profile: str, account_id: str) -> boto3.Session:
        """Log into AWS and return the current session."""
        session = boto3.Session(profile_name=profile)
        while True:
            try:
                account = session.client("sts").get_caller_identity()
                if account["Account"] == account_id:
                    return session
            except TokenRetrievalError:
                # FIXME: Can boto do this?
                subprocess.check_call(
                    ["aws", "sso", "login", "--profile", profile]
                )

    @cached_property
    def ec2(self) -> "EC2Client":
        """Return a boto3 EC2 client."""
        return self.session.client("ec2")

    def ec2_instances(self, name: str | None = None) -> Generator[EC2Instance]:
        """Return EC2 instances matching kv_filters."""
        filters: list["FilterTypeDef"] = [
            {"Name": "tag:role", "Values": ["playground"]},
        ]
        if name:
            filters.append({"Name": "tag:Name", "Values": [name]})

        paginator = self.ec2.get_paginator("describe_instances")
        for page in paginator.paginate(Filters=filters):
            for reservation in page["Reservations"]:
                for instance in reservation["Instances"]:
                    # Skip tombstones
                    if instance["State"]["Name"] == "terminated":
                        continue
                    yield EC2Instance(
                        instance_id=instance["InstanceId"],
                        name=self.get_instance_name(instance),
                        state=instance["State"]["Name"],
                        public_ipv4=instance.get("PublicIpAddress", None),
                        public_ipv6=self.get_instance_public_ipv6(instance),
                    )

    def ec2_instance(self, name: str) -> EC2Instance:
        """Return the EC2 Instance with name."""
        instances = self.ec2_instances(name=name)
        try:
            instance = next(instances)
        except StopIteration:
            raise InstanceDoesNotExist(f"Instance {name!r} does not exist")
        try:
            next(instances)
        except StopIteration:
            return instance
        else:
            raise Fail(f"More than one instance found with name {name!r}")

    def get_instance_name(self, instance: "InstanceTypeDef") -> str:
        """Extract an instance's name."""
        for tag in instance["Tags"]:
            if tag["Key"] == "Name":
                return tag["Value"]
        raise AssertionError(f"No name found for {instance['InstanceId']}")

    def get_instance_public_ipv6(self, instance: "InstanceTypeDef") -> str:
        """Extract an instance's IPv6 address."""
        for interface in instance["NetworkInterfaces"]:
            for address in interface["Ipv6Addresses"]:
                return address["Ipv6Address"]
        raise AssertionError(
            f"No IPv6 address found for {instance['InstanceId']}"
        )

    def ec2_launch(
        self,
        *,
        name: str,
        ami: str,
        instance_type: "InstanceTypeType",
        subnet: str,
        security_groups: list[str],
        key_name: str,
        ipv4: bool,
    ) -> EC2Instance:
        """Launch an EC2 instance."""
        instance = self.ec2.run_instances(
            BlockDeviceMappings=[
                {
                    "Ebs": {
                        "DeleteOnTermination": True,
                        "VolumeType": "gp3",
                        "VolumeSize": 20,
                    },
                    "DeviceName": "/dev/xvda",
                }
            ],
            ImageId=ami,
            InstanceType=instance_type,
            KeyName=key_name,
            MinCount=1,
            MaxCount=1,
            NetworkInterfaces=[
                {
                    "AssociatePublicIpAddress": ipv4,
                    "DeviceIndex": 0,
                    "Groups": security_groups,
                    "Ipv6AddressCount": 1,
                    "SubnetId": subnet,
                }
            ],
            TagSpecifications=[
                {
                    "ResourceType": "instance",
                    "Tags": [
                        {
                            "Key": "Name",
                            "Value": name,
                        },
                        {
                            "Key": "role",
                            "Value": "playground",
                        },
                    ],
                }
            ],
        )["Instances"][0]
        if ipv4:
            print("Waiting for an IPv4 address to be assigned...")
            while not instance.get("PublicIpAddress", None):
                tm.sleep(1)
                instance = self.ec2.describe_instances(
                    InstanceIds=[instance["InstanceId"]]
                )["Reservations"][0]["Instances"][0]
        return EC2Instance(
            instance_id=instance["InstanceId"],
            name=name,
            state=instance["State"]["Name"],
            public_ipv4=instance.get("PublicIpAddress", None),
            public_ipv6=self.get_instance_public_ipv6(instance),
        )

    def ec2_terminate(self, instance: EC2Instance) -> None:
        """Terminate an EC2 instance."""
        self.ec2.terminate_instances(InstanceIds=[instance.instance_id])

    @cached_property
    def route53(self) -> "Route53Client":
        """Return a boto3 EC2 client."""
        return self.session.client("route53")

    def route53_zone(self, name: str) -> Route53Zone:
        """Return the Route 53 zone for name."""
        name = name.rstrip(".") + "."
        # No need to paginate, we're only interested in the first entry
        zones = self.route53.list_hosted_zones_by_name(
            DNSName=name, MaxItems="1"
        )["HostedZones"]
        # DNSName just makes that zone appear first, if it exists
        if zones and zones[0]["Name"] == name:
            zone = zones[0]
            return Route53Zone(
                zone_id=zone["Id"],
                name=name,
            )
        raise Fail(f"Zone {name!r} does not exist")

    def route53_zone_records(self, zone: Route53Zone) -> Generator[DNSRecord]:
        """Get all the DNS records in zone."""
        paginator = self.route53.get_paginator("list_resource_record_sets")
        for page in paginator.paginate(HostedZoneId=zone.zone_id):
            for rs in page["ResourceRecordSets"]:
                for value in rs.get("ResourceRecords", []):
                    if rs["Type"] in {"A", "AAAA", "CNAME"}:
                        assert rs["Name"].endswith(zone.name)
                        yield DNSRecord.from_dns(
                            name=rs["Name"][: -(len(zone.name) + 1)],
                            ttl=rs["TTL"],
                            dns_type=cast(DNSType, rs["Type"]),
                            value=value["Value"],
                        )

    def route53_zone_records_for_name(
        self, zone: Route53Zone, name: str
    ) -> Generator[DNSRecord]:
        """Return all DNS records in zone matching name."""
        # We can't paginate the filtered endpoint, but we aren't expecting >300
        # results
        for rs in self.route53.list_resource_record_sets(
            HostedZoneId=zone.zone_id, StartRecordName=name
        )["ResourceRecordSets"]:
            for value in rs.get("ResourceRecords", []):
                # The name filter is just a starting point
                if name and rs["Name"] != name:
                    return
                if rs["Type"] in {"A", "AAAA", "CNAME"}:
                    yield DNSRecord.from_dns(
                        name=rs["Name"],
                        ttl=rs["TTL"],
                        dns_type=cast(DNSType, rs["Type"]),
                        value=value["Value"],
                    )

    def _route53_change_batch(
        self, action: "ChangeActionType", records: list[DNSRecord], zone: str
    ) -> "ChangeBatchTypeDef":
        assert zone.endswith(".")
        changes: list["ChangeTypeDef"] = []
        for record in records:
            changes.append(
                {
                    "Action": action,
                    "ResourceRecordSet": {
                        "Name": f"{record.name}.{zone}",
                        "Type": record.type,
                        "TTL": record.ttl,
                        "ResourceRecords": [
                            {
                                "Value": record.value,
                            }
                        ],
                    },
                }
            )
        return {"Changes": changes}

    def route53_set_records(
        self, zone: Route53Zone, records: list[DNSRecord]
    ) -> None:
        """Upsert records into zone."""
        self.route53.change_resource_record_sets(
            HostedZoneId=zone.zone_id,
            ChangeBatch=self._route53_change_batch(
                "UPSERT", records, zone.name
            ),
        )

    def route53_delete_records(
        self, zone: Route53Zone, records: list[DNSRecord]
    ) -> None:
        """Delete records from zone."""
        self.route53.change_resource_record_sets(
            HostedZoneId=zone.zone_id,
            ChangeBatch=self._route53_change_batch(
                "DELETE", records, zone.name
            ),
        )


class InstanceName(NamedTuple):
    """Parsed instance name."""

    mr: int
    type: str = "playground"
    variant: str = "default"

    def __str__(self) -> str:
        """Format the instance name."""
        if self.variant == "default":
            return f"{self.type}-{self.mr}"
        else:
            return f"{self.type}-{self.mr}-{self.variant}"

    @classmethod
    def parse(cls, text: str) -> Self:
        """Parse an instance name."""
        match text.count("-"):
            case 0:
                raise ValueError(f"Instance name {text!r} contains no dashes")
            case 1:
                server_type, mr = text.split("-", 1)
                return cls(type=server_type, mr=int(mr))
            case _:
                server_type, mr, variant = text.split("-", 2)
                return cls(type=server_type, mr=int(mr), variant=variant)


class Playground(contextlib.ExitStack):
    """
    Common infrastructure to manage one playground VM.

    A playground VM is identified by the number of a Debusine merge request and
    optionally a variant identifier.
    """

    args: argparse.Namespace
    aws: AWSClient
    config: ConfigParser
    debusine: "Project"
    domain: str
    gitlab: gitlab.Gitlab
    playground_password: str
    ttl: int

    def __init__(self, args: argparse.Namespace) -> None:
        """Construct a Playground object."""
        super().__init__()
        self.args = args
        self.config = load_config()

        self.gitlab = gitlab.Gitlab("https://salsa.debian.org")
        self.debusine = self.gitlab.projects.get("freexian-team/debusine")

        profile_name = self.config.get("aws", "profile", fallback="debusine")
        account_id = self.config.get("aws", "account", fallback="694521941919")
        self.aws = AWSClient(profile_name, account_id)

        self.domain = "aws.debusine.dev"
        self.ttl = 300
        self.key_name = self.config.get("user", "nick", fallback=getuser())
        self.playground_password = self.config.get(
            "tokens", "debusine_playground_password"
        )

    @cached_property
    def mr(self) -> "ProjectMergeRequest":
        """Get the gitlab merge request object."""
        return self.debusine.mergerequests.get(self.args.mr)

    @cached_property
    def mr_source_repository_path(self) -> str:
        """Get the merge request's source repository path."""
        return str(
            self.gitlab.projects.get(
                self.mr.source_project_id
            ).path_with_namespace
        )

    @cached_property
    def instance_name(self) -> InstanceName:
        """Get the server name given a MR and a variant."""
        return InstanceName(mr=int(self.args.mr), variant=self.args.variant)

    @cached_property
    def instance_fqdn(self) -> str:
        """Get the instance's FQDN."""
        return f"{self.instance_name}.{self.domain}"

    @cached_property
    def instance(self) -> EC2Instance:
        """Return the instance for the given MR (by server name)."""
        return self.aws.ec2_instance(name=str(self.instance_name))

    @cached_property
    def zone(self) -> Route53Zone:
        """Return the route 53 zone for our domain."""
        return self.aws.route53_zone(self.domain)

    def zone_records(self) -> Generator[DNSRecord]:
        """Return the all route 53 records for our zone."""
        yield from self.aws.route53_zone_records(self.zone)

    @cached_property
    def address_ssh(self) -> str:
        """Return the address to connect to the server via ssh."""
        if self.instance.public_ipv4:
            return self.instance.public_ipv4
        return self.instance.public_ipv6

    def create_instance_dns_record(
        self, instance: EC2Instance | None = None
    ) -> None:
        """Create DNS records for an EC2 instance."""
        if instance is None:
            instance = self.instance
        log.info("Creating DNS records for %s", instance.name)
        self.aws.route53_set_records(
            self.zone,
            records=list(instance.dns_records(self.domain, ttl=self.ttl)),
        )

    def delete_instance_dns_record(self, instance: EC2Instance) -> None:
        """Delete DNS records for an EC2 instance."""
        log.info("Deleting DNS records for %s", instance.name)
        records = list(
            self.aws.route53_zone_records_for_name(self.zone, instance.name)
        )
        if records:
            self.aws.route53_delete_records(self.zone, records)

    def print_instance_status(self) -> None:
        """Output the status of a server."""
        grid = Table.grid(padding=(0, 1, 0, 0))
        grid.add_column(style="bold", justify="right")
        grid.add_column()
        fqdn = self.instance_fqdn
        grid.add_row("Name: ", f"[link=https://{fqdn}]{fqdn}[/link]")
        grid.add_row("Status: ", self.instance.state)
        grid.add_row("IPv6: ", self.instance.public_ipv6)
        grid.add_row("IPv4: ", self.instance.public_ipv4)

        rich.print(grid)

    def terminate_instance(self) -> None:
        """Delete a server."""
        addresses = [
            self.instance_fqdn,
            self.instance.public_ipv6,
        ]
        if self.instance.public_ipv4:
            addresses.append(self.instance.public_ipv4)

        self.aws.ec2_terminate(self.instance)
        self.delete_instance_dns_record(self.instance)

        # Remove any cached known_host keys
        for address in addresses:
            subprocess.run(
                [
                    "ssh-keygen",
                    "-f",
                    os.path.expanduser("~/.ssh/known_hosts"),
                    "-R",
                    address,
                ],
            )


class AnsibleEnvironment(contextlib.ExitStack):
    """Temporary ansible environment for remote provisioning."""

    workdir: Path
    path_inventory: Path
    path_playbook: Path

    def __init__(self, playground: Playground) -> None:
        """Construct an AnsibleEnvironment object."""
        super().__init__()
        self.playground = playground
        self.playbook = yaml.safe_load(DEPLOY_PLAYBOOK)
        self.playbook[0]["vars"].update(
            {
                "hostname": self.playground.instance_fqdn,
                "source_repository_path": (
                    self.playground.mr_source_repository_path
                ),
                "source_branch": self.playground.mr.source_branch,
                "playground_password": self.playground.playground_password,
            }
        )

    def __enter__(self) -> Self:
        """Enter context."""
        super().__enter__()
        self.workdir = Path(self.enter_context(tempfile.TemporaryDirectory()))
        self.path_inventory = self.workdir / "hosts"
        self.path_playbook = self.workdir / "deploy.yml"
        return self

    def run_playbook(self) -> None:
        """Run the Ansible playbook for this environment."""
        address = self.playground.address_ssh
        with self.path_inventory.open("w") as fd:
            print(
                f"{self.playground.instance_name}"
                " ansible_user=root"
                f" ansible_host={address}",
                file=fd,
            )

        with self.path_playbook.open("w") as fd:
            yaml.safe_dump(self.playbook, stream=fd)

        env = dict(os.environ)
        env["ANSIBLE_NOCOWS"] = "1"
        env["ANSIBLE_STRATEGY"] = "linear"
        subprocess.run(
            [
                "ansible-playbook",
                "-i",
                str(self.path_inventory),
                str(self.path_playbook),
            ],
            check=True,
            cwd=self.workdir,
            env=env,
        )


class Command(contextlib.ExitStack, abc.ABC):
    """Base class for actions run from command line."""

    NAME: str | None = None

    def __init__(self, args: argparse.Namespace):
        """Initialize this subcommand."""
        super().__init__()
        if self.NAME is None:
            self.NAME = self.__class__.__name__.lower()
        self.args = args
        self.setup_logging()
        self.playground = Playground(args)

    def __enter__(self) -> Self:
        """Enter context."""
        super().__enter__()
        self.enter_context(self.playground)
        return self

    @classmethod
    def add_subparser(cls, subparsers: "SubParsers") -> argparse.ArgumentParser:
        """Create a subparser for this command."""
        if cls.NAME is None:
            cls.NAME = cls.__name__.lower()
        assert cls.__doc__
        parser = subparsers.add_parser(cls.NAME, help=cls.__doc__.strip())
        parser.set_defaults(command=cls)
        parser.add_argument(
            "--quiet", "-q", action="store_true", help="quiet output"
        )
        parser.add_argument("--debug", action="store_true", help="debug output")
        return parser

    def setup_logging(self) -> None:
        """Set up logging."""
        log_format = "%(levelname)s %(message)s"
        level = logging.INFO
        if self.args.debug:
            level = logging.DEBUG
        elif self.args.quiet:
            level = logging.WARN
        logging.basicConfig(level=level, stream=sys.stderr, format=log_format)

    @abc.abstractmethod
    def run(self) -> None:
        """Run this subcommand."""
        ...


class InstanceCommand(Command):
    """Base class for commands that act on an instance."""

    @classmethod
    def add_subparser(cls, subparsers: "SubParsers") -> argparse.ArgumentParser:
        """Add generic options for instance commands."""
        parser = super().add_subparser(subparsers)
        parser.add_argument("mr", help="merge request number")
        parser.add_argument(
            "variant", nargs="?", default="default", help="playground variant"
        )
        return parser


class List(Command):
    """List existing playground instances."""

    def run(self) -> None:
        """Run `list` command."""
        # Load information on opened merge requests
        mrs = {}
        for mr in self.playground.debusine.mergerequests.list(state="opened"):
            mrs[mr.iid] = mr

        # Load information on existing instances
        instances: dict[int, dict[str, EC2Instance]] = defaultdict(dict)
        for instance in self.playground.aws.ec2_instances():
            try:
                instance_name = InstanceName.parse(instance.name)
            except ValueError as e:
                log.warning(  # noqa: G200
                    "Invalid instance name %r: %s", instance.name, e
                )
            if instance_name.type != "playground":
                log.warning(
                    "Instance %r has unsupported type prefix %r",
                    instance.name,
                    instance_name.type,
                )
                continue
            instances[instance_name.mr][instance_name.variant] = instance

        mr_table = Table(box=box.SIMPLE)
        mr_table.add_column("MR")
        mr_table.add_column("Author")
        mr_table.add_column("Branch")
        mr_table.add_column("Title")
        mr_table.add_column("Instances")

        for mr in mrs.values():
            if (variants := instances.get(mr.iid, None)) is None:
                instance_names = "none"
            else:
                instance_names = ", ".join(
                    f"[link=https://{instance.name}.{self.playground.domain}]"
                    f"{name}[/link]"
                    for name, instance in sorted(variants.items())
                )
            mr_table.add_row(
                f"[link={mr.web_url}]!{mr.iid}[/link]",
                f"[link={mr.author['web_url']}]{mr.author['name']}[/link]",
                f"{mr.source_branch}",
                mr.title,
                instance_names,
            )

        instances_table = Table(box=box.SIMPLE)
        instances_table.add_column("MR")
        instances_table.add_column("Variant")
        instances_table.add_column("Name")
        instances_table.add_column("Status")
        for mr_id, variants in instances.items():
            for variant, instance in variants.items():
                instances_table.add_row(
                    f"!{mr_id}", variant, instance.name, instance.state
                )

        print("* Open merge requests")
        rich.print(mr_table)

        print("* Instances")
        rich.print(instances_table)


class Create(InstanceCommand):
    """Create a new playground instance."""

    @classmethod
    def add_subparser(cls, subparsers: "SubParsers") -> argparse.ArgumentParser:
        """Add options for `create` command."""
        parser = super().add_subparser(subparsers)
        parser.add_argument(
            "--type", default="t3a.large", help="EC2 instance type"
        )
        parser.add_argument(
            "--force", "-f", action="store_true", help="force creation"
        )
        return parser

    def run(self) -> None:
        """Run `create` command."""
        if not self.args.force and self.playground.mr.state != "opened":
            raise Fail(f"!{self.args.mr} is not an open merge request")

        try:
            self.playground.instance
        except InstanceDoesNotExist:
            pass
        else:
            raise Fail(
                f"Instance {self.playground.instance_name} already exists"
            )

        self.playground.aws.ec2_launch(
            name=str(self.playground.instance_name),
            ami=AMI,
            instance_type=self.args.type,
            subnet=SUBNET,
            security_groups=SECURITY_GROUPS,
            key_name=self.playground.key_name,
            ipv4=True,
        )
        self.playground.create_instance_dns_record()
        self.playground.print_instance_status()


class Delete(InstanceCommand):
    """Delete a playground server."""

    def run(self) -> None:
        """Run `delete` command."""
        self.playground.terminate_instance()


class Login(InstanceCommand):
    """Log into a server."""

    def run(self) -> None:
        """Run `login` command."""
        address = self.playground.address_ssh
        # TODO: run ssh-keygen to edit host keys to auth?
        os.execlp("ssh", "ssh", f"admin@{address}")


class Provision(InstanceCommand):
    """Provision a newly created server."""

    def run(self) -> None:
        """Run `provision` command."""
        # Enable ssh ssh as root
        address = self.playground.address_ssh
        subprocess.check_call(
            [
                "ssh",
                f"admin@{address}",
                "sudo sed -i 's/^.*\" ssh-/ssh-/' /root/.ssh/authorized_keys",
            ]
        )
        with AnsibleEnvironment(self.playground) as env:
            env.run_playbook()


class Status(InstanceCommand):
    """Status of a running server."""

    def run(self) -> None:
        """Run `status` command."""
        self.playground.print_instance_status()


class Cleanup(Command):
    """Remove servers for closed MRs."""

    @classmethod
    def add_subparser(cls, subparsers: "SubParsers") -> argparse.ArgumentParser:
        """Add options for `cleanup` command."""
        parser = super().add_subparser(subparsers)
        parser.add_argument(
            "--dry-run",
            "-n",
            action="store_true",
            help="check only, do not delete",
        )
        return parser

    def run(self) -> None:
        """Run `cleanup` command."""
        # Load list of opened merge requests
        mrs: set[int] = set()
        for mr in self.playground.debusine.mergerequests.list(state="opened"):
            mrs.add(mr.iid)

        # Load list of existing instances
        instances: dict[int, list[EC2Instance]] = defaultdict(list)
        for instance in self.playground.aws.ec2_instances():
            instance_name = InstanceName.parse(instance.name)
            if instance_name.type != "playground":
                log.warning(
                    "Instance %r has unsupported type prefix %r",
                    instance_name,
                    instance_name.type,
                )
                continue
            instances[instance_name.mr].append(instance)

        for mr_id in instances.keys() - mrs:
            for instance in instances[mr_id]:
                print(f"Expired instance: {instance.name}")
                if not self.args.dry_run:
                    self.playground.aws.ec2_terminate(instance)


class DNSList(Command):
    """List DNS records."""

    NAME = "dns-list"

    def run(self) -> None:
        """Run `dns-list` command."""
        for record in self.playground.zone_records():
            print(record)


class DNSCheck(Command):
    """Check and fix DNS records."""

    NAME = "dns-check"

    @classmethod
    def add_subparser(cls, subparsers: "SubParsers") -> argparse.ArgumentParser:
        """Add options for `dns-check` command."""
        parser = super().add_subparser(subparsers)
        parser.add_argument(
            "--fix", "-f", action="store_true", help="perform changes to DNS"
        )
        return parser

    def run(self) -> None:  # noqa: C901
        """Run `dns-check` command."""
        existing: set[DNSRecord] = set(self.playground.zone_records())
        wanted: set[DNSRecord] = set()

        instances: dict[str, EC2Instance] = {}
        for instance in self.playground.aws.ec2_instances():
            instances[instance.name] = instance
            wanted.update(
                instance.dns_records(
                    self.playground.domain, ttl=self.playground.ttl
                )
            )

        # Current status table
        status_table = Table(box=box.SIMPLE)
        status_table.add_column("Instance")
        status_table.add_column("Name")
        status_table.add_column("Type")
        status_table.add_column("TTL")
        status_table.add_column("Value")
        status_table.add_column("State")
        stale: list[DNSRecord] = []
        missing: list[DNSRecord] = []
        for record in sorted(existing | wanted):
            if record in existing and record in wanted:
                status = "ok"
            elif record in existing:
                status = "stale"
                stale.append(record)
            elif record in wanted:
                status = "missing"
                missing.append(record)
            url = f"https://{record.instance_name}.{self.playground.domain}"
            status_table.add_row(
                f"[link={url}]{record.instance_name}[/link]",
                record.name,
                record.type,
                str(record.ttl),
                record.value,
                status,
            )
        rich.print(status_table)

        # Delete stale or incorrect DNS records first
        if self.args.fix:
            log.info("Deleting %d stale DNS records", len(stale))
            self.playground.aws.route53_delete_records(
                self.playground.zone, stale
            )

        # Create missing or correct records
        instance_names_to_fix = {x.instance_name for x in missing}
        for name in instance_names_to_fix:
            instance = instances[name]
            log.info("Recreating DNS records for %s", name)
            self.playground.create_instance_dns_record(instance)


def main() -> None:
    """Run the playground-vm program."""
    parser = argparse.ArgumentParser(
        description="Manage ephemeral Hetzner machines"
    )
    subparsers = parser.add_subparsers(
        help="actions", required=True, dest="command_name"
    )

    Create.add_subparser(subparsers)
    Delete.add_subparser(subparsers)
    List.add_subparser(subparsers)
    Login.add_subparser(subparsers)
    Provision.add_subparser(subparsers)
    Status.add_subparser(subparsers)
    Cleanup.add_subparser(subparsers)
    DNSList.add_subparser(subparsers)
    DNSCheck.add_subparser(subparsers)

    args = parser.parse_args()

    with args.command(args) as cmd:
        cmd.run()


if __name__ == "__main__":
    try:
        main()
    except Fail as e:
        print(e, file=sys.stderr)
        sys.exit(1)
    except Exception:
        log.exception("uncaught exception")