# Background jobs. Cron entries poll Stripe's event streams every minute: the

# platform's, to keep subscriptions and their licenses in sync, and each

# Connect account's, to keep memberships and payments in sync.

    {Oban.Plugins.Cron,

     crontab: [

       {"* * * * *", MyApp.Billing.Stripe.EventsPoller},

       {"* * * * *", MyApp.Connect.EnqueueEventPolling}

     ]}

    * `:platform_fee_percent` — the platform's cut of what the user's own

      customers pay them through Stripe Connect

  @enforce_keys [:key, :name, :rank, :platform_fee_percent]

  defstruct [

    :key,

    :name,

    :description,

    :rank,

    :platform_fee_percent,

    public?: true,

    limits: %{},

    features: []

  ]

  Every plan defines every limit, only known features, and a platform fee, or

  the app fails to compile.

      platform_fee_percent: 10.0,

      platform_fee_percent: 8.0,

      platform_fee_percent: 5.0,

      platform_fee_percent: 3.0,

    unless is_number(plan.platform_fee_percent) and plan.platform_fee_percent >= 0 do

      raise ArgumentError, "plan #{inspect(plan.key)} needs a platform_fee_percent"

    end

  @doc """

  Formats a percentage for display, without a trailing ".0".

  ## Examples

      format_percent(8.0)

      #=> "8%"

      format_percent(7.5)

      #=> "7.5%"

  """

  def format_percent(percent) when percent == trunc(percent), do: "#{trunc(percent)}%"

  def format_percent(percent), do: "#{percent}%"

  `:limit` (default 100, Stripe's max). `:account` lists a Stripe Connect

  account's events instead of the platform's.

      request_opts =

        case opts[:account] do

          nil -> []

          account_id -> [headers: %{"Stripe-Account" => account_id}]

        end

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

  Drains one of Stripe's `/v1/events` streams and dispatches each event.

  checkout without a public webhook endpoint. There's a stream per source:

    * **The platform's own account**, the default, polled by an Oban cron

      entry every minute (see `config/config.exs`). Events go to

      `MyApp.Billing.Stripe.handle_event/1`.

    * **Each Stripe Connect account**, with `%{"account" => "acct_..."}` as

      job args, enqueued by `MyApp.Connect.EnqueueEventPolling`. Events go to

      `MyApp.Connect.handle_stripe_event/2`.

  A checkpoint per stream remembers the last event processed, so each run

  fetches only newer events:

  The `unique` window covers the cron interval and keys on the args, so a

  slow poll can't pile up duplicates, and each stream dedupes on its own.

  def perform(%Oban.Job{args: %{"account" => account_id}}) when is_binary(account_id) do

    poll({:connect, account_id})

  end

  def perform(%Oban.Job{}), do: poll(:platform)

  defp poll(stream) do

    case Stripe.get_checkpoint(source(stream)) do

        drain(stream, checkpoint)

        bootstrap(stream)

  defp bootstrap(stream) do

    case list_events(stream, limit: 1) do

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

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

  defp drain(stream, checkpoint) do

    case list_events(stream, ending_before: checkpoint.last_event_id) do

        save(stream, checkpoint.last_event_id)

        process(stream, events)

        recover(stream, checkpoint)

        Logger.error(

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

        )

  defp recover(stream, checkpoint) do

      "[Stripe.EventsPoller] #{source(stream)}: 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(stream, checkpoint.last_polled_at) do

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

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

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

    case list_events(stream, opts) do

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

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

    |> Enum.each(&dispatch(stream, &1))

    save(stream, newest["id"])

  defp dispatch(:platform, event), do: Stripe.handle_event(event)

  defp dispatch({:connect, account_id}, event),

    do: MyApp.Connect.handle_stripe_event(event, account_id)

  defp list_events(:platform, opts), do: Stripe.client().list_events(opts)

  defp list_events({:connect, account_id}, opts),

    do: Stripe.client().list_events([account: account_id] ++ opts)

  defp source(:platform), do: "platform"

  defp source({:connect, account_id}), do: "connect:#{account_id}"

  defp save(stream, last_event_id) do

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

defmodule MyApp.Connect do

  @moduledoc """

  Stripe Connect — letting your users get paid by *their* audience, with the

  platform taking an application fee.

  A user onboards a Connect Express account (`MyApp.Connect.Account`), sets a

  recurring membership price and/or a one-time payment price, and shares a public

  page where anyone can pay them by email. The platform collects

  `platform_fee_percent/1` on every transaction.

  This is a separate domain from `MyApp.Billing`, which is the platform charging

  its own users. The two meet in two places:

    * **The fee comes from the seller's plan.** `platform_fee_percent/1` reads

      the plan their licenses grant, so a better plan, however it was obtained,

      pays a lower cut.

    * **One event poller.** `MyApp.Billing.Stripe.EventsPoller` polls a stream

      per connected account alongside the platform's, and routes those events

      here via `handle_stripe_event/2`. `MyApp.Connect.EnqueueEventPolling`

      starts the per-account polls.

  """

  import Ecto.Query, warn: false

  alias MyApp.Accounts.User

  alias MyApp.Billing

  alias MyApp.Connect.Account

  alias MyApp.Connect.Payment

  alias MyApp.Connect.Stripe, as: StripeAPI

  alias MyApp.Connect.Subscription

  alias MyApp.Repo

  @doc """

  The platform's cut of a seller's sales, as a percentage, from the plan their

  licenses grant.

  The fee is set when a buyer checks out, so changing plans affects new

  memberships and payments, not existing memberships.

  """

  def platform_fee_percent(%User{} = user) do

    Billing.entitlements_for_user(user).plan.platform_fee_percent

  end

  def platform_fee_percent(%Account{} = account) do

    account

    |> Repo.preload(:user)

    |> Map.fetch!(:user)

    |> platform_fee_percent()

  end

  ## Account reads

  @doc """

  The connected account for a `User` (or by id string, for the public subscribe

  page). Returns nil if there isn't one.

  """

  def get_account(%User{} = user), do: Repo.get_by(Account, user_id: user.id)

  def get_account(id) when is_binary(id), do: Repo.get(Account, id)

  defp get_account_by_stripe_id(account_id),

    do: Repo.get_by(Account, stripe_account_id: account_id)

  @doc """

  The Stripe ids of every account that finished onboarding, whose event

  streams are polled.

  """

  def list_onboarded_stripe_account_ids do

    Repo.all(Account.onboarded_stripe_ids_query())

  end

  ## Onboarding

  @doc """

  Starts (or resumes) Stripe Connect onboarding for a user. Creates the local

  account row and the Stripe account on first call, then returns a one-time

  onboarding URL to redirect the seller to.

  """

  def start_onboarding(%User{} = user, return_url, refresh_url) do

    with {:ok, account} <- ensure_account(user),

         {:ok, account} <- ensure_stripe_account(account, user) do

      case StripeAPI.create_account_link(account.stripe_account_id, return_url, refresh_url) do

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

        {:error, _} = err -> err

      end

    end

  end

  defp ensure_account(user) do

    case get_account(user) do

      nil -> %Account{} |> Account.changeset(%{user_id: user.id}) |> Repo.insert()

      account -> {:ok, account}

    end

  end

  defp ensure_stripe_account(%Account{stripe_account_id: nil} = account, user) do

    with {:ok, account_id} <- StripeAPI.create_account(user) do

      update_account(account, %{stripe_account_id: account_id})

    end

  end

  defp ensure_stripe_account(%Account{} = account, _user), do: {:ok, account}

  @doc "Generates a fresh onboarding link for an account that's mid-onboarding."

  def refresh_account_link(%Account{stripe_account_id: account_id}, return_url, refresh_url)

      when is_binary(account_id) do

    StripeAPI.create_account_link(account_id, return_url, refresh_url)

  end

  @doc """

  Re-reads onboarding status from Stripe and updates the local flags. Call this

  when the seller returns from the Connect flow.

  """

  def refresh_onboarding_status(%Account{stripe_account_id: account_id} = account)

      when is_binary(account_id) do

    case StripeAPI.fetch_account(account_id) do

      {:ok, %{details_submitted: submitted, charges_enabled: charges}} ->

        update_account(account, %{onboarding_complete: submitted, charges_enabled: charges})

      {:error, _} = err ->

        err

    end

  end

  def refresh_onboarding_status(%Account{} = account), do: {:ok, account}

  @doc """

  Disconnects the account from Stripe: deletes the connected account, cancels

  local active memberships, and resets the local Stripe fields (payment history

  is preserved).

  """

  def disconnect(%Account{} = account) do

    _ = StripeAPI.disconnect_account(account.stripe_account_id)

    now = DateTime.utc_now(:second)

    Subscription

    |> where(connect_account_id: ^account.id)

    |> where([s], s.status in ^[:active, :trialing, :past_due])

    |> Repo.update_all(set: [status: :canceled, canceled_at: now])

    update_account(account, %{

      stripe_account_id: nil,

      onboarding_complete: false,

      charges_enabled: false,

      membership_price_cents: nil,

      one_time_price_cents: nil

    })

  end

  ## Pricing

  @doc "Sets the recurring membership price (in cents) for the seller."

  def set_membership_price(%Account{} = account, cents),

    do: update_account(account, %{membership_price_cents: cents})

  @doc "Sets the one-time payment price (in cents) for the seller."

  def set_one_time_price(%Account{} = account, cents),

    do: update_account(account, %{one_time_price_cents: cents})

  defp update_account(account, attrs) do

    account |> Account.changeset(attrs) |> Repo.update()

  end

  ## Buyer checkout

  @doc """

  Starts a recurring-membership checkout for a buyer (identified by email) on

  the seller's connected account. Returns `{:ok, %{id, url}}`.

  """

  def create_membership_checkout(%Account{} = account, email, success_url, cancel_url) do

    if Account.membership_enabled?(account) do

      StripeAPI.create_membership_checkout_session(

        account.stripe_account_id,

        email,

        account.membership_price_cents,

        account.currency,

        platform_fee_percent(account),

        success_url,

        cancel_url

      )

    else

      {:error, :membership_not_enabled}

    end

  end

  @doc """

  Starts a one-time payment checkout for a buyer on the seller's connected

  account. The application fee is computed from `platform_fee_percent/1`.

  """

  def create_payment_checkout(%Account{} = account, email, success_url, cancel_url) do

    if Account.one_time_enabled?(account) do

      fee_amount = round(account.one_time_price_cents * platform_fee_percent(account) / 100)

      StripeAPI.create_payment_checkout_session(

        account.stripe_account_id,

        email,

        account.one_time_price_cents,

        account.currency,

        fee_amount,

        success_url,

        cancel_url

      )

    else

      {:error, :one_time_not_enabled}

    end

  end

  @doc """

  Polls a completed Checkout Session and records the result — a membership

  subscription or a one-time payment, depending on the session mode. Returns

  `{:ok, %Subscription{}}`, `{:ok, %Payment{}}`, `{:ok, :pending}`, or

  `{:error, reason}`.

  """

  def sync_checkout_session(%Account{} = account, session_id) do

    with {:ok, session} <- StripeAPI.fetch_checkout_session(account.stripe_account_id, session_id) do

      case session.mode do

        "subscription" -> sync_membership(account, session)

        "payment" -> record_payment(account, session)

        _ -> {:ok, :pending}

      end

    end

  end

  defp sync_membership(account, %{subscription: sub_id} = session) when is_binary(sub_id) do

    case StripeAPI.fetch_subscription(account.stripe_account_id, sub_id) do

      {:ok, sub} -> upsert_subscription(account, sub, session.customer_email)

      {:error, _} = err -> err

    end

  end

  defp sync_membership(_account, _session), do: {:ok, :pending}

  ## Sync

  @doc """

  Inserts or updates a membership subscription from a normalized Stripe

  subscription map. Upserts on the Stripe subscription id (idempotent across the

  return poll and the event poller). `email`, when given, is recorded; nil never

  overwrites an existing email.

  """

  def upsert_subscription(%Account{} = account, sub, email \\ nil) do

    attrs =

      %{

        connect_account_id: account.id,

        stripe_subscription_id: sub.id,

        stripe_customer_id: sub.customer,

        status: status_from_stripe(sub.status),

        current_period_end: sub.current_period_end,

        cancel_at_period_end: sub.cancel_at_period_end || false,

        canceled_at: sub.canceled_at

      }

      |> maybe_put_email(email)

    case Repo.get_by(Subscription, stripe_subscription_id: sub.id) do

      nil -> %Subscription{}

      existing -> existing

    end

    |> Subscription.changeset(attrs)

    |> Repo.insert_or_update()

  end

  defp maybe_put_email(attrs, nil), do: attrs

  defp maybe_put_email(attrs, email), do: Map.put(attrs, :customer_email, email)

  @doc """

  Records a one-time payment from a Checkout Session. Idempotent on the session

  id, so the return poll and a later event can't double-insert.

  """

  def record_payment(%Account{} = account, session) do

    attrs = %{

      connect_account_id: account.id,

      stripe_session_id: session.id,

      stripe_payment_intent_id: session.payment_intent,

      stripe_customer_id: session.customer,

      customer_email: session.customer_email,

      amount_cents: session.amount_total,

      currency: account.currency,

      status: if(session.payment_status == "paid", do: :succeeded, else: :pending)

    }

    case Repo.get_by(Payment, stripe_session_id: session.id) do

      nil -> %Payment{}

      existing -> existing

    end

    |> Payment.changeset(attrs)

    |> Repo.insert_or_update()

  end

  defp status_from_stripe("active"), do: :active

  defp status_from_stripe("trialing"), do: :trialing

  defp status_from_stripe("past_due"), do: :past_due

  defp status_from_stripe("unpaid"), do: :past_due

  defp status_from_stripe("canceled"), do: :canceled

  defp status_from_stripe("incomplete_expired"), do: :canceled

  defp status_from_stripe("incomplete"), do: :incomplete

  defp status_from_stripe(_), do: :incomplete

  ## Queries (seller dashboard)

  def list_subscriptions(%Account{} = account) do

    Subscription

    |> where(connect_account_id: ^account.id)

    |> order_by(desc: :inserted_at)

    |> Repo.all()

  end

  def count_active_subscriptions(%Account{} = account) do

    Subscription

    |> where(connect_account_id: ^account.id)

    |> where([s], s.status in ^[:active, :trialing, :past_due])

    |> Repo.aggregate(:count)

  end

  @doc "Monthly recurring revenue (in cents) from active memberships."

  def monthly_revenue_cents(%Account{} = account) do

    (account.membership_price_cents || 0) * count_active_subscriptions(account)

  end

  def list_payments(%Account{} = account) do

    Payment

    |> where(connect_account_id: ^account.id)

    |> order_by(desc: :inserted_at)

    |> Repo.all()

  end

  @doc "Total of all succeeded one-time payments (in cents)."

  def total_payments_cents(%Account{} = account) do

    Payment

    |> where(connect_account_id: ^account.id, status: :succeeded)

    |> Repo.aggregate(:sum, :amount_cents) || 0

  end

  ## Customer Portal

  @doc "Creates a Customer Portal session for a buyer to manage their membership."

  def create_customer_portal_session(%Account{} = account, %Subscription{} = sub, return_url) do

    StripeAPI.create_customer_portal_session(

      account.stripe_account_id,

      sub.stripe_customer_id,

      return_url

    )

  end

  ## Event handling (called by the poller, per connected account)

  @doc """

  Handles a Stripe event for one connected account. Acts on

  `customer.subscription.*` (resync the membership) and `charge.refunded` (mark

  the one-time payment refunded); ignores everything else. Always returns `:ok`.

  """

  def handle_stripe_event(event, account_id) when is_binary(account_id) do

    case get_account_by_stripe_id(account_id) do

      nil -> :ok

      account -> dispatch_event(account, event)

    end

    :ok

  end

  defp dispatch_event(account, %{"type" => "customer.subscription." <> _} = event) do

    sub_id = get_in(event, ["data", "object", "id"])

    with sub_id when is_binary(sub_id) <- sub_id,

         {:ok, sub} <- StripeAPI.fetch_subscription(account.stripe_account_id, sub_id) do

      upsert_subscription(account, sub)

    end

  end

  defp dispatch_event(account, %{"type" => "charge.refunded"} = event) do

    payment_intent = get_in(event, ["data", "object", "payment_intent"])

    if is_binary(payment_intent) do

      Payment

      |> where(connect_account_id: ^account.id, stripe_payment_intent_id: ^payment_intent)

      |> Repo.update_all(set: [status: :refunded])

    end

  end

  defp dispatch_event(_account, _event), do: :ok

end

defmodule MyApp.Connect.Account do

  @moduledoc """

  A user's Stripe Connect (Express) account — the user acting as a *seller* who

  collects payments from their audience, with the platform taking an application

  fee on each transaction.

  A seller can offer two products, independently: a recurring `membership` and a

  one-time `payment`. Each is "enabled" once onboarding is complete and a price

  is set for that mode.

  """

  use Ecto.Schema

  import Ecto.Changeset

  import Ecto.Query

  @derive {Flop.Schema,

           filterable: [:email],

           sortable: [:email, :inserted_at],

           default_limit: 20,

           max_limit: 100,

           default_order: %{order_by: [:inserted_at], order_directions: [:desc]},

           pagination_types: [:page],

           adapter_opts: [

             join_fields: [email: [binding: :user, field: :email, ecto_type: :string]]

           ]}

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

  @foreign_key_type :binary_id

  schema "connect_accounts" do

    field :stripe_account_id, :string

    field :onboarding_complete, :boolean, default: false

    field :charges_enabled, :boolean, default: false

    field :membership_price_cents, :integer

    field :one_time_price_cents, :integer

    field :currency, :string, default: "usd"

    belongs_to :user, MyApp.Accounts.User

    timestamps(type: :utc_datetime)

  end

  def changeset(account, attrs) do

    account

    |> cast(attrs, [

      :user_id,

      :stripe_account_id,

      :onboarding_complete,

      :charges_enabled,

      :membership_price_cents,

      :one_time_price_cents,

      :currency

    ])

    |> validate_required([:user_id])

    |> validate_number(:membership_price_cents, greater_than: 0)

    |> validate_number(:one_time_price_cents, greater_than: 0)

    |> unique_constraint(:user_id)

    |> unique_constraint(:stripe_account_id)

  end

  ## Queries

  def with_user_query(query \\ __MODULE__) do

    from a in query, join: u in assoc(a, :user), as: :user, preload: [user: u]

  end

  def onboarded_stripe_ids_query do

    from a in __MODULE__,

      where: not is_nil(a.stripe_account_id) and a.onboarding_complete,

      select: a.stripe_account_id

  end

  ## View functions

  @doc "True once the seller can actually accept charges."

  def ready?(%__MODULE__{} = account),

    do: account.onboarding_complete and account.charges_enabled

  @doc "True when the seller is set up to sell recurring memberships."

  def membership_enabled?(%__MODULE__{} = account),

    do: ready?(account) and is_integer(account.membership_price_cents)

  @doc "True when the seller is set up to accept one-time payments."

  def one_time_enabled?(%__MODULE__{} = account),

    do: ready?(account) and is_integer(account.one_time_price_cents)

  @doc "True when onboarding was started but Stripe hasn't reported it complete."

  def onboarding_in_progress?(%__MODULE__{} = account),

    do: is_binary(account.stripe_account_id) and not account.onboarding_complete

  @doc "Where the seller is in setting up: `:not_started`, `:onboarding`, or `:ready`."

  def status(%__MODULE__{} = account) do

    cond do

      ready?(account) -> :ready

      is_binary(account.stripe_account_id) -> :onboarding

      true -> :not_started

    end

  end

end

defmodule MyApp.Connect.EnqueueEventPolling do

  @moduledoc """

  Cron worker that enqueues one `MyApp.Billing.Stripe.EventsPoller` job per

  onboarded Connect account, every minute.

  Separate jobs mean one slow or failing account never blocks another, and

  the poller's `unique` option dedupes overlapping runs per account.

  """

  use Oban.Worker, queue: :stripe, max_attempts: 1

  alias MyApp.Billing.Stripe.EventsPoller

  alias MyApp.Connect

  @impl Oban.Worker

  def perform(_job) do

    for account_id <- Connect.list_onboarded_stripe_account_ids() do

      %{account: account_id} |> EventsPoller.new() |> Oban.insert!()

    end

    :ok

  end

end

defmodule MyApp.Connect.Payment do

  @moduledoc """

  A one-time payment a buyer made to a seller's connected account.

  Recorded from the Checkout Session on return and reconciled by the event

  poller (`charge.refunded` flips it to `:refunded`).

  """

  use Ecto.Schema

  import Ecto.Changeset

  import Ecto.Query

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

  @foreign_key_type :binary_id

  @statuses ~w(succeeded pending refunded)a

  schema "connect_payments" do

    field :stripe_session_id, :string

    field :stripe_payment_intent_id, :string

    field :stripe_customer_id, :string

    field :customer_email, :string

    field :amount_cents, :integer

    field :currency, :string

    field :status, Ecto.Enum, values: @statuses

    belongs_to :connect_account, MyApp.Connect.Account

    timestamps(type: :utc_datetime)

  end

  def statuses, do: @statuses

  def changeset(payment, attrs) do

    payment

    |> cast(attrs, [

      :connect_account_id,

      :stripe_session_id,

      :stripe_payment_intent_id,

      :stripe_customer_id,

      :customer_email,

      :amount_cents,

      :currency,

      :status

    ])

    |> validate_required([

      :connect_account_id,

      :stripe_session_id,

      :amount_cents,

      :currency,

      :status

    ])

    |> unique_constraint(:stripe_session_id)

  end

  ## Queries

  @doc "Totals of succeeded payments, as `{connect_account_id, cents}`."

  def succeeded_total_by_account_query(account_ids) do

    from p in __MODULE__,

      where: p.connect_account_id in ^account_ids and p.status == :succeeded,

      group_by: p.connect_account_id,

      select: {p.connect_account_id, sum(p.amount_cents)}

  end

end

defmodule MyApp.Connect.Stripe do

  @moduledoc """

  The Stripe API surface for Connect — payments made *on behalf of* a seller's

  connected account.

  The key difference from `MyApp.Billing.Stripe.Client` is scope: every call here is

  made against a connected account via the `Stripe-Account` header

  (`connect_opts/1`), and checkout sessions carry an application fee that's

  routed to the platform.

  Same conventions as the billing wrapper: normalized return maps (never raw

  `Stripe.*` structs) and a stub mode that threads the checkout mode + amount

  through the round-trip so the whole flow is clickable without API keys.

  """

  require Logger

  alias MyApp.Accounts.User

  @stub_email "stub-buyer@example.com"

  defp configured? do

    is_binary(Application.get_env(:stripity_stripe, :api_key))

  end

  defp connect_opts(account_id), do: [headers: %{"Stripe-Account" => account_id}]

  ## Accounts

  @doc "Creates a Stripe Connect Express account for a user. Returns `{:ok, account_id}`."

  def create_account(%User{} = user) do

    if configured?() do

      params = %{

        type: "express",

        capabilities: %{

          card_payments: %{requested: true},

          transfers: %{requested: true}

        },

        metadata: %{user_id: user.id}

      }

      case Stripe.Account.create(params) do

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

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

      end

    else

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

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

    end

  end

  @doc """

  Creates an Account Link the seller visits to provide identity, banking, and

  tax info. Returns `{:ok, url}`.

  """

  def create_account_link(account_id, return_url, refresh_url) do

    if configured?() do

      params = %{

        account: account_id,

        type: "account_onboarding",

        return_url: return_url,

        refresh_url: refresh_url

      }

      case Stripe.AccountLink.create(params) do

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

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

      end

    else

      stub_log("create_account_link", account: account_id)

      # The stub redirects straight back to the return url as if onboarding

      # finished, so the local flow completes in one click.

      {:ok, return_url <> "?stub_onboarding=complete"}

    end

  end

  @doc "Fetches onboarding status. Returns `{:ok, %{details_submitted, charges_enabled}}`."

  def fetch_account(account_id) do

    if configured?() do

      case Stripe.Account.retrieve(account_id) do

        {:ok, account} ->

          {:ok,

           %{

             details_submitted: account.details_submitted,

             charges_enabled: account.charges_enabled

           }}

        {:error, err} ->

          log_and_wrap("fetch_account", err)

      end

    else

      stub_log("fetch_account", account: account_id)

      {:ok, %{details_submitted: true, charges_enabled: true}}

    end

  end

  @doc "Disconnects (deletes) a connected account."

  def disconnect_account(nil), do: :ok

  def disconnect_account(account_id) do

    if configured?() do

      case Stripe.Account.delete(account_id) do

        {:ok, _} -> :ok

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

      end

    else

      stub_log("disconnect_account", account: account_id)

      :ok

    end

  end

  ## Checkout

  @doc """

  Creates a subscription-mode Checkout Session on the connected account, with a

  recurring monthly price built inline and `application_fee_percent` routed to

  the platform. Returns `{:ok, %{id, url}}`.

  """

  def create_membership_checkout_session(

        account_id,

        email,

        amount_cents,

        currency,

        fee_percent,

        success_url,

        cancel_url

      ) do

    if configured?() do

      params = %{

        mode: "subscription",

        line_items: [

          %{

            price_data: %{

              currency: currency,

              unit_amount: amount_cents,

              recurring: %{interval: "month"},

              product_data: %{name: "Membership"}

            },

            quantity: 1

          }

        ],

        customer_email: email,

        subscription_data: %{application_fee_percent: fee_percent},

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

        cancel_url: cancel_url

      }

      create_session(params, account_id, "create_membership_checkout_session")

    else

      stub_checkout("sub", amount_cents, success_url)

    end

  end

  @doc """

  Creates a payment-mode Checkout Session on the connected account, with

  `application_fee_amount` (in cents) routed to the platform.

  Returns `{:ok, %{id, url}}`.

  """

  def create_payment_checkout_session(

        account_id,

        email,

        amount_cents,

        currency,

        fee_amount,

        success_url,

        cancel_url

      ) do

    if configured?() do

      params = %{

        mode: "payment",

        line_items: [

          %{

            price_data: %{

              currency: currency,

              unit_amount: amount_cents,

              product_data: %{name: "Payment"}

            },

            quantity: 1

          }

        ],

        customer_email: email,

        payment_intent_data: %{application_fee_amount: fee_amount},

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

        cancel_url: cancel_url

      }

      create_session(params, account_id, "create_payment_checkout_session")

    else

      stub_checkout("pay", amount_cents, success_url)

    end

  end

  defp create_session(params, account_id, action) do

    case Stripe.Checkout.Session.create(params, connect_opts(account_id)) do

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

      {:error, err} -> log_and_wrap(action, err)

    end

  end

  @doc """

  Fetches a Checkout Session on the connected account. Returns a normalized map:

  `%{id, mode, payment_status, customer, customer_email, subscription,

  payment_intent, amount_total}`.

  """

  def fetch_checkout_session(account_id, session_id) do

    if configured?() do

      case Stripe.Checkout.Session.retrieve(session_id, %{}, connect_opts(account_id)) do

        {:ok, session} ->

          {:ok,

           %{

             id: session.id,

             mode: session.mode,

             payment_status: session.payment_status,

             customer: session.customer,

             customer_email: session_email(session),

             subscription: session.subscription,

             payment_intent: session.payment_intent,

             amount_total: session.amount_total

           }}

        {:error, err} ->

          log_and_wrap("fetch_checkout_session", err)

      end

    else

      stub_log("fetch_checkout_session", account: account_id, session: session_id)

      {:ok, stub_session(session_id)}

    end

  end

  defp session_email(session) do

    case session do

      %{customer_details: %{email: email}} when is_binary(email) -> email

      _ -> session.customer_email

    end

  end

  ## Subscriptions

  @doc "Fetches a subscription on the connected account. Returns a normalized map."

  def fetch_subscription(account_id, subscription_id) do

    if configured?() do

      case Stripe.Subscription.retrieve(subscription_id, %{}, connect_opts(account_id)) do

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

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

      end

    else

      stub_log("fetch_subscription", account: account_id, subscription: subscription_id)

      {:ok, stub_subscription(subscription_id)}

    end

  end

  ## Customer Portal

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

  def create_customer_portal_session(account_id, customer_id, return_url) do

    if configured?() do

      params = %{customer: customer_id, return_url: return_url}

      case Stripe.BillingPortal.Session.create(params, connect_opts(account_id)) do

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

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

      end

    else

      stub_log("create_customer_portal_session", account: account_id, customer: customer_id)

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

    end

  end

  ## Normalization

  defp normalize_subscription(sub) do

    %{

      id: sub.id,

      customer: sub.customer,

      status: sub.status,

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

      cancel_at_period_end: Map.get(sub, :cancel_at_period_end) || false,

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

    }

  end

  defp from_unix(nil), do: nil

  defp from_unix(0), do: nil

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

  ## Stub helpers — encode the mode + amount into the session id so the stubbed

  ## return leg can reconstruct a believable session.

  defp stub_checkout(kind, amount_cents, success_url) do

    session_id = "cs_stub_#{kind}_#{amount_cents}"

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

  end

  defp stub_session("cs_stub_sub_" <> amount = session_id) do

    %{

      id: session_id,

      mode: "subscription",

      payment_status: "paid",

      customer: "cus_stub",

      customer_email: @stub_email,

      subscription: "sub_stub_#{amount}",

      payment_intent: nil,

      amount_total: String.to_integer(amount)

    }

  end

  defp stub_session("cs_stub_pay_" <> amount = session_id) do

    %{

      id: session_id,

      mode: "payment",

      payment_status: "paid",

      customer: "cus_stub",

      customer_email: @stub_email,

      subscription: nil,

      payment_intent: "pi_stub_#{amount}",

      amount_total: String.to_integer(amount)

    }

  end

  defp stub_session(session_id) do

    %{

      id: session_id,

      mode: "payment",

      payment_status: "paid",

      customer: "cus_stub",

      customer_email: @stub_email,

      subscription: nil,

      payment_intent: "pi_stub",

      amount_total: 0

    }

  end

  defp stub_subscription(sub_id) do

    %{

      id: sub_id,

      customer: "cus_stub",

      status: "active",

      current_period_end: DateTime.add(DateTime.utc_now(:second), 30, :day),

      cancel_at_period_end: false,

      canceled_at: nil

    }

  end

  ## Logging

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

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

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

  end

  defp log_and_wrap(action, err) do

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

    {:error, "Stripe request failed"}

  end

  defp stub_log(action, meta) do

    Logger.info("[Connect.Stripe stub] #{action} #{inspect(meta)}")

  end

end

defmodule MyApp.Connect.Subscription do

  @moduledoc """

  A recurring membership a buyer holds on a seller's connected account.

  This is the Connect analogue of `MyApp.Billing.Stripe.Subscription`, but it

  lives on a *connected account* rather than the platform, and the buyer is

  identified by email + Stripe Customer instead of an app `User`. It grants no

  license: what buyers get from a seller is up to the seller.

  """

  use Ecto.Schema

  import Ecto.Changeset

  import Ecto.Query

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

  @foreign_key_type :binary_id

  @statuses ~w(active trialing past_due canceled incomplete)a

  schema "connect_subscriptions" do

    field :stripe_subscription_id, :string

    field :stripe_customer_id, :string

    field :customer_email, :string

    field :status, Ecto.Enum, values: @statuses

    field :current_period_end, :utc_datetime

    field :cancel_at_period_end, :boolean, default: false

    field :canceled_at, :utc_datetime

    belongs_to :connect_account, MyApp.Connect.Account

    timestamps(type: :utc_datetime)

  end

  def statuses, do: @statuses

  def changeset(subscription, attrs) do

    subscription

    |> cast(attrs, [

      :connect_account_id,

      :stripe_subscription_id,

      :stripe_customer_id,

      :customer_email,

      :status,

      :current_period_end,

      :cancel_at_period_end,

      :canceled_at

    ])

    |> validate_required([

      :connect_account_id,

      :stripe_subscription_id,

      :stripe_customer_id,

      :status

    ])

    |> unique_constraint(:stripe_subscription_id)

  end

  ## Queries

  @doc "Active membership counts, as `{connect_account_id, count}`."

  def active_count_by_account_query(account_ids) do

    from s in __MODULE__,

      where: s.connect_account_id in ^account_ids,

      where: s.status in [:active, :trialing, :past_due],

      group_by: s.connect_account_id,

      select: {s.connect_account_id, count(s.id)}

  end

  ## View functions

  def active?(%__MODULE__{status: status}), do: status in [:active, :trialing, :past_due]

  def active?(_), do: false

end

defmodule MyAppAdmin.Query.ConnectAccounts do

  alias MyApp.Repo

  alias MyApp.Connect.{Account, Payment, Subscription}

  @doc """

  Returns a paginated, sorted, and filtered list of Connect accounts with

  users preloaded.

  Returns `{:ok, {accounts, meta}}` or `{:error, meta}`.

  """

  def list(params \\ %{}) do

    Flop.validate_and_run(Account.with_user_query(), params, for: Account)

  end

  @doc """

  Active memberships and one-time payment totals for a page of accounts, as

  maps keyed by account id.

  """

  def stats(accounts) do

    ids = Enum.map(accounts, & &1.id)

    %{

      active_members:

        ids |> Subscription.active_count_by_account_query() |> Repo.all() |> Map.new(),

      payments_cents: ids |> Payment.succeeded_total_by_account_query() |> Repo.all() |> Map.new()

    }

  end

  @doc """

  Gets a single account with its user preloaded.

  """

  def get!(id), do: Repo.get!(Account.with_user_query(), id)

end

defmodule MyAppAdminWeb.ConnectComponents do

  @moduledoc """

  Components shared by the admin Connect pages.

  """

  use MyAppAdminWeb, :html

  alias MyApp.Connect.Account

  @doc """

  Renders where a seller is in setting up their Connect account.

  """

  attr :account, Account, required: true

  def account_status(assigns) do

    ~H"""

    <span class={["badge badge-sm", status_class(Account.status(@account))]}>

      {status_label(Account.status(@account))}

    </span>

    """

  end

  defp status_class(:ready), do: "badge-success"

  defp status_class(:onboarding), do: "badge-warning"

  defp status_class(:not_started), do: "badge-ghost"

  defp status_label(:ready), do: "Ready"

  defp status_label(:onboarding), do: "Onboarding"

  defp status_label(:not_started), do: "Not started"

  def format_cents(nil), do: "-"

  def format_cents(cents), do: "$" <> :erlang.float_to_binary(cents / 100, decimals: 2)

end

        <.sidebar_item

          icon="hero-banknotes-mini"

          navigate={~p"/admin/connect"}

          current={@page_selected == :connect}

        >

          Connect

        </.sidebar_item>

defmodule MyAppAdminWeb.ConnectAccountLive.Index do

  use MyAppAdminWeb, :live_view

  import MyAppAdminWeb.ConnectComponents

  alias MyApp.Connect

  alias MyAppAdmin.Query

  def render(assigns) do

    ~H"""

    <Layouts.admin_ui flash={@flash} current_scope={@current_scope} page_selected={:connect}>

      <.header>

        Connect accounts

        <:subtitle>Users getting paid by their own customers. {@meta.total_count} total</:subtitle>

        <:actions>

          <form phx-change="search" class="relative">

            <.icon

              name="hero-magnifying-glass-micro"

              class="absolute left-3 top-1/2 -translate-y-1/2 h-4 w-4 text-base-content/50"

            />

            <input

              type="search"

              name="email"

              value={Filter.get_value(@meta.flop.filters, :email)}

              placeholder="Search by email..."

              phx-debounce="300"

              class="input input-bordered input-sm pl-9 w-64"

            />

          </form>

        </:actions>

      </.header>

      <.data_table

        items={@accounts}

        meta={@meta}

        path={~p"/admin/connect"}

        id="connect-accounts"

        row_click={fn account -> JS.navigate(~p"/admin/connect/#{account}") end}

      >

        <:col :let={account} label="Email" field={:email}>{account.user.email}</:col>

        <:col :let={account} label="Status"><.account_status account={account} /></:col>

        <:col :let={account} label="Members">

          {Map.get(@stats.active_members, account.id, 0)}

        </:col>

        <:col :let={account} label="Monthly revenue">

          {format_cents(

            (account.membership_price_cents || 0) * Map.get(@stats.active_members, account.id, 0)

          )}

        </:col>

        <:col :let={account} label="One-time total">

          {format_cents(Map.get(@stats.payments_cents, account.id, 0))}

        </:col>

        <:col :let={account} label="Fee">

          {MyApp.Billing.Plans.format_percent(Connect.platform_fee_percent(account.user))}

        </:col>

        <:action :let={account}>

          <.link navigate={~p"/admin/connect/#{account}"}>Show</.link>

        </:action>

      </.data_table>

      <.pagination meta={@meta} path={~p"/admin/connect"} />

    </Layouts.admin_ui>

    """

  end

  def mount(_params, _session, socket) do

    {:ok, socket}

  end

  def handle_params(params, _uri, socket) do

    case Query.ConnectAccounts.list(params) do

      {:ok, {accounts, meta}} ->

        {:noreply,

         assign(socket,

           accounts: accounts,

           meta: meta,

           stats: Query.ConnectAccounts.stats(accounts)

         )}