takahe/activities/models/fan_out.py

230 lines
8.2 KiB
Python
Raw Normal View History

2022-12-17 19:29:48 +00:00
import httpx
2022-11-12 06:04:43 +00:00
from asgiref.sync import sync_to_async
from django.db import models
from activities.models.timeline_event import TimelineEvent
from core.ld import canonicalise
from stator.models import State, StateField, StateGraph, StatorModel
2022-12-20 14:20:11 +00:00
from users.models import FollowStates
2022-11-12 06:04:43 +00:00
class FanOutStates(StateGraph):
2022-12-20 06:23:50 +00:00
new = State(try_interval=600)
2022-11-12 06:04:43 +00:00
sent = State()
2022-12-20 06:23:50 +00:00
failed = State()
2022-11-12 06:04:43 +00:00
new.transitions_to(sent)
2022-12-20 06:23:50 +00:00
new.times_out_to(failed, seconds=86400 * 3)
2022-11-12 06:04:43 +00:00
@classmethod
async def handle_new(cls, instance: "FanOut"):
"""
Sends the fan-out to the right inbox.
"""
2022-11-27 18:09:46 +00:00
2022-11-12 06:04:43 +00:00
fan_out = await instance.afetch_full()
2022-11-27 18:09:46 +00:00
2022-12-20 07:01:30 +00:00
# Don't try to fan out to identities that are not fetched yet
if not (fan_out.identity.local or fan_out.identity.inbox_uri):
return
2022-11-27 18:09:46 +00:00
match (fan_out.type, fan_out.identity.local):
# Handle creating/updating local posts
2022-11-27 19:09:08 +00:00
case ((FanOut.Types.post | FanOut.Types.post_edited), True):
2022-11-27 18:09:46 +00:00
post = await fan_out.subject_post.afetch_full()
2022-11-14 01:42:47 +00:00
# Make a timeline event directly
# If it's a reply, we only add it if we follow at least one
2022-12-20 14:20:11 +00:00
# of the people mentioned AND the author, or we're mentioned,
# or it's a reply to us or the author
add = True
mentioned = {identity.id for identity in post.mentions.all()}
2022-11-27 19:20:58 +00:00
followed = await sync_to_async(set)(
2022-12-20 14:20:11 +00:00
fan_out.identity.outbound_follows.filter(
state__in=FollowStates.group_active()
).values_list("target_id", flat=True)
2022-11-27 19:20:58 +00:00
)
if post.in_reply_to:
2022-12-20 14:20:11 +00:00
interested_in = followed.union(
{post.author_id, fan_out.identity_id}
)
add = (post.author_id in followed) and (
bool(mentioned.intersection(interested_in))
2022-11-27 19:20:58 +00:00
)
if add:
await sync_to_async(TimelineEvent.add_post)(
identity=fan_out.identity,
post=post,
)
# We might have been mentioned
if (
fan_out.identity.id in mentioned
and fan_out.identity_id != post.author_id
):
2022-11-26 01:35:43 +00:00
await sync_to_async(TimelineEvent.add_mentioned)(
identity=fan_out.identity,
post=post,
)
2022-11-27 18:09:46 +00:00
# Handle sending remote posts create
2022-11-27 19:09:08 +00:00
case (FanOut.Types.post, False):
2022-11-27 18:09:46 +00:00
post = await fan_out.subject_post.afetch_full()
2022-11-14 01:42:47 +00:00
# Sign it and send it
2022-12-17 19:29:48 +00:00
try:
await post.author.signed_request(
method="post",
uri=fan_out.identity.inbox_uri,
body=canonicalise(post.to_create_ap()),
)
except httpx.RequestError:
return
2022-11-27 18:09:46 +00:00
# Handle sending remote posts update
2022-11-27 19:09:08 +00:00
case (FanOut.Types.post_edited, False):
2022-11-27 18:09:46 +00:00
post = await fan_out.subject_post.afetch_full()
# Sign it and send it
2022-12-17 19:29:48 +00:00
try:
await post.author.signed_request(
method="post",
uri=fan_out.identity.inbox_uri,
body=canonicalise(post.to_update_ap()),
)
except httpx.RequestError:
return
2022-11-27 18:09:46 +00:00
# Handle deleting local posts
2022-11-27 19:09:08 +00:00
case (FanOut.Types.post_deleted, True):
2022-11-27 18:09:46 +00:00
post = await fan_out.subject_post.afetch_full()
if fan_out.identity.local:
# Remove all timeline events mentioning it
await TimelineEvent.objects.filter(
identity=fan_out.identity,
subject_post=post,
).adelete()
# Handle sending remote post deletes
2022-11-27 19:09:08 +00:00
case (FanOut.Types.post_deleted, False):
2022-11-27 18:09:46 +00:00
post = await fan_out.subject_post.afetch_full()
# Send it to the remote inbox
2022-12-17 19:29:48 +00:00
try:
await post.author.signed_request(
method="post",
uri=fan_out.identity.inbox_uri,
body=canonicalise(post.to_delete_ap()),
)
except httpx.RequestError:
return
2022-11-27 18:09:46 +00:00
# Handle local boosts/likes
2022-11-27 19:09:08 +00:00
case (FanOut.Types.interaction, True):
2022-11-27 18:09:46 +00:00
interaction = await fan_out.subject_post_interaction.afetch_full()
# Make a timeline event directly
await sync_to_async(TimelineEvent.add_post_interaction)(
identity=fan_out.identity,
interaction=interaction,
)
2022-11-27 18:09:46 +00:00
# Handle sending remote boosts/likes
2022-11-27 19:09:08 +00:00
case (FanOut.Types.interaction, False):
2022-11-27 18:09:46 +00:00
interaction = await fan_out.subject_post_interaction.afetch_full()
# Send it to the remote inbox
2022-12-17 19:29:48 +00:00
try:
await interaction.identity.signed_request(
method="post",
uri=fan_out.identity.inbox_uri,
body=canonicalise(interaction.to_ap()),
)
except httpx.RequestError:
return
2022-11-27 18:09:46 +00:00
# Handle undoing local boosts/likes
2022-11-27 19:09:08 +00:00
case (FanOut.Types.undo_interaction, True): # noqa:F841
2022-11-27 18:09:46 +00:00
interaction = await fan_out.subject_post_interaction.afetch_full()
# Delete any local timeline events
await sync_to_async(TimelineEvent.delete_post_interaction)(
identity=fan_out.identity,
interaction=interaction,
)
2022-11-27 18:09:46 +00:00
# Handle sending remote undoing boosts/likes
2022-11-27 19:09:08 +00:00
case (FanOut.Types.undo_interaction, False): # noqa:F841
2022-11-27 18:09:46 +00:00
interaction = await fan_out.subject_post_interaction.afetch_full()
# Send an undo to the remote inbox
2022-12-17 19:29:48 +00:00
try:
await interaction.identity.signed_request(
method="post",
uri=fan_out.identity.inbox_uri,
body=canonicalise(interaction.to_undo_ap()),
)
except httpx.RequestError:
return
2022-11-27 18:09:46 +00:00
case _:
raise ValueError(
f"Cannot fan out with type {fan_out.type} local={fan_out.identity.local}"
)
2022-11-18 01:52:00 +00:00
return cls.sent
2022-11-12 06:04:43 +00:00
class FanOut(StatorModel):
"""
An activity that needs to get to an inbox somewhere.
"""
class Types(models.TextChoices):
post = "post"
post_edited = "post_edited"
post_deleted = "post_deleted"
2022-11-14 01:42:47 +00:00
interaction = "interaction"
undo_interaction = "undo_interaction"
2022-11-12 06:04:43 +00:00
state = StateField(FanOutStates)
# The user this event is targeted at
identity = models.ForeignKey(
"users.Identity",
on_delete=models.CASCADE,
related_name="fan_outs",
)
# What type of activity it is
type = models.CharField(max_length=100, choices=Types.choices)
# Links to the appropriate objects
subject_post = models.ForeignKey(
"activities.Post",
on_delete=models.CASCADE,
blank=True,
null=True,
related_name="fan_outs",
)
2022-11-14 01:42:47 +00:00
subject_post_interaction = models.ForeignKey(
"activities.PostInteraction",
on_delete=models.CASCADE,
blank=True,
null=True,
related_name="fan_outs",
)
2022-11-12 06:04:43 +00:00
created = models.DateTimeField(auto_now_add=True)
updated = models.DateTimeField(auto_now=True)
### Async helpers ###
async def afetch_full(self):
"""
Returns a version of the object with all relations pre-loaded
"""
2022-12-15 07:50:54 +00:00
return (
await FanOut.objects.select_related(
"identity",
"subject_post",
"subject_post_interaction",
)
.prefetch_related(
"subject_post__emojis",
)
.aget(pk=self.pk)
)