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