4 files changed, 105 insertions, 10 deletions
diff --git a/users/models/__init__.py b/users/models/__init__.py
index d46003f..28d62b0 100644
--- a/users/models/__init__.py
+++ b/users/models/__init__.py
@@ -2,5 +2,6 @@ from .block import Block  # noqa
 from .domain import Domain  # noqa
 from .follow import Follow, FollowStates  # noqa
 from .identity import Identity, IdentityStates  # noqa
+from .inbox_message import InboxMessage, InboxMessageStates  # noqa
 from .user import User  # noqa
 from .user_event import UserEvent  # noqa
diff --git a/users/models/follow.py b/users/models/follow.py
index 3325a0b..6f62481 100644
--- a/users/models/follow.py
+++ b/users/models/follow.py
@@ -6,16 +6,20 @@ from stator.models import State, StateField, StateGraph, StatorModel
 
 
 class FollowStates(StateGraph):
-    pending = State(try_interval=30)
-    requested = State()
+    unrequested = State(try_interval=30)
+    requested = State(try_interval=24 * 60 * 60)
     accepted = State()
 
-    @pending.add_transition(requested)
-    async def try_request(instance: "Follow"):  # type:ignore
+    unrequested.transitions_to(requested)
+    requested.transitions_to(accepted)
+
+    @classmethod
+    async def handle_unrequested(cls, instance: "Follow"):
         print("Would have tried to follow on", instance)
-        return False
 
-    requested.add_manual_transition(accepted)
+    @classmethod
+    async def handle_requested(cls, instance: "Follow"):
+        print("Would have tried to requested on", instance)
 
 
 class Follow(StatorModel):
@@ -73,3 +77,17 @@ class Follow(StatorModel):
                 follow.state = FollowStates.accepted
             follow.save()
         return follow
+
+    @classmethod
+    def remote_created(cls, source, target, uri):
+        follow = cls.maybe_get(source=source, target=target)
+        if follow is None:
+            follow = Follow.objects.create(source=source, target=target, uri=uri)
+        if follow.state == FollowStates.fresh:
+            follow.transition_perform(FollowStates.requested)
+
+    @classmethod
+    def remote_accepted(cls, source, target):
+        follow = cls.maybe_get(source=source, target=target)
+        if follow and follow.state == FollowStates.requested:
+            follow.transition_perform(FollowStates.accepted)
diff --git a/users/models/identity.py b/users/models/identity.py
index 5e2cd06..7dff492 100644
--- a/users/models/identity.py
+++ b/users/models/identity.py
@@ -22,11 +22,16 @@ class IdentityStates(StateGraph):
     outdated = State(try_interval=3600)
     updated = State()
 
-    @outdated.add_transition(updated)
-    async def fetch_identity(identity: "Identity"):  # type:ignore
+    outdated.transitions_to(updated)
+
+    @classmethod
+    async def handle_outdated(cls, identity: "Identity"):
+        # Local identities never need fetching
         if identity.local:
-            return True
-        return await identity.fetch_actor()
+            return "updated"
+        # Run the actor fetch and progress to updated if it succeeds
+        if await identity.fetch_actor():
+            return "updated"
 
 
 def upload_namer(prefix, instance, filename):
diff --git a/users/models/inbox_message.py b/users/models/inbox_message.py
new file mode 100644
index 0000000..0dbdc3a
--- /dev/null
+++ b/users/models/inbox_message.py
@@ -0,0 +1,71 @@
+from asgiref.sync import sync_to_async
+from django.db import models
+
+from stator.models import State, StateField, StateGraph, StatorModel
+from users.models import Follow, Identity
+
+
+class InboxMessageStates(StateGraph):
+    received = State(try_interval=300)
+    processed = State()
+
+    received.transitions_to(processed)
+
+    @classmethod
+    async def handle_received(cls, instance: "InboxMessage"):
+        type = instance.message["type"].lower()
+        if type == "follow":
+            await instance.follow_request()
+        elif type == "accept":
+            inner_type = instance.message["object"]["type"].lower()
+            if inner_type == "follow":
+                await instance.follow_accepted()
+            else:
+                raise ValueError(f"Cannot handle activity of type accept.{inner_type}")
+        elif type == "undo":
+            inner_type = instance.message["object"]["type"].lower()
+            if inner_type == "follow":
+                await instance.follow_undo()
+            else:
+                raise ValueError(f"Cannot handle activity of type undo.{inner_type}")
+        else:
+            raise ValueError(f"Cannot handle activity of type {type}")
+
+
+class InboxMessage(StatorModel):
+    """
+    an incoming inbox message that needs processing.
+
+    Yes, this is kind of its own message queue built on the state graph system.
+    It's fine. It'll scale up to a decent point.
+    """
+
+    message = models.JSONField()
+
+    state = StateField(InboxMessageStates)
+
+    @sync_to_async
+    def follow_request(self):
+        """
+        Handles an incoming follow request
+        """
+        Follow.remote_created(
+            source=Identity.by_actor_uri_with_create(self.message["actor"]),
+            target=Identity.by_actor_uri(self.message["object"]),
+            uri=self.message["id"],
+        )
+
+    @sync_to_async
+    def follow_accepted(self):
+        """
+        Handles an incoming acceptance of one of our follow requests
+        """
+        Follow.remote_accepted(
+            source=Identity.by_actor_uri_with_create(self.message["actor"]),
+            target=Identity.by_actor_uri(self.message["object"]),
+        )
+
+    async def follow_undo(self):
+        """
+        Handles an incoming follow undo
+        """