stripe and email after signup

config/runtime.exs

@@ -219,7 +219,9 @@ if config_env() == :prod do
          # System insights (agent offline detection) every 5 minutes
          {"*/5 * * * *", Towerops.Workers.SystemInsightWorker},
          # Gaiia reconciliation insights nightly at 4:30 AM
-         {"30 4 * * *", Towerops.Workers.GaiiaInsightWorker}
+         {"30 4 * * *", Towerops.Workers.GaiiaInsightWorker},
+         # Sync device usage to Stripe daily at 3 AM UTC
+         {"0 3 * * *", Towerops.Workers.BillingSyncWorker}
        ]},
       # Automatically delete completed jobs after 60 seconds
       {Oban.Plugins.Pruner, max_age: 60},
@@ -348,4 +350,39 @@ if config_env() == :prod do
   if owm_key = System.get_env("OPENWEATHERMAP_API_KEY") do
     config :towerops, :openweathermap_api_key, owm_key
   end
+
+  # Stripe configuration for billing
+  stripe_secret_key =
+    System.get_env("STRIPE_SECRET_KEY") ||
+      raise """
+      environment variable STRIPE_SECRET_KEY is missing.
+      Get this from Stripe dashboard and store in K8s secrets.
+      """
+
+  stripe_webhook_secret =
+    System.get_env("STRIPE_WEBHOOK_SECRET") ||
+      raise """
+      environment variable STRIPE_WEBHOOK_SECRET is missing.
+      Get this from Stripe webhook settings and store in K8s secrets.
+      """
+
+  stripe_price_id =
+    System.get_env("STRIPE_PRICE_ID") ||
+      raise """
+      environment variable STRIPE_PRICE_ID is missing.
+      This is the Stripe price ID for the metered billing plan.
+      """
+
+  config :towerops,
+    stripe_secret_key: stripe_secret_key,
+    stripe_webhook_secret: stripe_webhook_secret,
+    stripe_price_id: stripe_price_id
+end
+
+# Test environment Stripe configuration
+if config_env() == :test do
+  config :towerops,
+    stripe_secret_key: "sk_test_fake",
+    stripe_webhook_secret: "whsec_test_fake",
+    stripe_price_id: "price_test_fake"
 end

lib/towerops/accounts/user_notifier.ex

@@ -46,7 +46,7 @@ defmodule Towerops.Accounts.UserNotifier do
     body =
       t_email(
         """
-        Hi %{email},
+        Hi %{name},
 
         You can change your email by visiting the URL below:
 
@@ -54,7 +54,7 @@ defmodule Towerops.Accounts.UserNotifier do
 
         If you didn't request this change, please ignore this.
         """,
-        email: user.email,
+        name: user.first_name || user.email,
         url: url
       )
 
@@ -77,7 +77,7 @@ defmodule Towerops.Accounts.UserNotifier do
     body =
       t_email(
         """
-        Hi %{email},
+        Hi %{name},
 
         You can log into your account by visiting the URL below:
 
@@ -85,7 +85,7 @@ defmodule Towerops.Accounts.UserNotifier do
 
         If you didn't request this email, please ignore this.
         """,
-        email: user.email,
+        name: user.first_name || user.email,
         url: url
       )
 
@@ -101,7 +101,7 @@ defmodule Towerops.Accounts.UserNotifier do
     body =
       t_email(
         """
-        Hi %{email},
+        Hi %{name},
 
         You can confirm your account by visiting the URL below:
 
@@ -109,7 +109,7 @@ defmodule Towerops.Accounts.UserNotifier do
 
         If you didn't create an account with us, please ignore this.
         """,
-        email: user.email,
+        name: user.first_name || user.email,
         url: url
       )
 
@@ -139,6 +139,44 @@ defmodule Towerops.Accounts.UserNotifier do
     deliver(invitation.email, subject, body)
   end
 
+  @doc """
+  Deliver a welcome email to a new user.
+  """
+  def deliver_welcome_email(user) do
+    subject = t_email("Welcome to Towerops")
+
+    body =
+      t_email(
+        """
+        Hi %{name},
+
+        Thank you for signing up for Towerops. Even though this email is automated, you can reply and it'll go directly to me.
+
+        I created Towerops to fill a void I found while running my own small WISP. It's still a work in progress, but I hope you find it useful. Feel free to reply to this email and let me know any improvements or new features you would like to see and I'll do my best to implement them.
+
+        Graham McIntire
+        """,
+        name: user.first_name || user.email
+      )
+
+    email =
+      new()
+      |> to(user.email)
+      |> from({"Graham McIntire", "graham@towerops.net"})
+      |> subject(subject)
+      |> text_body(body)
+
+    case Mailer.deliver(email) do
+      {:ok, _metadata} ->
+        Logger.info("Welcome email sent to #{user.email}")
+        {:ok, email}
+
+      {:error, reason} = error ->
+        Logger.error("Failed to send welcome email to #{user.email}: #{inspect(reason)}")
+        error
+    end
+  end
+
   @doc """
   Deliver instructions to reset a user password.
   """
@@ -148,7 +186,7 @@ defmodule Towerops.Accounts.UserNotifier do
     body =
       t_email(
         """
-        Hi %{email},
+        Hi %{name},
 
         You can reset your password by visiting the URL below:
 
@@ -158,7 +196,7 @@ defmodule Towerops.Accounts.UserNotifier do
 
         This link will expire in 1 hour.
         """,
-        email: user.email,
+        name: user.first_name || user.email,
         url: url
       )
 

lib/towerops/billing.ex

@@ -0,0 +1,191 @@
+defmodule Towerops.Billing do
+  @moduledoc """
+  Billing context for subscription and payment management.
+
+  Integrates with Stripe for metered billing based on device count.
+  Pricing: $1/device/month after first 10 free devices.
+  """
+
+  alias Towerops.Billing.StripeClient
+  alias Towerops.Organizations.Organization
+  alias Towerops.Organizations.SubscriptionLimits
+  alias Towerops.Repo
+
+  require Logger
+
+  # Public API
+
+  @doc """
+  Create Stripe checkout session for organization to start paid subscription.
+
+  Returns URL to redirect user to Stripe-hosted checkout page.
+  """
+  def create_checkout_session(organization, opts \\ []) do
+    success_url = Keyword.fetch!(opts, :success_url)
+    cancel_url = Keyword.fetch!(opts, :cancel_url)
+
+    with {:ok, customer_id} <- ensure_stripe_customer(organization),
+         {:ok, checkout_url} <-
+           StripeClient.create_checkout_session(customer_id,
+             success_url: success_url,
+             cancel_url: cancel_url
+           ) do
+      {:ok, checkout_url}
+    else
+      {:error, reason} ->
+        Logger.error("Failed to create checkout session for org #{organization.id}: #{inspect(reason)}")
+
+        {:error, reason}
+    end
+  end
+
+  @doc """
+  Create Stripe Customer Portal session for billing management.
+
+  Returns URL to redirect user to Stripe-hosted portal (invoices, payment method, cancel).
+  """
+  def create_portal_session(organization, return_url) do
+    case organization.stripe_customer_id do
+      nil ->
+        {:error, :no_stripe_customer}
+
+      customer_id ->
+        case StripeClient.create_portal_session(customer_id, return_url) do
+          {:ok, portal_url} ->
+            {:ok, portal_url}
+
+          {:error, reason} ->
+            Logger.error("Failed to create portal session: #{inspect(reason)}")
+            {:error, reason}
+        end
+    end
+  end
+
+  @doc """
+  Calculate billable device count (devices over free tier).
+
+  ## Examples
+      iex> billable_device_count(organization)
+      15  # Organization has 25 devices, 10 free = 15 billable
+  """
+  def billable_device_count(organization) do
+    total = SubscriptionLimits.count_organization_devices(organization.id)
+    max(0, total - 10)
+  end
+
+  @doc """
+  Calculate estimated monthly cost based on current device count.
+
+  ## Examples
+      iex> estimated_monthly_cost(organization)
+      {:ok, %{devices: 25, billable: 15, cost_usd: #Decimal<15.00>}}
+  """
+  def estimated_monthly_cost(organization) do
+    total = SubscriptionLimits.count_organization_devices(organization.id)
+    billable = max(0, total - 10)
+    cost = Decimal.new(billable)
+
+    {:ok,
+     %{
+       devices: total,
+       billable: billable,
+       cost_usd: cost
+     }}
+  end
+
+  @doc """
+  Report current device usage to Stripe for metered billing.
+
+  Called daily by BillingSyncWorker.
+  """
+  def sync_usage_to_stripe(organization) do
+    if Organization.has_active_subscription?(organization) do
+      billable = billable_device_count(organization)
+
+      case StripeClient.report_usage(organization.stripe_customer_id, billable) do
+        :ok ->
+          update_billing_sync(organization, billable)
+          Logger.info("Synced usage for org #{organization.slug}: #{billable} billable devices")
+          :ok
+
+        {:error, reason} ->
+          Logger.error("Failed to sync usage for org #{organization.slug}: #{inspect(reason)}")
+          {:error, reason}
+      end
+    else
+      {:ok, :no_active_subscription}
+    end
+  end
+
+  @doc """
+  Update organization billing status from Stripe subscription object.
+
+  Called by webhook processor when subscription events occur.
+  """
+  def update_subscription_from_stripe(organization, stripe_subscription) do
+    attrs = %{
+      stripe_subscription_id: stripe_subscription["id"],
+      subscription_status: stripe_subscription["status"],
+      subscription_current_period_start: unix_to_datetime(stripe_subscription["current_period_start"]),
+      subscription_current_period_end: unix_to_datetime(stripe_subscription["current_period_end"]),
+      subscription_plan: subscription_plan_from_status(stripe_subscription["status"])
+    }
+
+    organization
+    |> Organization.changeset(attrs)
+    |> Repo.update()
+  end
+
+  @doc """
+  Handle payment method status change (valid, failed, requires action).
+  """
+  def update_payment_method_status(organization, status) when status in ["valid", "missing", "requires_action"] do
+    organization
+    |> Organization.changeset(%{payment_method_status: status})
+    |> Repo.update()
+  end
+
+  # Private Helpers
+
+  defp ensure_stripe_customer(organization) do
+    case organization.stripe_customer_id do
+      nil ->
+        # Create Stripe customer
+        case StripeClient.create_customer(organization) do
+          {:ok, %{"id" => customer_id}} ->
+            # Store customer ID
+            organization
+            |> Organization.changeset(%{stripe_customer_id: customer_id})
+            |> Repo.update()
+
+            {:ok, customer_id}
+
+          {:error, reason} ->
+            {:error, reason}
+        end
+
+      customer_id ->
+        {:ok, customer_id}
+    end
+  end
+
+  defp update_billing_sync(organization, device_count) do
+    organization
+    |> Organization.changeset(%{
+      last_billing_sync_at: DateTime.utc_now(),
+      last_synced_device_count: device_count
+    })
+    |> Repo.update()
+  end
+
+  defp subscription_plan_from_status("active"), do: "paid"
+  defp subscription_plan_from_status("trialing"), do: "paid"
+  defp subscription_plan_from_status("past_due"), do: "paid"
+  defp subscription_plan_from_status(_), do: "free"
+
+  defp unix_to_datetime(nil), do: nil
+
+  defp unix_to_datetime(unix_timestamp) do
+    DateTime.from_unix!(unix_timestamp)
+  end
+end

lib/towerops/billing/stripe_client.ex

