{:oban, "~> 2.18"},

      {:stripity_stripe, "~> 3.2"},

# Background jobs. A cron entry polls Stripe's event stream every minute to

# keep subscriptions, and the licenses they grant, in sync.

config :my_app, Oban,

  engine: Oban.Engines.Basic,

  repo: MyApp.Repo,

  queues: [stripe: 1],

  plugins: [

    {Oban.Plugins.Cron, crontab: [{"* * * * *", MyApp.Billing.Stripe.EventsPoller}]}

  ]

# Stripe. Tests use fixture prices and a fake client from config/test.exs.

#

#   * STRIPE_SECRET_KEY: leave unset to stub Stripe calls, so the subscribe

#     flow can be clicked through locally.

#   * STRIPE_PRICES_<PLAN>: the plan's Stripe Price ids, comma-separated,

#     newest first. The first is sold at checkout; older ones still map

#     existing subscribers to the plan. A plan without prices isn't for sale.

if config_env() != :test do

  if api_key = System.get_env("STRIPE_SECRET_KEY") do

    config :stripity_stripe, api_key: api_key

  end

  config :my_app, MyApp.Billing.Stripe,

    prices:

      Map.new([:pro, :business], fn plan ->

        env = "STRIPE_PRICES_#{plan |> Atom.to_string() |> String.upcase()}"

        {plan, env |> System.get_env("") |> String.split(",", trim: true)}

      end)

end

# Run Oban jobs only when tests ask, and fake Stripe responses.

config :my_app, Oban, testing: :manual

config :my_app, MyApp.Billing.Stripe,

  client: MyApp.Billing.Stripe.FakeClient,

  prices: %{pro: ["price_pro", "price_pro_2025"], business: ["price_business"]}

      {Oban, Application.fetch_env!(:my_app, Oban)},

  # Payment providers add their source here.

  @sources [:manual, :stripe]

