Restructure code base

lib/matrix_server/schema/account.ex

@@ -0,0 +1,112 @@
+defmodule MatrixServer.Account do
+  use Ecto.Schema
+
+  import Ecto.{Changeset, Query}
+
+  alias MatrixServer.{Repo, Account, Device}
+  alias MatrixServerWeb.Request.{Register, Login}
+  alias Ecto.Multi
+
+  @max_mxid_length 255
+
+  @primary_key {:localpart, :string, []}
+  schema "accounts" do
+    field :password_hash, :string, redact: true
+    has_many :devices, Device, foreign_key: :localpart
+    timestamps(updated_at: false)
+  end
+
+  def available?(localpart) when is_binary(localpart) do
+    if Regex.match?(MatrixServer.localpart_regex(), localpart) and
+         String.length(localpart) <= localpart_length() do
+      if Repo.one!(
+           Account
+           |> where([a], a.localpart == ^localpart)
+           |> select([a], count(a))
+         ) == 0 do
+        :ok
+      else
+        {:error, :user_in_use}
+      end
+    else
+      {:error, :invalid_username}
+    end
+  end
+
+  def register(%Register{} = input) do
+    account_params = %{
+      localpart: input.username || MatrixServer.random_string(10, ?a..?z),
+      password_hash: Bcrypt.hash_pwd_salt(input.password)
+    }
+
+    Multi.new()
+    |> Multi.insert(:account, changeset(%Account{}, account_params))
+    |> Multi.insert(:device, fn %{account: account} ->
+      device_params = %{
+        display_name: input.initial_device_display_name,
+        device_id: input.device_id || Device.generate_device_id(account.localpart)
+      }
+
+      Ecto.build_assoc(account, :devices)
+      |> Device.changeset(device_params)
+    end)
+    |> Multi.run(:device_with_access_token, &Device.insert_new_access_token/2)
+  end
+
+  def login(%Login{} = input) do
+    localpart = try_get_localpart(input.identifier.user)
+
+    fn repo ->
+      case repo.one(from a in Account, where: a.localpart == ^localpart) do
+        %Account{password_hash: hash} = account ->
+          if Bcrypt.verify_pass(input.password, hash) do
+            case Device.login(input, account) do
+              {:ok, device} ->
+                device
+
+              {:error, _cs} ->
+                repo.rollback(:forbidden)
+            end
+          else
+            repo.rollback(:forbidden)
+          end
+
+        nil ->
+          repo.rollback(:forbidden)
+      end
+    end
+  end
+
+  def by_access_token(access_token) do
+    Device
+    |> where([d], d.access_token == ^access_token)
+    |> join(:inner, [d], a in assoc(d, :account))
+    |> select([d, a], {a, d})
+    |> Repo.one()
+  end
+
+  def changeset(account, params \\ %{}) do
+    # TODO: fix password_hash in params
+    account
+    |> cast(params, [:localpart, :password_hash])
+    |> validate_required([:localpart, :password_hash])
+    |> validate_length(:password_hash, max: 60)
+    |> validate_format(:localpart, MatrixServer.localpart_regex())
+    |> validate_length(:localpart, max: localpart_length())
+    |> unique_constraint(:localpart, name: :accounts_pkey)
+  end
+
+  defp localpart_length do
+    # Subtract the "@" and ":" in the MXID.
+    @max_mxid_length - 2 - String.length(MatrixServer.server_name())
+  end
+
+  defp try_get_localpart("@" <> rest = user_id) do
+    case String.split(rest, ":", parts: 2) do
+      [localpart, _] -> localpart
+      _ -> user_id
+    end
+  end
+
+  defp try_get_localpart(localpart), do: localpart
+end

lib/matrix_server/schema/alias.ex