@@ -0,0 +1,258 @@
+defmodule Towerops.Billing.StripeClient do
+  @moduledoc """
+  Direct Stripe API client using :req library.
+
+  Handles:
+  - Customer creation and management
+  - Checkout session creation (Stripe-hosted payment page)
+  - Customer Portal session creation (Stripe-hosted billing management)
+  - Subscription management
+  - Usage reporting for metered billing
+  """
+
+  require Logger
+
+  @base_url "https://api.stripe.com"
+  @api_version "2024-11-20.acacia"
+
+  # Customer Management
+
+  @doc "Create a Stripe customer for an organization"
+  def create_customer(organization) do
+    primary_owner = Towerops.Organizations.get_primary_owner(organization)
+
+    params = %{
+      email: primary_owner.email,
+      name: organization.name,
+      metadata: %{
+        towerops_org_id: organization.id,
+        towerops_org_slug: organization.slug
+      }
+    }
+
+    case post("/v1/customers", params) do
+      {:ok, customer} -> {:ok, customer}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  @doc "Update customer metadata"
+  def update_customer(stripe_customer_id, params) do
+    post("/v1/customers/#{stripe_customer_id}", params)
+  end
+
+  # Checkout & Portal
+
+  @doc """
+  Create a Checkout Session for subscription signup.
+
+  Returns URL to redirect user to Stripe-hosted payment page.
+  """
+  def create_checkout_session(stripe_customer_id, opts \\ []) do
+    price_id = stripe_price_id()
+    success_url = Keyword.fetch!(opts, :success_url)
+    cancel_url = Keyword.fetch!(opts, :cancel_url)
+
+    params = %{
+      customer: stripe_customer_id,
+      mode: "subscription",
+      line_items: [
+        %{
+          price: price_id
+        }
+      ],
+      success_url: success_url,
+      cancel_url: cancel_url,
+      allow_promotion_codes: true
+    }
+
+    case post("/v1/checkout/sessions", params) do
+      {:ok, %{"url" => url}} -> {:ok, url}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  @doc """
+  Create a Customer Portal session for billing management.
+
+  Returns URL to redirect user to Stripe-hosted portal.
+  """
+  def create_portal_session(stripe_customer_id, return_url) do
+    params = %{
+      customer: stripe_customer_id,
+      return_url: return_url
+    }
+
+    case post("/v1/billing_portal/sessions", params) do
+      {:ok, %{"url" => url}} -> {:ok, url}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  # Subscription Management
+
+  @doc "Get subscription details"
+  def get_subscription(stripe_subscription_id) do
+    get("/v1/subscriptions/#{stripe_subscription_id}")
+  end
+
+  @doc "Cancel subscription at period end"
+  def cancel_subscription(stripe_subscription_id) do
+    params = %{cancel_at_period_end: true}
+    post("/v1/subscriptions/#{stripe_subscription_id}", params)
+  end
+
+  # Usage Reporting
+
+  @doc """
+  Report device usage to Stripe for metered billing.
+
+  This creates a meter event that Stripe will use to calculate the monthly invoice.
+  """
+  def report_usage(stripe_customer_id, billable_device_count, timestamp \\ DateTime.utc_now()) do
+    params = %{
+      event_name: "devices_monitored",
+      payload: %{
+        stripe_customer_id: stripe_customer_id,
+        value: to_string(billable_device_count)
+      },
+      timestamp: DateTime.to_unix(timestamp)
+    }
+
+    case post("/v1/billing/meter_events", params) do
+      {:ok, _event} -> :ok
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  # HTTP Request Helpers
+
+  defp get(path) do
+    request(:get, path, [])
+  end
+
+  defp post(path, params) do
+    encoded = encode_params(params)
+    request(:post, path, body: encoded)
+  end
+
+  defp request(method, path, opts) do
+    url = @base_url <> path
+
+    req_opts =
+      [
+        method: method,
+        url: url,
+        headers: auth_headers(),
+        connect_options: [timeout: 10_000],
+        receive_timeout: 15_000,
+        decode_json: [keys: :strings]
+      ] ++ opts
+
+    # Use Req.Test in test environment (following existing pattern)
+    req_opts =
+      if Application.get_env(:towerops, :env) == :test do
+        Keyword.put(req_opts, :plug, {Req.Test, __MODULE__})
+      else
+        req_opts
+      end
+
+    try do
+      case Req.request(req_opts) do
+        {:ok, %Req.Response{status: status, body: body}} when status in 200..299 ->
+          {:ok, body}
+
+        {:ok, %Req.Response{status: 400, body: %{"error" => error}}} ->
+          Logger.warning("Stripe API bad request: #{inspect(error)}")
+          {:error, {:bad_request, error["message"]}}
+
+        {:ok, %Req.Response{status: 401}} ->
+          Logger.error("Stripe API unauthorized - check STRIPE_SECRET_KEY")
+          {:error, :unauthorized}
+
+        {:ok, %Req.Response{status: 429, headers: headers}} ->
+          retry_after = get_retry_after(headers)
+          Logger.warning("Stripe API rate limited, retry after #{retry_after}s")
+          {:error, {:rate_limited, retry_after}}
+
+        {:ok, %Req.Response{status: status, body: body}} ->
+          Logger.error("Stripe API error #{status}: #{inspect(body)}")
+          {:error, {:http_error, status, body}}
+
+        {:error, %Req.TransportError{reason: reason}} ->
+          Logger.error("Stripe API transport error: #{inspect(reason)}")
+          {:error, {:transport_error, reason}}
+      end
+    rescue
+      e in Req.TransportError ->
+        Logger.error("Stripe API transport error: #{inspect(e.reason)}")
+        {:error, {:transport_error, e.reason}}
+    end
+  end
+
+  defp auth_headers do
+    secret_key = Application.fetch_env!(:towerops, :stripe_secret_key)
+
+    [
+      {"authorization", "Bearer #{secret_key}"},
+      {"stripe-version", @api_version},
+      {"content-type", "application/x-www-form-urlencoded"}
+    ]
+  end
+
+  # Stripe uses form-encoded params, not JSON
+  defp encode_params(params) when is_map(params) do
+    params
+    |> flatten_params()
+    |> URI.encode_query()
+  end
+
+  # Flatten nested maps to Stripe's bracket notation
+  # %{metadata: %{org_id: "123"}} -> %{"metadata[org_id]" => "123"}
+  # %{line_items: [%{price: "price_123"}]} -> %{"line_items[0][price]" => "price_123"}
+  defp flatten_params(params, prefix \\ nil) do
+    Enum.reduce(params, %{}, fn {key, value}, acc ->
+      key_str = to_string(key)
+      full_key = if prefix, do: "#{prefix}[#{key_str}]", else: key_str
+
+      cond do
+        is_list(value) ->
+          flatten_list(value, full_key, acc)
+
+        is_map(value) ->
+          Map.merge(acc, flatten_params(value, full_key))
+
+        true ->
+          Map.put(acc, full_key, to_string(value))
+      end
+    end)
+  end
+
+  defp flatten_list(list, key, acc) do
+    list
+    |> Enum.with_index()
+    |> Enum.reduce(acc, fn {item, index}, list_acc ->
+      indexed_key = "#{key}[#{index}]"
+      Map.merge(list_acc, flatten_params(%{indexed_key => item}, nil))
+    end)
+  end
+
+  defp get_retry_after(headers) when is_list(headers) do
+    case List.keyfind(headers, "retry-after", 0) do
+      {_, value} -> String.to_integer(value)
+      nil -> 60
+    end
+  end
+
+  defp get_retry_after(headers) when is_map(headers) do
+    case Map.get(headers, "retry-after") do
+      [value | _] -> String.to_integer(value)
+      value when is_binary(value) -> String.to_integer(value)
+      nil -> 60
+    end
+  end
+
+  defp stripe_price_id do
+    Application.fetch_env!(:towerops, :stripe_price_id)
+  end
+end

lib/towerops/billing/stripe_webhook_event.ex

@@ -0,0 +1,27 @@
+defmodule Towerops.Billing.StripeWebhookEvent do
+  @moduledoc """
+  Schema for tracking processed Stripe webhook events.
+
+  Used for idempotency checking to prevent duplicate processing
+  when Stripe retries webhook delivery.
+  """
+  use Ecto.Schema
+
+  import Ecto.Changeset
+
+  @primary_key {:id, :string, autogenerate: false}
+  schema "stripe_webhook_events" do
+    field :event_type, :string
+    field :processed_at, :utc_datetime
+    field :created_at, :utc_datetime
+
+    belongs_to :organization, Towerops.Organizations.Organization, type: :binary_id
+  end
+
+  @doc false
+  def changeset(event, attrs) do
+    event
+    |> cast(attrs, [:id, :event_type, :organization_id, :processed_at, :created_at])
+    |> validate_required([:id, :event_type, :processed_at, :created_at])
+  end
+end

lib/towerops/billing/webhook_processor.ex

@@ -0,0 +1,168 @@
+defmodule Towerops.Billing.WebhookProcessor do
+  @moduledoc """
+  Processes Stripe webhook events.
+
+  Handles subscription lifecycle events with idempotency checking.
+  """
+
+  import Ecto.Query
+
+  alias Towerops.Billing
+  alias Towerops.Billing.BillingNotifier
+  alias Towerops.Billing.StripeWebhookEvent
+  alias Towerops.Organizations
+  alias Towerops.Organizations.Organization
+  alias Towerops.Repo
+
+  require Logger
+
+  @doc """
+  Process a Stripe webhook event.
+
+  Checks idempotency before processing to handle Stripe retries.
+  """
+  def process(%{"id" => event_id, "type" => event_type, "data" => data}) do
+    if event_already_processed?(event_id) do
+      Logger.info("Webhook #{event_id} already processed, skipping")
+      :ok
+    else
+      result = handle_event(event_type, data["object"])
+      mark_event_processed(event_id, event_type, extract_org_id(data["object"]))
+      result
+    end
+  end
+
+  # Subscription Events
+
+  defp handle_event("customer.subscription.created", subscription) do
+    Logger.info("Subscription created: #{subscription["id"]}")
+    update_organization_subscription(subscription)
+  end
+
+  defp handle_event("customer.subscription.updated", subscription) do
+    Logger.info("Subscription updated: #{subscription["id"]}")
+    update_organization_subscription(subscription)
+  end
+
+  defp handle_event("customer.subscription.deleted", subscription) do
+    Logger.info("Subscription deleted: #{subscription["id"]}")
+
+    with {:ok, org} <- find_organization_by_stripe_id(subscription),
+         {:ok, _org} <- downgrade_to_free(org) do
+      :ok
+    end
+  end
+
+  # Payment Events
+
+  defp handle_event("invoice.payment_succeeded", invoice) do
+    Logger.info("Payment succeeded for invoice #{invoice["id"]}")
+
+    with {:ok, org} <- find_organization_by_stripe_id(invoice),
+         {:ok, _org} <- Billing.update_payment_method_status(org, "valid") do
+      :ok
+    end
+  end
+
+  defp handle_event("invoice.payment_failed", invoice) do
+    Logger.warning("Payment failed for invoice #{invoice["id"]}")
+
+    with {:ok, org} <- find_organization_by_stripe_id(invoice),
+         {:ok, _org} <- Billing.update_payment_method_status(org, "requires_action") do
+      # Send email notification to org owners and admins
+      case BillingNotifier.deliver_payment_failed_notification(org) do
+        {:ok, _results} ->
+          Logger.info("Payment failure notification sent for organization #{org.slug}")
+          :ok
+
+        {:error, reason} ->
+          Logger.error("Failed to send payment failure notification for organization #{org.slug}: #{inspect(reason)}")
+
+          :ok
+      end
+    end
+  end
+
+  # Payment Method Events
+
+  defp handle_event("customer.updated", customer) do
+    # Check if default payment method changed
+    if customer["invoice_settings"]["default_payment_method"] do
+      with {:ok, org} <- find_organization_by_customer_id(customer["id"]),
+           {:ok, _org} <- Billing.update_payment_method_status(org, "valid") do
+        :ok
+      end
+    end
+
+    :ok
+  end
+
+  # Unhandled events
+  defp handle_event(event_type, _object) do
+    Logger.debug("Unhandled webhook event: #{event_type}")
+    :ok
+  end
+
+  # Helpers
+
+  defp update_organization_subscription(subscription) do
+    with {:ok, org} <- find_organization_by_stripe_id(subscription),
+         {:ok, _org} <- Billing.update_subscription_from_stripe(org, subscription) do
+      :ok
+    else
+      {:error, :organization_not_found} ->
+        Logger.warning("Organization not found for subscription #{subscription["id"]}")
+        :ok
+
+      error ->
+        Logger.error("Failed to update organization subscription: #{inspect(error)}")
+        :ok
+    end
+  end
+
+  defp downgrade_to_free(organization) do
+    organization
+    |> Organization.changeset(%{
+      subscription_plan: "free",
+      subscription_status: "canceled",
+      stripe_subscription_id: nil
+    })
+    |> Repo.update()
+  end
+
+  defp find_organization_by_stripe_id(%{"customer" => customer_id}) do
+    find_organization_by_customer_id(customer_id)
+  end
+
+  defp find_organization_by_customer_id(customer_id) do
+    case Organizations.get_organization_by_stripe_customer_id(customer_id) do
+      nil -> {:error, :organization_not_found}
+      org -> {:ok, org}
+    end
+  end
+
+  defp extract_org_id(%{"customer" => customer_id}) do
+    case Organizations.get_organization_by_stripe_customer_id(customer_id) do
+      nil -> nil
+      org -> org.id
+    end
+  end
+
+  defp extract_org_id(_), do: nil
+
+  defp event_already_processed?(event_id) do
+    Repo.exists?(from e in StripeWebhookEvent, where: e.id == ^event_id)
+  end
+
+  defp mark_event_processed(event_id, event_type, org_id) do
+    %StripeWebhookEvent{}
+    |> StripeWebhookEvent.changeset(%{
+      id: event_id,
+      event_type: event_type,
+      organization_id: org_id,
+      processed_at: DateTime.utc_now(),
+      created_at: DateTime.utc_now()
+    })
+    |> Repo.insert(on_conflict: :nothing)
+  end
+end

