[#1149] Refactored Oban workers API (introduced `enqueue/3`).
[akkoma] / lib / pleroma / web / activity_pub / activity_pub.ex
1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2019 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.Config
8 alias Pleroma.Conversation
9 alias Pleroma.Notification
10 alias Pleroma.Object
11 alias Pleroma.Object.Containment
12 alias Pleroma.Object.Fetcher
13 alias Pleroma.Pagination
14 alias Pleroma.Repo
15 alias Pleroma.Upload
16 alias Pleroma.User
17 alias Pleroma.Web.ActivityPub.MRF
18 alias Pleroma.Web.ActivityPub.Transmogrifier
19 alias Pleroma.Web.WebFinger
20 alias Pleroma.Workers.BackgroundWorker
21
22 import Ecto.Query
23 import Pleroma.Web.ActivityPub.Utils
24 import Pleroma.Web.ActivityPub.Visibility
25
26 require Logger
27 require Pleroma.Constants
28
29 # For Announce activities, we filter the recipients based on following status for any actors
30 # that match actual users. See issue #164 for more information about why this is necessary.
31 defp get_recipients(%{"type" => "Announce"} = data) do
32 to = Map.get(data, "to", [])
33 cc = Map.get(data, "cc", [])
34 bcc = Map.get(data, "bcc", [])
35 actor = User.get_cached_by_ap_id(data["actor"])
36
37 recipients =
38 Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
39 case User.get_cached_by_ap_id(recipient) do
40 nil -> true
41 user -> User.following?(user, actor)
42 end
43 end)
44
45 {recipients, to, cc}
46 end
47
48 defp get_recipients(%{"type" => "Create"} = data) do
49 to = Map.get(data, "to", [])
50 cc = Map.get(data, "cc", [])
51 bcc = Map.get(data, "bcc", [])
52 actor = Map.get(data, "actor", [])
53 recipients = [to, cc, bcc, [actor]] |> Enum.concat() |> Enum.uniq()
54 {recipients, to, cc}
55 end
56
57 defp get_recipients(data) do
58 to = Map.get(data, "to", [])
59 cc = Map.get(data, "cc", [])
60 bcc = Map.get(data, "bcc", [])
61 recipients = Enum.concat([to, cc, bcc])
62 {recipients, to, cc}
63 end
64
65 defp check_actor_is_active(actor) do
66 if not is_nil(actor) do
67 with user <- User.get_cached_by_ap_id(actor),
68 false <- user.info.deactivated do
69 true
70 else
71 _e -> false
72 end
73 else
74 true
75 end
76 end
77
78 defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
79 limit = Config.get([:instance, :remote_limit])
80 String.length(content) <= limit
81 end
82
83 defp check_remote_limit(_), do: true
84
85 def increase_note_count_if_public(actor, object) do
86 if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
87 end
88
89 def decrease_note_count_if_public(actor, object) do
90 if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
91 end
92
93 def increase_replies_count_if_reply(%{
94 "object" => %{"inReplyTo" => reply_ap_id} = object,
95 "type" => "Create"
96 }) do
97 if is_public?(object) do
98 Object.increase_replies_count(reply_ap_id)
99 end
100 end
101
102 def increase_replies_count_if_reply(_create_data), do: :noop
103
104 def decrease_replies_count_if_reply(%Object{
105 data: %{"inReplyTo" => reply_ap_id} = object
106 }) do
107 if is_public?(object) do
108 Object.decrease_replies_count(reply_ap_id)
109 end
110 end
111
112 def decrease_replies_count_if_reply(_object), do: :noop
113
114 def increase_poll_votes_if_vote(%{
115 "object" => %{"inReplyTo" => reply_ap_id, "name" => name},
116 "type" => "Create"
117 }) do
118 Object.increase_vote_count(reply_ap_id, name)
119 end
120
121 def increase_poll_votes_if_vote(_create_data), do: :noop
122
123 def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
124 with nil <- Activity.normalize(map),
125 map <- lazy_put_activity_defaults(map, fake),
126 true <- bypass_actor_check || check_actor_is_active(map["actor"]),
127 {_, true} <- {:remote_limit_error, check_remote_limit(map)},
128 {:ok, map} <- MRF.filter(map),
129 {recipients, _, _} = get_recipients(map),
130 {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
131 :ok <- Containment.contain_child(map),
132 {:ok, map, object} <- insert_full_object(map) do
133 {:ok, activity} =
134 Repo.insert(%Activity{
135 data: map,
136 local: local,
137 actor: map["actor"],
138 recipients: recipients
139 })
140
141 # Splice in the child object if we have one.
142 activity =
143 if not is_nil(object) do
144 Map.put(activity, :object, object)
145 else
146 activity
147 end
148
149 BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
150
151 Notification.create_notifications(activity)
152
153 participations =
154 activity
155 |> Conversation.create_or_bump_for()
156 |> get_participations()
157
158 stream_out(activity)
159 stream_out_participations(participations)
160 {:ok, activity}
161 else
162 %Activity{} = activity ->
163 {:ok, activity}
164
165 {:fake, true, map, recipients} ->
166 activity = %Activity{
167 data: map,
168 local: local,
169 actor: map["actor"],
170 recipients: recipients,
171 id: "pleroma:fakeid"
172 }
173
174 Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
175 {:ok, activity}
176
177 error ->
178 {:error, error}
179 end
180 end
181
182 defp get_participations({:ok, %{participations: participations}}), do: participations
183 defp get_participations(_), do: []
184
185 def stream_out_participations(participations) do
186 participations =
187 participations
188 |> Repo.preload(:user)
189
190 Enum.each(participations, fn participation ->
191 Pleroma.Web.Streamer.stream("participation", participation)
192 end)
193 end
194
195 def stream_out_participations(%Object{data: %{"context" => context}}, user) do
196 with %Conversation{} = conversation <- Conversation.get_for_ap_id(context),
197 conversation = Repo.preload(conversation, :participations),
198 last_activity_id =
199 fetch_latest_activity_id_for_context(conversation.ap_id, %{
200 "user" => user,
201 "blocking_user" => user
202 }) do
203 if last_activity_id do
204 stream_out_participations(conversation.participations)
205 end
206 end
207 end
208
209 def stream_out_participations(_, _), do: :noop
210
211 def stream_out(activity) do
212 if activity.data["type"] in ["Create", "Announce", "Delete"] do
213 object = Object.normalize(activity)
214 # Do not stream out poll replies
215 unless object.data["type"] == "Answer" do
216 Pleroma.Web.Streamer.stream("user", activity)
217 Pleroma.Web.Streamer.stream("list", activity)
218
219 if get_visibility(activity) == "public" do
220 Pleroma.Web.Streamer.stream("public", activity)
221
222 if activity.local do
223 Pleroma.Web.Streamer.stream("public:local", activity)
224 end
225
226 if activity.data["type"] in ["Create"] do
227 object.data
228 |> Map.get("tag", [])
229 |> Enum.filter(fn tag -> is_bitstring(tag) end)
230 |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
231
232 if object.data["attachment"] != [] do
233 Pleroma.Web.Streamer.stream("public:media", activity)
234
235 if activity.local do
236 Pleroma.Web.Streamer.stream("public:local:media", activity)
237 end
238 end
239 end
240 else
241 if get_visibility(activity) == "direct",
242 do: Pleroma.Web.Streamer.stream("direct", activity)
243 end
244 end
245 end
246 end
247
248 def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do
249 additional = params[:additional] || %{}
250 # only accept false as false value
251 local = !(params[:local] == false)
252 published = params[:published]
253
254 with create_data <-
255 make_create_data(
256 %{to: to, actor: actor, published: published, context: context, object: object},
257 additional
258 ),
259 {:ok, activity} <- insert(create_data, local, fake),
260 {:fake, false, activity} <- {:fake, fake, activity},
261 _ <- increase_replies_count_if_reply(create_data),
262 _ <- increase_poll_votes_if_vote(create_data),
263 # Changing note count prior to enqueuing federation task in order to avoid
264 # race conditions on updating user.info
265 {:ok, _actor} <- increase_note_count_if_public(actor, activity),
266 :ok <- maybe_federate(activity) do
267 {:ok, activity}
268 else
269 {:fake, true, activity} ->
270 {:ok, activity}
271
272 {:error, message} ->
273 {:error, message}
274 end
275 end
276
277 def accept(%{to: to, actor: actor, object: object} = params) do
278 # only accept false as false value
279 local = !(params[:local] == false)
280
281 with data <- %{"to" => to, "type" => "Accept", "actor" => actor.ap_id, "object" => object},
282 {:ok, activity} <- insert(data, local),
283 :ok <- maybe_federate(activity) do
284 {:ok, activity}
285 end
286 end
287
288 def reject(%{to: to, actor: actor, object: object} = params) do
289 # only accept false as false value
290 local = !(params[:local] == false)
291
292 with data <- %{"to" => to, "type" => "Reject", "actor" => actor.ap_id, "object" => object},
293 {:ok, activity} <- insert(data, local),
294 :ok <- maybe_federate(activity) do
295 {:ok, activity}
296 end
297 end
298
299 def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
300 # only accept false as false value
301 local = !(params[:local] == false)
302
303 with data <- %{
304 "to" => to,
305 "cc" => cc,
306 "type" => "Update",
307 "actor" => actor,
308 "object" => object
309 },
310 {:ok, activity} <- insert(data, local),
311 :ok <- maybe_federate(activity) do
312 {:ok, activity}
313 end
314 end
315
316 # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
317 def like(
318 %User{ap_id: ap_id} = user,
319 %Object{data: %{"id" => _}} = object,
320 activity_id \\ nil,
321 local \\ true
322 ) do
323 with nil <- get_existing_like(ap_id, object),
324 like_data <- make_like_data(user, object, activity_id),
325 {:ok, activity} <- insert(like_data, local),
326 {:ok, object} <- add_like_to_object(activity, object),
327 :ok <- maybe_federate(activity) do
328 {:ok, activity, object}
329 else
330 %Activity{} = activity -> {:ok, activity, object}
331 error -> {:error, error}
332 end
333 end
334
335 def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
336 with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
337 unlike_data <- make_unlike_data(actor, like_activity, activity_id),
338 {:ok, unlike_activity} <- insert(unlike_data, local),
339 {:ok, _activity} <- Repo.delete(like_activity),
340 {:ok, object} <- remove_like_from_object(like_activity, object),
341 :ok <- maybe_federate(unlike_activity) do
342 {:ok, unlike_activity, like_activity, object}
343 else
344 _e -> {:ok, object}
345 end
346 end
347
348 def announce(
349 %User{ap_id: _} = user,
350 %Object{data: %{"id" => _}} = object,
351 activity_id \\ nil,
352 local \\ true,
353 public \\ true
354 ) do
355 with true <- is_public?(object),
356 announce_data <- make_announce_data(user, object, activity_id, public),
357 {:ok, activity} <- insert(announce_data, local),
358 {:ok, object} <- add_announce_to_object(activity, object),
359 :ok <- maybe_federate(activity) do
360 {:ok, activity, object}
361 else
362 error -> {:error, error}
363 end
364 end
365
366 def unannounce(
367 %User{} = actor,
368 %Object{} = object,
369 activity_id \\ nil,
370 local \\ true
371 ) do
372 with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
373 unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
374 {:ok, unannounce_activity} <- insert(unannounce_data, local),
375 :ok <- maybe_federate(unannounce_activity),
376 {:ok, _activity} <- Repo.delete(announce_activity),
377 {:ok, object} <- remove_announce_from_object(announce_activity, object) do
378 {:ok, unannounce_activity, object}
379 else
380 _e -> {:ok, object}
381 end
382 end
383
384 def follow(follower, followed, activity_id \\ nil, local \\ true) do
385 with data <- make_follow_data(follower, followed, activity_id),
386 {:ok, activity} <- insert(data, local),
387 :ok <- maybe_federate(activity),
388 _ <- User.set_follow_state_cache(follower.ap_id, followed.ap_id, activity.data["state"]) do
389 {:ok, activity}
390 end
391 end
392
393 def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
394 with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
395 {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
396 unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
397 {:ok, activity} <- insert(unfollow_data, local),
398 :ok <- maybe_federate(activity) do
399 {:ok, activity}
400 end
401 end
402
403 def delete(%User{ap_id: ap_id, follower_address: follower_address} = user) do
404 with data <- %{
405 "to" => [follower_address],
406 "type" => "Delete",
407 "actor" => ap_id,
408 "object" => %{"type" => "Person", "id" => ap_id}
409 },
410 {:ok, activity} <- insert(data, true, true, true),
411 :ok <- maybe_federate(activity) do
412 {:ok, user}
413 end
414 end
415
416 def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
417 user = User.get_cached_by_ap_id(actor)
418 to = (object.data["to"] || []) ++ (object.data["cc"] || [])
419
420 with {:ok, object, activity} <- Object.delete(object),
421 data <- %{
422 "type" => "Delete",
423 "actor" => actor,
424 "object" => id,
425 "to" => to,
426 "deleted_activity_id" => activity && activity.id
427 },
428 {:ok, activity} <- insert(data, local, false),
429 stream_out_participations(object, user),
430 _ <- decrease_replies_count_if_reply(object),
431 # Changing note count prior to enqueuing federation task in order to avoid
432 # race conditions on updating user.info
433 {:ok, _actor} <- decrease_note_count_if_public(user, object),
434 :ok <- maybe_federate(activity) do
435 {:ok, activity}
436 end
437 end
438
439 def block(blocker, blocked, activity_id \\ nil, local \\ true) do
440 outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
441 unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
442
443 if unfollow_blocked do
444 follow_activity = fetch_latest_follow(blocker, blocked)
445 if follow_activity, do: unfollow(blocker, blocked, nil, local)
446 end
447
448 with true <- outgoing_blocks,
449 block_data <- make_block_data(blocker, blocked, activity_id),
450 {:ok, activity} <- insert(block_data, local),
451 :ok <- maybe_federate(activity) do
452 {:ok, activity}
453 else
454 _e -> {:ok, nil}
455 end
456 end
457
458 def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
459 with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
460 unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
461 {:ok, activity} <- insert(unblock_data, local),
462 :ok <- maybe_federate(activity) do
463 {:ok, activity}
464 end
465 end
466
467 def flag(
468 %{
469 actor: actor,
470 context: context,
471 account: account,
472 statuses: statuses,
473 content: content
474 } = params
475 ) do
476 # only accept false as false value
477 local = !(params[:local] == false)
478 forward = !(params[:forward] == false)
479
480 additional = params[:additional] || %{}
481
482 params = %{
483 actor: actor,
484 context: context,
485 account: account,
486 statuses: statuses,
487 content: content
488 }
489
490 additional =
491 if forward do
492 Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
493 else
494 Map.merge(additional, %{"to" => [], "cc" => []})
495 end
496
497 with flag_data <- make_flag_data(params, additional),
498 {:ok, activity} <- insert(flag_data, local),
499 :ok <- maybe_federate(activity) do
500 Enum.each(User.all_superusers(), fn superuser ->
501 superuser
502 |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
503 |> Pleroma.Emails.Mailer.deliver_async()
504 end)
505
506 {:ok, activity}
507 end
508 end
509
510 defp fetch_activities_for_context_query(context, opts) do
511 public = [Pleroma.Constants.as_public()]
512
513 recipients =
514 if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
515
516 from(activity in Activity)
517 |> maybe_preload_objects(opts)
518 |> maybe_preload_bookmarks(opts)
519 |> maybe_set_thread_muted_field(opts)
520 |> restrict_blocked(opts)
521 |> restrict_recipients(recipients, opts["user"])
522 |> where(
523 [activity],
524 fragment(
525 "?->>'type' = ? and ?->>'context' = ?",
526 activity.data,
527 "Create",
528 activity.data,
529 ^context
530 )
531 )
532 |> exclude_poll_votes(opts)
533 |> exclude_id(opts)
534 |> order_by([activity], desc: activity.id)
535 end
536
537 @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
538 def fetch_activities_for_context(context, opts \\ %{}) do
539 context
540 |> fetch_activities_for_context_query(opts)
541 |> Repo.all()
542 end
543
544 @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
545 Pleroma.FlakeId.t() | nil
546 def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
547 context
548 |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
549 |> limit(1)
550 |> select([a], a.id)
551 |> Repo.one()
552 end
553
554 def fetch_public_activities(opts \\ %{}) do
555 q = fetch_activities_query([Pleroma.Constants.as_public()], opts)
556
557 q
558 |> restrict_unlisted()
559 |> Pagination.fetch_paginated(opts)
560 |> Enum.reverse()
561 end
562
563 @valid_visibilities ~w[direct unlisted public private]
564
565 defp restrict_visibility(query, %{visibility: visibility})
566 when is_list(visibility) do
567 if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
568 query =
569 from(
570 a in query,
571 where:
572 fragment(
573 "activity_visibility(?, ?, ?) = ANY (?)",
574 a.actor,
575 a.recipients,
576 a.data,
577 ^visibility
578 )
579 )
580
581 query
582 else
583 Logger.error("Could not restrict visibility to #{visibility}")
584 end
585 end
586
587 defp restrict_visibility(query, %{visibility: visibility})
588 when visibility in @valid_visibilities do
589 from(
590 a in query,
591 where:
592 fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
593 )
594 end
595
596 defp restrict_visibility(_query, %{visibility: visibility})
597 when visibility not in @valid_visibilities do
598 Logger.error("Could not restrict visibility to #{visibility}")
599 end
600
601 defp restrict_visibility(query, _visibility), do: query
602
603 defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
604 do: query
605
606 defp restrict_thread_visibility(
607 query,
608 %{"user" => %User{info: %{skip_thread_containment: true}}},
609 _
610 ),
611 do: query
612
613 defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
614 from(
615 a in query,
616 where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
617 )
618 end
619
620 defp restrict_thread_visibility(query, _, _), do: query
621
622 def fetch_user_activities(user, reading_user, params \\ %{}) do
623 params =
624 params
625 |> Map.put("type", ["Create", "Announce"])
626 |> Map.put("user", reading_user)
627 |> Map.put("actor_id", user.ap_id)
628 |> Map.put("whole_db", true)
629 |> Map.put("pinned_activity_ids", user.info.pinned_activities)
630
631 recipients =
632 user_activities_recipients(%{
633 "godmode" => params["godmode"],
634 "reading_user" => reading_user
635 })
636
637 fetch_activities(recipients, params)
638 |> Enum.reverse()
639 end
640
641 defp user_activities_recipients(%{"godmode" => true}) do
642 []
643 end
644
645 defp user_activities_recipients(%{"reading_user" => reading_user}) do
646 if reading_user do
647 [Pleroma.Constants.as_public()] ++ [reading_user.ap_id | reading_user.following]
648 else
649 [Pleroma.Constants.as_public()]
650 end
651 end
652
653 defp restrict_since(query, %{"since_id" => ""}), do: query
654
655 defp restrict_since(query, %{"since_id" => since_id}) do
656 from(activity in query, where: activity.id > ^since_id)
657 end
658
659 defp restrict_since(query, _), do: query
660
661 defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
662 raise "Can't use the child object without preloading!"
663 end
664
665 defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
666 when is_list(tag_reject) and tag_reject != [] do
667 from(
668 [_activity, object] in query,
669 where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
670 )
671 end
672
673 defp restrict_tag_reject(query, _), do: query
674
675 defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
676 raise "Can't use the child object without preloading!"
677 end
678
679 defp restrict_tag_all(query, %{"tag_all" => tag_all})
680 when is_list(tag_all) and tag_all != [] do
681 from(
682 [_activity, object] in query,
683 where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
684 )
685 end
686
687 defp restrict_tag_all(query, _), do: query
688
689 defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
690 raise "Can't use the child object without preloading!"
691 end
692
693 defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
694 from(
695 [_activity, object] in query,
696 where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
697 )
698 end
699
700 defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
701 from(
702 [_activity, object] in query,
703 where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
704 )
705 end
706
707 defp restrict_tag(query, _), do: query
708
709 defp restrict_recipients(query, [], _user), do: query
710
711 defp restrict_recipients(query, recipients, nil) do
712 from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
713 end
714
715 defp restrict_recipients(query, recipients, user) do
716 from(
717 activity in query,
718 where: fragment("? && ?", ^recipients, activity.recipients),
719 or_where: activity.actor == ^user.ap_id
720 )
721 end
722
723 defp restrict_local(query, %{"local_only" => true}) do
724 from(activity in query, where: activity.local == true)
725 end
726
727 defp restrict_local(query, _), do: query
728
729 defp restrict_actor(query, %{"actor_id" => actor_id}) do
730 from(activity in query, where: activity.actor == ^actor_id)
731 end
732
733 defp restrict_actor(query, _), do: query
734
735 defp restrict_type(query, %{"type" => type}) when is_binary(type) do
736 from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
737 end
738
739 defp restrict_type(query, %{"type" => type}) do
740 from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
741 end
742
743 defp restrict_type(query, _), do: query
744
745 defp restrict_state(query, %{"state" => state}) do
746 from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
747 end
748
749 defp restrict_state(query, _), do: query
750
751 defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
752 from(
753 [_activity, object] in query,
754 where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
755 )
756 end
757
758 defp restrict_favorited_by(query, _), do: query
759
760 defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
761 raise "Can't use the child object without preloading!"
762 end
763
764 defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
765 from(
766 [_activity, object] in query,
767 where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
768 )
769 end
770
771 defp restrict_media(query, _), do: query
772
773 defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
774 from(
775 activity in query,
776 where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
777 )
778 end
779
780 defp restrict_replies(query, _), do: query
781
782 defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
783 from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
784 end
785
786 defp restrict_reblogs(query, _), do: query
787
788 defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
789
790 defp restrict_muted(query, %{"muting_user" => %User{info: info}} = opts) do
791 mutes = info.mutes
792
793 query =
794 from([activity] in query,
795 where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
796 where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
797 )
798
799 unless opts["skip_preload"] do
800 from([thread_mute: tm] in query, where: is_nil(tm))
801 else
802 query
803 end
804 end
805
806 defp restrict_muted(query, _), do: query
807
808 defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
809 blocks = info.blocks || []
810 domain_blocks = info.domain_blocks || []
811
812 query =
813 if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
814
815 from(
816 [activity, object: o] in query,
817 where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
818 where: fragment("not (? && ?)", activity.recipients, ^blocks),
819 where:
820 fragment(
821 "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
822 activity.data,
823 activity.data,
824 ^blocks
825 ),
826 where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
827 where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
828 )
829 end
830
831 defp restrict_blocked(query, _), do: query
832
833 defp restrict_unlisted(query) do
834 from(
835 activity in query,
836 where:
837 fragment(
838 "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
839 activity.data,
840 ^[Pleroma.Constants.as_public()]
841 )
842 )
843 end
844
845 defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
846 from(activity in query, where: activity.id in ^ids)
847 end
848
849 defp restrict_pinned(query, _), do: query
850
851 defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
852 muted_reblogs = info.muted_reblogs || []
853
854 from(
855 activity in query,
856 where:
857 fragment(
858 "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
859 activity.data,
860 activity.actor,
861 ^muted_reblogs
862 )
863 )
864 end
865
866 defp restrict_muted_reblogs(query, _), do: query
867
868 defp exclude_poll_votes(query, %{"include_poll_votes" => "true"}), do: query
869
870 defp exclude_poll_votes(query, _) do
871 if has_named_binding?(query, :object) do
872 from([activity, object: o] in query,
873 where: fragment("not(?->>'type' = ?)", o.data, "Answer")
874 )
875 else
876 query
877 end
878 end
879
880 defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
881 from(activity in query, where: activity.id != ^id)
882 end
883
884 defp exclude_id(query, _), do: query
885
886 defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
887
888 defp maybe_preload_objects(query, _) do
889 query
890 |> Activity.with_preloaded_object()
891 end
892
893 defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
894
895 defp maybe_preload_bookmarks(query, opts) do
896 query
897 |> Activity.with_preloaded_bookmark(opts["user"])
898 end
899
900 defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
901
902 defp maybe_set_thread_muted_field(query, opts) do
903 query
904 |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
905 end
906
907 defp maybe_order(query, %{order: :desc}) do
908 query
909 |> order_by(desc: :id)
910 end
911
912 defp maybe_order(query, %{order: :asc}) do
913 query
914 |> order_by(asc: :id)
915 end
916
917 defp maybe_order(query, _), do: query
918
919 def fetch_activities_query(recipients, opts \\ %{}) do
920 config = %{
921 skip_thread_containment: Config.get([:instance, :skip_thread_containment])
922 }
923
924 Activity
925 |> maybe_preload_objects(opts)
926 |> maybe_preload_bookmarks(opts)
927 |> maybe_set_thread_muted_field(opts)
928 |> maybe_order(opts)
929 |> restrict_recipients(recipients, opts["user"])
930 |> restrict_tag(opts)
931 |> restrict_tag_reject(opts)
932 |> restrict_tag_all(opts)
933 |> restrict_since(opts)
934 |> restrict_local(opts)
935 |> restrict_actor(opts)
936 |> restrict_type(opts)
937 |> restrict_state(opts)
938 |> restrict_favorited_by(opts)
939 |> restrict_blocked(opts)
940 |> restrict_muted(opts)
941 |> restrict_media(opts)
942 |> restrict_visibility(opts)
943 |> restrict_thread_visibility(opts, config)
944 |> restrict_replies(opts)
945 |> restrict_reblogs(opts)
946 |> restrict_pinned(opts)
947 |> restrict_muted_reblogs(opts)
948 |> Activity.restrict_deactivated_users()
949 |> exclude_poll_votes(opts)
950 end
951
952 def fetch_activities(recipients, opts \\ %{}) do
953 list_memberships = Pleroma.List.memberships(opts["user"])
954
955 fetch_activities_query(recipients ++ list_memberships, opts)
956 |> Pagination.fetch_paginated(opts)
957 |> Enum.reverse()
958 |> maybe_update_cc(list_memberships, opts["user"])
959 end
960
961 defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
962 when is_list(list_memberships) and length(list_memberships) > 0 do
963 Enum.map(activities, fn
964 %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
965 if Enum.any?(bcc, &(&1 in list_memberships)) do
966 update_in(activity.data["cc"], &[user_ap_id | &1])
967 else
968 activity
969 end
970
971 activity ->
972 activity
973 end)
974 end
975
976 defp maybe_update_cc(activities, _, _), do: activities
977
978 def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
979 from(activity in query,
980 where:
981 fragment("? && ?", activity.recipients, ^recipients) or
982 (fragment("? && ?", activity.recipients, ^recipients_with_public) and
983 ^Pleroma.Constants.as_public() in activity.recipients)
984 )
985 end
986
987 def fetch_activities_bounded(recipients, recipients_with_public, opts \\ %{}) do
988 fetch_activities_query([], opts)
989 |> fetch_activities_bounded_query(recipients, recipients_with_public)
990 |> Pagination.fetch_paginated(opts)
991 |> Enum.reverse()
992 end
993
994 def upload(file, opts \\ []) do
995 with {:ok, data} <- Upload.store(file, opts) do
996 obj_data =
997 if opts[:actor] do
998 Map.put(data, "actor", opts[:actor])
999 else
1000 data
1001 end
1002
1003 Repo.insert(%Object{data: obj_data})
1004 end
1005 end
1006
1007 defp object_to_user_data(data) do
1008 avatar =
1009 data["icon"]["url"] &&
1010 %{
1011 "type" => "Image",
1012 "url" => [%{"href" => data["icon"]["url"]}]
1013 }
1014
1015 banner =
1016 data["image"]["url"] &&
1017 %{
1018 "type" => "Image",
1019 "url" => [%{"href" => data["image"]["url"]}]
1020 }
1021
1022 fields =
1023 data
1024 |> Map.get("attachment", [])
1025 |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1026 |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1027
1028 locked = data["manuallyApprovesFollowers"] || false
1029 data = Transmogrifier.maybe_fix_user_object(data)
1030
1031 user_data = %{
1032 ap_id: data["id"],
1033 info: %{
1034 ap_enabled: true,
1035 source_data: data,
1036 banner: banner,
1037 fields: fields,
1038 locked: locked
1039 },
1040 avatar: avatar,
1041 name: data["name"],
1042 follower_address: data["followers"],
1043 following_address: data["following"],
1044 bio: data["summary"]
1045 }
1046
1047 # nickname can be nil because of virtual actors
1048 user_data =
1049 if data["preferredUsername"] do
1050 Map.put(
1051 user_data,
1052 :nickname,
1053 "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1054 )
1055 else
1056 Map.put(user_data, :nickname, nil)
1057 end
1058
1059 {:ok, user_data}
1060 end
1061
1062 def fetch_follow_information_for_user(user) do
1063 with {:ok, following_data} <-
1064 Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1065 following_count when is_integer(following_count) <- following_data["totalItems"],
1066 {:ok, hide_follows} <- collection_private(following_data),
1067 {:ok, followers_data} <-
1068 Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1069 followers_count when is_integer(followers_count) <- followers_data["totalItems"],
1070 {:ok, hide_followers} <- collection_private(followers_data) do
1071 {:ok,
1072 %{
1073 hide_follows: hide_follows,
1074 follower_count: followers_count,
1075 following_count: following_count,
1076 hide_followers: hide_followers
1077 }}
1078 else
1079 {:error, _} = e ->
1080 e
1081
1082 e ->
1083 {:error, e}
1084 end
1085 end
1086
1087 defp maybe_update_follow_information(data) do
1088 with {:enabled, true} <-
1089 {:enabled, Pleroma.Config.get([:instance, :external_user_synchronization])},
1090 {:ok, info} <- fetch_follow_information_for_user(data) do
1091 info = Map.merge(data.info, info)
1092 Map.put(data, :info, info)
1093 else
1094 {:enabled, false} ->
1095 data
1096
1097 e ->
1098 Logger.error(
1099 "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1100 )
1101
1102 data
1103 end
1104 end
1105
1106 defp collection_private(data) do
1107 if is_map(data["first"]) and
1108 data["first"]["type"] in ["CollectionPage", "OrderedCollectionPage"] do
1109 {:ok, false}
1110 else
1111 with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1112 Fetcher.fetch_and_contain_remote_object_from_id(data["first"]) do
1113 {:ok, false}
1114 else
1115 {:error, {:ok, %{status: code}}} when code in [401, 403] ->
1116 {:ok, true}
1117
1118 {:error, _} = e ->
1119 e
1120
1121 e ->
1122 {:error, e}
1123 end
1124 end
1125 end
1126
1127 def user_data_from_user_object(data) do
1128 with {:ok, data} <- MRF.filter(data),
1129 {:ok, data} <- object_to_user_data(data) do
1130 {:ok, data}
1131 else
1132 e -> {:error, e}
1133 end
1134 end
1135
1136 def fetch_and_prepare_user_from_ap_id(ap_id) do
1137 with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1138 {:ok, data} <- user_data_from_user_object(data),
1139 data <- maybe_update_follow_information(data) do
1140 {:ok, data}
1141 else
1142 e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1143 end
1144 end
1145
1146 def make_user_from_ap_id(ap_id) do
1147 if _user = User.get_cached_by_ap_id(ap_id) do
1148 Transmogrifier.upgrade_user_from_ap_id(ap_id)
1149 else
1150 with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1151 User.insert_or_update_user(data)
1152 else
1153 e -> {:error, e}
1154 end
1155 end
1156 end
1157
1158 def make_user_from_nickname(nickname) do
1159 with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1160 make_user_from_ap_id(ap_id)
1161 else
1162 _e -> {:error, "No AP id in WebFinger"}
1163 end
1164 end
1165
1166 # filter out broken threads
1167 def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1168 entire_thread_visible_for_user?(activity, user)
1169 end
1170
1171 # do post-processing on a specific activity
1172 def contain_activity(%Activity{} = activity, %User{} = user) do
1173 contain_broken_threads(activity, user)
1174 end
1175
1176 def fetch_direct_messages_query do
1177 Activity
1178 |> restrict_type(%{"type" => "Create"})
1179 |> restrict_visibility(%{visibility: "direct"})
1180 |> order_by([activity], asc: activity.id)
1181 end
1182 end