Phase 2: droplet create/destroy saga

The most load-bearing write workflow — droplet provisioning is the spine
of phase 4a deployment onboarding.

DigitalOcean.Client: create_droplet, get_droplet, list_droplets_by_tag,
destroy_droplet. list_paginated/3 now threads caller-supplied params
(opts[:params]) through pagination so tag-filtered listing works.

Four droplet saga steps:
- CreateDroplet — POST a droplet, tagged arcadia-saga-<saga8> +
  managed-by-arcadia-cloud. Idempotency: re-run checks context for
  droplet_id, then queries DO by the saga tag, so a crash between POST
  and context-save adopts the existing droplet. compensate destroys it.
- WaitDropletActive — polls get_droplet until status "active" (96x5s);
  records the public IP. No compensation (waiting has no side effect).
- RegisterDroplet — fetches the droplet, upserts it into cloud_resources
  (inventory consistent immediately, not at next 15-min sync) and writes
  cloud_provisioned desired-state {size_slug, region, image}. compensate
  removes the DB rows (the droplet itself is destroyed by CreateDroplet's
  compensate).
- DestroyDroplet — DELETE the droplet + mark its cloud_resources row
  deleted. Terminal/irreversible: compensate is a logged noop, per the
  saga design destroy-class steps don't roll back.

Provisioning helpers:
- provision_droplet/1 — [CreateDroplet, WaitDropletActive, RegisterDroplet]
- destroy_droplet/2   — [DestroyDroplet]

Live smoke verified end-to-end (full create + destroy on a real
s-1vcpu-512mb-10gb droplet in syd1):
- provision saga completed: droplet 572017320 created, reached active
  with public IP, registered into cloud_resources (status=active) +
  cloud_provisioned (spec recorded).
- destroy saga completed: cloud_resources row marked deleted; droplet
  confirmed 404 on DO afterward. Account back to its original 5
  droplets, zero leftover, ~1 cent total cost.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

lib/arcadia_cloud/provisioning/steps/register_droplet.ex

@@ -0,0 +1,103 @@
+defmodule ArcadiaCloud.Provisioning.Steps.RegisterDroplet do
+  @moduledoc """
+  Makes arcadia-cloud's own DB reflect a freshly-provisioned droplet
+  without waiting for the next 15-minute sync:
+
+    1. fetch the droplet from DO
+    2. upsert it into cloud_resources (inventory immediately consistent)
+    3. record desired-state in cloud_provisioned (so drift detection has
+       a baseline)
+
+  Compensation: marks the cloud_resources row deleted and removes the
+  cloud_provisioned row. The droplet itself is destroyed by the
+  CreateDroplet step's compensate — this step only undoes DB rows.
+  """
+
+  @behaviour ArcadiaCloud.Provisioning.Step
+
+  import Ecto.Query
+
+  alias ArcadiaCloud.{Cloud, Provisioning, Repo}
+  alias ArcadiaCloud.Cloud.CloudResource
+  alias ArcadiaCloud.DigitalOcean.Client
+  alias ArcadiaCloud.Provisioning.{CloudProvisioned, SagaState}
+
+  @impl true
+  def name, do: "register_droplet"
+
+  @impl true
+  def execute(state) do
+    case SagaState.get_output(state, :droplet_id) do
+      nil ->
+        {:error, :no_droplet_id_in_context}
+
+      droplet_id ->
+        with {:ok, droplet} <- Client.get_droplet(droplet_id),
+             {:ok, resource} <- Cloud.upsert_resource(normalize(droplet)),
+             {:ok, _prov} <- record_provisioned(state, droplet, resource) do
+          {:ok, SagaState.put_output(state, :cloud_resource_id, resource.id)}
+        end
+    end
+  end
+
+  @impl true
+  def compensate(state) do
+    case SagaState.get_output(state, :cloud_resource_id) do
+      nil ->
+        :ok
+
+      resource_id ->
+        Repo.delete_all(from(p in CloudProvisioned, where: p.resource_id == ^resource_id))
+
+        from(r in CloudResource, where: r.id == ^resource_id)
+        |> Repo.update_all(set: [deleted_at: DateTime.utc_now() |> DateTime.truncate(:second)])
+
+        :ok
+    end
+  end
+
+  # ---- internals ------------------------------------------------------------
+
+  defp record_provisioned(state, droplet, resource) do
+    Provisioning.record_provisioned(
+      resource.id,
+      %{
+        "size_slug" => droplet["size_slug"],
+        "region" => get_in(droplet, ["region", "slug"]),
+        "image" => get_in(droplet, ["image", "slug"])
+      },
+      provisioned_by: "saga:#{state.saga_id}",
+      saga_id: state.saga_id
+    )
+  end
+
+  defp normalize(d) do
+    now = DateTime.utc_now() |> DateTime.truncate(:second)
+
+    %{
+      provider: "digitalocean",
+      provider_id: to_string(d["id"]),
+      kind: "droplet",
+      name: d["name"],
+      region: get_in(d, ["region", "slug"]),
+      status: normalize_status(d["status"]),
+      size_slug: d["size_slug"],
+      tags: d["tags"] || [],
+      attrs: %{
+        memory_mb: d["memory"],
+        vcpus: d["vcpus"],
+        disk_gb: d["disk"],
+        networks: d["networks"],
+        do_created_at: d["created_at"]
+      },
+      first_seen_at: now,
+      last_seen_at: now
+    }
+  end
+
+  defp normalize_status("active"), do: "active"
+  defp normalize_status("off"), do: "off"
+  defp normalize_status("new"), do: "provisioning"
+  defp normalize_status(other) when is_binary(other), do: other
+  defp normalize_status(_), do: "unknown"
+end