lib/towerops/organizations.ex

@@ -5,6 +5,7 @@ defmodule Towerops.Organizations do
 
   import Ecto.Query
 
+  alias Towerops.Accounts.User
   alias Towerops.Agents.AgentAssignment
   alias Towerops.Contexts.ConfigChangeTracker
   alias Towerops.Devices.Device
@@ -262,7 +263,7 @@ defmodule Towerops.Organizations do
   """
   def list_organization_notification_recipients(organization_id) do
     Repo.all(
-      from(u in Towerops.Accounts.User,
+      from(u in User,
         join: m in Membership,
         on: m.user_id == u.id,
         where: m.organization_id == ^organization_id,
@@ -522,4 +523,50 @@ defmodule Towerops.Organizations do
       false
   """
   defdelegate can?(membership, action, resource), to: Policy
+
+  ## Billing
+
+  @doc """
+  Gets an organization by Stripe customer ID.
+
+  Returns nil if not found.
+  """
+  @spec get_organization_by_stripe_customer_id(String.t()) :: Organization.t() | nil
+  def get_organization_by_stripe_customer_id(stripe_customer_id) do
+    Repo.get_by(Organization, stripe_customer_id: stripe_customer_id)
+  end
+
+  @doc """
+  Lists all organizations with active paid subscriptions.
+
+  Returns organizations where subscription_status is "active" or "trialing".
+  Used by BillingSyncWorker to sync usage to Stripe.
+  """
+  @spec list_organizations_with_active_subscriptions() :: [Organization.t()]
+  def list_organizations_with_active_subscriptions do
+    Repo.all(
+      from o in Organization,
+        where: o.subscription_status in ["active", "trialing"],
+        where: not is_nil(o.stripe_customer_id)
+    )
+  end
+
+  @doc """
+  Gets the primary owner of an organization.
+
+  Returns the user with the :owner role. If multiple owners exist (shouldn't happen),
+  returns the first one by insertion date.
+  """
+  @spec get_primary_owner(Organization.t()) :: User.t()
+  def get_primary_owner(%Organization{} = organization) do
+    Repo.one!(
+      from m in Membership,
+        join: u in assoc(m, :user),
+        where: m.organization_id == ^organization.id,
+        where: m.role == :owner,
+        order_by: [asc: m.inserted_at],
+        limit: 1,
+        select: u
+    )
+  end
 end

lib/towerops/organizations/organization.ex

@@ -49,6 +49,16 @@ defmodule Towerops.Organizations.Organization do
     field :mikrotik_use_ssl, :boolean, default: true
     field :mikrotik_enabled, :boolean, default: false
 
+    # Billing fields
+    field :stripe_customer_id, :string
+    field :stripe_subscription_id, :string
+    field :subscription_status, :string
+    field :subscription_current_period_start, :utc_datetime
+    field :subscription_current_period_end, :utc_datetime
+    field :payment_method_status, :string
+    field :last_billing_sync_at, :utc_datetime
+    field :last_synced_device_count, :integer
+
     belongs_to :default_agent_token, AgentToken
 
     has_many :memberships, Membership
@@ -80,6 +90,14 @@ defmodule Towerops.Organizations.Organization do
           mikrotik_ssh_port: integer() | nil,
           mikrotik_use_ssl: boolean() | nil,
           mikrotik_enabled: boolean() | nil,
+          stripe_customer_id: String.t() | nil,
+          stripe_subscription_id: String.t() | nil,
+          subscription_status: String.t() | nil,
+          subscription_current_period_start: DateTime.t() | nil,
+          subscription_current_period_end: DateTime.t() | nil,
+          payment_method_status: String.t() | nil,
+          last_billing_sync_at: DateTime.t() | nil,
+          last_synced_device_count: integer() | nil,
           default_agent_token_id: Ecto.UUID.t() | nil,
           default_agent_token: NotLoaded.t() | AgentToken.t() | nil,
           memberships: NotLoaded.t() | [Membership.t()],
@@ -112,11 +130,19 @@ defmodule Towerops.Organizations.Organization do
       :mikrotik_port,
       :mikrotik_ssh_port,
       :mikrotik_use_ssl,
-      :mikrotik_enabled
+      :mikrotik_enabled,
+      :stripe_customer_id,
+      :stripe_subscription_id,
+      :subscription_status,
+      :subscription_current_period_start,
+      :subscription_current_period_end,
+      :payment_method_status,
+      :last_billing_sync_at,
+      :last_synced_device_count
     ])
     |> validate_required([:name])
     |> validate_length(:name, min: 2, max: 100)
-    |> validate_inclusion(:subscription_plan, ["free"])
+    |> validate_inclusion(:subscription_plan, ["free", "paid"])
     |> validate_inclusion(:snmp_version, ["1", "2c", "3"], message: "must be 1, 2c, or 3")
     |> validate_number(:snmp_port, greater_than: 0, less_than: 65_536)
     |> validate_snmpv3_fields()
@@ -238,4 +264,21 @@ defmodule Towerops.Organizations.Organization do
       put_change(changeset, :slug, slug)
     end
   end
+
+  @doc """
+  Check if organization has active paid subscription.
+
+  Returns true if subscription status is "active" or "trialing".
+  """
+  def has_active_subscription?(%__MODULE__{subscription_status: status}) when status in ["active", "trialing"], do: true
+
+  def has_active_subscription?(_), do: false
+
+  @doc """
+  Check if payment method is valid.
+
+  Returns true if payment_method_status is "valid".
+  """
+  def has_valid_payment_method?(%__MODULE__{payment_method_status: "valid"}), do: true
+  def has_valid_payment_method?(_), do: false
 end

lib/towerops/workers/billing_sync_worker.ex

@@ -0,0 +1,46 @@
+defmodule Towerops.Workers.BillingSyncWorker do
+  @moduledoc """
+  Daily Oban job to sync device usage to Stripe for metered billing.
+
+  Runs at 3 AM UTC daily via Oban cron.
+  """
+
+  use Oban.Worker, queue: :maintenance, max_attempts: 3
+
+  alias Towerops.Billing
+  alias Towerops.Organizations
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(_job) do
+    Logger.info("Starting daily billing sync")
+
+    # Get all organizations with active paid subscriptions
+    organizations = Organizations.list_organizations_with_active_subscriptions()
+
+    results =
+      Enum.map(organizations, fn org ->
+        case Billing.sync_usage_to_stripe(org) do
+          :ok -> {:ok, org.slug}
+          {:error, reason} -> {:error, org.slug, reason}
+        end
+      end)
+
+    successes = Enum.count(results, &match?({:ok, _}, &1))
+    failures = Enum.count(results, &match?({:error, _, _}, &1))
+
+    Logger.info("Billing sync complete: #{successes} success, #{failures} failures")
+
+    if failures > 0 do
+      failed_orgs =
+        results
+        |> Enum.filter(&match?({:error, _, _}, &1))
+        |> Enum.map_join(", ", fn {:error, slug, reason} -> "#{slug}: #{inspect(reason)}" end)
+
+      Logger.error("Failed to sync: #{failed_orgs}")
+    end
+
+    :ok
+  end
+end

lib/towerops/workers/welcome_email_worker.ex

@@ -0,0 +1,34 @@
+defmodule Towerops.Workers.WelcomeEmailWorker do
+  @moduledoc """
+  Sends a welcome email to new users a few minutes after registration.
+  """
+  use Oban.Worker,
+    queue: :notifications,
+    max_attempts: 3
+
+  alias Towerops.Accounts
+  alias Towerops.Accounts.UserNotifier
+
+  require Logger
+
+  @delay_minutes 3
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{"user_id" => user_id}}) do
+    case Accounts.get_user(user_id) do
+      nil ->
+        Logger.warning("Welcome email skipped: user #{user_id} not found")
+        :ok
+
+      user ->
+        UserNotifier.deliver_welcome_email(user)
+        :ok
+    end
+  end
+
+  def enqueue(user_id) do
+    %{user_id: user_id}
+    |> new(schedule_in: @delay_minutes * 60)
+    |> Oban.insert()
+  end
+end

lib/towerops_web/controllers/api/v1/stripe_webhook_controller.ex

@@ -0,0 +1,103 @@
+defmodule ToweropsWeb.Api.V1.StripeWebhookController do
+  use ToweropsWeb, :controller
+
+  alias Towerops.Billing.WebhookProcessor
+
+  require Logger
+
+  @doc """
+  Stripe webhook endpoint.
+
+  Receives events like payment_intent.succeeded, customer.subscription.updated.
+  Verifies signature before processing.
+  """
+  def create(conn, _params) do
+    signature = conn |> get_req_header("stripe-signature") |> List.first()
+    raw_body = conn.private[:raw_body] || ""
+
+    case verify_signature(raw_body, signature) do
+      :ok ->
+        event = Jason.decode!(raw_body)
+
+        case WebhookProcessor.process(event) do
+          :ok ->
+            json(conn, %{received: true})
+
+          {:error, reason} ->
+            Logger.error("Webhook processing failed: #{inspect(reason)}")
+            json(conn, %{received: true})
+        end
+
+      {:error, :invalid_signature} ->
+        Logger.warning("Invalid webhook signature")
+
+        conn
+        |> put_status(401)
+        |> json(%{error: "Invalid signature"})
+
+      {:error, reason} ->
+        Logger.error("Webhook verification failed: #{inspect(reason)}")
+
+        conn
+        |> put_status(400)
+        |> json(%{error: "Verification failed"})
+    end
+  end
+
+  defp verify_signature(payload, signature_header) do
+    webhook_secret = Application.fetch_env!(:towerops, :stripe_webhook_secret)
+
+    with {:ok, timestamp, signature} <- parse_signature_header(signature_header),
+         :ok <- check_timestamp_tolerance(timestamp, 300) do
+      signed_payload = "#{timestamp}.#{payload}"
+
+      expected_signature =
+        :hmac
+        |> :crypto.mac(:sha256, webhook_secret, signed_payload)
+        |> Base.encode16(case: :lower)
+
+      if Plug.Crypto.secure_compare(signature, expected_signature) do
+        :ok
+      else
+        {:error, :invalid_signature}
+      end
+    end
+  end
+
+  defp parse_signature_header(header) do
+    parts = String.split(header, ",")
+
+    timestamp =
+      Enum.find_value(parts, fn part ->
+        case String.split(part, "=") do
+          ["t", t] -> t
+          _ -> nil
+        end
+      end)
+
+    signature =
+      Enum.find_value(parts, fn part ->
+        case String.split(part, "=") do
+          ["v1", sig] -> sig
+          _ -> nil
+        end
+      end)
+
+    if timestamp && signature do
+      {:ok, timestamp, signature}
+    else
+      {:error, :malformed_header}
+    end
+  end
+
+  defp check_timestamp_tolerance(timestamp_str, max_age_seconds) do
+    timestamp = String.to_integer(timestamp_str)
+    now = System.system_time(:second)
+
+    if abs(now - timestamp) <= max_age_seconds do
+      :ok
+    else
+      {:error, :timestamp_too_old}
+    end
+  end
+end

lib/towerops_web/controllers/user_registration_controller.ex

@@ -6,6 +6,7 @@ defmodule ToweropsWeb.UserRegistrationController do
   alias Towerops.Accounts
   alias Towerops.Accounts.User
   alias Towerops.Organizations
+  alias Towerops.Workers.WelcomeEmailWorker
   alias ToweropsWeb.UserAuth
 
   def new(conn, params) do
@@ -37,6 +38,8 @@ defmodule ToweropsWeb.UserRegistrationController do
   defp create_with_organization(conn, user_params) do
     case Accounts.register_user_with_organization(user_params) do
       {:ok, user} ->
+        WelcomeEmailWorker.enqueue(user.id)
+
         conn
         |> put_flash(:info, t("Account created successfully."))
         |> UserAuth.log_in_user(user)