defmodule MyApp.Billing.Stripe do

  @moduledoc """

  Stripe subscriptions as one source of licenses.

  Everything Stripe lives under this namespace, in its own tables. It touches

  the rest of billing in one place: `sync_subscription/2` mirrors a Stripe

  subscription and calls `MyApp.Billing.sync_license/4` with

  `source: :stripe` and the subscription id as `source_ref`. The app keeps

  reading `MyApp.Billing.entitlements/1`, so Stripe licenses sit alongside

  manual ones (enterprise deals, AppSumo codes) and other providers', and

  best wins.

  Subscription state reaches us two ways, both landing in `sync_subscription/2`:

    1. **On return from Checkout** (`sync_checkout_session/2`), so the page

       reflects the purchase right away.

    2. **From the event poller** (`MyApp.Billing.Stripe.EventsPoller`), which

       catches renewals, plan changes, failed payments, and cancellations

       without a public webhook endpoint.

  ## Translating a subscription into a license

    * active, trialing, and past due grant the plan. Past due keeps access

      while Stripe retries the card.

    * Set to cancel: the license ends when access does (`cancel_at`).

    * Canceled, unpaid, paused, and expired cancel the license.

    * Incomplete (first payment not made) grants nothing yet.

  """

  require Logger

  alias Ecto.Multi

  alias MyApp.Repo

  alias MyApp.Accounts.Scope

  alias MyApp.Accounts.User

  alias MyApp.Billing

  alias MyApp.Billing.Plans

  alias MyApp.Billing.Stripe.{Checkpoint, Customer, Subscription}

  @doc false

  def client, do: Keyword.get(config(), :client, MyApp.Billing.Stripe.Client)

  ## Prices

  @doc """

  The Stripe Price sold at checkout for a plan, or nil if the plan isn't for

  sale on Stripe. Locally, with no API key, every paid public plan gets a stub

  price so the flow can be clicked through.

  """

  def price_for_plan(plan_key) do

    case Map.get(prices(), plan_key) do

      [price | _older] -> price

      _ -> if stub_mode?(), do: stub_price(plan_key)

    end

  end

  @doc """

  The plan a Stripe Price grants. Older prices still map to their plan, so

  changing a price never strands existing subscribers.

  """

  def plan_for_price(nil), do: nil

  def plan_for_price(price_id) do

    Enum.find_value(prices(), fn {plan_key, price_ids} -> price_id in price_ids && plan_key end) ||

      if(stub_mode?(), do: plan_for_stub_price(price_id))

  end

  @doc """

  Public plans that can be bought through Stripe, lowest rank first.

  """

  def purchasable_plans do

    Enum.filter(Plans.public(), &(&1.key != :free and price_for_plan(&1.key)))

  end

  defp prices, do: Keyword.get(config(), :prices, %{})

  defp config, do: Application.get_env(:my_app, __MODULE__, [])

  defp stub_mode?, do: client() == MyApp.Billing.Stripe.Client and not client().configured?()

  defp stub_price(plan_key), do: "price_stub_#{plan_key}"

  defp plan_for_stub_price("price_stub_" <> key) do

    Enum.find_value(Plans.public(), &(Atom.to_string(&1.key) == key && &1.key))

  end

  defp plan_for_stub_price(_price_id), do: nil

  ## Subscriptions

  @doc """

  The current user's subscription that Stripe still bills, or nil.

  """

  def get_live_subscription(%Scope{user: user}), do: get_live_subscription_for_user(user)

  def get_live_subscription_for_user(%User{} = user) do

    Repo.one(Subscription.live_for_user_query(user))

  end

  ## Checkout

  @doc """

  Starts a Stripe Checkout for a plan. Returns `{:ok, %{id, url}}`: send the

  user to `url`.

  Returns `{:error, :already_subscribed}` when the user already has a live

  Stripe subscription. Plan changes go through the Customer Portal instead, so

  a user never ends up paying for two subscriptions.

  """

  def create_checkout_session(%Scope{user: user} = scope, plan_key, success_url, cancel_url) do

    with %{} = plan <- find_purchasable_plan(plan_key),

         nil <- get_live_subscription(scope),

         {:ok, customer_id} <- ensure_customer(user) do

      client().create_checkout_session(%{

        user_id: user.id,

        customer: customer_id,

        price: price_for_plan(plan.key),

        success_url: success_url,

        cancel_url: cancel_url

      })

    else

      nil -> {:error, :not_purchasable}

      %Subscription{} -> {:error, :already_subscribed}

      {:error, _reason} = error -> error

    end

  end

  # Takes an atom or a string from a form, without creating atoms.

  defp find_purchasable_plan(plan_key) do

    Enum.find(purchasable_plans(), &(Atom.to_string(&1.key) == to_string(plan_key)))

  end

  @doc """

  Syncs the subscription a finished Checkout created. Returns

  `{:ok, %Subscription{}}`, `{:ok, :pending}` before Stripe has created it,

  or `{:error, reason}`, including `:not_yours` for another user's session.

  """

  def sync_checkout_session(%Scope{user: user}, session_id) do

    with {:ok, session} <- client().fetch_checkout_session(session_id),

         :ok <- verify_owner(session, user),

         sub_id when is_binary(sub_id) <- session.subscription || {:ok, :pending},

         {:ok, sub} <- client().fetch_subscription(sub_id) do

      sync_subscription(user, sub)

    end

  end

  defp verify_owner(%{client_reference_id: user_id}, %User{id: user_id}), do: :ok

  defp verify_owner(_session, _user), do: {:error, :not_yours}

  ## Customer Portal

  @doc """

  Starts a Customer Portal session for managing the subscription: changing

  plan, updating the card, canceling. Returns `{:ok, url}`, or

  `{:error, :no_customer}` before the user has checked out.

  """

  def create_portal_session(%Scope{user: user}, return_url) do

    case Repo.get_by(Customer, user_id: user.id) do

      nil -> {:error, :no_customer}

      customer -> client().create_portal_session(customer.stripe_customer_id, return_url)

    end

  end

  defp ensure_customer(user) do

    case Repo.get_by(Customer, user_id: user.id) do

      %Customer{stripe_customer_id: id} -> {:ok, id}

      nil -> create_customer(user)

    end

  end

  defp create_customer(user) do

    with {:ok, customer_id} <- client().create_customer(user) do

      %Customer{}

      |> Customer.changeset(%{user_id: user.id, stripe_customer_id: customer_id})

      |> Repo.insert(on_conflict: :nothing, conflict_target: :user_id)

      # On a double click, the first request's customer wins.

      {:ok, Repo.get_by!(Customer, user_id: user.id).stripe_customer_id}

    end

  end

  ## Sync

  @doc """

  Mirrors a normalized Stripe subscription (see `MyApp.Billing.Stripe.Client`)

  and syncs the license it grants, in one transaction. Safe to repeat, and

  safe when the checkout return and the poller race.

  Returns `{:ok, %Subscription{}}`.

  """

  def sync_subscription(%User{} = user, sub) do

    Multi.new()

    |> Multi.insert(

      :subscription,

      Subscription.changeset(%Subscription{}, %{

        user_id: user.id,

        stripe_subscription_id: sub.id,

        stripe_customer_id: sub.customer,

        stripe_price_id: sub.price_id,

        status: sub.status,

        current_period_end: sub.current_period_end,

        cancel_at: sub.cancel_at,

        canceled_at: sub.canceled_at

      }),

      on_conflict: {:replace_all_except, [:id, :user_id, :inserted_at]},

      conflict_target: :stripe_subscription_id,

      returning: true

    )

    |> Multi.run(:license, fn _repo, _changes -> sync_license(user, sub) end)

    |> Repo.transaction()

    |> case do

      {:ok, %{subscription: subscription}} -> {:ok, subscription}

      {:error, _step, reason, _changes} -> {:error, reason}

    end

  end

  defp sync_license(user, sub) do

    case {plan_for_price(sub.price_id), license_status(sub.status)} do

      {_plan, :none} ->

        {:ok, nil}

      {nil, _status} ->

        Logger.warning(

          "[Billing.Stripe] #{sub.id} has price #{inspect(sub.price_id)}, which maps to no plan. " <>

            "Add it to config :my_app, MyApp.Billing.Stripe, prices: ..."

        )

        {:ok, nil}

      {plan, status} ->

        Billing.sync_license(user, :stripe, sub.id, %{

          plan: plan,

          status: status,

          starts_at: sub.start_date || DateTime.utc_now(:second),

          ends_at: if(status == :active, do: sub.cancel_at),

          canceled_at: if(status == :canceled, do: sub.canceled_at || DateTime.utc_now(:second))

        })

    end

  end

  defp license_status(status) do

    cond do

      Subscription.live_status?(status) -> :active

      status == "incomplete" -> :none

      true -> :canceled

    end

  end

  ## Events

  @doc """

  Handles one event from Stripe's event stream. Acts on

  `customer.subscription.*` and ignores the rest. Always returns `:ok`, so a

  bad event never stalls the poller.

  The subscription is re-fetched rather than read from the event, because an

  event is a snapshot that may already be stale.

  """

  def handle_event(%{"type" => "customer.subscription." <> _} = event) do

    object = get_in(event, ["data", "object"]) || %{}

    with sub_id when is_binary(sub_id) <- object["id"],

         {:ok, sub} <- client().fetch_subscription(sub_id),

         %User{} = user <- find_user(sub) do

      sync_subscription(user, sub)

    else

      nil -> Logger.warning("[Billing.Stripe] no user for subscription #{object["id"]}")

      _error -> :ok

    end

    :ok

  end

  def handle_event(_event), do: :ok

  # Known subscriptions keep their user. New ones match the Stripe Customer

  # we created, or the user id set at checkout.

  defp find_user(sub) do

    with nil <- user_by_subscription(sub.id),

         nil <- user_by_customer(sub.customer) do

      sub.user_id && Repo.get(User, sub.user_id)

    end

  end

  defp user_by_subscription(sub_id) do

    case Repo.get_by(Subscription, stripe_subscription_id: sub_id) do

      %Subscription{user_id: user_id} -> Repo.get(User, user_id)

      nil -> nil

    end

  end

  defp user_by_customer(nil), do: nil

  defp user_by_customer(customer_id) do

    case Repo.get_by(Customer, stripe_customer_id: customer_id) do

      %Customer{user_id: user_id} -> Repo.get(User, user_id)

      nil -> nil

    end

  end

  ## Checkpoints

  def get_checkpoint(source), do: Repo.get_by(Checkpoint, source: source)

  @doc """

  Records the newest event seen for a stream and when it was polled.

  """

  def save_checkpoint(source, last_event_id) do

    (get_checkpoint(source) || %Checkpoint{source: source})

    |> Checkpoint.changeset(%{

      last_event_id: last_event_id,

      last_polled_at: DateTime.utc_now(:second)

    })

    |> Repo.insert_or_update()

  end