@@ -0,0 +1,28 @@
+defmodule MatrixServer.Alias do
+  use Ecto.Schema
+
+  import Ecto.Changeset
+
+  alias MatrixServer.{Repo, Alias, Room}
+  alias Ecto.Changeset
+
+  @primary_key {:alias, :string, []}
+  schema "aliases" do
+    belongs_to :room, Room, foreign_key: :room_id, references: :id, type: :string
+  end
+
+  def create(alias, room_id) do
+    change(%Alias{}, alias: alias, room_id: room_id)
+    |> assoc_constraint(:room)
+    |> unique_constraint(:alias, name: :aliases_pkey)
+    |> Repo.insert()
+  end
+
+  def get_error(%Changeset{errors: [error | _]}), do: get_error(error)
+  def get_error({:alias, {_, [{:constraint, :unique} | _]}}), do: :room_alias_exists
+
+  def get_error({:room, {_, [{:constraint, :assoc} | _]}}),
+    do: {:not_found, "The room was not found."}
+
+  def get_error(_), do: :bad_json
+end

lib/matrix_server/schema/device.ex

@@ -0,0 +1,75 @@
+defmodule MatrixServer.Device do
+  use Ecto.Schema
+
+  import Ecto.{Changeset, Query}
+
+  alias MatrixServer.{Account, Device, Repo}
+  alias MatrixServerWeb.Request.Login
+
+  @primary_key false
+  schema "devices" do
+    field :device_id, :string, primary_key: true
+    field :access_token, :string, redact: true
+    field :display_name, :string
+
+    belongs_to :account, Account,
+      foreign_key: :localpart,
+      references: :localpart,
+      type: :string,
+      primary_key: true
+  end
+
+  def changeset(device, params \\ %{}) do
+    device
+    |> cast(params, [:display_name, :device_id])
+    |> validate_required([:localpart, :device_id])
+    |> unique_constraint([:localpart, :device_id], name: :devices_pkey)
+  end
+
+  def insert_new_access_token(repo, %{
+        device: %Device{localpart: localpart, device_id: device_id} = device
+      }) do
+    access_token = generate_access_token(localpart, device_id)
+
+    device
+    |> change(%{access_token: access_token})
+    |> repo.update()
+  end
+
+  def generate_access_token(localpart, device_id) do
+    Phoenix.Token.encrypt(MatrixServerWeb.Endpoint, "access_token", {localpart, device_id})
+  end
+
+  def generate_device_id(localpart) do
+    # TODO: use random string instead
+    time_string = System.os_time(:millisecond) |> Integer.to_string()
+
+    "#{localpart}_#{time_string}"
+  end
+
+  def login(%Login{} = input, account) do
+    device_id = input.device_id || generate_device_id(account.localpart)
+    access_token = generate_access_token(account.localpart, device_id)
+
+    update_query =
+      from(d in Device)
+      |> update(set: [access_token: ^access_token, device_id: ^device_id])
+      |> then(fn q ->
+        if input.initial_device_display_name do
+          update(q, set: [display_name: ^input.initial_device_display_name])
+        else
+          q
+        end
+      end)
+
+    device_params = %{
+      device_id: device_id,
+      display_name: input.initial_device_display_name
+    }
+
+    Ecto.build_assoc(account, :devices)
+    |> Device.changeset(device_params)
+    |> put_change(:access_token, access_token)
+    |> Repo.insert(on_conflict: update_query, conflict_target: [:localpart, :device_id])
+  end
+end

lib/matrix_server/schema/event.ex