@@ -54,6 +57,8 @@ defmodule ToweropsWeb.UserRegistrationController do
     with {:ok, invitation} <- get_valid_invitation(invitation_token),
          {:ok, user} <- Accounts.register_user(user_params),
          {:ok, _membership} <- Organizations.accept_invitation(invitation, user.id) do
+      WelcomeEmailWorker.enqueue(user.id)
+
       conn
       |> put_flash(:info, t("Account created successfully. Welcome to %{org}!", org: invitation.organization.name))
       |> UserAuth.log_in_user(user)

lib/towerops_web/live/user_registration_live.ex

@@ -8,6 +8,7 @@ defmodule ToweropsWeb.UserRegistrationLive do
   alias Towerops.Accounts.HIBP
   alias Towerops.Accounts.User
   alias Towerops.Organizations
+  alias Towerops.Workers.WelcomeEmailWorker
 
   def mount(params, session, socket) do
     # Check if registering via invitation
@@ -97,6 +98,8 @@ defmodule ToweropsWeb.UserRegistrationLive do
           &url(~p"/users/confirm/#{&1}")
         )
 
+        WelcomeEmailWorker.enqueue(user.id)
+
         {:noreply,
          socket
          |> put_flash(:info, t("Account created! Please check your email to verify your account."))
@@ -120,6 +123,8 @@ defmodule ToweropsWeb.UserRegistrationLive do
         &url(~p"/users/confirm/#{&1}")
       )
 
+      WelcomeEmailWorker.enqueue(user.id)
+
       {:noreply,
        socket
        |> put_flash(:info, t("Account created! Please check your email to verify your account."))

lib/towerops_web/router.ex

@@ -123,6 +123,13 @@ defmodule ToweropsWeb.Router do
     get "/devices/:id", MobileController, :get_equipment
   end
 
+  # Stripe webhook endpoint (no authentication, verified by signature)
+  scope "/api/v1", V1 do
+    pipe_through :api
+
+    post "/webhooks/stripe", StripeWebhookController, :create
+  end
+
   # GraphQL API endpoint (requires API token authentication)
   scope "/api/graphql" do
     pipe_through [:api_v1, :rate_limit_api, :graphql_context]

priv/repo/migrations/20260306145558_add_billing_to_organizations.exs

@@ -0,0 +1,31 @@
+defmodule Towerops.Repo.Migrations.AddBillingToOrganizations do
+  use Ecto.Migration
+
+  def change do
+    alter table(:organizations) do
+      # Stripe customer and subscription IDs
+      add :stripe_customer_id, :string
+      add :stripe_subscription_id, :string
+
+      # Subscription status: "active", "past_due", "canceled", "incomplete"
+      add :subscription_status, :string
+
+      # Billing period tracking
+      add :subscription_current_period_start, :utc_datetime
+      add :subscription_current_period_end, :utc_datetime
+
+      # Payment method status: "valid", "missing", "requires_action"
+      add :payment_method_status, :string
+
+      # Last successful billing sync timestamp
+      add :last_billing_sync_at, :utc_datetime
+
+      # Device count at last sync (for debugging/reconciliation)
+      add :last_synced_device_count, :integer
+    end
+
+    create index(:organizations, [:stripe_customer_id])
+    create index(:organizations, [:stripe_subscription_id])
+    create index(:organizations, [:subscription_status])
+  end
+end

priv/repo/migrations/20260306145559_create_stripe_webhook_events.exs

@@ -0,0 +1,17 @@
+defmodule Towerops.Repo.Migrations.CreateStripeWebhookEvents do
+  use Ecto.Migration
+
+  def change do
+    create table(:stripe_webhook_events, primary_key: false) do
+      # Stripe event ID (evt_xxx)
+      add :id, :string, primary_key: true
+      add :event_type, :string, null: false
+      add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all)
+      add :processed_at, :utc_datetime, null: false
+      add :created_at, :utc_datetime, null: false
+    end
+
+    create index(:stripe_webhook_events, [:event_type])
+    create index(:stripe_webhook_events, [:organization_id])
+  end
+end

test/support/fixtures/billing_fixtures.ex

@@ -0,0 +1,151 @@
+defmodule Towerops.BillingFixtures do
+  @moduledoc """
+  Test fixtures for billing-related entities.
+  """
+
+  alias Towerops.Organizations.Organization
+  alias Towerops.Repo
+
+  @doc """
+  Update organization with Stripe customer ID.
+  """
+  def organization_with_stripe_customer(organization, stripe_customer_id \\ "cus_test_123") do
+    organization
+    |> Organization.changeset(%{stripe_customer_id: stripe_customer_id})
+    |> Repo.update!()
+  end
+
+  @doc """
+  Update organization with active subscription.
+  """
+  def organization_with_active_subscription(organization) do
+    organization
+    |> Organization.changeset(%{
+      stripe_customer_id: "cus_test_#{System.unique_integer()}",
+      stripe_subscription_id: "sub_test_#{System.unique_integer()}",
+      subscription_status: "active",
+      subscription_plan: "paid",
+      subscription_current_period_start: DateTime.utc_now(),
+      subscription_current_period_end: DateTime.add(DateTime.utc_now(), 30, :day),
+      payment_method_status: "valid"
+    })
+    |> Repo.update!()
+  end
+
+  @doc """
+  Update organization with past_due subscription.
+  """
+  def organization_with_past_due_subscription(organization) do
+    organization
+    |> Organization.changeset(%{
+      stripe_customer_id: "cus_test_#{System.unique_integer()}",
+      stripe_subscription_id: "sub_test_#{System.unique_integer()}",
+      subscription_status: "past_due",
+      subscription_plan: "paid",
+      payment_method_status: "requires_action"
+    })
+    |> Repo.update!()
+  end
+
+  @doc """
+  Stripe customer object (mock).
+  """
+  def stripe_customer_object(attrs \\ %{}) do
+    %{
+      "id" => Map.get(attrs, :id, "cus_test_123"),
+      "object" => "customer",
+      "email" => Map.get(attrs, :email, "test@example.com"),
+      "name" => Map.get(attrs, :name, "Test Org"),
+      "metadata" => %{
+        "towerops_org_id" => Map.get(attrs, :org_id, "org_123"),
+        "towerops_org_slug" => Map.get(attrs, :org_slug, "testorg")
+      },
+      "invoice_settings" => %{
+        "default_payment_method" => Map.get(attrs, :payment_method, "pm_test_123")
+      }
+    }
+  end
+
+  @doc """
+  Stripe subscription object (mock).
+  """
+  def stripe_subscription_object(attrs \\ %{}) do
+    now = System.system_time(:second)
+
+    base = %{
+      "id" => Map.get(attrs, :id, "sub_test_123"),
+      "object" => "subscription",
+      "customer" => Map.get(attrs, :customer, "cus_test_123"),
+      "status" => Map.get(attrs, :status, "active"),
+      "current_period_start" => Map.get(attrs, :period_start, now),
+      "current_period_end" => Map.get(attrs, :period_end, now + 30 * 86_400),
+      "items" => %{
+        "data" => [
+          %{
+            "price" => %{
+              "id" => "price_test_123"
+            }
+          }
+        ]
+      }
+    }
+
+    # Merge any additional string-key attrs (like "cancel_at_period_end")
+    attrs
+    |> Enum.filter(fn {k, _v} -> is_binary(k) or is_atom(k) end)
+    |> Map.new(fn {k, v} -> {to_string(k), v} end)
+    |> Map.merge(base, fn _k, v1, _v2 -> v1 end)
+  end
+
+  @doc """
+  Stripe checkout session object (mock).
+  """
+  def stripe_checkout_session_object(attrs \\ %{}) do
+    %{
+      "id" => Map.get(attrs, :id, "cs_test_123"),
+      "object" => "checkout.session",
+      "url" => Map.get(attrs, :url, "https://checkout.stripe.com/c/pay/cs_test_123"),
+      "customer" => Map.get(attrs, :customer, "cus_test_123")
+    }
+  end
+
+  @doc """
+  Stripe portal session object (mock).
+  """
+  def stripe_portal_session_object(attrs \\ %{}) do
+    %{
+      "id" => Map.get(attrs, :id, "bps_test_123"),
+      "object" => "billing_portal.session",
+      "url" => Map.get(attrs, :url, "https://billing.stripe.com/p/session/bps_test_123"),
+      "customer" => Map.get(attrs, :customer, "cus_test_123")
+    }
+  end
+
+  @doc """
+  Stripe invoice object (mock).
+  """
+  def stripe_invoice_object(attrs \\ %{}) do
+    %{
+      "id" => Map.get(attrs, :id, "in_test_123"),
+      "object" => "invoice",
+      "customer" => Map.get(attrs, :customer, "cus_test_123"),
+      "status" => Map.get(attrs, :status, "paid"),
+      "subscription" => Map.get(attrs, :subscription, "sub_test_123")
+    }
+  end
+
+  @doc """
+  Stripe webhook event object (mock).
+  """
+  def stripe_webhook_event_object(type, data) do
+    %{
+      "id" => "evt_test_#{System.unique_integer()}",
+      "object" => "event",
+      "type" => type,
+      "data" => %{
+        "object" => data
+      },
+      "created" => System.system_time(:second)
+    }
+  end
+end

test/towerops/billing/stripe_client_test.exs

