{"* * * * *", MyApp.Workers.EnqueueStripeEventPolling}

# Stripe Connect: the platform's application fee on each transaction made on

# behalf of a connected account.

config :my_app, :connect, platform_fee_percent: 10.0

  def source_key({:connect, account_id}) when is_binary(account_id), do: "connect:#{account_id}"

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/0` on every transaction.

  This builds directly on the Stripe billing feature: it reuses Oban, the Stripe

  wrapper conventions, and the event poller — the poller just fans out an extra

  job per connected account and routes those events here via

  `handle_stripe_event/2`.

  """

  import Ecto.Query, warn: false

  alias MyApp.Accounts.User

  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 application fee, as a percentage of each transaction."

  def platform_fee_percent do

    :my_app

    |> Application.get_env(:connect, [])

    |> Keyword.get(:platform_fee_percent, 10.0)

  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)

  ## 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(),

        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/0`.

  """

  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() / 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

  @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

  @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

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

  @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

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` 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

  ## Events (consumed by the poller, scoped to one connected account)

  @doc """

  Lists events for one connected account, newest first. Same contract as

  `MyApp.Billing.Stripe.list_events/1`, but scoped via the `Stripe-Account`

  header. In stub mode returns `{:ok, []}`.

  """

  def list_events(account_id, opts \\ []) do

    if configured?() do

      params = %{limit: Keyword.get(opts, :limit, 100)}

      params =

        case Keyword.get(opts, :ending_before) do

          nil -> params

          cursor when is_binary(cursor) -> Map.put(params, :ending_before, cursor)

        end

      case Stripe.Event.list(params, connect_opts(account_id)) do

        {:ok, %{data: events}} -> {:ok, Enum.map(events, &normalize_event/1)}

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

      end

    else

      stub_log("list_events", account: account_id, opts: opts)

      {:ok, []}

    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 normalize_event(%Stripe.Event{} = event) do

    %{

      "id" => event.id,

      "type" => event.type,

      "created" => event.created,

      "account" => Map.get(event, :account),

      "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(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.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`.

  """

  use Ecto.Schema

  import Ecto.Changeset

  @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

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

  def active?(_), do: false

end

defmodule MyApp.Workers.EnqueueStripeEventPolling do

  @moduledoc """

  Cron worker — fires every minute and enqueues one `StripeEventsPoller` job for

  the platform event stream plus one for each connected account that has

  completed onboarding.

  Fanning out into separate jobs means the platform poll and the per-account

  polls run in parallel, one slow/failed account never blocks another, and the

  poller's `unique` clause dedupes overlapping cycles per scope.

  """

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

  import Ecto.Query

  alias MyApp.Connect.Account

  alias MyApp.Repo

  alias MyApp.Workers.StripeEventsPoller

  @impl Oban.Worker

  def perform(_job) do

    StripeEventsPoller.new(%{scope: "platform"}) |> Oban.insert!()

    for account_id <- connected_account_ids() do

      StripeEventsPoller.new(%{scope: "connect", account_id: account_id}) |> Oban.insert!()

    end

    :ok

  end

  defp connected_account_ids do

    from(a in Account,

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

      select: a.stripe_account_id,

      distinct: true

    )

    |> Repo.all()

  end

