Implement transfer job models

testbed/core/admin.py

@@ -6,6 +6,7 @@
     LikeActivity,
     FollowActivity,
     PortabilityOutbox,
+    TransferJob,
 )
 
 
@@ -123,3 +124,30 @@ def has_add_permission(self, request):
 
     def has_delete_permission(self, request, obj=None):
         return False
+
+
+@admin.register(TransferJob)
+class TransferJobAdmin(admin.ModelAdmin):
+    list_display = (
+        "id",
+        "user",
+        "destination_actor",
+        "state",
+        "retry_when",
+        "created_at",
+        "updated_at",
+    )
+    list_filter = ("state", "retry_when", "created_at")
+    search_fields = (
+        "destination_actor__user__username",
+        "source_base_url",
+        "source_actor_url",
+        "authorized_actor_url",
+    )
+    readonly_fields = ("policy", "progress", "artifacts", "created_at", "updated_at")
+
+    list_select_related = ("destination_actor", "destination_actor__user")
+
+    @admin.display(ordering="destination_actor__user__username", description="User")
+    def user(self, obj):
+        return obj.user

testbed/core/factories.py

@@ -13,6 +13,7 @@
     Following,
     Followers,
     TokenActorBinding,
+    TransferJob,
 )
 
 # Base factory for creating Users without associated actors
@@ -263,3 +264,28 @@ class Meta:
     actor = factory.LazyAttribute(
         lambda o: o.token.user.actors.get(role=Actor.ROLE_SOURCE)
     )
+
+
+class TransferJobFactory(DjangoModelFactory):
+    """
+    A destination-side transfer job.
+
+    The destination Actor is the job's only owner link (`TransferJob.user` reads through it), so
+    there is no user to pass and no way to build a job whose owner disagrees with its actor's.
+
+    Reuses the destination Actor the post_save signal already created, for the same reason
+    TokenActorBindingFactory reuses the source one: creating a second would collide on the unique
+    username constraint and on the one-actor-per-role invariant.
+    """
+
+    class Meta:
+        model = TransferJob
+
+    destination_actor = factory.LazyFunction(
+        lambda: UserWithActorsFactory().actors.get(role=Actor.ROLE_DESTINATION)
+    )
+
+    source_base_url = "https://source.example"
+    policy = factory.LazyFunction(lambda: {"dry_run": True})
+
+

testbed/core/migrations/0010_transfer_job_and_item.py

@@ -0,0 +1,38 @@
+# Generated by Django 5.1.3 on 2026-09-14 19:12
+
+import django.db.models.deletion
+import testbed.core.models
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0009_token_actor_binding'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='TransferJob',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('source_base_url', models.URLField(help_text='Base URL of the source server. Everything else about the source is discovered from it.', max_length=500)),
+                ('source_actor_url', models.URLField(blank=True, help_text='Source Actor URL as supplied or resolved by discovery. Advisory, not authoritative.', max_length=500, null=True)),
+                ('authorized_actor_url', models.URLField(blank=True, help_text='The Actor URL the source returned in `activitypub_actor` on the callback (LOLA §5.3).', max_length=500, null=True)),
+                ('state', models.CharField(choices=[('active', 'Active'), ('finished', 'Finished'), ('failed', 'Failed')], default='active', help_text='The working phase is derived from the data, not stored.', max_length=16)),
+                ('retry_when', models.DateTimeField(blank=True, help_text="Set from a 429's Retry-After. A job is paused.", null=True)),
+                ('policy', models.JSONField(blank=True, default=testbed.core.models.default_transfer_policy, help_text='Chosen when the job is created. Changing the decision means a new job rather than an edit.')),
+                ('progress', models.JSONField(blank=True, default=dict, help_text='Per-collection resume state. Read and written only.')),
+                ('artifacts', models.JSONField(blank=True, default=dict, help_text='References to captured raw payloads, never the payload bodies.')),
+                ('error', models.TextField(blank=True, default='', help_text='The single terminal reason this job failed: source unreachable, authorization revoked, etc.')),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+                ('destination_actor', models.ForeignKey(help_text='The destination Actor imported content is written to.', on_delete=django.db.models.deletion.CASCADE, related_name='inbound_transfers', to='core.actor')),
+            ],
+            options={
+                'ordering': ['-created_at'],
+                'indexes': [models.Index(fields=['destination_actor', 'state'], name='transferjob_actor_state_idx'), models.Index(fields=['state', 'retry_when'], name='transferjob_state_retry_idx')],
+                'constraints': [models.CheckConstraint(condition=models.Q(('error', ''), ('state', 'failed'), _negated=True), name='transferjob_failed_has_reason')],
+            },
+        ),
+    ]