@@ -0,0 +1,318 @@
+defmodule Towerops.Billing.StripeClientTest do
+  use Towerops.DataCase
+
+  import Towerops.AccountsFixtures
+  import Towerops.BillingFixtures
+  import Towerops.OrganizationsFixtures
+
+  alias Towerops.Billing.StripeClient
+
+  setup do
+    user = user_fixture()
+    organization = organization_fixture(user.id)
+    %{user: user, organization: organization}
+  end
+
+  describe "create_customer/1" do
+    test "creates a Stripe customer with organization metadata", %{organization: organization} do
+      Req.Test.stub(StripeClient, fn conn ->
+        assert conn.method == "POST"
+        assert conn.request_path == "/v1/customers"
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+
+        # Verify form-encoded body contains expected fields
+        params = URI.decode_query(body)
+        assert params["name"] == organization.name
+        assert params["metadata[towerops_org_id]"] == organization.id
+        assert params["metadata[towerops_org_slug]"] == organization.slug
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(stripe_customer_object()))
+      end)
+
+      assert {:ok, customer} = StripeClient.create_customer(organization)
+      assert customer["id"] == "cus_test_123"
+      assert customer["object"] == "customer"
+    end
+
+    test "returns error on API failure", %{organization: organization} do
+      Req.Test.stub(StripeClient, fn conn ->
+        error = %{
+          "error" => %{
+            "type" => "invalid_request_error",
+            "message" => "Email is invalid"
+          }
+        }
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(400, Jason.encode!(error))
+      end)
+
+      assert {:error, {:bad_request, "Email is invalid"}} =
+               StripeClient.create_customer(organization)
+    end
+
+    test "returns error on unauthorized", %{organization: organization} do
+      Req.Test.stub(StripeClient, fn conn ->
+        Plug.Conn.send_resp(conn, 401, "")
+      end)
+
+      assert {:error, :unauthorized} = StripeClient.create_customer(organization)
+    end
+  end
+
+  describe "update_customer/2" do
+    test "updates a Stripe customer" do
+      Req.Test.stub(StripeClient, fn conn ->
+        assert conn.method == "POST"
+        assert conn.request_path == "/v1/customers/cus_test_123"
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+
+        params = URI.decode_query(body)
+        assert params["name"] == "Updated Name"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(stripe_customer_object(%{name: "Updated Name"})))
+      end)
+
+      assert {:ok, customer} = StripeClient.update_customer("cus_test_123", %{name: "Updated Name"})
+      assert customer["name"] == "Updated Name"
+    end
+  end
+
+  describe "create_checkout_session/2" do
+    test "creates a checkout session with correct parameters" do
+      Req.Test.stub(StripeClient, fn conn ->
+        assert conn.method == "POST"
+        assert conn.request_path == "/v1/checkout/sessions"
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+
+        params = URI.decode_query(body)
+        assert params["customer"] == "cus_test_123"
+        assert params["mode"] == "subscription"
+        assert params["success_url"] == "https://example.com/success"
+        assert params["cancel_url"] == "https://example.com/cancel"
+        assert params["allow_promotion_codes"] == "true"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(stripe_checkout_session_object()))
+      end)
+
+      assert {:ok, url} =
+               StripeClient.create_checkout_session("cus_test_123",
+                 success_url: "https://example.com/success",
+                 cancel_url: "https://example.com/cancel"
+               )
+
+      assert url == "https://checkout.stripe.com/c/pay/cs_test_123"
+    end
+
+    test "returns error when session creation fails" do
+      Req.Test.stub(StripeClient, fn conn ->
+        error = %{
+          "error" => %{
+            "message" => "Customer not found"
+          }
+        }
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(400, Jason.encode!(error))
+      end)
+
+      assert {:error, {:bad_request, "Customer not found"}} =
+               StripeClient.create_checkout_session("cus_invalid",
+                 success_url: "https://example.com/success",
+                 cancel_url: "https://example.com/cancel"
+               )
+    end
+  end
+
+  describe "create_portal_session/2" do
+    test "creates a portal session with correct parameters" do
+      Req.Test.stub(StripeClient, fn conn ->
+        assert conn.method == "POST"
+        assert conn.request_path == "/v1/billing_portal/sessions"
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+
+        params = URI.decode_query(body)
+        assert params["customer"] == "cus_test_123"
+        assert params["return_url"] == "https://example.com/return"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(stripe_portal_session_object()))
+      end)
+
+      assert {:ok, url} = StripeClient.create_portal_session("cus_test_123", "https://example.com/return")
+      assert url == "https://billing.stripe.com/p/session/bps_test_123"
+    end
+  end
+
+  describe "get_subscription/1" do
+    test "retrieves a subscription" do
+      Req.Test.stub(StripeClient, fn conn ->
+        assert conn.method == "GET"
+        assert conn.request_path == "/v1/subscriptions/sub_test_123"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(stripe_subscription_object()))
+      end)
+
+      assert {:ok, subscription} = StripeClient.get_subscription("sub_test_123")
+      assert subscription["id"] == "sub_test_123"
+      assert subscription["status"] == "active"
+    end
+
+    test "returns error when subscription not found" do
+      Req.Test.stub(StripeClient, fn conn ->
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(404, Jason.encode!(%{"error" => %{"message" => "Not found"}}))
+      end)
+
+      assert {:error, {:http_error, 404, _}} = StripeClient.get_subscription("sub_invalid")
+    end
+  end
+
+  describe "cancel_subscription/1" do
+    test "cancels a subscription at period end" do
+      Req.Test.stub(StripeClient, fn conn ->
+        assert conn.method == "POST"
+        assert conn.request_path == "/v1/subscriptions/sub_test_123"
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+
+        params = URI.decode_query(body)
+        assert params["cancel_at_period_end"] == "true"
+
+        subscription = stripe_subscription_object(%{cancel_at_period_end: true})
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(subscription))
+      end)
+
+      assert {:ok, subscription} = StripeClient.cancel_subscription("sub_test_123")
+      assert subscription["cancel_at_period_end"] == true
+    end
+  end
+
+  describe "report_usage/3" do
+    test "reports usage to Stripe meter events" do
+      Req.Test.stub(StripeClient, fn conn ->
+        assert conn.method == "POST"
+        assert conn.request_path == "/v1/billing/meter_events"
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+
+        params = URI.decode_query(body)
+        assert params["event_name"] == "devices_monitored"
+        assert params["payload[stripe_customer_id]"] == "cus_test_123"
+        assert params["payload[value]"] == "15"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(%{"id" => "mtr_event_123"}))
+      end)
+
+      assert :ok = StripeClient.report_usage("cus_test_123", 15)
+    end
+
+    test "accepts custom timestamp" do
+      timestamp = DateTime.utc_now()
+
+      Req.Test.stub(StripeClient, fn conn ->
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+        params = URI.decode_query(body)
+        assert params["timestamp"] == to_string(DateTime.to_unix(timestamp))
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(%{"id" => "mtr_event_123"}))
+      end)
+
+      assert :ok = StripeClient.report_usage("cus_test_123", 10, timestamp)
+    end
+
+    test "returns error on API failure" do
+      Req.Test.stub(StripeClient, fn conn ->
+        error = %{"error" => %{"message" => "Rate limit exceeded"}}
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(400, Jason.encode!(error))
+      end)
+
+      assert {:error, {:bad_request, "Rate limit exceeded"}} =
+               StripeClient.report_usage("cus_test_123", 15)
+    end
+  end
+
+  describe "rate limiting" do
+    test "handles rate limit with retry-after header", %{organization: organization} do
+      Req.Test.stub(StripeClient, fn conn ->
+        conn
+        |> Plug.Conn.put_resp_header("retry-after", "120")
+        |> Plug.Conn.send_resp(429, "")
+      end)
+
+      assert {:error, {:rate_limited, 120}} = StripeClient.create_customer(organization)
+    end
+
+    test "uses default retry-after when header missing", %{organization: organization} do
+      Req.Test.stub(StripeClient, fn conn ->
+        Plug.Conn.send_resp(conn, 429, "")
+      end)
+
+      assert {:error, {:rate_limited, 60}} = StripeClient.create_customer(organization)
+    end
+  end
+
+  describe "parameter flattening" do
+    test "correctly flattens nested metadata" do
+      Req.Test.stub(StripeClient, fn conn ->
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+        params = URI.decode_query(body)
+
+        # Verify nested metadata is flattened with bracket notation
+        assert params["metadata[org_id]"] == "org_123"
+        assert params["metadata[org_slug]"] == "testorg"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(stripe_customer_object()))
+      end)
+
+      assert {:ok, _} =
+               StripeClient.update_customer("cus_test_123", %{
+                 metadata: %{
+                   org_id: "org_123",
+                   org_slug: "testorg"
+                 }
+               })
+    end
+  end
+
+  describe "transport errors" do
+    test "handles network timeout", %{organization: organization} do
+      Req.Test.stub(StripeClient, fn _conn ->
+        raise Req.TransportError, reason: :timeout
+      end)
+
+      assert {:error, {:transport_error, :timeout}} = StripeClient.create_customer(organization)
+    end
+
+    test "handles connection refused", %{organization: organization} do
+      Req.Test.stub(StripeClient, fn _conn ->
+        raise Req.TransportError, reason: :econnrefused
+      end)
+
+      assert {:error, {:transport_error, :econnrefused}} =
+               StripeClient.create_customer(organization)
+    end
+  end
+end

test/towerops/billing/webhook_processor_test.exs

@@ -0,0 +1,264 @@
+defmodule Towerops.Billing.WebhookProcessorTest do
+  use Towerops.DataCase
+
+  import Swoosh.TestAssertions
+  import Towerops.AccountsFixtures
+  import Towerops.BillingFixtures
+  import Towerops.OrganizationsFixtures
+
+  alias Towerops.Billing.StripeWebhookEvent
+  alias Towerops.Billing.WebhookProcessor
+  alias Towerops.Organizations
+  alias Towerops.Repo
+
+  setup do
+    user = user_fixture()
+    organization = organization_fixture(user.id)
+    organization = organization_with_stripe_customer(organization, "cus_webhook_test")
+    %{user: user, organization: organization}
+  end
+
+  describe "process/1" do
+    test "marks event as processed in database", %{organization: organization} do
+      subscription = stripe_subscription_object(%{customer: organization.stripe_customer_id})
+      event = stripe_webhook_event_object("customer.subscription.created", subscription)
+
+      assert :ok = WebhookProcessor.process(event)
+
+      # Verify event was recorded
+      assert Repo.get(StripeWebhookEvent, event["id"])
+    end
+
+    test "skips processing for duplicate events", %{organization: organization} do
+      subscription = stripe_subscription_object(%{customer: organization.stripe_customer_id})
+      event = stripe_webhook_event_object("customer.subscription.created", subscription)
+
+      # Process once
+      assert :ok = WebhookProcessor.process(event)
+
+      # Process again - should skip
+      assert :ok = WebhookProcessor.process(event)
+
+      # Verify only one record exists
+      assert Repo.aggregate(StripeWebhookEvent, :count) == 1
+    end
+  end
+
+  describe "customer.subscription.created" do
+    test "updates organization with new subscription", %{organization: organization} do
+      subscription =
+        stripe_subscription_object(%{
+          id: "sub_new_123",
+          customer: organization.stripe_customer_id,
+          status: "active"
+        })
+
+      event = stripe_webhook_event_object("customer.subscription.created", subscription)
+
+      assert :ok = WebhookProcessor.process(event)
+
+      updated_org = Organizations.get_organization!(organization.id)
+      assert updated_org.stripe_subscription_id == "sub_new_123"
+      assert updated_org.subscription_status == "active"
+      assert updated_org.subscription_plan == "paid"
+    end
+  end
+
+  describe "customer.subscription.updated" do
+    test "updates organization subscription status", %{organization: organization} do
+      organization = organization_with_active_subscription(organization)
+
+      subscription =
+        stripe_subscription_object(%{
+          id: organization.stripe_subscription_id,
+          customer: organization.stripe_customer_id,
+          status: "past_due"
+        })
+
+      event = stripe_webhook_event_object("customer.subscription.updated", subscription)
+
+      assert :ok = WebhookProcessor.process(event)
+
+      updated_org = Organizations.get_organization!(organization.id)
+      assert updated_org.subscription_status == "past_due"
+    end
+  end
+
+  describe "customer.subscription.deleted" do
+    test "downgrades organization to free plan", %{organization: organization} do
+      organization = organization_with_active_subscription(organization)
+
+      subscription =
+        stripe_subscription_object(%{
+          id: organization.stripe_subscription_id,
+          customer: organization.stripe_customer_id,
+          status: "canceled"
+        })
+
+      event = stripe_webhook_event_object("customer.subscription.deleted", subscription)
+
+      assert :ok = WebhookProcessor.process(event)
+
+      updated_org = Organizations.get_organization!(organization.id)
+      assert updated_org.subscription_plan == "free"
+      assert updated_org.subscription_status == "canceled"
+      assert is_nil(updated_org.stripe_subscription_id)
+    end
+  end
+
+  describe "invoice.payment_succeeded" do
+    test "updates payment method status to valid", %{organization: organization} do
+      organization = organization_with_past_due_subscription(organization)
+
+      invoice = stripe_invoice_object(%{customer: organization.stripe_customer_id})
+      event = stripe_webhook_event_object("invoice.payment_succeeded", invoice)
+
+      assert :ok = WebhookProcessor.process(event)
+
+      updated_org = Organizations.get_organization!(organization.id)
+      assert updated_org.payment_method_status == "valid"
+    end
+  end
+
+  describe "invoice.payment_failed" do
+    test "updates payment method status to requires_action", %{organization: organization} do
+      organization = organization_with_active_subscription(organization)
+
+      invoice =
+        stripe_invoice_object(%{
+          customer: organization.stripe_customer_id,
+          status: "open"
+        })
+
+      event = stripe_webhook_event_object("invoice.payment_failed", invoice)
+
+      assert :ok = WebhookProcessor.process(event)
+
+      updated_org = Organizations.get_organization!(organization.id)
+      assert updated_org.payment_method_status == "requires_action"
+    end
+
+    test "sends email notification to organization owners", %{organization: organization, user: user} do
+      organization = organization_with_active_subscription(organization)
+
+      invoice =
+        stripe_invoice_object(%{
+          customer: organization.stripe_customer_id,
+          status: "open"
+        })
+
+      event = stripe_webhook_event_object("invoice.payment_failed", invoice)
+
+      assert :ok = WebhookProcessor.process(event)
+
+      # Verify email was sent to organization owner
+      assert_email_sent(fn email ->
+        assert email.to == [{"", user.email}]
+        assert email.subject =~ "Payment failed"
+        assert email.subject =~ organization.name
+        assert email.text_body =~ "unable to process the payment"
+      end)
+    end
+  end
+
+  describe "customer.updated" do
+    test "updates payment method status when default payment method changed", %{
+      organization: organization
+    } do
+      customer =
+        stripe_customer_object(%{
+          id: organization.stripe_customer_id,
+          payment_method: "pm_new_123"
+        })
+
+      event = stripe_webhook_event_object("customer.updated", customer)
+
+      assert :ok = WebhookProcessor.process(event)
+
+      updated_org = Organizations.get_organization!(organization.id)
+      assert updated_org.payment_method_status == "valid"
+    end
+
+    test "succeeds when no payment method change", %{organization: organization} do
+      customer =
+        stripe_customer_object(%{
+          id: organization.stripe_customer_id,
+          payment_method: nil
+        })
+
+      event = stripe_webhook_event_object("customer.updated", customer)
+
+      assert :ok = WebhookProcessor.process(event)
+    end
+  end
+
+  describe "unhandled events" do
+    test "logs and succeeds for unknown event types", %{organization: organization} do
+      customer = stripe_customer_object(%{id: organization.stripe_customer_id})
+      event = stripe_webhook_event_object("customer.source.expiring", customer)
+
+      assert :ok = WebhookProcessor.process(event)
+    end
+  end
+
+  describe "organization not found" do
+    test "returns error when organization doesn't exist" do
+      subscription = stripe_subscription_object(%{customer: "cus_nonexistent"})
+      event = stripe_webhook_event_object("customer.subscription.created", subscription)
+
+      # Should still mark as processed but log error internally
+      assert :ok = WebhookProcessor.process(event)
+
+      # Event should be recorded with nil organization_id
+      webhook_event = Repo.get(StripeWebhookEvent, event["id"])
+      assert is_nil(webhook_event.organization_id)
+    end
+  end
+
+  describe "idempotency" do
+    test "processing same event multiple times only processes once", %{organization: organization} do
+      subscription =
+        stripe_subscription_object(%{
+          id: "sub_idempotent_test",
+          customer: organization.stripe_customer_id,
+          status: "active"
+        })
+
+      event = stripe_webhook_event_object("customer.subscription.created", subscription)
+
+      # Process multiple times
+      assert :ok = WebhookProcessor.process(event)
+      assert :ok = WebhookProcessor.process(event)
+      assert :ok = WebhookProcessor.process(event)
+
+      # Verify only one webhook event record
+      assert Repo.aggregate(StripeWebhookEvent, :count) == 1
+
+      # Verify organization was only updated once
+      updated_org = Organizations.get_organization!(organization.id)
+      assert updated_org.stripe_subscription_id == "sub_idempotent_test"
+    end
+  end
+
+  describe "event organization association" do
+    test "associates webhook event with correct organization", %{organization: organization} do
+      subscription = stripe_subscription_object(%{customer: organization.stripe_customer_id})
+      event = stripe_webhook_event_object("customer.subscription.created", subscription)
+
+      assert :ok = WebhookProcessor.process(event)
+
+      webhook_event = Repo.get(StripeWebhookEvent, event["id"])
+      assert webhook_event.organization_id == organization.id
+    end
+
+    test "creates webhook event without organization when customer not found" do
+      subscription = stripe_subscription_object(%{customer: "cus_unknown"})
+      event = stripe_webhook_event_object("customer.subscription.created", subscription)
+
+      assert :ok = WebhookProcessor.process(event)
+
+      webhook_event = Repo.get(StripeWebhookEvent, event["id"])
+      assert is_nil(webhook_event.organization_id)
+    end
+  end
+end

