add oauth flow for restream (#37)

.env.example

@@ -11,6 +11,9 @@ AWS_SECRET_ACCESS_KEY=""
 BUCKET_MEDIA=""
 BUCKET_ML=""
 
+RESTREAM_CLIENT_ID=""
+RESTREAM_CLIENT_SECRET=""
+
 REPLICATE_API_TOKEN=""
 HF_TOKEN=""
 

config/dev.exs

@@ -10,6 +10,10 @@ config :algora, :github,
   client_id: System.get_env("GITHUB_CLIENT_ID"),
   client_secret: System.get_env("GITHUB_CLIENT_SECRET")
 
+config :algora, :restream,
+  client_id: System.get_env("RESTREAM_CLIENT_ID"),
+  client_secret: System.get_env("RESTREAM_CLIENT_SECRET")
+
 config :algora, :event_sink, url: System.get_env("EVENT_SINK_URL")
 
 config :ex_aws,

config/runtime.exs

@@ -94,6 +94,10 @@ if config_env() == :prod do
     client_id: System.fetch_env!("GITHUB_CLIENT_ID"),
     client_secret: System.fetch_env!("GITHUB_CLIENT_SECRET")
 
+  config :algora, :restream,
+    client_id: System.fetch_env!("RESTREAM_CLIENT_ID"),
+    client_secret: System.fetch_env!("RESTREAM_CLIENT_SECRET")
+
   config :algora, :event_sink, url: System.get_env("EVENT_SINK_URL")
 
   config :ex_aws,

lib/algora/accounts.ex

@@ -2,7 +2,7 @@ defmodule Algora.Accounts do
   import Ecto.Query
   import Ecto.Changeset
 
-  alias Algora.Repo
+  alias Algora.{Repo, Restream}
   alias Algora.Accounts.{User, Identity, Destination}
 
   def list_users(opts) do
@@ -62,7 +62,7 @@ defmodule Algora.Accounts do
   ## User registration
 
   @doc """
-  Registers a user from their GithHub information.
+  Registers a user from their GitHub information.
   """
   def register_github_user(primary_email, info, emails, token) do
     if user = get_user_by_provider_email(:github, primary_email) do
@@ -74,6 +74,28 @@ defmodule Algora.Accounts do
     end
   end
 
+  def link_restream_account(user_id, info, tokens) do
+    user = get_user!(user_id)
+
+    identity =
+      from(u in User,
+        join: i in assoc(u, :identities),
+        where: i.provider == ^to_string(:restream) and u.id == ^user_id
+      )
+      |> Repo.one()
+
+    if identity do
+      update_restream_tokens(user, tokens)
+    else
+      {:ok, _} =
+        info
+        |> Identity.restream_oauth_changeset(user_id, tokens)
+        |> Repo.insert()
+
+      {:ok, Repo.preload(user, :identities, force: true)}
+    end
+  end
+
   def get_user_by_provider_email(provider, email) when provider in [:github] do
     query =
       from(u in User,
@@ -113,6 +135,29 @@ defmodule Algora.Accounts do
     {:ok, Repo.preload(user, :identities, force: true)}
   end
 
+  defp update_restream_tokens(%User{} = user, %{token: token, refresh_token: refresh_token}) do
+    identity =
+      Repo.one!(from(i in Identity, where: i.user_id == ^user.id and i.provider == "restream"))
+
+    {:ok, _} =
+      identity
+      |> change()
+      |> put_change(:provider_token, token)
+      |> put_change(:provider_refresh_token, refresh_token)
+      |> Repo.update()
+
+    {:ok, Repo.preload(user, :identities, force: true)}
+  end
+
+  def refresh_restream_tokens(%User{} = user) do
+    identity =
+      Repo.one!(from(i in Identity, where: i.user_id == ^user.id and i.provider == "restream"))
+
+    {:ok, tokens} = Restream.refresh_access_token(identity.provider_refresh_token)
+
+    update_restream_tokens(user, tokens)
+  end
+
   def gen_stream_key(%User{} = user) do
     user =
       Repo.one!(from(u in User, where: u.id == ^user.id))

lib/algora/accounts/identity.ex

@@ -6,11 +6,13 @@ defmodule Algora.Accounts.Identity do
 
   # providers
   @github "github"
+  @restream "restream"
 
-  @derive {Inspect, except: [:provider_token, :provider_meta]}
+  @derive {Inspect, except: [:provider_token, :provider_refresh_token, :provider_meta]}
   schema "identities" do
     field :provider, :string
     field :provider_token, :string
+    field :provider_refresh_token, :string
     field :provider_email, :string
     field :provider_login, :string
     field :provider_name, :string, virtual: true
@@ -45,4 +47,39 @@ defmodule Algora.Accounts.Identity do
     |> validate_required([:provider_token, :provider_email, :provider_name, :provider_id])
     |> validate_length(:provider_meta, max: 10_000)
   end
+
+  @doc """
+  A user changeset for restream oauth.
+  """
+  def restream_oauth_changeset(info, user_id, %{token: token, refresh_token: refresh_token}) do
+    params = %{
+      "provider_token" => token,
+      "provider_refresh_token" => refresh_token,
+      "provider_id" => to_string(info["id"]),
+      "provider_login" => info["username"],
+      "provider_name" => info["username"],
+      "provider_email" => info["email"],
+      "user_id" => user_id
+    }
+
+    %Identity{provider: @restream, provider_meta: %{"user" => info}}
+    |> cast(params, [
+      :provider_token,
+      :provider_refresh_token,
+      :provider_email,
+      :provider_login,
+      :provider_name,
+      :provider_id,
+      :user_id
+    ])
+    |> validate_required([
+      :provider_token,
+      :provider_refresh_token,
+      :provider_email,
+      :provider_name,
+      :provider_id,
+      :user_id
+    ])
+    |> validate_length(:provider_meta, max: 10_000)
+  end
 end

lib/algora/github.ex

@@ -5,7 +5,7 @@ defmodule Algora.Github do
     query =
       URI.encode_query(
         client_id: client_id(),
-        state: random_string(),
+        state: Algora.Util.random_string(),
         scope: "user:email",
         redirect_uri: "#{AlgoraWeb.Endpoint.url()}/oauth/callbacks/github?#{redirect_query}"
       )
@@ -82,18 +82,6 @@ defmodule Algora.Github do
     end
   end
 
-  def random_string do
-    binary = <<
-      System.system_time(:nanosecond)::64,
-      :erlang.phash2({node(), self()})::16,
-      :erlang.unique_integer()::16
-    >>
-
-    binary
-    |> Base.url_encode64()
-    |> String.replace(["/", "+"], "-")
-  end
-
   defp client_id, do: Algora.config([:github, :client_id])
   defp secret, do: Algora.config([:github, :client_secret])
 

lib/algora/restream.ex

@@ -0,0 +1,90 @@
+defmodule Algora.Restream do
+  def authorize_url(state) do
+    query =
+      URI.encode_query(
+        client_id: client_id(),
+        state: state,
+        response_type: "code",
+        redirect_uri: redirect_uri()
+      )
+
+    "https://api.restream.io/login?#{query}"
+  end
+
+  def exchange_access_token(opts) do
+    code = Keyword.fetch!(opts, :code)
+    state = Keyword.fetch!(opts, :state)
+
+    state
+    |> fetch_exchange_response(code)
+    |> fetch_user_info()
+  end
+
+  defp fetch_exchange_response(_state, code) do
+    body =
+      URI.encode_query(%{
+        grant_type: "authorization_code",
+        redirect_uri: redirect_uri(),
+        code: code
+      })
+
+    headers = [
+      {"Content-Type", "application/x-www-form-urlencoded"},
+      {"Authorization", "Basic " <> Base.encode64("#{client_id()}:#{secret()}")}
+    ]
+
+    resp = HTTPoison.post("https://api.restream.io/oauth/token", body, headers)
+
+    with {:ok, %HTTPoison.Response{status_code: 200, body: body}} <- resp,
+         %{"access_token" => token, "refresh_token" => refresh_token} <- Jason.decode!(body) do
+      {:ok, %{token: token, refresh_token: refresh_token}}
+    else
+      {:error, %HTTPoison.Error{reason: reason}} -> {:error, reason}
+      %{} = resp -> {:error, {:bad_response, resp}}
+    end
+  end
+
+  defp fetch_user_info({:error, _reason} = error), do: error
+
+  defp fetch_user_info({:ok, %{token: token} = tokens}) do
+    headers = [{"Authorization", "Bearer #{token}"}]
+
+    case HTTPoison.get("https://api.restream.io/v2/user/profile", headers) do
+      {:ok, %HTTPoison.Response{status_code: 200, body: body}} ->
+        {:ok, %{info: Jason.decode!(body), tokens: tokens}}
+
+      {:ok, %HTTPoison.Response{status_code: status_code, body: body}} ->
+        {:error, {status_code, body}}
+
+      {:error, %HTTPoison.Error{reason: reason}} ->
+        {:error, reason}
+    end
+  end
+
+  def refresh_access_token(refresh_token) do
+    body =
+      URI.encode_query(%{
+        grant_type: "refresh_token",
+        refresh_token: refresh_token
+      })
+
+    headers = [
+      {"Content-Type", "application/x-www-form-urlencoded"},
+      {"Authorization", "Basic " <> Base.encode64("#{client_id()}:#{secret()}")}
+    ]
+
+    resp = HTTPoison.post("https://api.restream.io/oauth/token", body, headers)
+
+    with {:ok, %HTTPoison.Response{status_code: 200, body: body}} <- resp,
+         %{"access_token" => token, "refresh_token" => refresh_token} <- Jason.decode!(body) do
+      {:ok, %{token: token, refresh_token: refresh_token}}
+    else
+      {:error, %HTTPoison.Error{reason: reason}} -> {:error, reason}
+      %{} = resp -> {:error, {:bad_response, resp}}
+    end
+  end
+
+  defp client_id, do: Algora.config([:restream, :client_id])
+  defp secret, do: Algora.config([:restream, :client_secret])
+  defp redirect_uri, do: "#{AlgoraWeb.Endpoint.url()}/oauth/callbacks/restream"
+end

lib/algora/util.ex

@@ -83,4 +83,16 @@ defmodule Algora.Util do
   ]
 
   def is_common_word(s), do: Enum.member?(@common_words, s)
+
+  def random_string do
+    binary = <<
+      System.system_time(:nanosecond)::64,
+      :erlang.phash2({node(), self()})::16,
+      :erlang.unique_integer()::16
+    >>
+
+    binary
+    |> Base.url_encode64()
+    |> String.replace(["/", "+"], "-")
+  end
 end

lib/algora_web/controllers/oauth_callback_controller.ex

@@ -27,7 +27,7 @@ defmodule AlgoraWeb.OAuthCallbackController do
         conn
         |> put_flash(
           :error,
-          "We were unable to fetch the necessary information from your GithHub account"
+          "We were unable to fetch the necessary information from your GitHub account"
         )
         |> redirect(to: "/")
 
@@ -44,6 +44,46 @@ defmodule AlgoraWeb.OAuthCallbackController do
     redirect(conn, to: "/")
   end
 
+  def new(conn, %{"provider" => "restream", "code" => code, "state" => state}) do
+    client = restream_client(conn)
+
+    user_id = get_session(conn, :user_id)
+
+    with {:ok, state} <- verify_session(conn, :restream_state, state),
+         {:ok, info} <- client.exchange_access_token(code: code, state: state),
+         %{info: info, tokens: tokens} = info,
+         {:ok, user} <- Accounts.link_restream_account(user_id, info, tokens) do
+      conn
+      |> put_flash(:info, "Restream account has been linked!")
+      |> AlgoraWeb.UserAuth.log_in_user(user)
+    else
+      {:error, %Ecto.Changeset{} = changeset} ->
+        Logger.debug("failed Restream insert #{inspect(changeset.errors)}")
+
+        conn
+        |> put_flash(
+          :error,
+          "We were unable to fetch the necessary information from your Restream account"
+        )
+        |> redirect(to: "/")
+
+      {:error, reason} ->
+        Logger.debug("failed Restream exchange #{inspect(reason)}")
+
+        conn
+        |> put_flash(:error, "We were unable to contact Restream. Please try again later")
+        |> redirect(to: "/")
+    end
+  end
+
+  defp verify_session(conn, key, token) do
+    if Plug.Crypto.secure_compare(token, get_session(conn, key)) do
+      {:ok, token}
+    else
+      {:error, "#{key} is invalid"}
+    end
+  end
+
   def sign_out(conn, _) do
     AlgoraWeb.UserAuth.log_out_user(conn)
   end
@@ -51,4 +91,8 @@ defmodule AlgoraWeb.OAuthCallbackController do
   defp github_client(conn) do
     conn.assigns[:github_client] || Algora.Github
   end
+
+  defp restream_client(conn) do
+    conn.assigns[:restream_client] || Algora.Restream
+  end
 end

lib/algora_web/controllers/oauth_login_controller.ex

@@ -0,0 +1,17 @@
+defmodule AlgoraWeb.OAuthLoginController do
+  use AlgoraWeb, :controller
+  require Logger
+
+  def new(conn, %{"provider" => "restream"} = params) do
+    if conn.assigns.current_user do
+      state = Algora.Util.random_string()
+
+      conn
+      |> put_session(:user_return_to, params["return_to"])
+      |> put_session(:restream_state, state)
+      |> redirect(external: Algora.Restream.authorize_url(state))
+    else
+      conn |> redirect(to: ~p"/auth/login")
+    end
+  end
+end

lib/algora_web/live/settings_live.ex

@@ -80,6 +80,21 @@ defmodule AlgoraWeb.SettingsLive do
           <.button phx-click="show_add_destination_modal">Add Destination</.button>
         </div>
       </div>
+      <div :if={Mix.env() == :dev} class="space-y-6 bg-white/5 rounded-lg p-6 ring-1 ring-white/15">
+        <.header>
+          Integrations
+          <:subtitle>
+            Connect with other apps
+          </:subtitle>
+        </.header>
+        <div class="space-y-6">
+          <.button>
+            <.link href={"/oauth/login/restream?#{URI.encode_query(return_to: "/channel/settings")}"}>
+              Restream
+            </.link>
+          </.button>
+        </div>
+      </div>
     </div>
     <!-- Add Destination Modal -->
     <.modal :if={@show_add_destination_modal} id="add-destination-modal" show>

lib/algora_web/router.ex

@@ -1,8 +1,7 @@
 defmodule AlgoraWeb.Router do
   use AlgoraWeb, :router
 
-  import AlgoraWeb.UserAuth,
+  import AlgoraWeb.UserAuth, only: [fetch_current_user: 2]
-    only: [redirect_if_user_is_authenticated: 2, fetch_current_user: 2]
 
   pipeline :browser do
     plug :accepts, ["html"]
@@ -23,9 +22,10 @@ defmodule AlgoraWeb.Router do
   end
 
   scope "/", AlgoraWeb do
-    pipe_through [:browser, :redirect_if_user_is_authenticated]
+    pipe_through :browser
 
     get "/oauth/callbacks/:provider", OAuthCallbackController, :new
+    get "/oauth/login/:provider", OAuthLoginController, :new
   end
 
   if Mix.env() in [:dev, :test] do

priv/repo/migrations/20240522125826_add_provider_refresh_token_to_identities.exs

@@ -0,0 +1,9 @@
+defmodule Algora.Repo.Local.Migrations.AddProviderRefreshTokenToIdentities do
+  use Ecto.Migration
+
+  def change do
+    alter table("identities") do
+      add :provider_refresh_token, :string
+    end
+  end
+end