diff options
24 files changed, 698 insertions, 241 deletions
diff --git a/miniq/admin.py b/miniq/admin.py deleted file mode 100644 index 1166f89..0000000 --- a/miniq/admin.py +++ /dev/null @@ -1,21 +0,0 @@ -from django.contrib import admin - -from miniq.models import Task - - -@admin.register(Task) -class TaskAdmin(admin.ModelAdmin): - - list_display = ["id", "created", "type", "subject", "completed", "failed"] - ordering = ["-created"] - actions = ["reset"] - - @admin.action(description="Reset Task") - def reset(self, request, queryset): - queryset.update( - failed=None, - completed=None, - locked=None, - locked_by=None, - error=None, - ) diff --git a/miniq/migrations/0001_initial.py b/miniq/migrations/0001_initial.py deleted file mode 100644 index dc6d42b..0000000 --- a/miniq/migrations/0001_initial.py +++ /dev/null @@ -1,48 +0,0 @@ -# Generated by Django 4.1.3 on 2022-11-07 04:19 - -from django.db import migrations, models - - -class Migration(migrations.Migration): - - initial = True - - dependencies = [] - - operations = [ - migrations.CreateModel( - name="Task", - fields=[ - ( - "id", - models.BigAutoField( - auto_created=True, - primary_key=True, - serialize=False, - verbose_name="ID", - ), - ), - ( - "type", - models.CharField( - choices=[ - ("identity_fetch", "Identity Fetch"), - ("inbox_item", "Inbox Item"), - ("follow_request", "Follow Request"), - ("follow_acknowledge", "Follow Acknowledge"), - ], - max_length=500, - ), - ), - ("priority", models.IntegerField(default=0)), - ("subject", models.TextField()), - ("payload", models.JSONField(blank=True, null=True)), - ("error", models.TextField(blank=True, null=True)), - ("created", models.DateTimeField(auto_now_add=True)), - ("completed", models.DateTimeField(blank=True, null=True)), - ("failed", models.DateTimeField(blank=True, null=True)), - ("locked", models.DateTimeField(blank=True, null=True)), - ("locked_by", models.CharField(blank=True, max_length=500, null=True)), - ], - ), - ] diff --git a/miniq/models.py b/miniq/models.py deleted file mode 100644 index 24d311c..0000000 --- a/miniq/models.py +++ /dev/null @@ -1,71 +0,0 @@ -from typing import Optional - -from django.db import models, transaction -from django.utils import timezone - - -class Task(models.Model): - """ - A task that must be done by a queue processor - """ - - class TypeChoices(models.TextChoices): - identity_fetch = "identity_fetch" - inbox_item = "inbox_item" - follow_request = "follow_request" - follow_acknowledge = "follow_acknowledge" - - type = models.CharField(max_length=500, choices=TypeChoices.choices) - priority = models.IntegerField(default=0) - subject = models.TextField() - payload = models.JSONField(blank=True, null=True) - error = models.TextField(blank=True, null=True) - - created = models.DateTimeField(auto_now_add=True) - completed = models.DateTimeField(blank=True, null=True) - failed = models.DateTimeField(blank=True, null=True) - locked = models.DateTimeField(blank=True, null=True) - locked_by = models.CharField(max_length=500, blank=True, null=True) - - def __str__(self): - return f"{self.id}/{self.type}({self.subject})" - - @classmethod - def get_one_available(cls, processor_id) -> Optional["Task"]: - """ - Gets one task off the list while reserving it, atomically. - """ - with transaction.atomic(): - next_task = cls.objects.filter(locked__isnull=True).first() - if next_task is None: - return None - next_task.locked = timezone.now() - next_task.locked_by = processor_id - next_task.save() - return next_task - - @classmethod - def submit(cls, type, subject: str, payload=None, deduplicate=True): - # Deduplication is done against tasks that have not started yet only, - # and only on tasks without payloads - if deduplicate and not payload: - if cls.objects.filter( - type=type, - subject=subject, - completed__isnull=True, - failed__isnull=True, - locked__isnull=True, - ).exists(): - return - cls.objects.create(type=type, subject=subject, payload=payload) - - async def complete(self): - await self.__class__.objects.filter(id=self.id).aupdate( - completed=timezone.now() - ) - - async def fail(self, error): - await self.__class__.objects.filter(id=self.id).aupdate( - failed=timezone.now(), - error=error, - ) diff --git a/miniq/tasks.py b/miniq/tasks.py deleted file mode 100644 index fedf8fd..0000000 --- a/miniq/tasks.py +++ /dev/null @@ -1,34 +0,0 @@ -import traceback - -from users.tasks.follow import handle_follow_request -from users.tasks.identity import handle_identity_fetch -from users.tasks.inbox import handle_inbox_item - - -class TaskHandler: - - handlers = { - "identity_fetch": handle_identity_fetch, - "inbox_item": handle_inbox_item, - "follow_request": handle_follow_request, - } - - def __init__(self, task): - self.task = task - self.subject = self.task.subject - self.payload = self.task.payload - - async def handle(self): - try: - print(f"Task {self.task}: Starting") - if self.task.type not in self.handlers: - raise ValueError(f"Cannot handle type {self.task.type}") - await self.handlers[self.task.type]( - self, - ) - await self.task.complete() - print(f"Task {self.task}: Complete") - except BaseException as e: - print(f"Task {self.task}: Error {e}") - traceback.print_exc() - await self.task.fail(f"{e}\n\n" + traceback.format_exc()) diff --git a/miniq/views.py b/miniq/views.py deleted file mode 100644 index 80c9ee2..0000000 --- a/miniq/views.py +++ /dev/null @@ -1,51 +0,0 @@ -import asyncio -import time -import uuid - -from asgiref.sync import sync_to_async -from django.http import HttpResponse -from django.views import View - -from miniq.models import Task -from miniq.tasks import TaskHandler - - -class QueueProcessor(View): - """ - A view that takes some items off the queue and processes them. - Tries to limit its own runtime so it's within HTTP timeout limits. - """ - - START_TIMEOUT = 30 - TOTAL_TIMEOUT = 60 - LOCK_TIMEOUT = 200 - MAX_TASKS = 20 - - async def get(self, request): - start_time = time.monotonic() - processor_id = uuid.uuid4().hex - handled = 0 - self.tasks = [] - # For the first time period, launch tasks - while (time.monotonic() - start_time) < self.START_TIMEOUT: - # Remove completed tasks - self.tasks = [t for t in self.tasks if not t.done()] - # See if there's a new task - if len(self.tasks) < self.MAX_TASKS: - # Pop a task off the queue and run it - task = await sync_to_async(Task.get_one_available)(processor_id) - if task is not None: - self.tasks.append(asyncio.create_task(TaskHandler(task).handle())) - handled += 1 - # Prevent busylooping - await asyncio.sleep(0.01) - # TODO: Clean up old locks here - # Then wait for tasks to finish - while (time.monotonic() - start_time) < self.TOTAL_TIMEOUT: - # Remove completed tasks - self.tasks = [t for t in self.tasks if not t.done()] - if not self.tasks: - break - # Prevent busylooping - await asyncio.sleep(1) - return HttpResponse(f"{handled} tasks handled") diff --git a/miniq/__init__.py b/stator/__init__.py index e69de29..e69de29 100644 --- a/miniq/__init__.py +++ b/stator/__init__.py diff --git a/stator/admin.py b/stator/admin.py new file mode 100644 index 0000000..c04d775 --- /dev/null +++ b/stator/admin.py @@ -0,0 +1,8 @@ +from django.contrib import admin + +from stator.models import StatorTask + + +@admin.register(StatorTask) +class DomainAdmin(admin.ModelAdmin): + list_display = ["id", "model_label", "instance_pk", "locked_until"] diff --git a/miniq/apps.py b/stator/apps.py index 4c7e773..8910ecb 100644 --- a/miniq/apps.py +++ b/stator/apps.py @@ -1,6 +1,6 @@ from django.apps import AppConfig -class MiniqConfig(AppConfig): +class StatorConfig(AppConfig): default_auto_field = "django.db.models.BigAutoField" - name = "miniq" + name = "stator" diff --git a/stator/graph.py b/stator/graph.py new file mode 100644 index 0000000..b06ffb8 --- /dev/null +++ b/stator/graph.py @@ -0,0 +1,162 @@ +import datetime +from functools import wraps +from typing import Callable, ClassVar, Dict, List, Optional, Set, Tuple, Union + +from django.db import models +from django.utils import timezone + + +class StateGraph: + """ + Represents a graph of possible states and transitions to attempt on them. + Does not support subclasses of existing graphs yet. + """ + + states: ClassVar[Dict[str, "State"]] + choices: ClassVar[List[Tuple[str, str]]] + initial_state: ClassVar["State"] + terminal_states: ClassVar[Set["State"]] + + def __init_subclass__(cls) -> None: + # Collect state memebers + cls.states = {} + for name, value in cls.__dict__.items(): + if name in ["__module__", "__doc__", "states"]: + pass + elif name in ["initial_state", "terminal_states", "choices"]: + raise ValueError(f"Cannot name a state {name} - this is reserved") + elif isinstance(value, State): + value._add_to_graph(cls, name) + elif callable(value) or isinstance(value, classmethod): + pass + else: + raise ValueError( + f"Graph has item {name} of unallowed type {type(value)}" + ) + # Check the graph layout + terminal_states = set() + initial_state = None + for state in cls.states.values(): + if state.initial: + if initial_state: + raise ValueError( + f"The graph has more than one initial state: {initial_state} and {state}" + ) + initial_state = state + if state.terminal: + terminal_states.add(state) + if initial_state is None: + raise ValueError("The graph has no initial state") + cls.initial_state = initial_state + cls.terminal_states = terminal_states + # Generate choices + cls.choices = [(name, name) for name in cls.states.keys()] + + +class State: + """ + Represents an individual state + """ + + def __init__(self, try_interval: float = 300): + self.try_interval = try_interval + self.parents: Set["State"] = set() + self.children: Dict["State", "Transition"] = {} + + def _add_to_graph(self, graph: StateGraph, name: str): + self.graph = graph + self.name = name + self.graph.states[name] = self + + def __repr__(self): + return f"<State {self.name}>" + + def add_transition( + self, + other: "State", + handler: Optional[Union[str, Callable]] = None, + priority: int = 0, + ) -> Callable: + def decorator(handler: Union[str, Callable]): + self.children[other] = Transition( + self, + other, + handler, + priority=priority, + ) + other.parents.add(self) + # All handlers should be class methods, so do that automatically. + if callable(handler): + return classmethod(handler) + + # If we're not being called as a decorator, invoke it immediately + if handler is not None: + decorator(handler) + return decorator + + def add_manual_transition(self, other: "State"): + self.children[other] = ManualTransition(self, other) + other.parents.add(self) + + @property + def initial(self): + return not self.parents + + @property + def terminal(self): + return not self.children + + def transitions(self, automatic_only=False) -> List["Transition"]: + """ + Returns all transitions from this State in priority order + """ + if automatic_only: + transitions = [t for t in self.children.values() if t.automatic] + else: + transitions = self.children.values() + return sorted(transitions, key=lambda t: t.priority, reverse=True) + + +class Transition: + """ + A possible transition from one state to another + """ + + def __init__( + self, + from_state: State, + to_state: State, + handler: Union[str, Callable], + priority: int = 0, + ): + self.from_state = from_state + self.to_state = to_state + self.handler = handler + self.priority = priority + self.automatic = True + + def get_handler(self) -> Callable: + """ + Returns the handler (it might need resolving from a string) + """ + if isinstance(self.handler, str): + self.handler = getattr(self.from_state.graph, self.handler) + return self.handler + + +class ManualTransition(Transition): + """ + A possible transition from one state to another that cannot be done by + the stator task runner, and must come from an external source. + """ + + def __init__( + self, + from_state: State, + to_state: State, + ): + self.from_state = from_state + self.to_state = to_state + self.handler = None + self.priority = 0 + self.automatic = False diff --git a/stator/migrations/0001_initial.py b/stator/migrations/0001_initial.py new file mode 100644 index 0000000..f485836 --- /dev/null +++ b/stator/migrations/0001_initial.py @@ -0,0 +1,31 @@ +# Generated by Django 4.1.3 on 2022-11-09 05:46 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [] + + operations = [ + migrations.CreateModel( + name="StatorTask", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("model_label", models.CharField(max_length=200)), + ("instance_pk", models.CharField(max_length=200)), + ("locked_until", models.DateTimeField(blank=True, null=True)), + ("priority", models.IntegerField(default=0)), + ], + ), + ] diff --git a/miniq/migrations/__init__.py b/stator/migrations/__init__.py index e69de29..e69de29 100644 --- a/miniq/migrations/__init__.py +++ b/stator/migrations/__init__.py diff --git a/stator/models.py b/stator/models.py new file mode 100644 index 0000000..3b0da0a --- /dev/null +++ b/stator/models.py @@ -0,0 +1,191 @@ +import datetime +from functools import reduce +from typing import Type, cast + +from asgiref.sync import sync_to_async +from django.apps import apps +from django.db import models, transaction +from django.utils import timezone +from django.utils.functional import classproperty + +from stator.graph import State, StateGraph + + +class StateField(models.CharField): + """ + A special field that automatically gets choices from a state graph + """ + + def __init__(self, graph: Type[StateGraph], **kwargs): + # Sensible default for state length + kwargs.setdefault("max_length", 100) + # Add choices and initial + self.graph = graph + kwargs["choices"] = self.graph.choices + kwargs["default"] = self.graph.initial_state.name + super().__init__(**kwargs) + + def deconstruct(self): + name, path, args, kwargs = super().deconstruct() + kwargs["graph"] = self.graph + return name, path, args, kwargs + + def from_db_value(self, value, expression, connection): + if value is None: + return value + return self.graph.states[value] + + def to_python(self, value): + if isinstance(value, State) or value is None: + return value + return self.graph.states[value] + + def get_prep_value(self, value): + if isinstance(value, State): + return value.name + return value + + +class StatorModel(models.Model): + """ + A model base class that has a state machine backing it, with tasks to work + out when to move the state to the next one. + + You need to provide a "state" field as an instance of StateField on the + concrete model yourself. + """ + + # When the state last actually changed, or the date of instance creation + state_changed = models.DateTimeField(auto_now_add=True) + + # When the last state change for the current state was attempted + # (and not successful, as this is cleared on transition) + state_attempted = models.DateTimeField(blank=True, null=True) + + class Meta: + abstract = True + + @classmethod + def schedule_overdue(cls, now=None) -> models.QuerySet: + """ + Finds instances of this model that need to run and schedule them. + """ + q = models.Q() + for transition in cls.state_graph.transitions(automatic_only=True): + q = q | transition.get_query(now=now) + return cls.objects.filter(q) + + @classproperty + def state_graph(cls) -> Type[StateGraph]: + return cls._meta.get_field("state").graph + + def schedule_transition(self, priority: int = 0): + """ + Adds this instance to the queue to get its state transition attempted. + + The scheduler will call this, but you can also call it directly if you + know it'll be ready and want to lower latency. + """ + StatorTask.schedule_for_execution(self, priority=priority) + + async def attempt_transition(self): + """ + Attempts to transition the current state by running its handler(s). + """ + # Try each transition in priority order + for transition in self.state_graph.states[self.state].transitions( + automatic_only=True + ): + success = await transition.get_handler()(self) + if success: + await self.perform_transition(transition.to_state.name) + return + await self.__class__.objects.filter(pk=self.pk).aupdate( + state_attempted=timezone.now() + ) + + async def perform_transition(self, state_name): + """ + Transitions the instance to the given state name + """ + if state_name not in self.state_graph.states: + raise ValueError(f"Invalid state {state_name}") + await self.__class__.objects.filter(pk=self.pk).aupdate( + state=state_name, + state_changed=timezone.now(), + state_attempted=None, + ) + + +class StatorTask(models.Model): + """ + The model that we use for an internal scheduling queue. + + Entries in this queue are up for checking and execution - it also performs + locking to ensure we get closer to exactly-once execution (but we err on + the side of at-least-once) + """ + + # appname.modelname (lowercased) label for the model this represents + model_label = models.CharField(max_length=200) + + # The primary key of that model (probably int or str) + instance_pk = models.CharField(max_length=200) + + # Locking columns (no runner ID, as we have no heartbeats - all runners + # only live for a short amount of time anyway) + locked_until = models.DateTimeField(null=True, blank=True) + + # Basic total ordering priority - higher is more important + priority = models.IntegerField(default=0) + + def __str__(self): + return f"#{self.pk}: {self.model_label}.{self.instance_pk}" + + @classmethod + def schedule_for_execution(cls, model_instance: StatorModel, priority: int = 0): + # We don't do a transaction here as it's fine to occasionally double up + model_label = model_instance._meta.label_lower + pk = model_instance.pk + # TODO: Increase priority of existing if present + if not cls.objects.filter( + model_label=model_label, instance_pk=pk, locked__isnull=True + ).exists(): + StatorTask.objects.create( + model_label=model_label, + instance_pk=pk, + priority=priority, + ) + + @classmethod + def get_for_execution(cls, number: int, lock_expiry: datetime.datetime): + """ + Returns up to `number` tasks for execution, having locked them. + """ + with transaction.atomic(): + selected = list( + cls.objects.filter(locked_until__isnull=True)[ + :number + ].select_for_update() + ) + cls.objects.filter(pk__in=[i.pk for i in selected]).update( + locked_until=timezone.now() + ) + return selected + + @classmethod + async def aget_for_execution(cls, number: int, lock_expiry: datetime.datetime): + return await sync_to_async(cls.get_for_execution)(number, lock_expiry) + + @classmethod + async def aclean_old_locks(cls): + await cls.objects.filter(locked_until__lte=timezone.now()).aupdate( + locked_until=None + ) + + async def aget_model_instance(self) -> StatorModel: + model = apps.get_model(self.model_label) + return cast(StatorModel, await model.objects.aget(pk=self.pk)) + + async def adelete(self): + self.__class__.objects.adelete(pk=self.pk) diff --git a/stator/runner.py b/stator/runner.py new file mode 100644 index 0000000..8c6e0f1 --- /dev/null +++ b/stator/runner.py @@ -0,0 +1,69 @@ +import asyncio +import datetime +import time +import uuid +from typing import List, Type + +from asgiref.sync import sync_to_async +from django.db import transaction +from django.utils import timezone + +from stator.models import StatorModel, StatorTask + + +class StatorRunner: + """ + Runs tasks on models that are looking for state changes. + Designed to run in a one-shot mode, living inside a request. + """ + + START_TIMEOUT = 30 + TOTAL_TIMEOUT = 60 + LOCK_TIMEOUT = 120 + + MAX_TASKS = 30 + + def __init__(self, models: List[Type[StatorModel]]): + self.models = models + self.runner_id = uuid.uuid4().hex + + async def run(self): + start_time = time.monotonic() + self.handled = 0 + self.tasks = [] + # Clean up old locks + await StatorTask.aclean_old_locks() + # Examine what needs scheduling + + # For the first time period, launch tasks + while (time.monotonic() - start_time) < self.START_TIMEOUT: + self.remove_completed_tasks() + space_remaining = self.MAX_TASKS - len(self.tasks) + # Fetch new tasks + if space_remaining > 0: + for new_task in await StatorTask.aget_for_execution( + space_remaining, + timezone.now() + datetime.timedelta(seconds=self.LOCK_TIMEOUT), + ): + self.tasks.append(asyncio.create_task(self.run_task(new_task))) + self.handled += 1 + # Prevent busylooping + await asyncio.sleep(0.01) + # Then wait for tasks to finish + while (time.monotonic() - start_time) < self.TOTAL_TIMEOUT: + self.remove_completed_tasks() + if not self.tasks: + break + # Prevent busylooping + await asyncio.sleep(1) + return self.handled + + async def run_task(self, task: StatorTask): + # Resolve the model instance + model_instance = await task.aget_model_instance() + await model_instance.attempt_transition() + # Remove ourselves from the database as complete + await task.adelete() + + def remove_completed_tasks(self): + self.tasks = [t for t in self.tasks if not t.done()] diff --git a/stator/tests/test_graph.py b/stator/tests/test_graph.py new file mode 100644 index 0000000..f6b8404 --- /dev/null +++ b/stator/tests/test_graph.py @@ -0,0 +1,66 @@ +import pytest + +from stator.graph import State, StateGraph + + +def test_declare(): + """ + Tests a basic graph declaration and various kinds of handler + lookups. + """ + + fake_handler = lambda: True + + class TestGraph(StateGraph): + initial = State() + second = State() + third = State() + fourth = State() + final = State() + + initial.add_transition(second, 60, handler=fake_handler) + second.add_transition(third, 60, handler="check_third") + + def check_third(cls): + return True + + @third.add_transition(fourth, 60) + def check_fourth(cls): + return True + + fourth.add_manual_transition(final) + + assert TestGraph.initial_state == TestGraph.initial + assert TestGraph.terminal_states == {TestGraph.final} + + assert TestGraph.initial.children[TestGraph.second].get_handler() == fake_handler + assert ( + TestGraph.second.children[TestGraph.third].get_handler() + == TestGraph.check_third + ) + assert ( + TestGraph.third.children[TestGraph.fourth].get_handler().__name__ + == "check_fourth" + ) + + +def test_bad_declarations(): + """ + Tests that you can't declare an invalid graph. + """ + # More than one initial state + with pytest.raises(ValueError): + + class TestGraph(StateGraph): + initial = State() + initial2 = State() + + # No initial states + with pytest.raises(ValueError): + + class TestGraph(StateGraph): + loop = State() + loop2 = State() + + loop.add_transition(loop2, 1, handler="fake") + loop2.add_transition(loop, 1, handler="fake") diff --git a/stator/views.py b/stator/views.py new file mode 100644 index 0000000..ef09b8e --- /dev/null +++ b/stator/views.py @@ -0,0 +1,17 @@ +from django.http import HttpResponse +from django.views import View + +from stator.runner import StatorRunner +from users.models import Follow + + +class RequestRunner(View): + """ + Runs a Stator runner within a HTTP request. For when you're on something + serverless. + """ + + async def get(self, request): + runner = StatorRunner([Follow]) + handled = await runner.run() + return HttpResponse(f"Handled {handled}") diff --git a/takahe/settings.py b/takahe/settings.py index 62065d2..cefbb35 100644 --- a/takahe/settings.py +++ b/takahe/settings.py @@ -26,7 +26,7 @@ INSTALLED_APPS = [ "core", "statuses", "users", - "miniq", + "stator", ] MIDDLEWARE = [ diff --git a/takahe/urls.py b/takahe/urls.py index 304bc23..764c8e9 100644 --- a/takahe/urls.py +++ b/takahe/urls.py @@ -2,7 +2,7 @@ from django.contrib import admin from django.urls import path from core import views as core -from miniq import views as miniq +from stator import views as stator from users.views import auth, identity urlpatterns = [ @@ -22,7 +22,7 @@ urlpatterns = [ # Well-known endpoints path(".well-known/webfinger", identity.Webfinger.as_view()), # Task runner - path(".queue/process/", miniq.QueueProcessor.as_view()), + path(".stator/runner/", stator.RequestRunner.as_view()), # Django admin path("djadmin/", admin.site.urls), ] diff --git a/users/admin.py b/users/admin.py index 0b0cc80..e517b0a 100644 --- a/users/admin.py +++ b/users/admin.py @@ -25,4 +25,4 @@ class IdentityAdmin(admin.ModelAdmin): @admin.register(Follow) class FollowAdmin(admin.ModelAdmin): - list_display = ["id", "source", "target", "requested", "accepted"] + list_display = ["id", "source", "target", "state"] diff --git a/users/migrations/0002_follow_state_follow_state_attempted_and_more.py b/users/migrations/0002_follow_state_follow_state_attempted_and_more.py new file mode 100644 index 0000000..b33236a --- /dev/null +++ b/users/migrations/0002_follow_state_follow_state_attempted_and_more.py @@ -0,0 +1,44 @@ +# Generated by Django 4.1.3 on 2022-11-07 19:22 + +import django.utils.timezone +from django.db import migrations, models + +import stator.models +import users.models.follow + + +class Migration(migrations.Migration): + + dependencies = [ + ("users", "0001_initial"), + ] + + operations = [ + migrations.AddField( + model_name="follow", + name="state", + field=stator.models.StateField( + choices=[ + ("pending", "pending"), + ("requested", "requested"), + ("accepted", "accepted"), + ], + default="pending", + graph=users.models.follow.FollowStates, + max_length=100, + ), + ), + migrations.AddField( + model_name="follow", + name="state_attempted", + field=models.DateTimeField(blank=True, null=True), + ), + migrations.AddField( + model_name="follow", + name="state_changed", + field=models.DateTimeField( + auto_now_add=True, default=django.utils.timezone.now + ), + preserve_default=False, + ), + ] diff --git a/users/migrations/0003_remove_follow_accepted_remove_follow_requested_and_more.py b/users/migrations/0003_remove_follow_accepted_remove_follow_requested_and_more.py new file mode 100644 index 0000000..180bfdd --- /dev/null +++ b/users/migrations/0003_remove_follow_accepted_remove_follow_requested_and_more.py @@ -0,0 +1,31 @@ +# Generated by Django 4.1.3 on 2022-11-08 03:58 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("users", "0002_follow_state_follow_state_attempted_and_more"), + ] + + operations = [ + migrations.RemoveField( + model_name="follow", + name="accepted", + ), + migrations.RemoveField( + model_name="follow", + name="requested", + ), + migrations.AddField( + model_name="follow", + name="state_locked", + field=models.DateTimeField(blank=True, null=True), + ), + migrations.AddField( + model_name="follow", + name="state_runner", + field=models.CharField(blank=True, max_length=100, null=True), + ), + ] diff --git a/users/migrations/0004_remove_follow_state_locked_and_more.py b/users/migrations/0004_remove_follow_state_locked_and_more.py new file mode 100644 index 0000000..bf98080 --- /dev/null +++ b/users/migrations/0004_remove_follow_state_locked_and_more.py @@ -0,0 +1,21 @@ +# Generated by Django 4.1.3 on 2022-11-09 05:15 + +from django.db import migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ("users", "0003_remove_follow_accepted_remove_follow_requested_and_more"), + ] + + operations = [ + migrations.RemoveField( + model_name="follow", + name="state_locked", + ), + migrations.RemoveField( + model_name="follow", + name="state_runner", + ), + ] diff --git a/users/models/follow.py b/users/models/follow.py index 29d036e..04f90ee 100644 --- a/users/models/follow.py +++ b/users/models/follow.py @@ -2,10 +2,23 @@ from typing import Optional from django.db import models -from miniq.models import Task +from stator.models import State, StateField, StateGraph, StatorModel -class Follow(models.Model): +class FollowStates(StateGraph): + pending = State(try_interval=3600) + requested = State() + accepted = State() + + @pending.add_transition(requested) + async def try_request(cls, instance): + print("Would have tried to follow") + return False + + requested.add_manual_transition(accepted) + + +class Follow(StatorModel): """ When one user (the source) follows other (the target) """ @@ -24,8 +37,7 @@ class Follow(models.Model): uri = models.CharField(blank=True, null=True, max_length=500) note = models.TextField(blank=True, null=True) - requested = models.BooleanField(default=False) - accepted = models.BooleanField(default=False) + state = StateField(FollowStates) created = models.DateTimeField(auto_now_add=True) updated = models.DateTimeField(auto_now=True) @@ -50,17 +62,15 @@ class Follow(models.Model): (which can be local or remote). """ if not source.local: - raise ValueError("You cannot initiate follows on a remote Identity") + raise ValueError("You cannot initiate follows from a remote Identity") try: follow = Follow.objects.get(source=source, target=target) except Follow.DoesNotExist: follow = Follow.objects.create(source=source, target=target, uri="") follow.uri = source.actor_uri + f"follow/{follow.pk}/" + # TODO: Local follow approvals if target.local: - follow.requested = True - follow.accepted = True - else: - Task.submit("follow_request", str(follow.pk)) + follow.state = FollowStates.accepted follow.save() return follow diff --git a/users/tasks/follow.py b/users/tasks/follow.py index 872b35f..0f802cf 100644 --- a/users/tasks/follow.py +++ b/users/tasks/follow.py @@ -27,3 +27,36 @@ async def handle_follow_request(task_handler): if response.status_code >= 400: raise ValueError(f"Request error: {response.status_code} {response.content}") await Follow.objects.filter(pk=follow.pk).aupdate(requested=True) + + +def send_follow_undo(id): + """ + Request a follow from a remote server + """ + follow = Follow.objects.select_related("source", "source__domain", "target").get( + pk=id + ) + # Construct the request + request = canonicalise( + { + "@context": "https://www.w3.org/ns/activitystreams", + "id": follow.uri + "#undo", + "type": "Undo", + "actor": follow.source.actor_uri, + "object": { + "id": follow.uri, + "type": "Follow", + "actor": follow.source.actor_uri, + "object": follow.target.actor_uri, + }, + } + ) + # Sign it and send it + from asgiref.sync import async_to_sync + + response = async_to_sync(HttpSignature.signed_request)( + follow.target.inbox_uri, request, follow.source + ) + if response.status_code >= 400: + raise ValueError(f"Request error: {response.status_code} {response.content}") + print(response) diff --git a/users/views/identity.py b/users/views/identity.py index 98fcdd6..41c7880 100644 --- a/users/views/identity.py +++ b/users/views/identity.py @@ -16,7 +16,6 @@ from django.views.generic import FormView, TemplateView, View from core.forms import FormHelper from core.ld import canonicalise from core.signatures import HttpSignature -from miniq.models import Task from users.decorators import identity_required from users.models import Domain, Follow, Identity from users.shortcuts import by_handle_or_404 |