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