end

defmodule MyApp.Billing.Stripe.Checkpoint do

  @moduledoc """

  Where event polling left off, so each poll asks Stripe only for newer

  events. One row per stream: `"platform"` for the account's own events. The

  table is keyed by source so a Stripe Connect feature can add

  `"connect:acct_..."` streams.

  See `MyApp.Billing.Stripe.EventsPoller`.

  """

  use Ecto.Schema

  import Ecto.Changeset

  @primary_key {:id, :binary_id, autogenerate: true}

  schema "stripe_event_checkpoints" do

    field :source, :string

    field :last_event_id, :string

    field :last_polled_at, :utc_datetime

    timestamps(type: :utc_datetime)

  end

  ## Changesets

  def changeset(checkpoint, attrs) do

    checkpoint

    |> cast(attrs, [:source, :last_event_id, :last_polled_at])

    |> validate_required([:source])

    |> unique_constraint(:source)

  end

end

defmodule MyApp.Billing.Stripe.Client do

  @moduledoc """

  The Stripe API, reduced to what billing needs.

    * **Normalized returns.** Every function returns plain maps with a fixed

      shape, never `Stripe.*` structs, so Stripe API changes stay in this

      module.

    * **Stub mode.** With no `config :stripity_stripe, api_key: ...`, calls

      return believable stubs instead of hitting the network. The stub threads

      the user and price through checkout → session → subscription, so the

      whole subscribe flow can be clicked through locally.

  A normalized subscription looks like:

      %{

        id: "sub_123",

        customer: "cus_123",

        price_id: "price_123",

        status: "active",

        start_date: ~U[2026-01-01 00:00:00Z],

        current_period_end: ~U[2026-02-01 00:00:00Z],

        cancel_at: nil,

        canceled_at: nil,

        user_id: "..."

      }

  `cancel_at` is when access ends, if the subscription is set to cancel.

  `user_id` comes from the subscription's metadata, set at checkout.

  """

  require Logger

  @stub_session_prefix "cs_stub_"

  @stub_subscription_prefix "sub_stub_"

  def configured?, do: is_binary(Application.get_env(:stripity_stripe, :api_key))

  ## Customers

  @doc "Creates a Stripe Customer for a user. Returns `{:ok, customer_id}`."

  def create_customer(user) do

    if configured?() do

      case Stripe.Customer.create(%{email: user.email, metadata: %{user_id: user.id}}) do

        {:ok, %{id: id}} -> {:ok, id}

        {:error, err} -> log_and_wrap("create_customer", err)

      end

    else

      stub_log("create_customer", user: user.id)

      {:ok, "cus_stub_#{user.id}"}

    end

  end

  ## Checkout

  @doc """

  Creates a subscription Checkout Session. Returns `{:ok, %{id, url}}`.

  Takes `:user_id`, `:customer`, `:price`, `:success_url`, and `:cancel_url`.

  The user id goes on the session and on the subscription, so either can be

  traced back to the user.

  """

  def create_checkout_session(params) do

    if configured?() do

      %{

        mode: "subscription",

        customer: params.customer,

        line_items: [%{price: params.price, quantity: 1}],

        allow_promotion_codes: true,

        success_url: params.success_url <> "?session_id={CHECKOUT_SESSION_ID}",

        cancel_url: params.cancel_url,

        client_reference_id: params.user_id,

        subscription_data: %{metadata: %{user_id: params.user_id}}

      }

      |> Stripe.Checkout.Session.create()

      |> case do

        {:ok, session} -> {:ok, %{id: session.id, url: session.url}}

        {:error, err} -> log_and_wrap("create_checkout_session", err)

      end

    else

      session_id = @stub_session_prefix <> encode_stub(params.user_id, params.price)

      stub_log("create_checkout_session", user: params.user_id, price: params.price)

      {:ok, %{id: session_id, url: params.success_url <> "?session_id=#{session_id}"}}

    end

  end

  @doc """

  Fetches a Checkout Session. Returns `{:ok, %{id, client_reference_id, subscription}}`,

  where `subscription` is the created subscription's id, or nil until there is one.

  """

  def fetch_checkout_session(session_id) do

    if configured?() do

      case Stripe.Checkout.Session.retrieve(session_id) do

        {:ok, session} ->

          {:ok,

           %{

             id: session.id,

             client_reference_id: session.client_reference_id,

             subscription: stripe_id(session.subscription)

           }}

        {:error, err} ->

          log_and_wrap("fetch_checkout_session", err)

      end

    else

      stub_log("fetch_checkout_session", session: session_id)

      case session_id do

        @stub_session_prefix <> encoded ->

          {user_id, _price} = decode_stub(encoded)

          {:ok,

           %{

             id: session_id,

             client_reference_id: user_id,

             subscription: @stub_subscription_prefix <> encoded

           }}

        _ ->

          {:error, "No such checkout session"}

      end

    end

  end

  ## Subscriptions

  @doc "Fetches a subscription, normalized (see the moduledoc)."

  def fetch_subscription(subscription_id) do

    if configured?() do

      case Stripe.Subscription.retrieve(subscription_id) do

        {:ok, sub} -> {:ok, normalize_subscription(sub)}

        {:error, err} -> log_and_wrap("fetch_subscription", err)

      end

    else

      stub_log("fetch_subscription", subscription: subscription_id)

      case subscription_id do

        @stub_subscription_prefix <> encoded -> {:ok, stub_subscription(subscription_id, encoded)}

        _ -> {:error, "No such subscription"}

      end

    end

  end

  ## Customer Portal

  @doc "Creates a Customer Portal session. Returns `{:ok, url}`."

  def create_portal_session(customer_id, return_url) do

    if configured?() do

      case Stripe.BillingPortal.Session.create(%{customer: customer_id, return_url: return_url}) do

        {:ok, %{url: url}} -> {:ok, url}

        {:error, err} -> log_and_wrap("create_portal_session", err)

      end

    else

      stub_log("create_portal_session", customer: customer_id)

      {:ok, return_url <> "?stub_portal=true"}

    end

  end

  ## Events

  @doc """

  Lists account events, newest first, normalized to string-keyed maps with

  `"id"`, `"type"`, `"created"`, and `"data"`.

  Takes `:ending_before` (events newer than this id), `:starting_after`

  (the next page, older than this id), `:created_after` (a `DateTime`), and

  `:limit` (default 100, Stripe's max).

  Returns `{:error, :cursor_expired}` when the `:ending_before` event has aged

  out of Stripe's 30-day event retention. Stub mode returns no events.

  """

  def list_events(opts \\ []) do

    if configured?() do

      case Stripe.Event.list(list_events_params(opts)) do

        {:ok, %{data: events}} ->

          {:ok, Enum.map(events, &normalize_event/1)}

        {:error,

         %Stripe.Error{

           extra: %{param: "ending_before", raw_error: %{"code" => "resource_missing"}}

         }} ->

          {:error, :cursor_expired}

        {:error, err} ->

          log_and_wrap("list_events", err)

      end

    else

      stub_log("list_events", opts: opts)

      {:ok, []}

    end

  end

  defp list_events_params(opts) do

    Enum.reduce(opts, %{limit: Keyword.get(opts, :limit, 100)}, fn

      {:ending_before, id}, params when is_binary(id) ->

        Map.put(params, :ending_before, id)

      {:starting_after, id}, params when is_binary(id) ->

        Map.put(params, :starting_after, id)

      {:created_after, %DateTime{} = dt}, params ->

        Map.put(params, :created, %{gt: DateTime.to_unix(dt)})

      _opt, params ->

        params

    end)

  end

  ## Normalization

  defp normalize_subscription(sub) do

    item = first_item(sub)

    %{

      id: sub.id,

      customer: stripe_id(sub.customer),

      price_id: item && stripe_id(item.price),

      status: sub.status,

      start_date: from_unix(Map.get(sub, :start_date)),

      # Newer Stripe API versions moved the period onto subscription items.

      current_period_end:

        from_unix(

          Map.get(sub, :current_period_end) || (item && Map.get(item, :current_period_end))

        ),

      cancel_at: cancel_at(sub, item),

      canceled_at: from_unix(Map.get(sub, :canceled_at)),

      user_id: get_in(sub, [Access.key(:metadata, %{}), "user_id"])

    }

  end

  defp first_item(%{items: %{data: [item | _]}}), do: item

  defp first_item(_sub), do: nil

  defp cancel_at(sub, item) do

    cond do

      Map.get(sub, :cancel_at) ->

        from_unix(sub.cancel_at)

      Map.get(sub, :cancel_at_period_end) ->

        from_unix(

          Map.get(sub, :current_period_end) || (item && Map.get(item, :current_period_end))

        )

      true ->

        nil

    end

  end

  # Expandable fields come back as an id string or an expanded struct.

  defp stripe_id(nil), do: nil

  defp stripe_id(id) when is_binary(id), do: id

  defp stripe_id(%{id: id}), do: id

  defp normalize_event(%Stripe.Event{} = event) do

    %{

      "id" => event.id,

      "type" => event.type,

      "created" => event.created,

      "data" => deep_stringify(event.data)

    }

  end

  defp deep_stringify(%_{} = struct), do: struct |> Map.from_struct() |> deep_stringify()

  defp deep_stringify(map) when is_map(map),

    do: Map.new(map, fn {k, v} -> {to_string(k), deep_stringify(v)} end)

  defp deep_stringify(list) when is_list(list), do: Enum.map(list, &deep_stringify/1)

  defp deep_stringify(other), do: other

  defp from_unix(nil), do: nil

  defp from_unix(unix) when is_integer(unix), do: DateTime.from_unix!(unix)

  ## Stub helpers

  defp encode_stub(user_id, price), do: Base.url_encode64("#{user_id}|#{price}", padding: false)

  defp decode_stub(encoded) do

    [user_id, price] =

      encoded |> Base.url_decode64!(padding: false) |> String.split("|", parts: 2)

    {user_id, price}

  end

  defp stub_subscription(subscription_id, encoded) do

    {user_id, price} = decode_stub(encoded)

    now = DateTime.utc_now(:second)

    %{

      id: subscription_id,

      customer: "cus_stub_#{user_id}",

      price_id: price,

      status: "active",

      start_date: now,

      current_period_end: DateTime.add(now, 30, :day),

      cancel_at: nil,

      canceled_at: nil,

      user_id: user_id

    }

  end

  ## Logging

  defp log_and_wrap(action, %Stripe.Error{} = err) do

    Logger.error("[Billing.Stripe] #{action} failed: #{err.message}")

    {:error, err.message || "Stripe request failed"}

  end

  defp log_and_wrap(action, err) do

    Logger.error("[Billing.Stripe] #{action} failed: #{inspect(err)}")

    {:error, "Stripe request failed"}

  end

  defp stub_log(action, meta), do: Logger.info("[Billing.Stripe stub] #{action} #{inspect(meta)}")