test/towerops/billing_test.exs

@@ -0,0 +1,329 @@
+defmodule Towerops.BillingTest do
+  use Towerops.DataCase
+
+  import Towerops.AccountsFixtures
+  import Towerops.BillingFixtures
+  import Towerops.DevicesFixtures
+  import Towerops.OrganizationsFixtures
+
+  alias Towerops.Billing
+  alias Towerops.Billing.StripeClient
+  alias Towerops.Organizations
+
+  setup do
+    user = user_fixture()
+    organization = organization_fixture(user.id)
+    %{user: user, organization: organization}
+  end
+
+  describe "create_checkout_session/2" do
+    test "creates Stripe customer and checkout session for new organization", %{
+      organization: organization
+    } do
+      Req.Test.stub(StripeClient, fn conn ->
+        cond do
+          # First call: create customer
+          conn.request_path == "/v1/customers" ->
+            conn
+            |> Plug.Conn.put_resp_content_type("application/json")
+            |> Plug.Conn.send_resp(200, Jason.encode!(stripe_customer_object(%{id: "cus_new_123"})))
+
+          # Second call: create checkout session
+          conn.request_path == "/v1/checkout/sessions" ->
+            conn
+            |> Plug.Conn.put_resp_content_type("application/json")
+            |> Plug.Conn.send_resp(
+              200,
+              Jason.encode!(stripe_checkout_session_object(%{url: "https://checkout.stripe.com/session"}))
+            )
+        end
+      end)
+
+      assert {:ok, url} =
+               Billing.create_checkout_session(organization,
+                 success_url: "https://example.com/success",
+                 cancel_url: "https://example.com/cancel"
+               )
+
+      assert url == "https://checkout.stripe.com/session"
+
+      # Verify customer ID was stored
+      updated_org = Organizations.get_organization!(organization.id)
+      assert updated_org.stripe_customer_id == "cus_new_123"
+    end
+
+    test "uses existing customer ID when already set", %{organization: organization} do
+      organization = organization_with_stripe_customer(organization, "cus_existing_123")
+
+      Req.Test.stub(StripeClient, fn conn ->
+        # Should only call checkout session, not customer creation
+        assert conn.request_path == "/v1/checkout/sessions"
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+        params = URI.decode_query(body)
+        assert params["customer"] == "cus_existing_123"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(
+          200,
+          Jason.encode!(stripe_checkout_session_object())
+        )
+      end)
+
+      assert {:ok, _url} =
+               Billing.create_checkout_session(organization,
+                 success_url: "https://example.com/success",
+                 cancel_url: "https://example.com/cancel"
+               )
+    end
+
+    test "returns error when customer creation fails", %{organization: organization} do
+      Req.Test.stub(StripeClient, fn conn ->
+        error = %{"error" => %{"message" => "Invalid email"}}
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(400, Jason.encode!(error))
+      end)
+
+      assert {:error, {:bad_request, "Invalid email"}} =
+               Billing.create_checkout_session(organization,
+                 success_url: "https://example.com/success",
+                 cancel_url: "https://example.com/cancel"
+               )
+    end
+
+    test "returns error when checkout session creation fails", %{organization: organization} do
+      Req.Test.stub(StripeClient, fn conn ->
+        cond do
+          conn.request_path == "/v1/customers" ->
+            conn
+            |> Plug.Conn.put_resp_content_type("application/json")
+            |> Plug.Conn.send_resp(200, Jason.encode!(stripe_customer_object()))
+
+          conn.request_path == "/v1/checkout/sessions" ->
+            error = %{"error" => %{"message" => "Price not found"}}
+
+            conn
+            |> Plug.Conn.put_resp_content_type("application/json")
+            |> Plug.Conn.send_resp(400, Jason.encode!(error))
+        end
+      end)
+
+      assert {:error, {:bad_request, "Price not found"}} =
+               Billing.create_checkout_session(organization,
+                 success_url: "https://example.com/success",
+                 cancel_url: "https://example.com/cancel"
+               )
+    end
+  end
+
+  describe "create_portal_session/2" do
+    test "creates portal session for organization with customer", %{organization: organization} do
+      organization = organization_with_stripe_customer(organization, "cus_test_123")
+
+      Req.Test.stub(StripeClient, fn conn ->
+        assert conn.request_path == "/v1/billing_portal/sessions"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(
+          200,
+          Jason.encode!(stripe_portal_session_object(%{url: "https://billing.stripe.com/portal"}))
+        )
+      end)
+
+      assert {:ok, url} = Billing.create_portal_session(organization, "https://example.com/return")
+      assert url == "https://billing.stripe.com/portal"
+    end
+
+    test "returns error when organization has no customer", %{organization: organization} do
+      assert {:error, :no_stripe_customer} =
+               Billing.create_portal_session(organization, "https://example.com/return")
+    end
+
+    test "returns error when portal session creation fails", %{organization: organization} do
+      organization = organization_with_stripe_customer(organization)
+
+      Req.Test.stub(StripeClient, fn conn ->
+        error = %{"error" => %{"message" => "Customer not found"}}
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(400, Jason.encode!(error))
+      end)
+
+      assert {:error, {:bad_request, "Customer not found"}} =
+               Billing.create_portal_session(organization, "https://example.com/return")
+    end
+  end
+
+  describe "billable_device_count/1" do
+    test "returns 0 for organization with no devices", %{organization: organization} do
+      assert Billing.billable_device_count(organization) == 0
+    end
+
+    test "returns 0 for organization with 10 or fewer devices", %{organization: organization} do
+      # Create 10 devices (at free tier limit)
+      for _ <- 1..10 do
+        device_fixture(%{organization_id: organization.id})
+      end
+
+      assert Billing.billable_device_count(organization) == 0
+    end
+
+    test "returns count over 10 for organization with more than 10 devices", %{
+      organization: organization
+    } do
+      # Create 25 devices
+      for _ <- 1..25 do
+        device_fixture(%{organization_id: organization.id})
+      end
+
+      # 25 - 10 free = 15 billable
+      assert Billing.billable_device_count(organization) == 15
+    end
+  end
+
+  describe "estimated_monthly_cost/1" do
+    test "returns correct cost calculation", %{organization: organization} do
+      # Create 25 devices (15 billable)
+      for _ <- 1..25 do
+        device_fixture(%{organization_id: organization.id})
+      end
+
+      assert {:ok, %{devices: 25, billable: 15, cost_usd: cost}} =
+               Billing.estimated_monthly_cost(organization)
+
+      assert Decimal.equal?(cost, Decimal.new(15))
+    end
+
+    test "returns zero cost for free tier usage", %{organization: organization} do
+      # Create 5 devices (under free tier)
+      for _ <- 1..5 do
+        device_fixture(%{organization_id: organization.id})
+      end
+
+      assert {:ok, %{devices: 5, billable: 0, cost_usd: cost}} =
+               Billing.estimated_monthly_cost(organization)
+
+      assert Decimal.equal?(cost, Decimal.new(0))
+    end
+  end
+
+  describe "sync_usage_to_stripe/1" do
+    test "reports usage for organization with active subscription", %{organization: organization} do
+      organization = organization_with_active_subscription(organization)
+
+      # Create 25 devices (15 billable)
+      for _ <- 1..25 do
+        device_fixture(%{organization_id: organization.id})
+      end
+
+      Req.Test.stub(StripeClient, fn conn ->
+        assert conn.request_path == "/v1/billing/meter_events"
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+        params = URI.decode_query(body)
+
+        # Verify reporting 15 billable devices
+        assert params["payload[value]"] == "15"
+        assert params["payload[stripe_customer_id]"] == organization.stripe_customer_id
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(%{"id" => "mtr_event_123"}))
+      end)
+
+      assert :ok = Billing.sync_usage_to_stripe(organization)
+
+      # Verify billing sync timestamp updated
+      updated_org = Organizations.get_organization!(organization.id)
+      assert updated_org.last_billing_sync_at
+      assert updated_org.last_synced_device_count == 15
+    end
+
+    test "skips sync for organization without active subscription", %{organization: organization} do
+      # Organization on free plan (no active subscription)
+      assert {:ok, :no_active_subscription} = Billing.sync_usage_to_stripe(organization)
+    end
+
+    test "returns error when usage reporting fails", %{organization: organization} do
+      organization = organization_with_active_subscription(organization)
+
+      Req.Test.stub(StripeClient, fn conn ->
+        error = %{"error" => %{"message" => "Rate limit exceeded"}}
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(429, Jason.encode!(error))
+      end)
+
+      assert {:error, {:rate_limited, _}} = Billing.sync_usage_to_stripe(organization)
+    end
+  end
+
+  describe "update_subscription_from_stripe/2" do
+    test "updates organization with subscription data", %{organization: organization} do
+      organization = organization_with_stripe_customer(organization)
+
+      subscription =
+        stripe_subscription_object(%{
+          id: "sub_new_123",
+          status: "active",
+          period_start: System.system_time(:second),
+          period_end: System.system_time(:second) + 30 * 86_400
+        })
+
+      assert {:ok, updated_org} = Billing.update_subscription_from_stripe(organization, subscription)
+
+      assert updated_org.stripe_subscription_id == "sub_new_123"
+      assert updated_org.subscription_status == "active"
+      assert updated_org.subscription_plan == "paid"
+      assert updated_org.subscription_current_period_start
+      assert updated_org.subscription_current_period_end
+    end
+
+    test "sets plan to paid for active status", %{organization: organization} do
+      subscription = stripe_subscription_object(%{status: "active"})
+      {:ok, org} = Billing.update_subscription_from_stripe(organization, subscription)
+      assert org.subscription_plan == "paid"
+    end
+
+    test "sets plan to paid for trialing status", %{organization: organization} do
+      subscription = stripe_subscription_object(%{status: "trialing"})
+      {:ok, org} = Billing.update_subscription_from_stripe(organization, subscription)
+      assert org.subscription_plan == "paid"
+    end
+
+    test "sets plan to paid for past_due status", %{organization: organization} do
+      subscription = stripe_subscription_object(%{status: "past_due"})
+      {:ok, org} = Billing.update_subscription_from_stripe(organization, subscription)
+      assert org.subscription_plan == "paid"
+    end
+
+    test "sets plan to free for canceled status", %{organization: organization} do
+      subscription = stripe_subscription_object(%{status: "canceled"})
+      {:ok, org} = Billing.update_subscription_from_stripe(organization, subscription)
+      assert org.subscription_plan == "free"
+    end
+  end
+
+  describe "update_payment_method_status/2" do
+    test "updates payment method status to valid", %{organization: organization} do
+      assert {:ok, updated_org} = Billing.update_payment_method_status(organization, "valid")
+      assert updated_org.payment_method_status == "valid"
+    end
+
+    test "updates payment method status to missing", %{organization: organization} do
+      assert {:ok, updated_org} = Billing.update_payment_method_status(organization, "missing")
+      assert updated_org.payment_method_status == "missing"
+    end
+
+    test "updates payment method status to requires_action", %{organization: organization} do
+      assert {:ok, updated_org} =
+               Billing.update_payment_method_status(organization, "requires_action")
+
+      assert updated_org.payment_method_status == "requires_action"
+    end
+  end
+end