@@ -0,0 +1,360 @@
+defmodule MatrixServer.Event do
+  use Ecto.Schema
+
+  import Ecto.Query
+
+  alias MatrixServer.{Repo, Room, Event, Account, OrderedMap, SigningServer}
+
+  @schema_meta_fields [:__meta__]
+  @primary_key {:event_id, :string, []}
+  schema "events" do
+    field :type, :string
+    field :origin_server_ts, :integer
+    field :state_key, :string
+    field :sender, :string
+    field :content, :map
+    field :prev_events, {:array, :string}
+    field :auth_events, {:array, :string}
+    # TODO: make these database fields eventually?
+    field :signing_keys, :map, virtual: true, default: %{}
+    field :unsigned, :map, virtual: true, default: %{}
+    field :signatures, :map, virtual: true, default: %{}
+    field :hashes, :map, virtual: true, default: %{}
+    belongs_to :room, Room, type: :string
+  end
+
+  def new(%Room{id: room_id}, %Account{localpart: localpart}) do
+    %Event{
+      room_id: room_id,
+      sender: MatrixServer.get_mxid(localpart),
+      event_id: generate_event_id(),
+      origin_server_ts: System.os_time(:millisecond),
+      prev_events: [],
+      auth_events: []
+    }
+  end
+
+  def create_room(room, %Account{localpart: localpart} = creator, room_version, auth_events \\ []) do
+    mxid = MatrixServer.get_mxid(localpart)
+
+    %Event{
+      new(room, creator)
+      | type: "m.room.create",
+        state_key: "",
+        content: %{
+          "creator" => mxid,
+          "room_version" => room_version || MatrixServer.default_room_version()
+        },
+        auth_events: auth_events
+    }
+  end
+
+  def join(room, %Account{localpart: localpart} = sender, auth_events \\ []) do
+    mxid = MatrixServer.get_mxid(localpart)
+
+    %Event{
+      new(room, sender)
+      | type: "m.room.member",
+        state_key: mxid,
+        content: %{
+          "membership" => "join"
+        },
+        auth_events: auth_events
+    }
+  end
+
+  def power_levels(room, %Account{localpart: localpart} = sender, auth_events \\ []) do
+    mxid = MatrixServer.get_mxid(localpart)
+
+    %Event{
+      new(room, sender)
+      | type: "m.room.power_levels",
+        state_key: "",
+        content: %{
+          "ban" => 50,
+          "events" => %{},
+          "events_default" => 0,
+          "invite" => 50,
+          "kick" => 50,
+          "redact" => 50,
+          "state_default" => 50,
+          "users" => %{
+            mxid => 50
+          },
+          "users_default" => 0,
+          "notifications" => %{
+            "room" => 50
+          }
+        },
+        auth_events: auth_events
+    }
+  end
+
+  def name(room, sender, name, auth_events \\ []) do
+    %Event{
+      new(room, sender)
+      | type: "m.room.name",
+        state_key: "",
+        content: %{
+          "name" => name
+        },
+        auth_events: auth_events
+    }
+  end
+
+  def topic(room, sender, topic, auth_events \\ []) do
+    %Event{
+      new(room, sender)
+      | type: "m.room.topic",
+        state_key: "",
+        content: %{
+          "topic" => topic
+        },
+        auth_events: auth_events
+    }
+  end
+
+  def join_rules(room, sender, join_rule, auth_events \\ []) do
+    %Event{
+      new(room, sender)
+      | type: "m.room.join_rules",
+        state_key: "",
+        content: %{
+          "join_rule" => join_rule
+        },
+        auth_events: auth_events
+    }
+  end
+
+  def history_visibility(room, sender, history_visibility, auth_events \\ []) do
+    %Event{
+      new(room, sender)
+      | type: "m.room.history_visibility",
+        state_key: "",
+        content: %{
+          "history_visibility" => history_visibility
+        },
+        auth_events: auth_events
+    }
+  end
+
+  def guest_access(room, sender, guest_access, auth_events \\ []) do
+    %Event{
+      new(room, sender)
+      | type: "m.room.guest_access",
+        state_key: "",
+        content: %{
+          "guest_access" => guest_access
+        },
+        auth_events: auth_events
+    }
+  end
+
+  def generate_event_id do
+    "$" <> MatrixServer.random_string(17) <> ":" <> MatrixServer.server_name()
+  end
+
+  def is_control_event(%Event{type: "m.room.power_levels", state_key: ""}), do: true
+  def is_control_event(%Event{type: "m.room.join_rules", state_key: ""}), do: true
+
+  def is_control_event(%Event{
+        type: "m.room.member",
+        state_key: state_key,
+        sender: sender,
+        content: %{membership: membership}
+      })
+      when sender != state_key and membership in ["leave", "ban"],
+      do: true
+
+  def is_control_event(_), do: false
+
+  def is_state_event(%Event{state_key: state_key}), do: state_key != nil
+
+  # Perform validations that can be done before state resolution.
+  # For example checking the domain of the sender.
+  # We assume that required keys, as well as in the content, is already validated.
+
+  # Rule 1.4 is left to changeset validation.
+  def prevalidate(%Event{
+        type: "m.room.create",
+        prev_events: prev_events,
+        auth_events: auth_events,
+        room_id: room_id,
+        sender: sender
+      }) do
+    # TODO: error check on domains?
+    # TODO: rule 1.3
+
+    # Check rules: 1.1, 1.2
+    prev_events == [] and
+      auth_events == [] and
+      MatrixServer.get_domain(sender) == MatrixServer.get_domain(room_id)
+  end
+
+  def prevalidate(%Event{auth_events: auth_event_ids, prev_events: prev_event_ids} = event) do
+    prev_events =
+      Event
+      |> where([e], e.event_id in ^prev_event_ids)
+      |> Repo.all()
+
+    auth_events =
+      Event
+      |> where([e], e.event_id in ^auth_event_ids)
+      |> Repo.all()
+
+    state_pairs = Enum.map(auth_events, &{&1.type, &1.state_key})
+
+    # Check rules: 2.1, 2.2, 3
+    length(auth_events) == length(auth_event_ids) and
+      length(prev_events) == length(prev_event_ids) and
+      not MatrixServer.has_duplicates?(state_pairs) and
+      valid_auth_events?(event, auth_events) and
+      Enum.find_value(state_pairs, &(&1 == {"m.room.create", ""})) and
+      do_prevalidate(event, auth_events, prev_events)
+  end
+
+  # Rule 4.1 is left to changeset validation.
+  defp do_prevalidate(%Event{type: "m.room.aliases", sender: sender, state_key: state_key}, _, _) do
+    # Check rule: 4.2
+    MatrixServer.get_domain(sender) == MatrixServer.get_domain(state_key)
+  end
+
+  # Rule 5.1 is left to changeset validation.
+  # Rules 5.2.3, 5.2.4, 5.2.5 is left to state resolution.
+  # Check rule: 5.2.1
+  defp do_prevalidate(
+         %Event{type: "m.room.member", content: %{"membership" => "join"}, sender: sender},
+         _,
+         [%Event{type: "m.room.create", state_key: sender}]
+       ),
+       do: true
+
+  # Check rule: 5.2.2
+  defp do_prevalidate(
+         %Event{
+           type: "m.room.member",
+           content: %{"membership" => "join"},
+           sender: sender,
+           state_key: state_key
+         },
+         _,
+         _
+       )
+       when sender != state_key,
+       do: false
+
+  # All other rules will be checked during state resolution.
+  defp do_prevalidate(_, _, _), do: true
+
+  defp valid_auth_events?(
+         %Event{type: type, sender: sender, state_key: state_key, content: content},
+         auth_events
+       ) do
+    Enum.all?(auth_events, fn
+      %Event{type: "m.room.create", state_key: ""} ->
+        true
+
+      %Event{type: "m.room.power_levels", state_key: ""} ->
+        true
+
+      %Event{type: "m.room.member", state_key: ^sender} ->
+        true
+
+      %Event{type: auth_type, state_key: auth_state_key} ->
+        if type == "m.room.member" do
+          %{"membership" => membership} = content
+
+          (auth_type == "m.room.member" and auth_state_key == state_key) or
+            (membership in ["join", "invite"] and auth_type == "m.room.join_rules" and
+               auth_state_key == "") or
+            (membership == "invite" and auth_type == "m.room.third_party_invite" and
+               auth_state_key == "")
+        else
+          false
+        end
+    end)
+  end
+
+  def sign(event) do
+    content_hash = calculate_content_hash(event)
+
+    event
+    |> Map.put(:hashes, %{"sha256" => content_hash})
+    |> redact()
+    |> SigningServer.sign_object()
+  end
+
+  defp calculate_content_hash(event) do
+    result =
+      event
+      |> to_map()
+      |> Map.drop([:unsigned, :signature, :hashes])
+      |> OrderedMap.from_map()
+      |> Jason.encode()
+
+    case result do
+      {:ok, json} ->
+        :crypto.hash(:sha256, json)
+        |> MatrixServer.encode_unpadded_base64()
+
+      error ->
+        error
+    end
+  end
+
+  defp redact(%Event{type: type, content: content} = event) do
+    redacted_event =
+      event
+      |> to_map()
+      |> Map.take([
+        :event_id,
+        :type,
+        :room_id,
+        :sender,
+        :state_key,
+        :content,
+        :hashes,
+        :signatures,
+        :depth,
+        :prev_events,
+        :prev_state,
+        :auth_events,
+        :origin,
+        :origin_server_ts,
+        :membership
+      ])
+
+    %{redacted_event | content: redact_content(type, content)}
+  end
+
+  defp redact_content("m.room.member", content), do: Map.take(content, ["membership"])
+  defp redact_content("m.room.create", content), do: Map.take(content, ["creator"])
+  defp redact_content("m.room.join_rules", content), do: Map.take(content, ["join_rule"])
+  defp redact_content("m.room.aliases", content), do: Map.take(content, ["aliases"])
+
+  defp redact_content("m.room.history_visibility", content),
+    do: Map.take(content, ["history_visibility"])
+
+  defp redact_content("m.room.power_levels", content),
+    do:
+      Map.take(content, [
+        "ban",
+        "events",
+        "events_default",
+        "kick",
+        "redact",
+        "state_default",
+        "users",
+        "users_default"
+      ])
+
+  # https://stackoverflow.com/questions/41523762/41671211
+  def to_map(event) do
+    association_fields = event.__struct__.__schema__(:associations)
+    waste_fields = association_fields ++ @schema_meta_fields
+
+    event
+    |> Map.from_struct()
+    |> Map.drop(waste_fields)
+  end
+end