end

defmodule MyApp.Billing.Stripe.Customer do

  @moduledoc """

  The Stripe Customer for a user, kept so every checkout and portal session

  reuses it instead of creating duplicates.

  """

  use Ecto.Schema

  import Ecto.Changeset

  @primary_key {:id, :binary_id, autogenerate: true}

  @foreign_key_type :binary_id

  schema "stripe_customers" do

    field :stripe_customer_id, :string

    belongs_to :user, MyApp.Accounts.User

    timestamps(type: :utc_datetime)

  end

  ## Changesets

  def changeset(customer, attrs) do

    customer

    |> cast(attrs, [:user_id, :stripe_customer_id])

    |> validate_required([:user_id, :stripe_customer_id])

    |> unique_constraint(:user_id)

    |> unique_constraint(:stripe_customer_id)

  end

end

defmodule MyApp.Billing.Stripe.EventsPoller do

  @moduledoc """

  Drains Stripe's `/v1/events` stream into `MyApp.Billing.Stripe.handle_event/1`.

  This keeps subscriptions, and the licenses they grant, in sync after

  checkout without a public webhook endpoint. An Oban cron entry runs it

  every minute (see `config/config.exs`).

  A checkpoint remembers the last event processed, so each run fetches only

  newer events:

    * **First run:** record the latest event without processing it, rather

      than replaying the account's whole history.

    * **Cursor expired:** Stripe keeps events for 30 days. If polling stopped

      for longer, the checkpoint's event is gone, so replay every event still

      kept since the last successful poll. Handlers are idempotent, so replays

      are harmless. Anything older is lost, which is logged loudly: reconcile

      by hand.

  The `unique` window covers the cron interval, so a slow poll can't pile up

  duplicates.

  """

  use Oban.Worker,

    queue: :stripe,

    max_attempts: 3,

    unique: [period: 120, fields: [:worker, :args]]

  require Logger

  alias MyApp.Billing.Stripe

  @source "platform"

  @page_size 100

  @impl Oban.Worker

  def perform(_job) do

    case Stripe.get_checkpoint(@source) do

      %{last_event_id: last_event_id} = checkpoint when is_binary(last_event_id) ->

        drain(checkpoint)

      _none ->

        bootstrap()

    end

  end

  defp bootstrap do

    case Stripe.client().list_events(limit: 1) do

      {:ok, [%{"id" => id} | _]} -> save(id)

      # No events yet: record the poll so the next run bootstraps again cheaply.

      {:ok, []} -> save(nil)

      {:error, reason} -> {:error, reason}

    end

  end

  defp drain(checkpoint) do

    case Stripe.client().list_events(ending_before: checkpoint.last_event_id) do

      {:ok, []} ->

        save(checkpoint.last_event_id)

      {:ok, events} ->

        process(events)

      {:error, :cursor_expired} ->

        recover(checkpoint)

      {:error, reason} ->

        Logger.error("[Stripe.EventsPoller] list_events failed: #{inspect(reason)}")

        {:error, reason}

    end

  end

  defp recover(checkpoint) do

    Logger.warning(

      "[Stripe.EventsPoller] event #{checkpoint.last_event_id} aged out of Stripe's retention. " <>

        "Replaying events since #{checkpoint.last_polled_at}; anything older is gone, so " <>

        "reconcile by hand if polling stopped for more than 30 days."

    )

    case fetch_since(checkpoint.last_polled_at) do

      {:ok, []} -> bootstrap()

      {:ok, events} -> process(events)

      {:error, reason} -> {:error, reason}

    end

  end

  # Every retained event created after `since`, newest first, across pages.

  defp fetch_since(since, acc \\ []) do

    opts =

      [limit: @page_size, created_after: since]

      |> Keyword.merge(if oldest = List.last(acc), do: [starting_after: oldest["id"]], else: [])

    case Stripe.client().list_events(opts) do

      {:ok, events} when length(events) == @page_size -> fetch_since(since, acc ++ events)

      {:ok, events} -> {:ok, acc ++ events}

      {:error, reason} -> {:error, reason}

    end

  end

  # Stripe returns newest first. Process oldest first, then move the cursor to

  # the newest.

  defp process([newest | _] = events) do

    events

    |> Enum.reverse()

    |> Enum.each(&Stripe.handle_event/1)

    save(newest["id"])

  end

  defp save(last_event_id) do

    with {:ok, _checkpoint} <- Stripe.save_checkpoint(@source, last_event_id), do: :ok

  end