test/towerops/organizations/organization_test.exs

@@ -318,4 +318,58 @@ defmodule Towerops.Organizations.OrganizationTest do
       assert changeset_valid.valid?
     end
   end
+
+  describe "has_active_subscription?/1" do
+    test "returns true for active subscription" do
+      org = %Organization{subscription_status: "active"}
+      assert Organization.has_active_subscription?(org)
+    end
+
+    test "returns true for trialing subscription" do
+      org = %Organization{subscription_status: "trialing"}
+      assert Organization.has_active_subscription?(org)
+    end
+
+    test "returns false for past_due subscription" do
+      org = %Organization{subscription_status: "past_due"}
+      refute Organization.has_active_subscription?(org)
+    end
+
+    test "returns false for canceled subscription" do
+      org = %Organization{subscription_status: "canceled"}
+      refute Organization.has_active_subscription?(org)
+    end
+
+    test "returns false for nil subscription_status" do
+      org = %Organization{subscription_status: nil}
+      refute Organization.has_active_subscription?(org)
+    end
+
+    test "returns false for incomplete subscription" do
+      org = %Organization{subscription_status: "incomplete"}
+      refute Organization.has_active_subscription?(org)
+    end
+  end
+
+  describe "has_valid_payment_method?/1" do
+    test "returns true when payment_method_status is valid" do
+      org = %Organization{payment_method_status: "valid"}
+      assert Organization.has_valid_payment_method?(org)
+    end
+
+    test "returns false when payment_method_status is missing" do
+      org = %Organization{payment_method_status: "missing"}
+      refute Organization.has_valid_payment_method?(org)
+    end
+
+    test "returns false when payment_method_status is requires_action" do
+      org = %Organization{payment_method_status: "requires_action"}
+      refute Organization.has_valid_payment_method?(org)
+    end
+
+    test "returns false when payment_method_status is nil" do
+      org = %Organization{payment_method_status: nil}
+      refute Organization.has_valid_payment_method?(org)
+    end
+  end
 end

test/towerops/organizations_test.exs

@@ -5,6 +5,7 @@ defmodule Towerops.OrganizationsTest do
 
   alias Towerops.Devices.Device
   alias Towerops.Organizations
+  alias Towerops.Organizations.Organization
 
   describe "organizations" do
     setup do
@@ -853,4 +854,123 @@ defmodule Towerops.OrganizationsTest do
       assert {:error, _changeset} = Organizations.accept_invitation(invitation, new_user.id)
     end
   end
