Relay support

activities/models/post.py

@@ -44,6 +44,7 @@
 from users.models.hashtag_follow import HashtagFollow
 from users.models.identity import Identity, IdentityStates
 from users.models.inbox_message import InboxMessage
+from users.models.relay import Relay
 from users.models.system_actor import SystemActor
 
 logger = logging.getLogger(__name__)
@@ -77,6 +78,30 @@ def targets_fan_out(cls, post: "Post", type_: str) -> None:
                 type=type_,
                 subject_post=post,
             )
+        cls.fan_out_to_relay(post, type_)
+
+    @classmethod
+    def fan_out_to_relay(cls, post: "Post", type_: str) -> None:
+        if not post.local or post.visibility != Post.Visibilities.public:
+            return
+        relay_uris = Relay.active_inbox_uris()
+        if not relay_uris:
+            return
+        obj = None
+        match type_:
+            case FanOut.Types.post:
+                obj = canonicalise(post.to_create_ap())
+            case FanOut.Types.post_edited:
+                obj = canonicalise(post.to_update_ap())
+            case FanOut.Types.post_deleted:
+                obj = canonicalise(post.to_delete_ap())
+        if not obj:
+            return
+        for uri in relay_uris:
+            try:
+                post.author.signed_request(method="post", uri=uri, body=obj)
+            except Exception as e:
+                logger.warning(f"Error sending relay: {uri} {e}")
 
     @classmethod
     def handle_new(cls, instance: "Post"):

takahe/urls.py

@@ -153,6 +153,11 @@
         admin.FederationEdit.as_view(),
         name="admin_federation_edit",
     ),
+    path(
+        "admin/relays/",
+        admin.RelaysRoot.as_view(),
+        name="admin_relays",
+    ),
     path(
         "admin/users/",
         admin.UsersRoot.as_view(),

templates/admin/_menu.html

@@ -45,6 +45,10 @@ <h3>Administration</h3>
             <i class="fa-solid fa-diagram-project"></i>
             <span>Federation</span>
         </a>
+        <a href="{% url "admin_relays" %}" {% if section == "relays" %}class="selected"{% endif %} title="Relays">
+            <i class="fa-solid fa-tower-broadcast"></i>
+            <span>Relays</span>
+        </a>
         <a href="{% url "admin_users" %}" {% if section == "users" %}class="selected"{% endif %} title="Users">
             <i class="fa-solid fa-users"></i>
             <span>Users</span>

templates/admin/relays.html

@@ -0,0 +1,53 @@
+{% extends "admin/base_main.html" %}
+{% load activity_tags %}
+{% block subtitle %}Relay{% endblock %}
+{% block settings_content %}
+    <form action="?subscribe" method="post" class="search">
+        <input type="url"
+               name="inbox_uri"
+               pattern="^https?://.+"
+               placeholder="Relay inbox URI, e.g. https://relay.server/inbox">
+        {% csrf_token %}
+        <button>Subscribe</button>
+    </form>
+    <table class="items">
+        {% for relay in page_obj %}
+            <tr>
+                <td class="icon">
+                    {% if relay.state == 'subscribed' %}
+                        <i class="fa-regular fa-circle-check"></i>
+                    {% elif relay.state == 'failed' or relay.state == 'rejected' or relay.state == 'unsubscribed' %}
+                        <i class="fa-solid fa-circle-exclamation"></i>
+                    {% else %}
+                        <i class="fa-solid fa-cog fa-spin"></i>
+                    {% endif %}
+                </td>
+                <td class="name">{{ relay.inbox_uri }}</td>
+                <td class="stat">{{ relay.state }}</td>
+                <td class="actions">
+                    <form action="?unsubscribe" method="post">
+                        <input type="hidden" name="id" value="{{ relay.id }}">
+                        {% csrf_token %}
+                        <button {% if relay.state == 'failed' or relay.state == 'rejected' %}disabled{% endif %}>Unsubscribe</button>
+                    </form>
+                </td>
+                <td class="actions">
+                    <form action="?remove" method="post">
+                        <input type="hidden" name="id" value="{{ relay.id }}">
+                        {% csrf_token %}
+                        <button onclick="return confirm('Sure to force remove?')"
+                                {% if relay.state == 'subscribed' %}disabled{% endif %}>Remove</button>
+                    </form>
+                </td>
+            </tr>
+        {% empty %}
+            <tr class="empty">
+                <td>There are no relay yet.</td>
+            </tr>
+        {% endfor %}
+    </table>
+    <div class="view-options">
+        <small><i class="fa-regular fa-lightbulb"></i>&nbsp; Use remove only when it's stuck in (un)subscribing state for more than 10 minutes.</small>
+    </div>
+    {% include "admin/_pagination.html" with nouns="relay,relays" %}
+{% endblock %}

users/migrations/0023_add_relay.py

@@ -0,0 +1,60 @@
+# Generated by Django 4.2.8 on 2024-01-02 16:20
+
+from django.db import migrations, models
+
+import stator.models
+import users.models.relay
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("users", "0022_follow_request"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="Relay",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("state_changed", models.DateTimeField(auto_now_add=True)),
+                ("state_next_attempt", models.DateTimeField(blank=True, null=True)),
+                (
+                    "state_locked_until",
+                    models.DateTimeField(blank=True, db_index=True, null=True),
+                ),
+                ("inbox_uri", models.CharField(max_length=500, unique=True)),
+                (
+                    "state",
+                    stator.models.StateField(
+                        choices=[
+                            ("new", "new"),
+                            ("subscribed", "subscribed"),
+                            ("unsubscribing", "unsubscribing"),
+                            ("unsubscribed", "unsubscribed"),
+                        ],
+                        default="new",
+                        graph=users.models.relay.RelayStates,
+                        max_length=100,
+                    ),
+                ),
+                ("created", models.DateTimeField(auto_now_add=True)),
+                ("updated", models.DateTimeField(auto_now=True)),
+            ],
+            options={
+                "indexes": [
+                    models.Index(
+                        fields=["state", "state_next_attempt", "state_locked_until"],
+                        name="ix_relay_state_next",
+                    )
+                ],
+            },
+        ),
+    ]