end

defmodule MyApp.Billing.Stripe.Subscription do

  @moduledoc """

  A local mirror of a Stripe Subscription.

  Stripe is the source of truth: rows are only ever written from what Stripe

  reports, and each one is translated into a license by

  `MyApp.Billing.Stripe.sync_subscription/2`. The mirror keeps the

  Stripe-only details, like the renewal date, out of the license.

  `status` is Stripe's own status string, so nothing breaks if Stripe adds one.

  """

  use Ecto.Schema

  import Ecto.Changeset

  import Ecto.Query

  @live_statuses ~w(active trialing past_due)

  @primary_key {:id, :binary_id, autogenerate: true}

  @foreign_key_type :binary_id

  schema "stripe_subscriptions" do

    field :stripe_subscription_id, :string

    field :stripe_customer_id, :string

    field :stripe_price_id, :string

    field :status, :string

    field :current_period_end, :utc_datetime

    field :cancel_at, :utc_datetime

    field :canceled_at, :utc_datetime

    belongs_to :user, MyApp.Accounts.User

    timestamps(type: :utc_datetime)

  end

  ## Changesets

  def changeset(subscription, attrs) do

    subscription

    |> cast(attrs, [

      :user_id,

      :stripe_subscription_id,

      :stripe_customer_id,

      :stripe_price_id,

      :status,

      :current_period_end,

      :cancel_at,

      :canceled_at

    ])

    |> validate_required([:user_id, :stripe_subscription_id, :stripe_customer_id, :status])

  end

  ## Queries

  @doc """

  The user's newest subscription that Stripe still bills: active, trialing,

  or past due.

  """

  def live_for_user_query(user) do

    from s in __MODULE__,

      where: s.user_id == ^user.id and s.status in @live_statuses,

      order_by: [desc: s.inserted_at],

      limit: 1

  end

  ## View functions

  @doc """

  Whether Stripe still bills the subscription. Past due counts: Stripe is

  retrying the payment, and access continues until it gives up.

  """

  def live?(%__MODULE__{status: status}), do: status in @live_statuses

  def live_status?(status), do: status in @live_statuses