+
+  describe "billing query helpers" do
+    setup do
+      user = user_fixture()
+      %{user: user}
+    end
+
+    test "get_organization_by_stripe_customer_id/1 returns organization", %{user: user} do
+      {:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
+
+      organization =
+        organization
+        |> Organization.changeset(%{stripe_customer_id: "cus_test_123"})
+        |> Repo.update!()
+
+      assert found = Organizations.get_organization_by_stripe_customer_id("cus_test_123")
+      assert found.id == organization.id
+    end
+
+    test "get_organization_by_stripe_customer_id/1 returns nil for nonexistent customer", %{
+      user: user
+    } do
+      {:ok, _organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
+
+      assert Organizations.get_organization_by_stripe_customer_id("cus_nonexistent") == nil
+    end
+
+    test "list_organizations_with_active_subscriptions/0 returns only active/trialing orgs", %{
+      user: user
+    } do
+      # Create organizations with different subscription statuses
+      {:ok, org_active} =
+        Organizations.create_organization(%{name: "Active Org"}, user.id, bypass_limits: true)
+
+      org_active =
+        org_active
+        |> Organization.changeset(%{
+          stripe_customer_id: "cus_active",
+          subscription_status: "active"
+        })
+        |> Repo.update!()
+
+      {:ok, org_trialing} =
+        Organizations.create_organization(%{name: "Trialing Org"}, user.id, bypass_limits: true)
+
+      org_trialing =
+        org_trialing
+        |> Organization.changeset(%{
+          stripe_customer_id: "cus_trialing",
+          subscription_status: "trialing"
+        })
+        |> Repo.update!()
+
+      {:ok, org_past_due} =
+        Organizations.create_organization(%{name: "Past Due Org"}, user.id, bypass_limits: true)
+
+      _org_past_due =
+        org_past_due
+        |> Organization.changeset(%{
+          stripe_customer_id: "cus_past_due",
+          subscription_status: "past_due"
+        })
+        |> Repo.update!()
+
+      {:ok, org_canceled} =
+        Organizations.create_organization(%{name: "Canceled Org"}, user.id, bypass_limits: true)
+
+      _org_canceled =
+        org_canceled
+        |> Organization.changeset(%{
+          stripe_customer_id: "cus_canceled",
+          subscription_status: "canceled"
+        })
+        |> Repo.update!()
+
+      {:ok, _org_free} =
+        Organizations.create_organization(%{name: "Free Org"}, user.id, bypass_limits: true)
+
+      active_orgs = Organizations.list_organizations_with_active_subscriptions()
+
+      assert length(active_orgs) == 2
+      org_ids = Enum.map(active_orgs, & &1.id)
+      assert org_active.id in org_ids
+      assert org_trialing.id in org_ids
+    end
+
+    test "list_organizations_with_active_subscriptions/0 returns empty when none active", %{
+      user: user
+    } do
+      {:ok, _org} = Organizations.create_organization(%{name: "Free Org"}, user.id)
+
+      assert Organizations.list_organizations_with_active_subscriptions() == []
+    end
+
+    test "get_primary_owner/1 returns the owner", %{user: user} do
+      {:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
+
+      primary_owner = Organizations.get_primary_owner(organization)
+      assert primary_owner.id == user.id
+    end
+
+    test "get_primary_owner/1 returns first owner when multiple exist", %{user: user} do
+      {:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
+
+      # Add another owner
+      second_owner = user_fixture()
+
+      {:ok, _membership} =
+        Organizations.create_membership(%{
+          organization_id: organization.id,
+          user_id: second_owner.id,
+          role: :owner
+        })
+
+      primary_owner = Organizations.get_primary_owner(organization)
+      # Should return the first owner (by insertion date)
+      assert primary_owner.id == user.id
+    end
+  end
 end

test/towerops/workers/billing_sync_worker_test.exs

@@ -0,0 +1,213 @@
+defmodule Towerops.Workers.BillingSyncWorkerTest do
+  use Towerops.DataCase
+  use Oban.Testing, repo: Towerops.Repo
+
+  import Towerops.AccountsFixtures
+  import Towerops.BillingFixtures
+  import Towerops.DevicesFixtures
+  import Towerops.OrganizationsFixtures
+
+  alias Towerops.Billing.StripeClient
+  alias Towerops.Organizations
+  alias Towerops.Workers.BillingSyncWorker
+
+  setup do
+    user = user_fixture()
+    %{user: user}
+  end
+
+  describe "perform/1" do
+    test "syncs usage for all organizations with active subscriptions", %{user: user} do
+      # Create 3 organizations - 2 with active subscriptions, 1 free
+      org1 = user.id |> organization_fixture() |> organization_with_active_subscription()
+      org2 = user.id |> organization_fixture(%{name: "Org 2"}) |> organization_with_active_subscription()
+      # Free plan
+      org3 = organization_fixture(user.id, %{name: "Org 3"})
+
+      # Add devices to each
+      # 5 billable
+      for _ <- 1..15, do: device_fixture(%{organization_id: org1.id})
+      # 15 billable
+      for _ <- 1..25, do: device_fixture(%{organization_id: org2.id})
+      # 0 billable (free plan)
+      for _ <- 1..5, do: device_fixture(%{organization_id: org3.id})
+
+      # Mock Stripe API calls
+      Req.Test.stub(StripeClient, fn conn ->
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+        params = URI.decode_query(body)
+
+        # Verify correct device counts reported
+        case params["payload[stripe_customer_id]"] do
+          customer_id when customer_id == org1.stripe_customer_id ->
+            assert params["payload[value]"] == "5"
+
+          customer_id when customer_id == org2.stripe_customer_id ->
+            assert params["payload[value]"] == "15"
+        end
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(%{"id" => "mtr_event_123"}))
+      end)
+
+      assert :ok = perform_job(BillingSyncWorker, %{})
+
+      # Verify sync timestamps updated for paid orgs
+      updated_org1 = Organizations.get_organization!(org1.id)
+      assert updated_org1.last_billing_sync_at
+      assert updated_org1.last_synced_device_count == 5
+
+      updated_org2 = Organizations.get_organization!(org2.id)
+      assert updated_org2.last_billing_sync_at
+      assert updated_org2.last_synced_device_count == 15
+
+      # Free org should not be synced
+      updated_org3 = Organizations.get_organization!(org3.id)
+      refute updated_org3.last_billing_sync_at
+    end
+
+    test "continues processing after individual failures", %{user: user} do
+      org1 = user.id |> organization_fixture() |> organization_with_active_subscription()
+      org2 = user.id |> organization_fixture(%{name: "Org 2"}) |> organization_with_active_subscription()
+
+      for _ <- 1..15, do: device_fixture(%{organization_id: org1.id})
+      for _ <- 1..20, do: device_fixture(%{organization_id: org2.id})
+
+      call_count = :counters.new(1, [:atomics])
+
+      Req.Test.stub(StripeClient, fn conn ->
+        count = :counters.get(call_count, 1)
+        :counters.add(call_count, 1, 1)
+
+        # First call fails, second succeeds
+        if count == 0 do
+          error = %{"error" => %{"message" => "Rate limit"}}
+
+          conn
+          |> Plug.Conn.put_resp_content_type("application/json")
+          |> Plug.Conn.send_resp(429, Jason.encode!(error))
+        else
+          conn
+          |> Plug.Conn.put_resp_content_type("application/json")
+          |> Plug.Conn.send_resp(200, Jason.encode!(%{"id" => "mtr_event_123"}))
+        end
+      end)
+
+      # Should succeed overall despite individual failure
+      assert :ok = perform_job(BillingSyncWorker, %{})
+
+      # At least one organization should have synced (whichever was processed second)
+      updated_org1 = Organizations.get_organization!(org1.id)
+      updated_org2 = Organizations.get_organization!(org2.id)
+
+      # Either org1 or org2 should have synced (one fails, one succeeds)
+      assert updated_org1.last_billing_sync_at || updated_org2.last_billing_sync_at
+    end
+
+    test "succeeds when no organizations have active subscriptions", %{user: user} do
+      # Create organizations on free plan
+      _org1 = organization_fixture(user.id)
+      _org2 = organization_fixture(user.id, %{name: "Org 2"})
+
+      # Should succeed without making any Stripe API calls
+      assert :ok = perform_job(BillingSyncWorker, %{})
+    end
+
+    test "handles organizations with zero billable devices", %{user: user} do
+      org = user.id |> organization_fixture() |> organization_with_active_subscription()
+
+      # Add exactly 10 devices (at free tier limit)
+      for _ <- 1..10, do: device_fixture(%{organization_id: org.id})
+
+      Req.Test.stub(StripeClient, fn conn ->
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+        params = URI.decode_query(body)
+
+        # Should report 0 billable devices
+        assert params["payload[value]"] == "0"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(%{"id" => "mtr_event_123"}))
+      end)
+
+      assert :ok = perform_job(BillingSyncWorker, %{})
+
+      updated_org = Organizations.get_organization!(org.id)
+      assert updated_org.last_synced_device_count == 0
+    end
+
+    test "handles organizations with many devices", %{user: user} do
+      org = user.id |> organization_fixture() |> organization_with_active_subscription()
+
+      # Add 150 devices (140 billable)
+      for _ <- 1..150, do: device_fixture(%{organization_id: org.id})
+
+      Req.Test.stub(StripeClient, fn conn ->
+        {:ok, body, _} = Plug.Conn.read_body(conn)
+        params = URI.decode_query(body)
+
+        # Should report 140 billable devices
+        assert params["payload[value]"] == "140"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(%{"id" => "mtr_event_123"}))
+      end)
+
+      assert :ok = perform_job(BillingSyncWorker, %{})
+
+      updated_org = Organizations.get_organization!(org.id)
+      assert updated_org.last_synced_device_count == 140
+    end
+
+    test "only syncs organizations with active or trialing status", %{user: user} do
+      org_active = user.id |> organization_fixture() |> organization_with_active_subscription()
+
+      org_trialing =
+        user.id
+        |> organization_fixture(%{name: "Org Trialing"})
+        |> organization_with_active_subscription()
+        |> Ecto.Changeset.change(subscription_status: "trialing")
+        |> Repo.update!()
+
+      org_past_due =
+        user.id
+        |> organization_fixture(%{name: "Org Past Due"})
+        |> organization_with_past_due_subscription()
+
+      org_canceled =
+        user.id
+        |> organization_fixture(%{name: "Org Canceled"})
+        |> organization_with_active_subscription()
+        |> Ecto.Changeset.change(subscription_status: "canceled")
+        |> Repo.update!()
+
+      for org <- [org_active, org_trialing, org_past_due, org_canceled] do
+        for _ <- 1..15, do: device_fixture(%{organization_id: org.id})
+      end
+
+      call_count = :counters.new(1, [:atomics])
+
+      Req.Test.stub(StripeClient, fn conn ->
+        :counters.add(call_count, 1, 1)
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.send_resp(200, Jason.encode!(%{"id" => "mtr_event_123"}))
+      end)
+
+      assert :ok = perform_job(BillingSyncWorker, %{})
+
+      # Only active and trialing should be synced (2 calls)
+      assert :counters.get(call_count, 1) == 2
+
+      # Verify only active/trialing updated
+      assert Organizations.get_organization!(org_active.id).last_billing_sync_at
+      assert Organizations.get_organization!(org_trialing.id).last_billing_sync_at
+      refute Organizations.get_organization!(org_past_due.id).last_billing_sync_at
+      refute Organizations.get_organization!(org_canceled.id).last_billing_sync_at
+    end
+  end
+end

test/towerops/workers/welcome_email_worker_test.exs

@@ -0,0 +1,43 @@
+defmodule Towerops.Workers.WelcomeEmailWorkerTest do
+  use Towerops.DataCase, async: true
+
+  import Swoosh.TestAssertions
+  import Towerops.AccountsFixtures
+
+  alias Towerops.Workers.WelcomeEmailWorker
+
+  describe "perform/1" do
+    test "sends welcome email to the user" do
+      user = user_fixture()
+
+      assert :ok = WelcomeEmailWorker.perform(%Oban.Job{args: %{"user_id" => user.id}})
+
+      assert_email_sent(fn email ->
+        assert email.to == [{"", user.email}]
+        assert email.from == {"Graham McIntire", "graham@towerops.net"}
+        assert email.subject =~ "Welcome"
+        assert email.text_body =~ "Hi"
+      end)
+    end
+
+    test "returns :ok when user no longer exists" do
+      assert :ok =
+               WelcomeEmailWorker.perform(%Oban.Job{
+                 args: %{"user_id" => Ecto.UUID.generate()}
+               })
+
+      assert_no_email_sent()
+    end
+  end
+
+  describe "enqueue/1" do
+    test "schedules job with delay" do
+      user = user_fixture()
+
+      assert {:ok, %Oban.Job{} = job} = WelcomeEmailWorker.enqueue(user.id)
+      assert job.args["user_id"] || job.args[:user_id] == user.id
+      assert job.queue == "notifications"
+      assert DateTime.diff(job.scheduled_at, DateTime.utc_now(), :second) > 0
+    end
+  end
+end

test/towerops_web/controllers/api/v1/stripe_webhook_controller_test.exs

@@ -0,0 +1,258 @@
+defmodule ToweropsWeb.Api.V1.StripeWebhookControllerTest do
+  use ToweropsWeb.ConnCase
+
+  import Towerops.AccountsFixtures
+  import Towerops.BillingFixtures
+  import Towerops.OrganizationsFixtures
+
+  alias Towerops.Billing.StripeWebhookEvent
+  alias Towerops.Repo
+
+  setup do
+    user = user_fixture()
+    organization = organization_fixture(user.id)
+    organization = organization_with_active_subscription(organization)
+    %{user: user, organization: organization}
+  end
+
+  describe "POST /api/v1/webhooks/stripe" do
+    test "processes valid webhook with correct signature", %{conn: conn, organization: organization} do
+      subscription =
+        stripe_subscription_object(%{
+          id: "sub_webhook_123",
+          customer: organization.stripe_customer_id,
+          status: "active"
+        })
+
+      event = stripe_webhook_event_object("customer.subscription.updated", subscription)
+      payload = Jason.encode!(event)
+      signature = generate_signature(payload)
+
+      conn =
+        conn
+        |> put_req_header("stripe-signature", signature)
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/stripe", payload)
+
+      assert json_response(conn, 200) == %{"received" => true}
+
+      # Verify event was processed
+      assert Repo.get(StripeWebhookEvent, event["id"])
+    end
+
+    test "returns 401 for invalid signature", %{conn: conn} do
+      event = stripe_webhook_event_object("customer.subscription.updated", %{})
+      payload = Jason.encode!(event)
+
+      # Use current timestamp with invalid signature
+      timestamp = System.system_time(:second)
+      invalid_signature = "t=#{timestamp},v1=invalid_signature_hash"
+
+      conn =
+        conn
+        |> put_req_header("stripe-signature", invalid_signature)
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/stripe", payload)
+
+      assert json_response(conn, 401) == %{"error" => "Invalid signature"}
+
+      # Verify event was not processed
+      refute Repo.get(StripeWebhookEvent, event["id"])
+    end
+
+    test "returns 400 for malformed signature header", %{conn: conn} do
+      event = stripe_webhook_event_object("customer.subscription.updated", %{})
+      payload = Jason.encode!(event)
+
+      conn =
+        conn
+        |> put_req_header("stripe-signature", "malformed")
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/stripe", payload)
+
+      assert json_response(conn, 400) == %{"error" => "Verification failed"}
+    end
+
+    test "returns 400 for timestamp too old", %{conn: conn} do
+      event = stripe_webhook_event_object("customer.subscription.updated", %{})
+      payload = Jason.encode!(event)
+
+      # Generate signature with old timestamp (> 5 minutes ago)
+      old_timestamp = System.system_time(:second) - 400
+      signature = generate_signature(payload, old_timestamp)
+
+      conn =
+        conn
+        |> put_req_header("stripe-signature", signature)
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/stripe", payload)
+
+      assert json_response(conn, 400) == %{"error" => "Verification failed"}
+    end
+
+    test "returns 200 even when event processing fails", %{conn: conn} do
+      # Event with invalid data that will fail processing
+      event = %{
+        "id" => "evt_fail_test",
+        "type" => "customer.subscription.updated",
+        "data" => %{
+          "object" => %{
+            "customer" => "cus_nonexistent"
+          }
+        }
+      }
+
+      payload = Jason.encode!(event)
+      signature = generate_signature(payload)
+
+      conn =
+        conn
+        |> put_req_header("stripe-signature", signature)
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/stripe", payload)
+
+      # Should still return 200 to Stripe (idempotency)
+      assert json_response(conn, 200) == %{"received" => true}
+    end
+
+    test "handles duplicate webhook deliveries correctly", %{conn: conn, organization: organization} do
+      subscription =
+        stripe_subscription_object(%{
+          customer: organization.stripe_customer_id,
+          status: "active"
+        })
+
+      event = stripe_webhook_event_object("customer.subscription.updated", subscription)
+      payload = Jason.encode!(event)
+      signature = generate_signature(payload)
+
+      # Send same event twice
+      conn1 =
+        conn
+        |> put_req_header("stripe-signature", signature)
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/stripe", payload)
+
+      conn2 =
+        build_conn()
+        |> put_req_header("stripe-signature", signature)
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/stripe", payload)
+
+      assert json_response(conn1, 200) == %{"received" => true}
+      assert json_response(conn2, 200) == %{"received" => true}
+
+      # Verify only one event record
+      assert Repo.aggregate(StripeWebhookEvent, :count) == 1
+    end
+
+    test "processes different event types", %{conn: _conn, organization: organization} do
+      event_types = [
+        "customer.subscription.created",
+        "customer.subscription.updated",
+        "customer.subscription.deleted",
+        "invoice.payment_succeeded",
+        "invoice.payment_failed",
+        "customer.updated"
+      ]
+
+      for event_type <- event_types do
+        data =
+          case event_type do
+            "invoice." <> _ ->
+              stripe_invoice_object(%{customer: organization.stripe_customer_id})
+
+            _ ->
+              stripe_subscription_object(%{customer: organization.stripe_customer_id})
+          end
+
+        event = stripe_webhook_event_object(event_type, data)
+        payload = Jason.encode!(event)
+        signature = generate_signature(payload)
+
+        conn_result =
+          build_conn()
+          |> put_req_header("stripe-signature", signature)
+          |> put_req_header("content-type", "application/json")
+          |> post(~p"/api/v1/webhooks/stripe", payload)
+
+        assert json_response(conn_result, 200) == %{"received" => true}
+        assert Repo.get(StripeWebhookEvent, event["id"])
+      end
+    end
+  end
+
+  describe "signature verification" do
+    test "accepts signature with correct timestamp and hash", %{conn: conn, organization: organization} do
+      subscription = stripe_subscription_object(%{customer: organization.stripe_customer_id})
+      event = stripe_webhook_event_object("customer.subscription.updated", subscription)
+      payload = Jason.encode!(event)
+
+      timestamp = System.system_time(:second)
+      signature = generate_signature(payload, timestamp)
+
+      conn =
+        conn
+        |> put_req_header("stripe-signature", signature)
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/stripe", payload)
+
+      assert json_response(conn, 200) == %{"received" => true}
+    end
+
+    test "rejects signature with timestamp in future", %{conn: conn} do
+      event = stripe_webhook_event_object("customer.subscription.updated", %{})
+      payload = Jason.encode!(event)
+
+      # Timestamp 10 minutes in future
+      future_timestamp = System.system_time(:second) + 600
+      signature = generate_signature(payload, future_timestamp)
+
+      conn =
+        conn
+        |> put_req_header("stripe-signature", signature)
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/stripe", payload)
+
+      assert json_response(conn, 400) == %{"error" => "Verification failed"}
+    end
+
+    test "rejects signature with wrong webhook secret", %{conn: conn} do
+      event = stripe_webhook_event_object("customer.subscription.updated", %{})
+      payload = Jason.encode!(event)
+
+      timestamp = System.system_time(:second)
+      signed_payload = "#{timestamp}.#{payload}"
+
+      # Sign with wrong secret
+      wrong_signature =
+        :hmac
+        |> :crypto.mac(:sha256, "wrong_secret", signed_payload)
+        |> Base.encode16(case: :lower)
+
+      signature = "t=#{timestamp},v1=#{wrong_signature}"
+
+      conn =
+        conn
+        |> put_req_header("stripe-signature", signature)
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/stripe", payload)
+
+      assert json_response(conn, 401) == %{"error" => "Invalid signature"}
+    end
+  end
+
+  # Helper to generate valid Stripe signature
+  defp generate_signature(payload, timestamp \\ nil) do
+    timestamp = timestamp || System.system_time(:second)
+    webhook_secret = Application.fetch_env!(:towerops, :stripe_webhook_secret)
+    signed_payload = "#{timestamp}.#{payload}"
+
+    signature =
+      :hmac
+      |> :crypto.mac(:sha256, webhook_secret, signed_payload)
+      |> Base.encode16(case: :lower)
+
+    "t=#{timestamp},v1=#{signature}"
+  end
+end