end

  Drains a Stripe `/v1/events` stream and dispatches each event to the right

  context. One worker handles two scopes, selected by job args:

    * `%{"scope" => "platform"}` — events on the platform account (subscription

      billing), dispatched to `MyApp.Billing.handle_stripe_event/1`.

    * `%{"scope" => "connect", "account_id" => "acct_..."}` — events on one

      connected account (Connect payments), dispatched to

      `MyApp.Connect.handle_stripe_event/2`.

  `MyApp.Workers.EnqueueStripeEventPolling` fans these out every minute — one

  platform job plus one per connected account — so a slow account never blocks

  the others.

  A `MyApp.Billing.Stripe.Checkpoint` row per scope remembers the last processed

  event id, so each run fetches only new events. The first run *bootstraps*:

  it records the latest event id without processing it, so we don't replay

  history.

  The `unique` clause covers the 60s cron interval (doubled for safety). Because

  it keys on `:args`, the platform job and each per-account job dedupe

  independently.

  alias MyApp.Connect

  def perform(%Oban.Job{args: %{"scope" => "connect", "account_id" => account_id}})

      when is_binary(account_id),

      do: poll({:connect, account_id})

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

  defp poll(scope) do

    source = Checkpoint.source_key(scope)

      nil -> bootstrap(scope, source)

      %Checkpoint{last_event_id: nil} -> bootstrap(scope, source)

      %Checkpoint{last_event_id: last_id} -> drain(scope, source, last_id)

  defp bootstrap(scope, source) do

    case list_events(scope, limit: 1) do

  defp drain(scope, source, last_event_id) do

    case list_events(scope, ending_before: last_event_id) do

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

  # Each scope reads from its own (platform- or connect-scoped) Stripe wrapper.

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

  defp list_events({:connect, account_id}, opts), do: Connect.Stripe.list_events(account_id, opts)

  # ...and dispatches to its own context.

  defp dispatch(:platform, event), do: Billing.handle_stripe_event(event)

  defp dispatch({:connect, account_id}, event), do: Connect.handle_stripe_event(event, account_id)

defmodule MyAppWeb.ConnectCheckoutController do

  @moduledoc """

  The return leg of a *buyer's* checkout on a seller's connected account.

  This is public — buyers don't have app accounts. Stripe redirects here with

  `?session_id=...`; we sync the session (recording a membership or a one-time

  payment) and send the buyer back to the seller's public page.

  """

  use MyAppWeb, :controller

  alias MyApp.Connect

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

    account = Connect.get_account(account_id)

    session_id = params["session_id"]

    cond do

      is_nil(account) ->

        conn

        |> put_flash(:error, "That page is no longer available.")

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

      is_nil(session_id) ->

        redirect(conn, to: ~p"/c/#{account_id}")

      true ->

        flash = sync_flash(Connect.sync_checkout_session(account, session_id))

        conn

        |> put_flash(elem(flash, 0), elem(flash, 1))

        |> redirect(to: ~p"/c/#{account_id}")

    end

  end

  defp sync_flash({:ok, %Connect.Subscription{}}), do: {:info, "You're subscribed — thank you!"}

  defp sync_flash({:ok, %Connect.Payment{}}), do: {:info, "Payment received — thank you!"}

  defp sync_flash({:ok, :pending}),

    do: {:info, "Payment is processing. It'll be confirmed shortly."}

  defp sync_flash({:error, _}),

    do: {:error, "We couldn't confirm your payment. Please contact the seller."}

end

defmodule MyAppWeb.ConnectController do

  @moduledoc """

  The return and refresh legs of Stripe Connect onboarding for the seller.

  Onboarding itself is kicked off from `MyAppWeb.ConnectLive` (it redirects the

  seller to Stripe). Stripe then sends them back to one of these:

    * `return` — onboarding finished (or was abandoned); we re-read the status

      from Stripe and bounce to the dashboard.

    * `refresh` — the onboarding link expired; we mint a fresh one.

  """

  use MyAppWeb, :controller

  alias MyApp.Connect

  def return(conn, _params) do

    user = conn.assigns.current_scope.user

    case Connect.get_account(user) do

      nil ->

        redirect(conn, to: ~p"/connect")

      account ->

        {:ok, account} = Connect.refresh_onboarding_status(account)

        flash =

          if account.onboarding_complete do

            {:info, "Stripe is connected. Set a price to start accepting payments."}

          else

            {:info,

             "Stripe onboarding isn't finished yet — complete the steps to enable payouts."}

          end

        conn

        |> put_flash(elem(flash, 0), elem(flash, 1))

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

    end

  end

  def refresh(conn, _params) do

    user = conn.assigns.current_scope.user

    return_url = url(~p"/connect/return")

    refresh_url = url(~p"/connect/refresh")

    with account when not is_nil(account) <- Connect.get_account(user),

         {:ok, onboarding_url} <- Connect.refresh_account_link(account, return_url, refresh_url) do

      redirect(conn, external: onboarding_url)

    else

      _ ->

        conn

        |> put_flash(:error, "Couldn't resume Stripe onboarding. Please try again.")

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

    end

  end