users/models/__init__.py

@@ -8,6 +8,7 @@
 from .inbox_message import InboxMessage, InboxMessageStates  # noqa
 from .invite import Invite  # noqa
 from .password_reset import PasswordReset  # noqa
+from .relay import Relay, RelayStates  # noqa
 from .report import Report  # noqa
 from .system_actor import SystemActor  # noqa
 from .user import User  # noqa

users/models/inbox_message.py

@@ -16,7 +16,7 @@ class InboxMessageStates(StateGraph):
     @classmethod
     def handle_received(cls, instance: "InboxMessage"):
         from activities.models import Post, PostInteraction, TimelineEvent
-        from users.models import Block, Follow, Identity, Report
+        from users.models import Block, Follow, Identity, Relay, Report
         from users.services import IdentityService
 
         try:
@@ -68,7 +68,10 @@ def handle_received(cls, instance: "InboxMessage"):
                 case "accept":
                     match instance.message_object_type:
                         case "follow":
-                            Follow.handle_accept_ap(instance.message)
+                            if Relay.is_ap_message_for_relay(instance.message):
+                                Relay.handle_accept_ap(instance.message)
+                            else:
+                                Follow.handle_accept_ap(instance.message)
                         case None:
                             # It's a string object, but these will only be for Follows
                             Follow.handle_accept_ap(instance.message)
@@ -77,7 +80,10 @@ def handle_received(cls, instance: "InboxMessage"):
                 case "reject":
                     match instance.message_object_type:
                         case "follow":
-                            Follow.handle_reject_ap(instance.message)
+                            if Relay.is_ap_message_for_relay(instance.message):
+                                Relay.handle_reject_ap(instance.message)
+                            else:
+                                Follow.handle_reject_ap(instance.message)
                         case None:
                             # It's a string object, but these will only be for Follows
                             Follow.handle_reject_ap(instance.message)

users/models/relay.py

