added notification constraints

lib/pleroma/migration_helper/notification_backfill.ex

@@ -19,13 +19,13 @@ def fill_in_notification_types do
     query
     |> Repo.chunk_stream(100)
     |> Enum.each(fn notification ->
-      type =
-        notification.activity
-        |> type_from_activity()
+      if notification.activity do
+        type = type_from_activity(notification.activity)
 
         notification
         |> Ecto.Changeset.change(%{type: type})
         |> Repo.update()
+      end
     end)
   end
 
@@ -72,8 +72,7 @@ defp type_from_activity(%{data: %{"type" => type}} = activity) do
         "pleroma:emoji_reaction"
 
       "Create" ->
-        activity
-        |> type_from_activity_object()
+        type_from_activity_object(activity)
 
       t ->
         raise "No notification type for activity type #{t}"

lib/pleroma/repo.ex

@@ -49,7 +49,7 @@ def get_assoc(resource, association) do
     end
   end
 
-  def chunk_stream(query, chunk_size) do
+  def chunk_stream(query, chunk_size, returns_as \\ :one) do
     # We don't actually need start and end funcitons of resource streaming,
     # but it seems to be the only way to not fetch records one-by-one and
     # have individual records be the elements of the stream, instead of
@@ -69,7 +69,12 @@ def chunk_stream(query, chunk_size) do
 
             records ->
               last_id = List.last(records).id
+
+              if returns_as == :one do
                 {records, last_id}
+              else
+                {[records], last_id}
+              end
           end
       end,
       fn _ -> :ok end

priv/repo/migrations/20200914105638_delete_notification_without_activity.exs

@@ -0,0 +1,30 @@
+defmodule Pleroma.Repo.Migrations.DeleteNotificationWithoutActivity do
+  use Ecto.Migration
+
+  import Ecto.Query
+  alias Pleroma.Repo
+
+  def up do
+    from(
+      q in Pleroma.Notification,
+      left_join: c in assoc(q, :activity),
+      select: %{id: type(q.id, :integer)},
+      where: is_nil(c.id)
+    )
+    |> Repo.chunk_stream(1_000, :bacthes)
+    |> Stream.each(fn records ->
+      notification_ids = Enum.map(records, fn %{id: id} -> id end)
+
+      Repo.delete_all(
+        from(n in "notifications",
+          where: n.id in ^notification_ids
+        )
+      )
+    end)
+    |> Stream.run()
+  end
+
+  def down do
+    :ok
+  end
+end

priv/repo/migrations/20200914105800_add_notification_constraints.exs

@@ -0,0 +1,23 @@
+defmodule Pleroma.Repo.Migrations.AddNotificationConstraints do
+  use Ecto.Migration
+
+  def up do
+    drop(constraint(:notifications, "notifications_activity_id_fkey"))
+
+    alter table(:notifications) do
+      modify(:activity_id, references(:activities, type: :uuid, on_delete: :delete_all),
+        null: false
+      )
+    end
+  end
+
+  def down do
+    drop(constraint(:notifications, "notifications_activity_id_fkey"))
+
+    alter table(:notifications) do
+      modify(:activity_id, references(:activities, type: :uuid, on_delete: :delete_all),
+        null: true
+      )
+    end
+  end
+end