end

defmodule MyAppWeb.StripeController do

  @moduledoc """

  The return leg of Stripe Checkout. Stripe redirects here with

  `?session_id=...`, and we sync the new subscription right away instead of

  waiting for the event poller.

  """

  use MyAppWeb, :controller

  alias MyApp.Billing.Stripe

  def return(conn, %{"session_id" => session_id}) do

    case Stripe.sync_checkout_session(conn.assigns.current_scope, session_id) do

      {:ok, %Stripe.Subscription{}} ->

        conn

        |> put_flash(:info, "You're subscribed. Thanks!")

        |> redirect(to: ~p"/billing")

      {:ok, :pending} ->

        conn

        |> put_flash(:info, "Your payment is processing. Your plan will update shortly.")

        |> redirect(to: ~p"/billing")

      {:error, _reason} ->

        conn

        |> put_flash(

          :error,

          "We couldn't confirm your subscription. If you were charged, contact us."

        )

        |> redirect(to: ~p"/billing")

    end

  end

  def return(conn, _params), do: redirect(conn, to: ~p"/billing")

end

  alias MyApp.Billing.Stripe

        <:actions :if={@subscription}>

          <.button id="manage-billing" phx-click="portal">Manage billing</.button>

        </:actions>

      <p :if={@subscription} id="subscription-status" class="text-sm">

        {subscription_status(@subscription)}

      </p>

      <p

        :if={!@subscription && @entitlements.license && @entitlements.license.ends_at}

        class="text-sm"

      >

          <:subtitle>{plans_subtitle(@subscription, @checkout_plans)}</:subtitle>

            checkout={plan in @checkout_plans}

            upgrade={plan.rank > @entitlements.plan.rank}

  attr :checkout, :boolean, required: true

  attr :upgrade, :boolean, required: true

        "rounded-box border p-5 space-y-3 flex flex-col",

      <ul class="text-sm space-y-1 grow">

      <.button

        :if={@checkout}

        phx-click="checkout"

        phx-value-plan={@plan.key}

        phx-disable-with="Redirecting..."

        class="btn btn-primary w-full"

      >

        {if @upgrade, do: "Upgrade", else: "Subscribe"}

      </.button>

    {:ok, assign_billing(socket)}

  end

  @impl true

  def handle_event("checkout", %{"plan" => plan}, socket) do

    scope = socket.assigns.current_scope

    case Stripe.create_checkout_session(

           scope,

           plan,

           url(~p"/billing/stripe/return"),

           url(~p"/billing")

         ) do

      {:ok, %{url: url}} ->

        {:noreply, redirect(socket, external: url)}

      {:error, :already_subscribed} ->

        open_portal(socket)

      {:error, :not_purchasable} ->

        {:noreply, put_flash(socket, :error, "That plan isn't available to buy.")}

      {:error, _reason} ->

        {:noreply,

         put_flash(socket, :error, "We couldn't reach our payment provider. Please try again.")}

    end

  def handle_event("portal", _params, socket), do: open_portal(socket)

    {:noreply, assign_billing(socket)}

  defp open_portal(socket) do

    case Stripe.create_portal_session(socket.assigns.current_scope, url(~p"/billing")) do

      {:ok, url} ->

        {:noreply, redirect(socket, external: url)}

      {:error, _reason} ->

        {:noreply,

         put_flash(socket, :error, "We couldn't open the billing portal. Please try again.")}

    end

  end

  defp assign_billing(socket) do

    scope = socket.assigns.current_scope

    entitlements = Billing.entitlements(scope)

    subscription = Stripe.get_live_subscription(scope)

    assign(socket,

      entitlements: entitlements,

      subscription: subscription,

      checkout_plans: checkout_plans(entitlements, subscription)

    )

  # Plans worth buying: anything better than the current plan, or the same

  # plan when the license granting it will end. Users already paying through

  # Stripe change plans in the portal, never through a second checkout.

  defp checkout_plans(_entitlements, %Stripe.Subscription{}), do: []

  defp checkout_plans(entitlements, nil) do

    current = entitlements.plan

    ending? = entitlements.license != nil and entitlements.license.ends_at != nil

    Enum.filter(Stripe.purchasable_plans(), fn plan ->

      plan.rank > current.rank or (plan.key == current.key and ending?)

    end)

  end

  defp subscription_status(%{status: "past_due"}) do

    "Your last payment failed. Update your card in the billing portal to keep your plan."

  end

  defp subscription_status(%{cancel_at: %DateTime{} = cancel_at}) do

    "Your subscription ends on #{format_date(cancel_at)}."

  end

  defp subscription_status(%{current_period_end: %DateTime{} = renews_at}) do

    "Your subscription renews on #{format_date(renews_at)}."

  end

  defp subscription_status(_subscription), do: "Your subscription is active."

  defp plans_subtitle(%Stripe.Subscription{}, _plans),

    do: "Change or cancel your plan in the billing portal."

  defp plans_subtitle(nil, []), do: "To change your plan, contact us."

  defp plans_subtitle(nil, _plans), do: "Upgrade anytime. Cancel whenever you like."

    # Stripe Checkout sends the user back here after paying.

    get "/billing/stripe/return", StripeController, :return