end

defmodule MyAppWeb.Connect.SubscribeLive do

  @moduledoc """

  The public payment page for a seller at `/c/:account_id`.

  Anyone — no app account required — enters their email and either subscribes to

  a recurring membership or makes a one-time payment. Both redirect out to Stripe

  Checkout on the seller's connected account; buyers come back through

  `MyAppWeb.ConnectCheckoutController`.

  """

  use MyAppWeb, :live_view

  alias MyApp.Connect

  alias MyApp.Connect.Account

  @impl true

  def mount(%{"account_id" => account_id}, _session, socket) do

    account = Connect.get_account(account_id)

    {:ok,

     socket

     |> assign(:page_title, "Subscribe")

     |> assign(:account, account)

     |> assign(:available?, account != nil and Account.ready?(account))}

  end

  @impl true

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

    account = socket.assigns.account

    success_url = url(~p"/c/#{account.id}/return")

    cancel_url = url(~p"/c/#{account.id}")

    result =

      case mode do

        "membership" ->

          Connect.create_membership_checkout(account, email, success_url, cancel_url)

        "payment" ->

          Connect.create_payment_checkout(account, email, success_url, cancel_url)

        _ ->

          {:error, :invalid_mode}

      end

    case result do

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

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

      {:error, _reason} ->

        {:noreply, put_flash(socket, :error, "Couldn't start checkout. Please try again.")}

    end

  end

  @impl true

  def render(assigns) do

    ~H"""

    <Layouts.app flash={@flash} current_scope={@current_scope}>

      <div :if={not @available?} class="mt-8 text-center">

        <p class="text-base-content/70">This payment page isn't available.</p>

      </div>

      <div :if={@available?} class="mt-8">

        <.header>Support this creator</.header>

        <form phx-submit="checkout" class="mt-6 space-y-5">

          <label class="block">

            <span class="text-sm font-medium">Your email</span>

            <input

              type="email"

              name="email"

              required

              placeholder="you@example.com"

              class="input mt-1 w-full"

            />

          </label>

          <div class="space-y-3">

            <button

              :if={Account.membership_enabled?(@account)}

              type="submit"

              name="mode"

              value="membership"

              class="btn btn-primary w-full"

              phx-disable-with="Redirecting…"

            >

              Subscribe — {money(@account.membership_price_cents)} / month

            </button>

            <button

              :if={Account.one_time_enabled?(@account)}

              type="submit"

              name="mode"

              value="payment"

              class="btn btn-soft btn-primary w-full"

              phx-disable-with="Redirecting…"

            >

              Pay once — {money(@account.one_time_price_cents)}

            </button>

          </div>

        </form>

      </div>

    </Layouts.app>

    """

  end

  defp money(cents) when is_integer(cents),

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

end