@@ -0,0 +1,146 @@
+import logging
+import re
+
+from django.db import models
+
+from stator.models import State, StateField, StateGraph, StatorModel
+from users.models.system_actor import SystemActor
+
+logger = logging.getLogger(__name__)
+
+
+class RelayStates(StateGraph):
+    new = State(try_interval=600)
+    subscribing = State(externally_progressed=True)
+    subscribed = State(externally_progressed=True)
+    failed = State(externally_progressed=True)
+    rejected = State(externally_progressed=True)
+    unsubscribing = State(try_interval=600)
+    unsubscribed = State(delete_after=1)
+
+    new.transitions_to(subscribing)
+    new.transitions_to(unsubscribing)
+    new.transitions_to(failed)
+    new.times_out_to(failed, seconds=38400)
+    subscribing.transitions_to(subscribed)
+    subscribing.transitions_to(unsubscribing)
+    subscribing.transitions_to(unsubscribed)
+    subscribing.transitions_to(rejected)
+    subscribing.transitions_to(failed)
+    subscribed.transitions_to(unsubscribing)
+    subscribed.transitions_to(rejected)
+    failed.transitions_to(unsubscribed)
+    rejected.transitions_to(unsubscribed)
+    unsubscribing.transitions_to(failed)
+    unsubscribing.transitions_to(unsubscribed)
+    unsubscribing.times_out_to(failed, seconds=38400)
+
+    @classmethod
+    def handle_new(cls, instance: "Relay"):
+        system_actor = SystemActor()
+        try:
+            response = system_actor.signed_request(
+                method="post",
+                uri=instance.inbox_uri,
+                body=instance.to_follow_ap(),
+            )
+        except Exception as e:
+            logger.error(f"Error sending follow request: {instance.inbox_uri} {e}")
+            return cls.failed
+        if response.status_code >= 200 and response.status_code < 300:
+            return cls.subscribing
+        else:
+            logger.error(f"Follow {instance.inbox_uri} HTTP {response.status_code}")
+            return cls.failed
+
+    @classmethod
+    def handle_unsubscribing(cls, instance: "Relay"):
+        system_actor = SystemActor()
+        try:
+            response = system_actor.signed_request(
+                method="post",
+                uri=instance.inbox_uri,
+                body=instance.to_unfollow_ap(),
+            )
+        except Exception as e:
+            logger.error(f"Error sending unfollow request: {instance.inbox_uri} {e}")
+            return cls.failed
+        if response.status_code >= 200 and response.status_code < 300:
+            return cls.unsubscribed
+        else:
+            logger.error(f"Unfollow {instance.inbox_uri} HTTP {response.status_code}")
+            return cls.failed
+
+
+class Relay(StatorModel):
+    inbox_uri = models.CharField(max_length=500, unique=True)
+
+    state = StateField(RelayStates)
+
+    created = models.DateTimeField(auto_now_add=True)
+    updated = models.DateTimeField(auto_now=True)
+
+    class Meta:
+        indexes: list = []
+
+    @classmethod
+    def active_inbox_uris(cls):
+        return list(
+            cls.objects.filter(state=RelayStates.subscribed).values_list(
+                "inbox_uri", flat=True
+            )
+        )
+
+    @classmethod
+    def subscribe(cls, inbox_uri: str) -> "Relay":
+        return cls.objects.get_or_create(inbox_uri=inbox_uri.strip())[0]
+
+    def unsubscribe(self):
+        self.transition_perform(RelayStates.unsubscribing)
+
+    def force_unsubscribe(self):
+        self.transition_perform(RelayStates.unsubscribed)
+
+    def to_follow_ap(self):
+        system_actor = SystemActor()
+        return {  # skip canonicalise here to keep Public addressing as full URI
+            "@context": ["https://www.w3.org/ns/activitystreams"],
+            "id": f"{system_actor.actor_uri}relay/{self.pk}/#follow",
+            "type": "Follow",
+            "actor": system_actor.actor_uri,
+            "object": "https://www.w3.org/ns/activitystreams#Public",
+        }
+
+    def to_unfollow_ap(self):
+        system_actor = SystemActor()
+        return {  # skip canonicalise here to keep Public addressing as full URI
+            "@context": ["https://www.w3.org/ns/activitystreams"],
+            "id": f"{system_actor.actor_uri}relay/{self.pk}/#unfollow",
+            "type": "Undo",
+            "actor": system_actor.actor_uri,
+            "object": self.to_follow_ap(),
+        }
+
+    @classmethod
+    def is_ap_message_for_relay(cls, message) -> bool:
+        return (
+            re.match(r".+/relay/(\d+)/#(follow|unfollow)$", message["object"]["id"])
+            is not None
+        )
+
+    @classmethod
+    def get_by_ap(cls, message) -> "Relay":
+        m = re.match(r".+/relay/(\d+)/#(follow|unfollow)$", message["object"]["id"])
+        if not m:
+            raise ValueError("Not a valid relay follow response")
+        return cls.objects.get(pk=int(m[1]))
+
+    @classmethod
+    def handle_accept_ap(cls, message):
+        relay = cls.get_by_ap(message)
+        relay.transition_perform(RelayStates.subscribed)
+
+    @classmethod
+    def handle_reject_ap(cls, message):
+        relay = cls.get_by_ap(message)
+        relay.transition_perform(RelayStates.rejected)

users/views/admin/__init__.py

@@ -31,6 +31,7 @@
 from users.views.admin.hashtags import HashtagEdit, HashtagEnable, Hashtags  # noqa
 from users.views.admin.identities import IdentitiesRoot, IdentityEdit  # noqa
 from users.views.admin.invites import InviteCreate, InvitesRoot, InviteView  # noqa
+from users.views.admin.relays import RelaysRoot  # noqa
 from users.views.admin.reports import ReportsRoot, ReportView  # noqa
 from users.views.admin.settings import (  # noqa
     BasicSettings,