defmodule MyApp.Repo.Migrations.AddObanJobsTable do

  use Ecto.Migration

  def up, do: Oban.Migration.up(version: 14)

  def down, do: Oban.Migration.down(version: 1)

end

defmodule MyApp.Repo.Migrations.CreateStripeTables do

  use Ecto.Migration

  def change do

    # One Stripe Customer per user, reused across checkouts and the portal.

    create table(:stripe_customers, primary_key: false) do

      add :id, :binary_id, primary_key: true

      add :user_id, references(:users, type: :binary_id, on_delete: :delete_all), null: false

      add :stripe_customer_id, :string, null: false

      timestamps(type: :utc_datetime)

    end

    create unique_index(:stripe_customers, [:user_id])

    create unique_index(:stripe_customers, [:stripe_customer_id])

    # Stripe's own record, mirrored. Each one is translated into a license,

    # which is all the rest of the app reads.

    create table(:stripe_subscriptions, primary_key: false) do

      add :id, :binary_id, primary_key: true

      add :user_id, references(:users, type: :binary_id, on_delete: :delete_all), null: false

      add :stripe_subscription_id, :string, null: false

      add :stripe_customer_id, :string, null: false

      add :stripe_price_id, :string

      add :status, :string, null: false

      add :current_period_end, :utc_datetime

      add :cancel_at, :utc_datetime

      add :canceled_at, :utc_datetime

      timestamps(type: :utc_datetime)

    end

    create unique_index(:stripe_subscriptions, [:stripe_subscription_id])

    create index(:stripe_subscriptions, [:user_id])

    # Where event polling left off, one row per stream ("platform" for now).

    create table(:stripe_event_checkpoints, primary_key: false) do

      add :id, :binary_id, primary_key: true

      add :source, :string, null: false

      add :last_event_id, :string

      add :last_polled_at, :utc_datetime

      timestamps(type: :utc_datetime)

    end

    create unique_index(:stripe_event_checkpoints, [:source])

  end

end

defmodule MyApp.Billing.Stripe.EventsPollerTest do

  use MyApp.DataCase, async: true

  use Oban.Testing, repo: MyApp.Repo

  import MyApp.AccountsFixtures

  alias MyApp.Billing

  alias MyApp.Billing.Stripe

  alias MyApp.Billing.Stripe.{EventsPoller, FakeClient}

  defp event(id, created, sub \\ nil) do

    data = if sub, do: %{"object" => %{"id" => sub.id}}, else: %{"object" => %{}}

    type = if sub, do: "customer.subscription.updated", else: "charge.succeeded"

    %{"id" => id, "type" => type, "created" => created, "data" => data}

  end

  defp setup_subscription do

    user = user_fixture()

    sub = FakeClient.subscription(user_id: user.id)

    FakeClient.put_subscription(sub)

    {user, sub}

  end

  test "the first run records the latest event without replaying history" do

    {user, sub} = setup_subscription()

    FakeClient.put_events([event("evt_2", 2, sub), event("evt_1", 1)])

    assert :ok = perform_job(EventsPoller, %{})

    assert Stripe.get_checkpoint("platform").last_event_id == "evt_2"

    assert Billing.list_licenses(user) == []

  end

  test "an empty account records the poll and bootstraps again later" do

    assert :ok = perform_job(EventsPoller, %{})

    checkpoint = Stripe.get_checkpoint("platform")

    assert checkpoint.last_event_id == nil

    assert checkpoint.last_polled_at

  end

  test "processes events newer than the checkpoint and advances it" do

    {user, sub} = setup_subscription()

    {:ok, _} = Stripe.save_checkpoint("platform", "evt_1")

    FakeClient.put_events([event("evt_3", 3, sub), event("evt_2", 2), event("evt_1", 1)])

    assert :ok = perform_job(EventsPoller, %{})

    assert Stripe.get_checkpoint("platform").last_event_id == "evt_3"

    assert Billing.entitlements_for_user(user).plan.key == :pro

  end

  test "recovers when the checkpoint's event has expired" do

    {user, sub} = setup_subscription()

    {:ok, checkpoint} = Stripe.save_checkpoint("platform", "evt_gone")

    since = DateTime.to_unix(checkpoint.last_polled_at)

    FakeClient.expire_cursor()

    FakeClient.put_events([event("evt_new", since + 10, sub), event("evt_old", since - 10)])

    ExUnit.CaptureLog.capture_log(fn -> assert :ok = perform_job(EventsPoller, %{}) end)

    assert Stripe.get_checkpoint("platform").last_event_id == "evt_new"

    assert Billing.entitlements_for_user(user).plan.key == :pro

  end

