git.squeep.com Git - akkoma/blob - lib/pleroma/repo.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Repo do
   6   use Ecto.Repo,
   7     otp_app: :pleroma,
   8     adapter: Ecto.Adapters.Postgres,
   9     migration_timestamps: [type: :naive_datetime_usec]
  10
  11   import Ecto.Query
  12   require Logger
  13
  14   defmodule Instrumenter do
  15     use Prometheus.EctoInstrumenter
  16   end
  17
  18   @doc """
  19   Dynamically loads the repository url from the
  20   DATABASE_URL environment variable.
  21   """
  22   def init(_, opts) do
  23     {:ok, Keyword.put(opts, :url, System.get_env("DATABASE_URL"))}
  24   end
  25
  26   @doc "find resource based on prepared query"
  27   @spec find_resource(Ecto.Query.t()) :: {:ok, struct()} | {:error, :not_found}
  28   def find_resource(%Ecto.Query{} = query) do
  29     case __MODULE__.one(query) do
  30       nil -> {:error, :not_found}
  31       resource -> {:ok, resource}
  32     end
  33   end
  34
  35   def find_resource(_query), do: {:error, :not_found}
  36
  37   @doc """
  38   Gets association from cache or loads if need
  39
  40   ## Examples
  41
  42     iex> Repo.get_assoc(token, :user)
  43     %User{}
  44
  45   """
  46   @spec get_assoc(struct(), atom()) :: {:ok, struct()} | {:error, :not_found}
  47   def get_assoc(resource, association) do
  48     case __MODULE__.preload(resource, association) do
  49       %{^association => assoc} when not is_nil(assoc) -> {:ok, assoc}
  50       _ -> {:error, :not_found}
  51     end
  52   end
  53
  54   def check_migrations_applied!() do
  55     unless Pleroma.Config.get(
  56              [:i_am_aware_this_may_cause_data_loss, :disable_migration_check],
  57              false
  58            ) do
  59       Ecto.Migrator.with_repo(__MODULE__, fn repo ->
  60         down_migrations =
  61           Ecto.Migrator.migrations(repo)
  62           |> Enum.reject(fn
  63             {:up, _, _} -> true
  64             {:down, _, _} -> false
  65           end)
  66
  67         if length(down_migrations) > 0 do
  68           down_migrations_text =
  69             Enum.map(down_migrations, fn {:down, id, name} -> "- #{name} (#{id})\n" end)
  70
  71           Logger.error(
  72             "The following migrations were not applied:\n#{down_migrations_text}If you want to start Pleroma anyway, set\nconfig :pleroma, :i_am_aware_this_may_cause_data_loss, disable_migration_check: true"
  73           )
  74
  75           raise Pleroma.Repo.UnappliedMigrationsError
  76         end
  77       end)
  78     else
  79       :ok
  80     end
  81   end
  82
  83   def chunk_stream(query, chunk_size) do
  84     # We don't actually need start and end funcitons of resource streaming,
  85     # but it seems to be the only way to not fetch records one-by-one and
  86     # have individual records be the elements of the stream, instead of
  87     # lists of records
  88     Stream.resource(
  89       fn -> 0 end,
  90       fn
  91         last_id ->
  92           query
  93           |> order_by(asc: :id)
  94           |> where([r], r.id > ^last_id)
  95           |> limit(^chunk_size)
  96           |> all()
  97           |> case do
  98             [] ->
  99               {:halt, last_id}
 100
 101             records ->
 102               last_id = List.last(records).id
 103               {records, last_id}
 104           end
 105       end,
 106       fn _ -> :ok end
 107     )
 108   end
 109 end
 110
 111 defmodule Pleroma.Repo.UnappliedMigrationsError do
 112   defexception message: "Unapplied Migrations detected"
 113 end