defmodule MyAppWeb.ConnectLive do

  @moduledoc """

  The seller's Connect dashboard at `/connect`: onboard a Stripe account, set

  membership/one-time prices, share the public payment page, and see subscribers,

  payments, and revenue.

  """

  use MyAppWeb, :live_view

  alias MyApp.Connect

  alias MyApp.Connect.Account

  @impl true

  def mount(_params, _session, socket) do

    {:ok, assign_dashboard(socket)}

  end

  defp assign_dashboard(socket) do

    user = socket.assigns.current_scope.user

    account = Connect.get_account(user)

    socket

    |> assign(:page_title, "Payments")

    |> assign(:account, account)

    |> assign(:fee_percent, Connect.platform_fee_percent())

    |> assign_account_data(account)

  end

  defp assign_account_data(socket, %Account{} = account) do

    if Account.ready?(account) do

      socket

      |> assign(:subscriptions, Connect.list_subscriptions(account))

      |> assign(:payments, Connect.list_payments(account))

      |> assign(:active_count, Connect.count_active_subscriptions(account))

      |> assign(:mrr_cents, Connect.monthly_revenue_cents(account))

      |> assign(:payments_total_cents, Connect.total_payments_cents(account))

    else

      socket

      |> assign(:subscriptions, [])

      |> assign(:payments, [])

      |> assign(:active_count, 0)

      |> assign(:mrr_cents, 0)

      |> assign(:payments_total_cents, 0)

    end

  end

  defp assign_account_data(socket, nil), do: assign_account_data(socket, %Account{})

  @impl true

  def handle_event("onboard", _params, socket) do

    user = socket.assigns.current_scope.user

    return_url = url(~p"/connect/return")

    refresh_url = url(~p"/connect/refresh")

    case Connect.start_onboarding(user, return_url, refresh_url) do

      {:ok, _account, onboarding_url} ->

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

      {:error, _reason} ->

        {:noreply,

         put_flash(socket, :error, "Couldn't start Stripe onboarding. Please try again.")}

    end

  end

  def handle_event("refresh_status", _params, socket) do

    {:ok, _account} = Connect.refresh_onboarding_status(socket.assigns.account)

    {:noreply, assign_dashboard(socket)}

  end

  def handle_event("disconnect", _params, socket) do

    {:ok, _account} = Connect.disconnect(socket.assigns.account)

    {:noreply,

     socket

     |> put_flash(:info, "Stripe disconnected.")

     |> assign_dashboard()}

  end

  def handle_event("set_membership_price", %{"price" => price}, socket) do

    update_price(socket, price, &Connect.set_membership_price/2, "Membership price updated.")

  end

  def handle_event("set_one_time_price", %{"price" => price}, socket) do

    update_price(socket, price, &Connect.set_one_time_price/2, "One-time price updated.")

  end

  defp update_price(socket, price, fun, ok_message) do

    case dollars_to_cents(price) do

      nil ->

        {:noreply, put_flash(socket, :error, "Enter a valid amount.")}

      cents ->

        case fun.(socket.assigns.account, cents) do

          {:ok, _account} ->

            {:noreply,

             socket

             |> put_flash(:info, ok_message)

             |> assign_dashboard()}

          {:error, _changeset} ->

            {:noreply, put_flash(socket, :error, "Enter a valid amount.")}

        end

    end

  end

  @impl true

  def render(assigns) do

    ~H"""

    <Layouts.app flash={@flash} current_scope={@current_scope}>

      <.header>

        Payments

        <:subtitle>Get paid by your audience with Stripe Connect.</:subtitle>

      </.header>

      <%= cond do %>

        <% @account == nil or @account.stripe_account_id == nil -> %>

          <.not_started fee_percent={@fee_percent} />

        <% Account.onboarding_in_progress?(@account) -> %>

          <.onboarding_incomplete />

        <% true -> %>

          <.dashboard

            account={@account}

            fee_percent={@fee_percent}

            subscriptions={@subscriptions}

            payments={@payments}

            active_count={@active_count}

            mrr_cents={@mrr_cents}

            payments_total_cents={@payments_total_cents}

          />

      <% end %>

    </Layouts.app>

    """

  end

  attr :fee_percent, :float, required: true

  defp not_started(assigns) do

    ~H"""

    <section class="border-base-300 mt-8 border p-5 sm:p-6">

      <p class="font-medium">Start accepting payments</p>

      <p class="text-base-content/70 mt-1 text-sm">

        Connect a Stripe account to charge your audience for memberships or one-time payments.

        We collect a {@fee_percent}% platform fee; the rest is paid out to you by Stripe.

      </p>

      <.button

        variant="primary"

        phx-click="onboard"

        phx-disable-with="Redirecting…"

        class="btn btn-primary mt-4"

      >

        Set up payouts with Stripe

      </.button>

    </section>

    """

  end

  defp onboarding_incomplete(assigns) do

    ~H"""

    <section class="border-base-300 mt-8 border p-5 sm:p-6">

      <p class="font-medium">Onboarding in progress</p>

      <p class="text-base-content/70 mt-1 text-sm">

        Your Stripe account isn't fully set up yet. Finish onboarding to enable payouts.

      </p>

      <div class="mt-4 flex flex-wrap gap-3">

        <.button href={~p"/connect/refresh"} variant="primary">Resume onboarding</.button>

        <.button phx-click="refresh_status">I've finished — refresh status</.button>