end

defmodule MyApp.Billing.StripeTest do

  use MyApp.DataCase, async: true

  import MyApp.AccountsFixtures

  import MyApp.BillingFixtures

  alias MyApp.Accounts.Scope

  alias MyApp.Billing

  alias MyApp.Billing.{License, Stripe}

  alias MyApp.Billing.Stripe.{Customer, FakeClient, Subscription}

  defp plan_for(user), do: Billing.entitlements_for_user(user).plan.key

  defp stripe_licenses(user),

    do: Enum.filter(Billing.list_licenses(user), &(&1.source == :stripe))

  describe "prices" do

    test "map plans to prices and back, including older prices" do

      assert Stripe.price_for_plan(:pro) == "price_pro"

      assert Stripe.plan_for_price("price_pro") == :pro

      assert Stripe.plan_for_price("price_pro_2025") == :pro

      assert Stripe.plan_for_price("price_unknown") == nil

    end

    test "only public paid plans with a price are purchasable" do

      assert Enum.map(Stripe.purchasable_plans(), & &1.key) == [:pro, :business]

    end

  end

  describe "sync_subscription/2" do

    test "an active subscription grants its plan through a Stripe license" do

      user = user_fixture()

      sub = FakeClient.subscription(price_id: "price_business")

      {:ok, %Subscription{status: "active"}} = Stripe.sync_subscription(user, sub)

      assert [%License{plan: :business, status: :active, source_ref: ref}] = stripe_licenses(user)

      assert ref == sub.id

      assert plan_for(user) == :business

    end

    test "replaying is harmless" do

      user = user_fixture()

      sub = FakeClient.subscription([])

      {:ok, first} = Stripe.sync_subscription(user, sub)

      {:ok, second} = Stripe.sync_subscription(user, sub)

      assert first.id == second.id

      assert length(stripe_licenses(user)) == 1

    end

    test "a plan change in the portal updates the same license" do

      user = user_fixture()

      sub = FakeClient.subscription(price_id: "price_pro")

      {:ok, _} = Stripe.sync_subscription(user, sub)

      {:ok, _} = Stripe.sync_subscription(user, %{sub | price_id: "price_business"})

      assert [%License{plan: :business}] = stripe_licenses(user)

    end

    test "a pending cancellation keeps access until it ends" do

      user = user_fixture()

      cancel_at = days_from_now(12)

      {:ok, _} = Stripe.sync_subscription(user, FakeClient.subscription(cancel_at: cancel_at))

      assert [%License{status: :active, ends_at: ^cancel_at}] = stripe_licenses(user)

      assert plan_for(user) == :pro

    end

    test "past due keeps access while Stripe retries" do

      user = user_fixture()

      {:ok, _} = Stripe.sync_subscription(user, FakeClient.subscription(status: "past_due"))

      assert plan_for(user) == :pro

    end

    test "canceled and unpaid subscriptions cancel the license" do

      for status <- ["canceled", "unpaid", "incomplete_expired"] do

        user = user_fixture()

        sub = FakeClient.subscription([])

        {:ok, _} = Stripe.sync_subscription(user, sub)

        {:ok, _} = Stripe.sync_subscription(user, %{sub | status: status})

        assert [%License{status: :canceled}] = stripe_licenses(user)

        assert plan_for(user) == :free

      end

    end

    test "an incomplete subscription grants nothing yet" do

      user = user_fixture()

      {:ok, _} = Stripe.sync_subscription(user, FakeClient.subscription(status: "incomplete"))

      assert stripe_licenses(user) == []

      assert plan_for(user) == :free

    end

    test "a price that maps to no plan grants nothing" do

      user = user_fixture()

      ExUnit.CaptureLog.capture_log(fn ->

        {:ok, _} =

          Stripe.sync_subscription(user, FakeClient.subscription(price_id: "price_unknown"))

      end)

      assert stripe_licenses(user) == []

    end

  end

  describe "alongside other licenses" do

    test "a manual enterprise deal outranks a Stripe subscription" do

      user = user_fixture()

      license_fixture(user, plan: :enterprise)

      sub = FakeClient.subscription([])

      {:ok, _} = Stripe.sync_subscription(user, sub)

      assert plan_for(user) == :enterprise

      {:ok, _} = Stripe.sync_subscription(user, %{sub | status: "canceled"})

      assert plan_for(user) == :enterprise

    end

    test "canceling Stripe falls back to the user's other license" do

      user = user_fixture()

      license_fixture(user, plan: :pro)

      sub = FakeClient.subscription(price_id: "price_business")

      {:ok, _} = Stripe.sync_subscription(user, sub)

      assert plan_for(user) == :business

      {:ok, _} = Stripe.sync_subscription(user, %{sub | status: "canceled"})

      assert plan_for(user) == :pro

    end

    test "admins can't edit or cancel Stripe licenses" do

      user = user_fixture()

      {:ok, _} = Stripe.sync_subscription(user, FakeClient.subscription([]))

      [license] = stripe_licenses(user)

      assert {:error, :not_editable} = Billing.update_license(license, %{plan: :business})

      assert {:error, :not_editable} = Billing.cancel_license(license)

    end

  end

  describe "create_checkout_session/4" do

    test "creates the Stripe customer once and returns the checkout url" do

      scope = user_scope_fixture()

      assert {:ok, %{url: "https://checkout.stripe.test/price_pro"}} =

               Stripe.create_checkout_session(scope, "pro", "http://s", "http://c")