projects / akkoma / blob
? search:


a38f9a3c8a101baa4c5d6cc5f3e165a852386dd4
[akkoma] / lib / pleroma / web / activity_pub / activity_pub.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.Web.ActivityPub.ActivityPub do
6 alias Pleroma.Activity
7 alias Pleroma.Activity.Ir.Topics
8 alias Pleroma.Config
9 alias Pleroma.Constants
10 alias Pleroma.Conversation
11 alias Pleroma.Conversation.Participation
12 alias Pleroma.Notification
13 alias Pleroma.Object
14 alias Pleroma.Object.Containment
15 alias Pleroma.Object.Fetcher
16 alias Pleroma.Pagination
17 alias Pleroma.Repo
18 alias Pleroma.Upload
19 alias Pleroma.User
20 alias Pleroma.Web.ActivityPub.MRF
21 alias Pleroma.Web.ActivityPub.Transmogrifier
22 alias Pleroma.Web.ActivityPub.Utils
23 alias Pleroma.Web.Streamer
24 alias Pleroma.Web.WebFinger
25 alias Pleroma.Workers.BackgroundWorker
26
27 import Ecto.Query
28 import Pleroma.Web.ActivityPub.Utils
29 import Pleroma.Web.ActivityPub.Visibility
30
31 require Logger
32 require Pleroma.Constants
33
34 # For Announce activities, we filter the recipients based on following status for any actors
35 # that match actual users. See issue #164 for more information about why this is necessary.
36 defp get_recipients(%{"type" => "Announce"} = data) do
37 to = Map.get(data, "to", [])
38 cc = Map.get(data, "cc", [])
39 bcc = Map.get(data, "bcc", [])
40 actor = User.get_cached_by_ap_id(data["actor"])
41
42 recipients =
43 Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
44 case User.get_cached_by_ap_id(recipient) do
45 nil -> true
46 user -> User.following?(user, actor)
47 end
48 end)
49
50 {recipients, to, cc}
51 end
52
53 defp get_recipients(%{"type" => "Create"} = data) do
54 to = Map.get(data, "to", [])
55 cc = Map.get(data, "cc", [])
56 bcc = Map.get(data, "bcc", [])
57 actor = Map.get(data, "actor", [])
58 recipients = [to, cc, bcc, [actor]] |> Enum.concat() |> Enum.uniq()
59 {recipients, to, cc}
60 end
61
62 defp get_recipients(data) do
63 to = Map.get(data, "to", [])
64 cc = Map.get(data, "cc", [])
65 bcc = Map.get(data, "bcc", [])
66 recipients = Enum.concat([to, cc, bcc])
67 {recipients, to, cc}
68 end
69
70 defp check_actor_is_active(actor) do
71 if not is_nil(actor) do
72 with user <- User.get_cached_by_ap_id(actor),
73 false <- user.deactivated do
74 true
75 else
76 _e -> false
77 end
78 else
79 true
80 end
81 end
82
83 defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
84 limit = Config.get([:instance, :remote_limit])
85 String.length(content) <= limit
86 end
87
88 defp check_remote_limit(_), do: true
89
90 def increase_note_count_if_public(actor, object) do
91 if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
92 end
93
94 def decrease_note_count_if_public(actor, object) do
95 if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
96 end
97
98 def increase_replies_count_if_reply(%{
99 "object" => %{"inReplyTo" => reply_ap_id} = object,
100 "type" => "Create"
101 }) do
102 if is_public?(object) do
103 Object.increase_replies_count(reply_ap_id)
104 end
105 end
106
107 def increase_replies_count_if_reply(_create_data), do: :noop
108
109 def decrease_replies_count_if_reply(%Object{
110 data: %{"inReplyTo" => reply_ap_id} = object
111 }) do
112 if is_public?(object) do
113 Object.decrease_replies_count(reply_ap_id)
114 end
115 end
116
117 def decrease_replies_count_if_reply(_object), do: :noop
118
119 def increase_poll_votes_if_vote(%{
120 "object" => %{"inReplyTo" => reply_ap_id, "name" => name},
121 "type" => "Create",
122 "actor" => actor
123 }) do
124 Object.increase_vote_count(reply_ap_id, name, actor)
125 end
126
127 def increase_poll_votes_if_vote(_create_data), do: :noop
128
129 @spec persist(map(), keyword()) :: {:ok, Activity.t() | Object.t()}
130 def persist(object, meta) do
131 with local <- Keyword.fetch!(meta, :local),
132 {recipients, _, _} <- get_recipients(object),
133 {:ok, activity} <-
134 Repo.insert(%Activity{
135 data: object,
136 local: local,
137 recipients: recipients,
138 actor: object["actor"]
139 }) do
140 {:ok, activity, meta}
141 end
142 end
143
144 @spec insert(map(), boolean(), boolean(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
145 def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
146 with nil <- Activity.normalize(map),
147 map <- lazy_put_activity_defaults(map, fake),
148 true <- bypass_actor_check || check_actor_is_active(map["actor"]),
149 {_, true} <- {:remote_limit_error, check_remote_limit(map)},
150 {:ok, map} <- MRF.filter(map),
151 {recipients, _, _} = get_recipients(map),
152 {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
153 {:containment, :ok} <- {:containment, Containment.contain_child(map)},
154 {:ok, map, object} <- insert_full_object(map) do
155 {:ok, activity} =
156 Repo.insert(%Activity{
157 data: map,
158 local: local,
159 actor: map["actor"],
160 recipients: recipients
161 })
162
163 # Splice in the child object if we have one.
164 activity =
165 if not is_nil(object) do
166 Map.put(activity, :object, object)
167 else
168 activity
169 end
170
171 BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
172
173 {:ok, activity}
174 else
175 %Activity{} = activity ->
176 {:ok, activity}
177
178 {:fake, true, map, recipients} ->
179 activity = %Activity{
180 data: map,
181 local: local,
182 actor: map["actor"],
183 recipients: recipients,
184 id: "pleroma:fakeid"
185 }
186
187 Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
188 {:ok, activity}
189
190 error ->
191 {:error, error}
192 end
193 end
194
195 def notify_and_stream(activity) do
196 Notification.create_notifications(activity)
197
198 conversation = create_or_bump_conversation(activity, activity.actor)
199 participations = get_participations(conversation)
200 stream_out(activity)
201 stream_out_participations(participations)
202 end
203
204 defp create_or_bump_conversation(activity, actor) do
205 with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
206 %User{} = user <- User.get_cached_by_ap_id(actor),
207 Participation.mark_as_read(user, conversation) do
208 {:ok, conversation}
209 end
210 end
211
212 defp get_participations({:ok, conversation}) do
213 conversation
214 |> Repo.preload(:participations, force: true)
215 |> Map.get(:participations)
216 end
217
218 defp get_participations(_), do: []
219
220 def stream_out_participations(participations) do
221 participations =
222 participations
223 |> Repo.preload(:user)
224
225 Streamer.stream("participation", participations)
226 end
227
228 def stream_out_participations(%Object{data: %{"context" => context}}, user) do
229 with %Conversation{} = conversation <- Conversation.get_for_ap_id(context),
230 conversation = Repo.preload(conversation, :participations),
231 last_activity_id =
232 fetch_latest_activity_id_for_context(conversation.ap_id, %{
233 "user" => user,
234 "blocking_user" => user
235 }) do
236 if last_activity_id do
237 stream_out_participations(conversation.participations)
238 end
239 end
240 end
241
242 def stream_out_participations(_, _), do: :noop
243
244 def stream_out(%Activity{data: %{"type" => data_type}} = activity)
245 when data_type in ["Create", "Announce", "Delete"] do
246 activity
247 |> Topics.get_activity_topics()
248 |> Streamer.stream(activity)
249 end
250
251 def stream_out(_activity) do
252 :noop
253 end
254
255 @spec create(map(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
256 def create(params, fake \\ false) do
257 with {:ok, result} <- Repo.transaction(fn -> do_create(params, fake) end) do
258 result
259 end
260 end
261
262 defp do_create(%{to: to, actor: actor, context: context, object: object} = params, fake) do
263 additional = params[:additional] || %{}
264 # only accept false as false value
265 local = !(params[:local] == false)
266 published = params[:published]
267 quick_insert? = Config.get([:env]) == :benchmark
268
269 with create_data <-
270 make_create_data(
271 %{to: to, actor: actor, published: published, context: context, object: object},
272 additional
273 ),
274 {:ok, activity} <- insert(create_data, local, fake),
275 {:fake, false, activity} <- {:fake, fake, activity},
276 _ <- increase_replies_count_if_reply(create_data),
277 _ <- increase_poll_votes_if_vote(create_data),
278 {:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
279 {:ok, _actor} <- increase_note_count_if_public(actor, activity),
280 _ <- notify_and_stream(activity),
281 :ok <- maybe_federate(activity) do
282 {:ok, activity}
283 else
284 {:quick_insert, true, activity} ->
285 {:ok, activity}
286
287 {:fake, true, activity} ->
288 {:ok, activity}
289
290 {:error, message} ->
291 Repo.rollback(message)
292 end
293 end
294
295 @spec listen(map()) :: {:ok, Activity.t()} | {:error, any()}
296 def listen(%{to: to, actor: actor, context: context, object: object} = params) do
297 additional = params[:additional] || %{}
298 # only accept false as false value
299 local = !(params[:local] == false)
300 published = params[:published]
301
302 with listen_data <-
303 make_listen_data(
304 %{to: to, actor: actor, published: published, context: context, object: object},
305 additional
306 ),
307 {:ok, activity} <- insert(listen_data, local),
308 _ <- notify_and_stream(activity),
309 :ok <- maybe_federate(activity) do
310 {:ok, activity}
311 end
312 end
313
314 @spec accept(map()) :: {:ok, Activity.t()} | {:error, any()}
315 def accept(params) do
316 accept_or_reject("Accept", params)
317 end
318
319 @spec reject(map()) :: {:ok, Activity.t()} | {:error, any()}
320 def reject(params) do
321 accept_or_reject("Reject", params)
322 end
323
324 @spec accept_or_reject(String.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
325 def accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
326 local = Map.get(params, :local, true)
327 activity_id = Map.get(params, :activity_id, nil)
328
329 with data <-
330 %{"to" => to, "type" => type, "actor" => actor.ap_id, "object" => object}
331 |> Utils.maybe_put("id", activity_id),
332 {:ok, activity} <- insert(data, local),
333 _ <- notify_and_stream(activity),
334 :ok <- maybe_federate(activity) do
335 {:ok, activity}
336 end
337 end
338
339 @spec update(map()) :: {:ok, Activity.t()} | {:error, any()}
340 def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
341 local = !(params[:local] == false)
342 activity_id = params[:activity_id]
343
344 with data <- %{
345 "to" => to,
346 "cc" => cc,
347 "type" => "Update",
348 "actor" => actor,
349 "object" => object
350 },
351 data <- Utils.maybe_put(data, "id", activity_id),
352 {:ok, activity} <- insert(data, local),
353 _ <- notify_and_stream(activity),
354 :ok <- maybe_federate(activity) do
355 {:ok, activity}
356 end
357 end
358
359 @spec follow(User.t(), User.t(), String.t() | nil, boolean()) ::
360 {:ok, Activity.t()} | {:error, any()}
361 def follow(follower, followed, activity_id \\ nil, local \\ true) do
362 with {:ok, result} <-
363 Repo.transaction(fn -> do_follow(follower, followed, activity_id, local) end) do
364 result
365 end
366 end
367
368 defp do_follow(follower, followed, activity_id, local) do
369 with data <- make_follow_data(follower, followed, activity_id),
370 {:ok, activity} <- insert(data, local),
371 _ <- notify_and_stream(activity),
372 :ok <- maybe_federate(activity) do
373 {:ok, activity}
374 else
375 {:error, error} -> Repo.rollback(error)
376 end
377 end
378
379 @spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
380 {:ok, Activity.t()} | nil | {:error, any()}
381 def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
382 with {:ok, result} <-
383 Repo.transaction(fn -> do_unfollow(follower, followed, activity_id, local) end) do
384 result
385 end
386 end
387
388 defp do_unfollow(follower, followed, activity_id, local) do
389 with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
390 {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
391 unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
392 {:ok, activity} <- insert(unfollow_data, local),
393 _ <- notify_and_stream(activity),
394 :ok <- maybe_federate(activity) do
395 {:ok, activity}
396 else
397 nil -> nil
398 {:error, error} -> Repo.rollback(error)
399 end
400 end
401
402 @spec block(User.t(), User.t(), String.t() | nil, boolean()) ::
403 {:ok, Activity.t()} | {:error, any()}
404 def block(blocker, blocked, activity_id \\ nil, local \\ true) do
405 with {:ok, result} <-
406 Repo.transaction(fn -> do_block(blocker, blocked, activity_id, local) end) do
407 result
408 end
409 end
410
411 defp do_block(blocker, blocked, activity_id, local) do
412 unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
413
414 if unfollow_blocked do
415 follow_activity = fetch_latest_follow(blocker, blocked)
416 if follow_activity, do: unfollow(blocker, blocked, nil, local)
417 end
418
419 with block_data <- make_block_data(blocker, blocked, activity_id),
420 {:ok, activity} <- insert(block_data, local),
421 _ <- notify_and_stream(activity),
422 :ok <- maybe_federate(activity) do
423 {:ok, activity}
424 else
425 {:error, error} -> Repo.rollback(error)
426 end
427 end
428
429 @spec flag(map()) :: {:ok, Activity.t()} | {:error, any()}
430 def flag(
431 %{
432 actor: actor,
433 context: _context,
434 account: account,
435 statuses: statuses,
436 content: content
437 } = params
438 ) do
439 # only accept false as false value
440 local = !(params[:local] == false)
441 forward = !(params[:forward] == false)
442
443 additional = params[:additional] || %{}
444
445 additional =
446 if forward do
447 Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
448 else
449 Map.merge(additional, %{"to" => [], "cc" => []})
450 end
451
452 with flag_data <- make_flag_data(params, additional),
453 {:ok, activity} <- insert(flag_data, local),
454 {:ok, stripped_activity} <- strip_report_status_data(activity),
455 _ <- notify_and_stream(activity),
456 :ok <- maybe_federate(stripped_activity) do
457 User.all_superusers()
458 |> Enum.filter(fn user -> not is_nil(user.email) end)
459 |> Enum.each(fn superuser ->
460 superuser
461 |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
462 |> Pleroma.Emails.Mailer.deliver_async()
463 end)
464
465 {:ok, activity}
466 end
467 end
468
469 @spec move(User.t(), User.t(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
470 def move(%User{} = origin, %User{} = target, local \\ true) do
471 params = %{
472 "type" => "Move",
473 "actor" => origin.ap_id,
474 "object" => origin.ap_id,
475 "target" => target.ap_id
476 }
477
478 with true <- origin.ap_id in target.also_known_as,
479 {:ok, activity} <- insert(params, local),
480 _ <- notify_and_stream(activity) do
481 maybe_federate(activity)
482
483 BackgroundWorker.enqueue("move_following", %{
484 "origin_id" => origin.id,
485 "target_id" => target.id
486 })
487
488 {:ok, activity}
489 else
490 false -> {:error, "Target account must have the origin in `alsoKnownAs`"}
491 err -> err
492 end
493 end
494
495 def fetch_activities_for_context_query(context, opts) do
496 public = [Constants.as_public()]
497
498 recipients =
499 if opts["user"],
500 do: [opts["user"].ap_id | User.following(opts["user"])] ++ public,
501 else: public
502
503 from(activity in Activity)
504 |> maybe_preload_objects(opts)
505 |> maybe_preload_bookmarks(opts)
506 |> maybe_set_thread_muted_field(opts)
507 |> restrict_blocked(opts)
508 |> restrict_recipients(recipients, opts["user"])
509 |> where(
510 [activity],
511 fragment(
512 "?->>'type' = ? and ?->>'context' = ?",
513 activity.data,
514 "Create",
515 activity.data,
516 ^context
517 )
518 )
519 |> exclude_poll_votes(opts)
520 |> exclude_id(opts)
521 |> order_by([activity], desc: activity.id)
522 end
523
524 @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
525 def fetch_activities_for_context(context, opts \\ %{}) do
526 context
527 |> fetch_activities_for_context_query(opts)
528 |> Repo.all()
529 end
530
531 @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
532 FlakeId.Ecto.CompatType.t() | nil
533 def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
534 context
535 |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
536 |> limit(1)
537 |> select([a], a.id)
538 |> Repo.one()
539 end
540
541 @spec fetch_public_or_unlisted_activities(map(), Pagination.type()) :: [Activity.t()]
542 def fetch_public_or_unlisted_activities(opts \\ %{}, pagination \\ :keyset) do
543 opts = Map.drop(opts, ["user"])
544
545 query = fetch_activities_query([Constants.as_public()], opts)
546
547 query =
548 if opts["restrict_unlisted"] do
549 restrict_unlisted(query)
550 else
551 query
552 end
553
554 Pagination.fetch_paginated(query, opts, pagination)
555 end
556
557 @spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
558 def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
559 opts
560 |> Map.put("restrict_unlisted", true)
561 |> fetch_public_or_unlisted_activities(pagination)
562 end
563
564 @valid_visibilities ~w[direct unlisted public private]
565
566 defp restrict_visibility(query, %{visibility: visibility})
567 when is_list(visibility) do
568 if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
569 query =
570 from(
571 a in query,
572 where:
573 fragment(
574 "activity_visibility(?, ?, ?) = ANY (?)",
575 a.actor,
576 a.recipients,
577 a.data,
578 ^visibility
579 )
580 )
581
582 query
583 else
584 Logger.error("Could not restrict visibility to #{visibility}")
585 end
586 end
587
588 defp restrict_visibility(query, %{visibility: visibility})
589 when visibility in @valid_visibilities do
590 from(
591 a in query,
592 where:
593 fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
594 )
595 end
596
597 defp restrict_visibility(_query, %{visibility: visibility})
598 when visibility not in @valid_visibilities do
599 Logger.error("Could not restrict visibility to #{visibility}")
600 end
601
602 defp restrict_visibility(query, _visibility), do: query
603
604 defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
605 when is_list(visibility) do
606 if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
607 from(
608 a in query,
609 where:
610 not fragment(
611 "activity_visibility(?, ?, ?) = ANY (?)",
612 a.actor,
613 a.recipients,
614 a.data,
615 ^visibility
616 )
617 )
618 else
619 Logger.error("Could not exclude visibility to #{visibility}")
620 query
621 end
622 end
623
624 defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
625 when visibility in @valid_visibilities do
626 from(
627 a in query,
628 where:
629 not fragment(
630 "activity_visibility(?, ?, ?) = ?",
631 a.actor,
632 a.recipients,
633 a.data,
634 ^visibility
635 )
636 )
637 end
638
639 defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
640 when visibility not in [nil | @valid_visibilities] do
641 Logger.error("Could not exclude visibility to #{visibility}")
642 query
643 end
644
645 defp exclude_visibility(query, _visibility), do: query
646
647 defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
648 do: query
649
650 defp restrict_thread_visibility(
651 query,
652 %{"user" => %User{skip_thread_containment: true}},
653 _
654 ),
655 do: query
656
657 defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
658 from(
659 a in query,
660 where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
661 )
662 end
663
664 defp restrict_thread_visibility(query, _, _), do: query
665
666 def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
667 params =
668 params
669 |> Map.put("user", reading_user)
670 |> Map.put("actor_id", user.ap_id)
671
672 recipients =
673 user_activities_recipients(%{
674 "godmode" => params["godmode"],
675 "reading_user" => reading_user
676 })
677
678 fetch_activities(recipients, params)
679 |> Enum.reverse()
680 end
681
682 def fetch_user_activities(user, reading_user, params \\ %{}) do
683 params =
684 params
685 |> Map.put("type", ["Create", "Announce"])
686 |> Map.put("user", reading_user)
687 |> Map.put("actor_id", user.ap_id)
688 |> Map.put("pinned_activity_ids", user.pinned_activities)
689
690 params =
691 if User.blocks?(reading_user, user) do
692 params
693 else
694 params
695 |> Map.put("blocking_user", reading_user)
696 |> Map.put("muting_user", reading_user)
697 end
698
699 recipients =
700 user_activities_recipients(%{
701 "godmode" => params["godmode"],
702 "reading_user" => reading_user
703 })
704
705 fetch_activities(recipients, params)
706 |> Enum.reverse()
707 end
708
709 def fetch_statuses(reading_user, params) do
710 params =
711 params
712 |> Map.put("type", ["Create", "Announce"])
713
714 recipients =
715 user_activities_recipients(%{
716 "godmode" => params["godmode"],
717 "reading_user" => reading_user
718 })
719
720 fetch_activities(recipients, params, :offset)
721 |> Enum.reverse()
722 end
723
724 defp user_activities_recipients(%{"godmode" => true}) do
725 []
726 end
727
728 defp user_activities_recipients(%{"reading_user" => reading_user}) do
729 if reading_user do
730 [Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
731 else
732 [Constants.as_public()]
733 end
734 end
735
736 defp restrict_since(query, %{"since_id" => ""}), do: query
737
738 defp restrict_since(query, %{"since_id" => since_id}) do
739 from(activity in query, where: activity.id > ^since_id)
740 end
741
742 defp restrict_since(query, _), do: query
743
744 defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
745 raise "Can't use the child object without preloading!"
746 end
747
748 defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
749 when is_list(tag_reject) and tag_reject != [] do
750 from(
751 [_activity, object] in query,
752 where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
753 )
754 end
755
756 defp restrict_tag_reject(query, _), do: query
757
758 defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
759 raise "Can't use the child object without preloading!"
760 end
761
762 defp restrict_tag_all(query, %{"tag_all" => tag_all})
763 when is_list(tag_all) and tag_all != [] do
764 from(
765 [_activity, object] in query,
766 where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
767 )
768 end
769
770 defp restrict_tag_all(query, _), do: query
771
772 defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
773 raise "Can't use the child object without preloading!"
774 end
775
776 defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
777 from(
778 [_activity, object] in query,
779 where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
780 )
781 end
782
783 defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
784 from(
785 [_activity, object] in query,
786 where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
787 )
788 end
789
790 defp restrict_tag(query, _), do: query
791
792 defp restrict_recipients(query, [], _user), do: query
793
794 defp restrict_recipients(query, recipients, nil) do
795 from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
796 end
797
798 defp restrict_recipients(query, recipients, user) do
799 from(
800 activity in query,
801 where: fragment("? && ?", ^recipients, activity.recipients),
802 or_where: activity.actor == ^user.ap_id
803 )
804 end
805
806 defp restrict_local(query, %{"local_only" => true}) do
807 from(activity in query, where: activity.local == true)
808 end
809
810 defp restrict_local(query, _), do: query
811
812 defp restrict_actor(query, %{"actor_id" => actor_id}) do
813 from(activity in query, where: activity.actor == ^actor_id)
814 end
815
816 defp restrict_actor(query, _), do: query
817
818 defp restrict_type(query, %{"type" => type}) when is_binary(type) do
819 from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
820 end
821
822 defp restrict_type(query, %{"type" => type}) do
823 from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
824 end
825
826 defp restrict_type(query, _), do: query
827
828 defp restrict_state(query, %{"state" => state}) do
829 from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
830 end
831
832 defp restrict_state(query, _), do: query
833
834 defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
835 from(
836 [_activity, object] in query,
837 where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
838 )
839 end
840
841 defp restrict_favorited_by(query, _), do: query
842
843 defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
844 raise "Can't use the child object without preloading!"
845 end
846
847 defp restrict_media(query, %{"only_media" => val}) when val in [true, "true", "1"] do
848 from(
849 [_activity, object] in query,
850 where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
851 )
852 end
853
854 defp restrict_media(query, _), do: query
855
856 defp restrict_replies(query, %{"exclude_replies" => val}) when val in [true, "true", "1"] do
857 from(
858 [_activity, object] in query,
859 where: fragment("?->>'inReplyTo' is null", object.data)
860 )
861 end
862
863 defp restrict_replies(query, %{
864 "reply_filtering_user" => user,
865 "reply_visibility" => "self"
866 }) do
867 from(
868 [activity, object] in query,
869 where:
870 fragment(
871 "?->>'inReplyTo' is null OR ? = ANY(?)",
872 object.data,
873 ^user.ap_id,
874 activity.recipients
875 )
876 )
877 end
878
879 defp restrict_replies(query, %{
880 "reply_filtering_user" => user,
881 "reply_visibility" => "following"
882 }) do
883 from(
884 [activity, object] in query,
885 where:
886 fragment(
887 "?->>'inReplyTo' is null OR ? && array_remove(?, ?) OR ? = ?",
888 object.data,
889 ^[user.ap_id | User.get_cached_user_friends_ap_ids(user)],
890 activity.recipients,
891 activity.actor,
892 activity.actor,
893 ^user.ap_id
894 )
895 )
896 end
897
898 defp restrict_replies(query, _), do: query
899
900 defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val in [true, "true", "1"] do
901 from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
902 end
903
904 defp restrict_reblogs(query, _), do: query
905
906 defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
907
908 defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
909 mutes = opts["muted_users_ap_ids"] || User.muted_users_ap_ids(user)
910
911 query =
912 from([activity] in query,
913 where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
914 where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
915 )
916
917 unless opts["skip_preload"] do
918 from([thread_mute: tm] in query, where: is_nil(tm.user_id))
919 else
920 query
921 end
922 end
923
924 defp restrict_muted(query, _), do: query
925
926 defp restrict_blocked(query, %{"blocking_user" => %User{} = user} = opts) do
927 blocked_ap_ids = opts["blocked_users_ap_ids"] || User.blocked_users_ap_ids(user)
928 domain_blocks = user.domain_blocks || []
929
930 following_ap_ids = User.get_friends_ap_ids(user)
931
932 query =
933 if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
934
935 from(
936 [activity, object: o] in query,
937 where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
938 where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
939 where:
940 fragment(
941 "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
942 activity.data,
943 activity.data,
944 ^blocked_ap_ids
945 ),
946 where:
947 fragment(
948 "(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
949 activity.actor,
950 ^domain_blocks,
951 activity.actor,
952 ^following_ap_ids
953 ),
954 where:
955 fragment(
956 "(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
957 o.data,
958 ^domain_blocks,
959 o.data,
960 ^following_ap_ids
961 )
962 )
963 end
964
965 defp restrict_blocked(query, _), do: query
966
967 defp restrict_unlisted(query) do
968 from(
969 activity in query,
970 where:
971 fragment(
972 "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
973 activity.data,
974 ^[Constants.as_public()]
975 )
976 )
977 end
978
979 # TODO: when all endpoints migrated to OpenAPI compare `pinned` with `true` (boolean) only,
980 # the same for `restrict_media/2`, `restrict_replies/2`, 'restrict_reblogs/2'
981 # and `restrict_muted/2`
982
983 defp restrict_pinned(query, %{"pinned" => pinned, "pinned_activity_ids" => ids})
984 when pinned in [true, "true", "1"] do
985 from(activity in query, where: activity.id in ^ids)
986 end
987
988 defp restrict_pinned(query, _), do: query
989
990 defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user} = opts) do
991 muted_reblogs = opts["reblog_muted_users_ap_ids"] || User.reblog_muted_users_ap_ids(user)
992
993 from(
994 activity in query,
995 where:
996 fragment(
997 "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
998 activity.data,
999 activity.actor,
1000 ^muted_reblogs
1001 )
1002 )
1003 end
1004
1005 defp restrict_muted_reblogs(query, _), do: query
1006
1007 defp restrict_instance(query, %{"instance" => instance}) do
1008 users =
1009 from(
1010 u in User,
1011 select: u.ap_id,
1012 where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
1013 )
1014 |> Repo.all()
1015
1016 from(activity in query, where: activity.actor in ^users)
1017 end
1018
1019 defp restrict_instance(query, _), do: query
1020
1021 defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
1022
1023 defp exclude_poll_votes(query, _) do
1024 if has_named_binding?(query, :object) do
1025 from([activity, object: o] in query,
1026 where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1027 )
1028 else
1029 query
1030 end
1031 end
1032
1033 defp exclude_invisible_actors(query, %{"invisible_actors" => true}), do: query
1034
1035 defp exclude_invisible_actors(query, _opts) do
1036 invisible_ap_ids =
1037 User.Query.build(%{invisible: true, select: [:ap_id]})
1038 |> Repo.all()
1039 |> Enum.map(fn %{ap_id: ap_id} -> ap_id end)
1040
1041 from([activity] in query, where: activity.actor not in ^invisible_ap_ids)
1042 end
1043
1044 defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
1045 from(activity in query, where: activity.id != ^id)
1046 end
1047
1048 defp exclude_id(query, _), do: query
1049
1050 defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
1051
1052 defp maybe_preload_objects(query, _) do
1053 query
1054 |> Activity.with_preloaded_object()
1055 end
1056
1057 defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
1058
1059 defp maybe_preload_bookmarks(query, opts) do
1060 query
1061 |> Activity.with_preloaded_bookmark(opts["user"])
1062 end
1063
1064 defp maybe_preload_report_notes(query, %{"preload_report_notes" => true}) do
1065 query
1066 |> Activity.with_preloaded_report_notes()
1067 end
1068
1069 defp maybe_preload_report_notes(query, _), do: query
1070
1071 defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
1072
1073 defp maybe_set_thread_muted_field(query, opts) do
1074 query
1075 |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1076 end
1077
1078 defp maybe_order(query, %{order: :desc}) do
1079 query
1080 |> order_by(desc: :id)
1081 end
1082
1083 defp maybe_order(query, %{order: :asc}) do
1084 query
1085 |> order_by(asc: :id)
1086 end
1087
1088 defp maybe_order(query, _), do: query
1089
1090 defp fetch_activities_query_ap_ids_ops(opts) do
1091 source_user = opts["muting_user"]
1092 ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []
1093
1094 ap_id_relationships =
1095 ap_id_relationships ++
1096 if opts["blocking_user"] && opts["blocking_user"] == source_user do
1097 [:block]
1098 else
1099 []
1100 end
1101
1102 preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)
1103
1104 restrict_blocked_opts = Map.merge(%{"blocked_users_ap_ids" => preloaded_ap_ids[:block]}, opts)
1105 restrict_muted_opts = Map.merge(%{"muted_users_ap_ids" => preloaded_ap_ids[:mute]}, opts)
1106
1107 restrict_muted_reblogs_opts =
1108 Map.merge(%{"reblog_muted_users_ap_ids" => preloaded_ap_ids[:reblog_mute]}, opts)
1109
1110 {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1111 end
1112
1113 def fetch_activities_query(recipients, opts \\ %{}) do
1114 {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1115 fetch_activities_query_ap_ids_ops(opts)
1116
1117 config = %{
1118 skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1119 }
1120
1121 Activity
1122 |> maybe_preload_objects(opts)
1123 |> maybe_preload_bookmarks(opts)
1124 |> maybe_preload_report_notes(opts)
1125 |> maybe_set_thread_muted_field(opts)
1126 |> maybe_order(opts)
1127 |> restrict_recipients(recipients, opts["user"])
1128 |> restrict_replies(opts)
1129 |> restrict_tag(opts)
1130 |> restrict_tag_reject(opts)
1131 |> restrict_tag_all(opts)
1132 |> restrict_since(opts)
1133 |> restrict_local(opts)
1134 |> restrict_actor(opts)
1135 |> restrict_type(opts)
1136 |> restrict_state(opts)
1137 |> restrict_favorited_by(opts)
1138 |> restrict_blocked(restrict_blocked_opts)
1139 |> restrict_muted(restrict_muted_opts)
1140 |> restrict_media(opts)
1141 |> restrict_visibility(opts)
1142 |> restrict_thread_visibility(opts, config)
1143 |> restrict_reblogs(opts)
1144 |> restrict_pinned(opts)
1145 |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1146 |> restrict_instance(opts)
1147 |> Activity.restrict_deactivated_users()
1148 |> exclude_poll_votes(opts)
1149 |> exclude_invisible_actors(opts)
1150 |> exclude_visibility(opts)
1151 end
1152
1153 def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1154 list_memberships = Pleroma.List.memberships(opts["user"])
1155
1156 fetch_activities_query(recipients ++ list_memberships, opts)
1157 |> Pagination.fetch_paginated(opts, pagination)
1158 |> Enum.reverse()
1159 |> maybe_update_cc(list_memberships, opts["user"])
1160 end
1161
1162 @doc """
1163 Fetch favorites activities of user with order by sort adds to favorites
1164 """
1165 @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1166 def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1167 user.ap_id
1168 |> Activity.Queries.by_actor()
1169 |> Activity.Queries.by_type("Like")
1170 |> Activity.with_joined_object()
1171 |> Object.with_joined_activity()
1172 |> select([_like, object, activity], %{activity | object: object})
1173 |> order_by([like, _, _], desc_nulls_last: like.id)
1174 |> Pagination.fetch_paginated(
1175 Map.merge(params, %{"skip_order" => true}),
1176 pagination,
1177 :object_activity
1178 )
1179 end
1180
1181 defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1182 when is_list(list_memberships) and length(list_memberships) > 0 do
1183 Enum.map(activities, fn
1184 %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1185 if Enum.any?(bcc, &(&1 in list_memberships)) do
1186 update_in(activity.data["cc"], &[user_ap_id | &1])
1187 else
1188 activity
1189 end
1190
1191 activity ->
1192 activity
1193 end)
1194 end
1195
1196 defp maybe_update_cc(activities, _, _), do: activities
1197
1198 def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1199 from(activity in query,
1200 where:
1201 fragment("? && ?", activity.recipients, ^recipients) or
1202 (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1203 ^Constants.as_public() in activity.recipients)
1204 )
1205 end
1206
1207 def fetch_activities_bounded(
1208 recipients,
1209 recipients_with_public,
1210 opts \\ %{},
1211 pagination \\ :keyset
1212 ) do
1213 fetch_activities_query([], opts)
1214 |> fetch_activities_bounded_query(recipients, recipients_with_public)
1215 |> Pagination.fetch_paginated(opts, pagination)
1216 |> Enum.reverse()
1217 end
1218
1219 @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1220 def upload(file, opts \\ []) do
1221 with {:ok, data} <- Upload.store(file, opts) do
1222 obj_data =
1223 if opts[:actor] do
1224 Map.put(data, "actor", opts[:actor])
1225 else
1226 data
1227 end
1228
1229 Repo.insert(%Object{data: obj_data})
1230 end
1231 end
1232
1233 @spec get_actor_url(any()) :: binary() | nil
1234 defp get_actor_url(url) when is_binary(url), do: url
1235 defp get_actor_url(%{"href" => href}) when is_binary(href), do: href
1236
1237 defp get_actor_url(url) when is_list(url) do
1238 url
1239 |> List.first()
1240 |> get_actor_url()
1241 end
1242
1243 defp get_actor_url(_url), do: nil
1244
1245 defp object_to_user_data(data) do
1246 avatar =
1247 data["icon"]["url"] &&
1248 %{
1249 "type" => "Image",
1250 "url" => [%{"href" => data["icon"]["url"]}]
1251 }
1252
1253 banner =
1254 data["image"]["url"] &&
1255 %{
1256 "type" => "Image",
1257 "url" => [%{"href" => data["image"]["url"]}]
1258 }
1259
1260 fields =
1261 data
1262 |> Map.get("attachment", [])
1263 |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1264 |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1265
1266 emojis =
1267 data
1268 |> Map.get("tag", [])
1269 |> Enum.filter(fn
1270 %{"type" => "Emoji"} -> true
1271 _ -> false
1272 end)
1273 |> Enum.reduce(%{}, fn %{"icon" => %{"url" => url}, "name" => name}, acc ->
1274 Map.put(acc, String.trim(name, ":"), url)
1275 end)
1276
1277 locked = data["manuallyApprovesFollowers"] || false
1278 data = Transmogrifier.maybe_fix_user_object(data)
1279 discoverable = data["discoverable"] || false
1280 invisible = data["invisible"] || false
1281 actor_type = data["type"] || "Person"
1282
1283 public_key =
1284 if is_map(data["publicKey"]) && is_binary(data["publicKey"]["publicKeyPem"]) do
1285 data["publicKey"]["publicKeyPem"]
1286 else
1287 nil
1288 end
1289
1290 shared_inbox =
1291 if is_map(data["endpoints"]) && is_binary(data["endpoints"]["sharedInbox"]) do
1292 data["endpoints"]["sharedInbox"]
1293 else
1294 nil
1295 end
1296
1297 user_data = %{
1298 ap_id: data["id"],
1299 uri: get_actor_url(data["url"]),
1300 ap_enabled: true,
1301 banner: banner,
1302 fields: fields,
1303 emoji: emojis,
1304 locked: locked,
1305 discoverable: discoverable,
1306 invisible: invisible,
1307 avatar: avatar,
1308 name: data["name"],
1309 follower_address: data["followers"],
1310 following_address: data["following"],
1311 bio: data["summary"],
1312 actor_type: actor_type,
1313 also_known_as: Map.get(data, "alsoKnownAs", []),
1314 public_key: public_key,
1315 inbox: data["inbox"],
1316 shared_inbox: shared_inbox
1317 }
1318
1319 # nickname can be nil because of virtual actors
1320 user_data =
1321 if data["preferredUsername"] do
1322 Map.put(
1323 user_data,
1324 :nickname,
1325 "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1326 )
1327 else
1328 Map.put(user_data, :nickname, nil)
1329 end
1330
1331 {:ok, user_data}
1332 end
1333
1334 def fetch_follow_information_for_user(user) do
1335 with {:ok, following_data} <-
1336 Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1337 {:ok, hide_follows} <- collection_private(following_data),
1338 {:ok, followers_data} <-
1339 Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1340 {:ok, hide_followers} <- collection_private(followers_data) do
1341 {:ok,
1342 %{
1343 hide_follows: hide_follows,
1344 follower_count: normalize_counter(followers_data["totalItems"]),
1345 following_count: normalize_counter(following_data["totalItems"]),
1346 hide_followers: hide_followers
1347 }}
1348 else
1349 {:error, _} = e -> e
1350 e -> {:error, e}
1351 end
1352 end
1353
1354 defp normalize_counter(counter) when is_integer(counter), do: counter
1355 defp normalize_counter(_), do: 0
1356
1357 def maybe_update_follow_information(user_data) do
1358 with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1359 {_, true} <- {:user_type_check, user_data[:type] in ["Person", "Service"]},
1360 {_, true} <-
1361 {:collections_available,
1362 !!(user_data[:following_address] && user_data[:follower_address])},
1363 {:ok, info} <-
1364 fetch_follow_information_for_user(user_data) do
1365 info = Map.merge(user_data[:info] || %{}, info)
1366
1367 user_data
1368 |> Map.put(:info, info)
1369 else
1370 {:user_type_check, false} ->
1371 user_data
1372
1373 {:collections_available, false} ->
1374 user_data
1375
1376 {:enabled, false} ->
1377 user_data
1378
1379 e ->
1380 Logger.error(
1381 "Follower/Following counter update for #{user_data.ap_id} failed.\n" <> inspect(e)
1382 )
1383
1384 user_data
1385 end
1386 end
1387
1388 defp collection_private(%{"first" => %{"type" => type}})
1389 when type in ["CollectionPage", "OrderedCollectionPage"],
1390 do: {:ok, false}
1391
1392 defp collection_private(%{"first" => first}) do
1393 with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1394 Fetcher.fetch_and_contain_remote_object_from_id(first) do
1395 {:ok, false}
1396 else
1397 {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1398 {:error, _} = e -> e
1399 e -> {:error, e}
1400 end
1401 end
1402
1403 defp collection_private(_data), do: {:ok, true}
1404
1405 def user_data_from_user_object(data) do
1406 with {:ok, data} <- MRF.filter(data),
1407 {:ok, data} <- object_to_user_data(data) do
1408 {:ok, data}
1409 else
1410 e -> {:error, e}
1411 end
1412 end
1413
1414 def fetch_and_prepare_user_from_ap_id(ap_id) do
1415 with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1416 {:ok, data} <- user_data_from_user_object(data),
1417 data <- maybe_update_follow_information(data) do
1418 {:ok, data}
1419 else
1420 {:error, "Object has been deleted"} = e ->
1421 Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1422 {:error, e}
1423
1424 e ->
1425 Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1426 {:error, e}
1427 end
1428 end
1429
1430 def make_user_from_ap_id(ap_id) do
1431 user = User.get_cached_by_ap_id(ap_id)
1432
1433 if user && !User.ap_enabled?(user) do
1434 Transmogrifier.upgrade_user_from_ap_id(ap_id)
1435 else
1436 with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1437 if user do
1438 user
1439 |> User.remote_user_changeset(data)
1440 |> User.update_and_set_cache()
1441 else
1442 data
1443 |> User.remote_user_changeset()
1444 |> Repo.insert()
1445 |> User.set_cache()
1446 end
1447 else
1448 e -> {:error, e}
1449 end
1450 end
1451 end
1452
1453 def make_user_from_nickname(nickname) do
1454 with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1455 make_user_from_ap_id(ap_id)
1456 else
1457 _e -> {:error, "No AP id in WebFinger"}
1458 end
1459 end
1460
1461 # filter out broken threads
1462 def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1463 entire_thread_visible_for_user?(activity, user)
1464 end
1465
1466 # do post-processing on a specific activity
1467 def contain_activity(%Activity{} = activity, %User{} = user) do
1468 contain_broken_threads(activity, user)
1469 end
1470
1471 def fetch_direct_messages_query do
1472 Activity
1473 |> restrict_type(%{"type" => "Create"})
1474 |> restrict_visibility(%{visibility: "direct"})
1475 |> order_by([activity], asc: activity.id)
1476 end
1477 end
Fork of https://akkoma.dev/AkkomaGang/akkoma.git