lib/matrix_server/schema/room.ex

@@ -0,0 +1,59 @@
+defmodule MatrixServer.Room do
+  use Ecto.Schema
+
+  import Ecto.Changeset
+  import Ecto.Query
+
+  alias MatrixServer.{Repo, Room, Event, Alias, RoomServer}
+  alias MatrixServerWeb.Request.CreateRoom
+
+  @primary_key {:id, :string, []}
+  schema "rooms" do
+    field :visibility, Ecto.Enum, values: [:public, :private]
+    field :state, {:array, {:array, :string}}
+    field :forward_extremities, {:array, :string}
+    has_many :events, Event, foreign_key: :event_id
+    has_many :aliases, Alias, foreign_key: :room_id
+  end
+
+  def changeset(room, params \\ %{}) do
+    cast(room, params, [:visibility])
+  end
+
+  def create_changeset(%CreateRoom{} = input) do
+    visibility = input.visibility || :public
+
+    %Room{id: generate_room_id()}
+    |> changeset(%{visibility: visibility})
+  end
+
+  def generate_room_id do
+    "!" <> MatrixServer.random_string(18) <> ":" <> MatrixServer.server_name()
+  end
+
+  def update_forward_extremities(
+        %Event{
+          event_id: event_id,
+          prev_events: prev_event_ids
+        },
+        %Room{id: room_id, forward_extremities: forward_extremities}
+      ) do
+    new_forward_extremities = [event_id | forward_extremities -- prev_event_ids]
+
+    # TODO: might not need to save to DB here.
+    {_, [room]} =
+      from(r in Room, where: r.id == ^room_id, select: r)
+      |> Repo.update_all(set: [forward_extremities: new_forward_extremities])
+
+    room
+  end
+
+  def create(account, input) do
+    with {:ok, %Room{id: room_id}} <- Repo.insert(create_changeset(input)),
+         {:ok, pid} <- RoomServer.get_room_server(room_id) do
+      RoomServer.create_room(pid, account, input)
+    else
+      _ -> {:error, :unknown}
+    end
+  end
+end