testbed/core/models.py

@@ -574,6 +574,141 @@ def __str__(self):
         return f"TokenActorBinding(token_id={self.token_id}, actor_id={self.actor_id})"
 
 
+def default_transfer_policy():
+    return {"dry_run": True}
+
+
+class TransferJob(models.Model):
+    """
+    One attempted account copy, from the destination server's point of view.
+
+    A LOLA transfer cannot complete inside one HTTP request. The OAuth approval is a human click, so
+    the callback arrives on a different request than the one that started the transfer; and per
+    decision a single request may never walk a whole collection, because the container runs
+    `--workers 1 --threads 8` and a Mode C self-call would occupy one thread while waiting on
+    another. So a transfer is many requests, and this row is what carries state between them.
+
+    LOLA §6.7: "The destination server has more state to maintain to keep track of what has been already
+    copied and what remains to be fetched."
+    """
+
+    class State(models.TextChoices):
+        ACTIVE = "active", "Active"
+        FINISHED = "finished", "Finished"
+        FAILED = "failed", "Failed"
+
+    destination_actor = models.ForeignKey(
+        "Actor",
+        on_delete=models.CASCADE,
+        related_name="inbound_transfers",
+        help_text="The destination Actor imported content is written to.",
+    )
+
+    source_base_url = models.URLField(
+        max_length=500,
+        help_text="Base URL of the source server. Everything else about the source is discovered from it.",
+    )
+    source_actor_url = models.URLField(
+        max_length=500,
+        null=True,
+        blank=True,
+        help_text="Source Actor URL as supplied or resolved by discovery. Advisory, not authoritative.",
+    )
+    authorized_actor_url = models.URLField(
+        max_length=500,
+        null=True,
+        blank=True,
+        help_text=(
+            "The Actor URL the source returned in `activitypub_actor` on the callback (LOLA §5.3)."
+        ),
+    )
+
+    state = models.CharField(
+        max_length=16,
+        choices=State.choices,
+        default=State.ACTIVE,
+        help_text="The working phase is derived from the data, not stored.",
+    )
+    retry_when = models.DateTimeField(
+        null=True,
+        blank=True,
+        help_text=(
+            "Set from a 429's Retry-After. A job is paused."
+        ),
+    )
+
+    policy = models.JSONField(
+        default=default_transfer_policy,
+        blank=True,
+        help_text=(
+            "Chosen when the job is created. Changing the decision means a new job rather than an edit."
+        ),
+    )
+    progress = models.JSONField(
+        default=dict,
+        blank=True,
+        help_text=(
+            "Per-collection resume state. Read and written only."
+        ),
+    )
+    artifacts = models.JSONField(
+        default=dict,
+        blank=True,
+        help_text=(
+            "References to captured raw payloads, never the payload bodies."
+        ),
+    )
+    error = models.TextField(
+        blank=True,
+        default="",
+        help_text=(
+            "The single terminal reason this job failed: source unreachable, authorization revoked, etc."
+        ),
+    )
+
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    class Meta:
+        ordering = ["-created_at"]
+        indexes = [
+            # This user's jobs
+            models.Index(
+                fields=["destination_actor", "state"],
+                name="transferjob_actor_state_idx",
+            ),
+            # Which jobs are ready to resume?
+            models.Index(
+                fields=["state", "retry_when"], name="transferjob_state_retry_idx"
+            ),
+        ]
+        constraints = [
+            # A failed job must say why. Nothing else ties `state` to `error`
+            models.CheckConstraint(
+                condition=~models.Q(state="failed", error=""),
+                name="transferjob_failed_has_reason",
+            ),
+        ]
+
+    def __str__(self):
+        return f"Transfer {self.pk} for {self.user.username}: {self.state}"
+
+    @property
+    def user(self):
+        # The owner of this job, read through the destination Actor
+        return self.destination_actor.user
+
+    def get_collection_progress(self, collection):
+        from testbed.core.transfer import jobs
+
+        return jobs.get_collection_progress(self, collection)
+
+    def set_collection_progress(self, collection, **values):
+        from testbed.core.transfer import jobs
+
+        return jobs.set_collection_progress(self, collection, **values)
+
+
 class PortabilityOutbox(models.Model):
     actor = models.OneToOneField(
         Actor, on_delete=models.CASCADE, related_name="portability_outbox"

testbed/core/tests/test_transfer_jobs.py

@@ -0,0 +1,127 @@
+import pytest
+
+from testbed.core.factories import TransferJobFactory
+from testbed.core.models import TransferJob
+from testbed.core.transfer.jobs import Collection
+
+
+# The collection vocabulary
+
+
+def test_collections_exclude_followers():
+    # LOLA §6.6 Not Fetched: the Followers collection is reconstructed by followers choosing
+    # to re-follow, never copied.
+    assert set(Collection.values) == {
+        "content",
+        "outbox",
+        "following",
+        "blocked",
+        "liked",
+    }
+    assert "followers" not in Collection.values
+
+
+def test_an_unknown_collection_is_rejected_on_write():
+    # Without this a typo writes progress under a key nothing ever reads, and that collection
+    # silently restarts from page 1 on every resume, re-importing what it already imported
+    job = TransferJobFactory()
+
+    with pytest.raises(ValueError):
+        job.set_collection_progress("contnet", cursor="page-2")
+
+
+def test_an_unknown_collection_is_rejected_on_read():
+    # Reading is validated too. An unknown name would otherwise return {}, which is
+    # indistinguishable from "this collection has not started yet"
+    job = TransferJobFactory()
+
+    with pytest.raises(ValueError):
+        job.get_collection_progress("contnet")
+
+
+# Reading and writing resume state
+
+
+def test_unrecorded_collection_reads_as_empty():
+    job = TransferJobFactory()
+
+    assert job.get_collection_progress("content") == {}
+
+
+def test_set_persists_to_the_database():
+    # The whole point of the model: state must survive the request that wrote it
+    job = TransferJobFactory()
+
+    job.set_collection_progress("content", cursor="page-2", seen=20)
+
+    reloaded = TransferJob.objects.get(pk=job.pk)
+    assert reloaded.get_collection_progress("content") == {
+        "cursor": "page-2",
+        "seen": 20,
+    }
+
+
+def test_set_merges_rather_than_replaces():
+    job = TransferJobFactory()
+    job.set_collection_progress("content", cursor="page-1", seen=10)
+
+    job.set_collection_progress("content", seen=20)
+
+    reloaded = TransferJob.objects.get(pk=job.pk)
+    assert reloaded.get_collection_progress("content") == {
+        "cursor": "page-1",
+        "seen": 20,
+    }
+
+
+def test_writing_one_collection_leaves_its_siblings_alone():
+    # A fetch walk moves through collections one at a time.
+    # Recording outbox's position must not lose where content got to
+    job = TransferJobFactory()
+    job.set_collection_progress("content", cursor="page-3")
+    job.set_collection_progress("following", cursor="page-1")
+
+    job.set_collection_progress("outbox", cursor="page-1")
+
+    reloaded = TransferJob.objects.get(pk=job.pk)
+    assert reloaded.get_collection_progress("content") == {"cursor": "page-3"}
+    assert reloaded.get_collection_progress("following") == {"cursor": "page-1"}
+    assert reloaded.get_collection_progress("outbox") == {"cursor": "page-1"}
+
+
+def test_both_accessors_return_copies():
+    # A caller mutating what either function handed back would change the job in memory without
+    # persisting it, giving a resume position that exists in the process and not in the database.
+    job = TransferJobFactory()
+    job.set_collection_progress("content", cursor="page-1")
+
+    from_get = job.get_collection_progress("content")
+    from_set = job.set_collection_progress("content", seen=5)
+
+    assert from_set == {"cursor": "page-1", "seen": 5}  # set returns the merged state
+
+    from_get["cursor"] = "page-99"
+    from_set["seen"] = 999
+
+    assert job.get_collection_progress("content") == {"cursor": "page-1", "seen": 5}
+
+
+def test_set_writes_only_the_progress_column():
+    job = TransferJobFactory()
+    TransferJob.objects.filter(pk=job.pk).update(state=TransferJob.State.FINISHED)
+
+    # `job` still holds the stale in-memory state
+    job.set_collection_progress("content", cursor="page-1")
+
+    reloaded = TransferJob.objects.get(pk=job.pk)
+    assert reloaded.state == TransferJob.State.FINISHED
+    assert reloaded.get_collection_progress("content") == {"cursor": "page-1"}
+
+
+def test_set_refreshes_updated_at():
+    job = TransferJobFactory()
+    before = TransferJob.objects.get(pk=job.pk).updated_at
+
+    job.set_collection_progress("content", cursor="page-1")
+
+    assert TransferJob.objects.get(pk=job.pk).updated_at > before