feat(learners): send a learner's next content delivery immediately

CHANGELOG.md

@@ -6,6 +6,12 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 Changes prior to v1.0.0 are available in the [git history](https://github.com/AvaCodeSolutions/django-email-learning/commits/master).
 
+## [Unreleased]
+
+### Added
+
+- **Send a learner's next content immediately from the Learners page** — The enrollment dialog now shows, under the course title, when the learner's next content is scheduled to arrive and which content it is, with a **send now** link beside it for organization admins. Sending runs the delivery there and then: the email goes out, the schedule is marked delivered, and the follow-up work happens exactly as it would during a job run — the next content is scheduled, or the enrollment graduates if that was the last one. Bringing the schedule's time forward was never equivalent, because the delivery job runs on a cron and the content would still wait for its next tick. Behind it is a new admin-only endpoint, `POST /api/platform/organizations/<id>/enrollments/<id>/delivery-schedules/<id>/send/`, and a `next_delivery` field on the enrollment detail response (`null` when nothing is scheduled). The schedule is claimed with the same `SCHEDULED → PROCESSING` compare-and-set the delivery queue uses, so a job run happening at the same moment cannot send the same content twice; a delivery that is no longer scheduled returns `409` and is left untouched, and one that fails to send is retried or blocked by the job's own retry logic.
+
 ## [5.0.0] - 2026-08-15
 
 > **Upgrading.** No migrations. One thing to check before you upgrade: if you call `POST /api/v1/enrollments/` and rely on the learner receiving a verification link, add `"verified": false` to the request body — that is now opt-in, and the default creates the enrollment active instead. Callers that read `status` from the `201` response get `active` rather than `unverified`. Nothing else in the API changed.

django_email_learning/jobs/deliver_contents_job.py

@@ -40,8 +40,13 @@ def _get_delivery_workers() -> int:
 
 
 class DeliverContentsJob:
-    def __init__(self) -> None:
-        self.delivery_queue: TaskQueueProtocol[DeliverySchedule] = self.get_delivery_queue()
+    def __init__(self, delivery_queue: TaskQueueProtocol[DeliverySchedule] | None = None) -> None:
+        # A caller that only needs `process_delivery` for a schedule it has
+        # already claimed can pass its own queue, so constructing the job does
+        # not claim a batch of work the running job should be handling.
+        self.delivery_queue: TaskQueueProtocol[DeliverySchedule] = (
+            delivery_queue if delivery_queue is not None else self.get_delivery_queue()
+        )
 
     def start(self) -> JobExecution | None:
         job_execution = JobExecution.start_if_not_running(job_name=JobName.DELIVER_CONTENTS.value)
@@ -80,7 +85,7 @@ def _run_sequential(self, job_execution: JobExecution) -> None:
             try:
                 self.process_delivery(delivery_schedule)
             except Exception as e:
-                self._block_delivery(delivery_schedule, e)
+                self.block_delivery(delivery_schedule, e)
 
     # ── threaded (workers > 1) ───────────────────────────────────────────────
 
@@ -106,7 +111,7 @@ def _run_threaded(self, job_execution: JobExecution, workers: int) -> None:
                     try:
                         future.result()
                     except Exception as e:
-                        self._block_delivery(delivery_schedule, e)
+                        self.block_delivery(delivery_schedule, e)
 
                 # Avoid a tight spin-wait when all workers are busy
                 if not done and futures:
@@ -123,10 +128,10 @@ def _worker(self, delivery_schedule: DeliverySchedule) -> None:
         try:
             self.process_delivery(delivery_schedule)
         except Exception as e:
-            self._block_delivery(delivery_schedule, e)
+            self.block_delivery(delivery_schedule, e)
             raise
 
-    def _block_delivery(self, delivery_schedule: DeliverySchedule, exc: Exception) -> None:
+    def block_delivery(self, delivery_schedule: DeliverySchedule, exc: Exception) -> None:
         """Mark a delivery as BLOCKED and emit a metric."""
         delivery_schedule.status = DeliveryStatus.BLOCKED
         delivery_schedule.save()

django_email_learning/platform/api/serializers/__init__.py

@@ -39,6 +39,7 @@
     Event,
     EventType,
     LearnerDetailResponse,
+    NextDeliveryResponse,
     QuizSubmitedEvent,
     ReminderSentEvent,
 )
@@ -143,6 +144,7 @@
     "EmailOpenedEvent",
     "Event",
     "EnrollmentResponse",
+    "NextDeliveryResponse",
     "LearnerDetailResponse",
     "GetOrCreateUserRequest",
     "UserResponse",

django_email_learning/platform/api/serializers/learners.py

@@ -8,6 +8,7 @@
     AssignmentSubmission,
     ContentDelivery,
     CourseContentType,
+    DeliverySchedule,
     DeliveryStatus,
     Enrollment,
     EnrollmentStatus,
@@ -101,12 +102,51 @@ class Event(BaseModel):
     ) = Field(discriminator="type")  # REGISTERED, VERIFIED, COURSE_COMPLETED have no additional data
 
 
+class NextDeliveryResponse(BaseModel):
+    """The next content the learner is scheduled to receive on this enrollment."""
+
+    delivery_schedule_id: int
+    course_content_id: int
+    course_content_title: str
+    course_content_type: str
+    scheduled_at: datetime
+
+
+def _next_delivery(enrollment: Enrollment) -> NextDeliveryResponse | None:
+    """The earliest still-scheduled delivery for `enrollment`, if any.
+
+    Ordered by time, then id, matching the order the delivery job would reach
+    them in, so this names the delivery a "send now" acts on.
+    """
+    schedule = (
+        DeliverySchedule.objects.filter(
+            delivery__enrollment=enrollment,
+            status=DeliveryStatus.SCHEDULED,
+        )
+        .select_related("delivery__course_content")
+        .order_by("time", "id")
+        .first()
+    )
+    if schedule is None:
+        return None
+
+    course_content = schedule.delivery.course_content
+    return NextDeliveryResponse(
+        delivery_schedule_id=schedule.id,
+        course_content_id=course_content.id,
+        course_content_title=course_content.title,
+        course_content_type=course_content.type,
+        scheduled_at=schedule.time,
+    )
+
+
 class EnrollmentResponse(BaseModel):
     id: int
     learner: LearnerResponse
     course: CourseSummaryResponse
     status: EnrollmentStatus
     events: list[Event]
+    next_delivery: NextDeliveryResponse | None = None
 
     @staticmethod
     def from_django_model(enrollment: Enrollment) -> "EnrollmentResponse":
@@ -267,6 +307,7 @@ def from_django_model(enrollment: Enrollment) -> "EnrollmentResponse":
                 "course": enrollment.course,
                 "status": enrollment.status,
                 "events": events,
+                "next_delivery": _next_delivery(enrollment),
             }
         )
 

django_email_learning/platform/api/urls.py

@@ -23,6 +23,7 @@
     OrganizationsView,
     OrganizationUsersView,
     ReorderCourseContentView,
+    SendDeliveryScheduleNowView,
     SendLessonToPlatformUser,
     SendoutView,
     SingleApiKeyView,
@@ -146,6 +147,12 @@
         EnrollmentView.as_view(),
         name="enrollments_detail",
     ),
+    path(
+        "organizations/<int:organization_id>/enrollments/<int:enrollment_id>/"
+        "delivery-schedules/<int:delivery_schedule_id>/send/",
+        SendDeliveryScheduleNowView.as_view(),
+        name="delivery_schedule_send_now",
+    ),
     path(
         "organizations/<int:organization_id>/courses/<int:course_id>/enrollments/",
         EnrollmentsView.as_view(),

django_email_learning/platform/api/views/__init__.py

@@ -18,6 +18,7 @@
     EnrollmentsView,
     EnrollmentView,
     LearnersView,
+    SendDeliveryScheduleNowView,
     SingleLearnerView,
 )
 from django_email_learning.platform.api.views.misc import (
@@ -84,6 +85,7 @@
     "EnrollmentsView",
     "EnrollmentView",
     "EnrollmentsStatisticsView",
+    "SendDeliveryScheduleNowView",
     "NewsletterView",
     "NewsletterEmbedSnippetView",
     "SingleNewsletterView",

django_email_learning/platform/api/views/learners.py

@@ -18,6 +18,7 @@
 from django_email_learning.models import (
     Certificate,
     Course,
+    DeliverySchedule,
     Enrollment,
     EnrollmentStatus,
     Learner,
@@ -37,6 +38,10 @@
 from django_email_learning.services.command_models.verify_enrollment_command import (
     VerifyEnrollmentCommand,
 )
+from django_email_learning.services.manual_delivery_service import (
+    ManualDeliveryOutcome,
+    send_delivery_schedule_now,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -197,6 +202,55 @@ def get(self, request, *args, **kwargs) -> JsonResponse:  # type: ignore[no-unty
             return JsonResponse({"error": e.json()}, status=400)
 
 
+@method_decorator(accessible_for(roles={"admin"}), name="post")
+class SendDeliveryScheduleNowView(View):
+    """Sends one scheduled content delivery immediately.
+
+    Admin-only: it puts an email in a learner's inbox and advances their
+    enrollment, which is a heavier action than the read access the rest of the
+    enrollment views grant. Changing the schedule's time would not do the same
+    job - the delivery job runs on a cron, so the content would still go out
+    whenever that next fires.
+    """
+
+    def post(self, request, *args, **kwargs) -> JsonResponse:  # type: ignore[no-untyped-def]
+        try:
+            enrollment = Enrollment.objects.get(
+                id=kwargs["enrollment_id"], course__organization_id=kwargs["organization_id"]
+            )
+        except Enrollment.DoesNotExist:
+            return JsonResponse({"error": "Enrollment not found"}, status=404)
+
+        try:
+            delivery_schedule = DeliverySchedule.objects.get(
+                id=kwargs["delivery_schedule_id"],
+                delivery__enrollment=enrollment,
+            )
+        except DeliverySchedule.DoesNotExist:
+            return JsonResponse({"error": "Delivery schedule not found"}, status=404)
+
+        result = send_delivery_schedule_now(delivery_schedule)
+
+        if result.outcome == ManualDeliveryOutcome.NOT_SCHEDULED:
+            return JsonResponse(
+                {"error": "Delivery is no longer scheduled", "delivery_status": result.delivery_status},
+                status=409,
+            )
+        if result.outcome == ManualDeliveryOutcome.FAILED:
+            logger.error(
+                f"Manual send of DeliverySchedule {delivery_schedule.id} failed with status {result.delivery_status}."
+            )
+            return JsonResponse(
+                {"error": "Delivery failed", "delivery_status": result.delivery_status},
+                status=500,
+            )
+
+        return JsonResponse(
+            {"status": result.delivery_status, "delivery_schedule_id": delivery_schedule.id},
+            status=200,
+        )
+
+
 @method_decorator(accessible_for(roles={"admin", "editor", "instructor", "viewer"}), name="get")
 class EnrollmentsStatisticsView(View):
     def get(self, request, *args, **kwargs) -> JsonResponse:  # type: ignore[no-untyped-def]

django_email_learning/platform/views/learners.py

@@ -65,4 +65,10 @@ def get_locale_messages(self) -> Dict[str, str]:
             "reset_filters": _("Reset Filters"),
             "progress": _("Progress"),
             "no_learners_found": _("No learners found."),
+            "next_delivery": _("Next delivery"),
+            "send_now": _("send now"),
+            "sending": _("Sending..."),
+            "content_sent_successfully": _("Content sent."),
+            "content_send_failed": _("The content could not be sent."),
+            "delivery_no_longer_scheduled": _("This delivery is no longer scheduled."),
         }

django_email_learning/services/manual_delivery_service.py

@@ -0,0 +1,120 @@
+"""Runs a single scheduled content delivery immediately, outside the job.
+
+The delivery job claims whatever is due and processes it, which means an
+operator who wants one learner to receive their next content *now* has no
+lever other than editing the schedule's time and waiting for the next job
+run - and the job runs on a cron, so "now" is really "some time within the
+next interval". This module does the same work the job does for exactly one
+`DeliverySchedule`, so the email goes out, the schedule is marked delivered,
+and the follow-up scheduling (next content, or graduation) happens in the
+same request.
+
+The row is claimed with the same `SCHEDULED -> PROCESSING` compare-and-set the
+database queue uses, so a job run happening at the same moment cannot pick up
+the same schedule and send the content twice.
+"""
+
+import logging
+from dataclasses import dataclass
+from enum import StrEnum
+
+from django.utils import timezone
+
+from django_email_learning.jobs.deliver_contents_job import DeliverContentsJob
+from django_email_learning.models import DeliverySchedule, DeliveryStatus
+from django_email_learning.ports.task_queue_protocol import TaskQueueProtocol
+from django_email_learning.services.metrics_service import metric_service
+
+logger = logging.getLogger(__name__)
+
+
+class ManualDeliveryOutcome(StrEnum):
+    DELIVERED = "delivered"
+    NOT_SCHEDULED = "not_scheduled"
+    FAILED = "failed"
+
+
+@dataclass(frozen=True)
+class ManualDeliveryResult:
+    outcome: ManualDeliveryOutcome
+    delivery_status: str
+
+
+class _EmptyDeliveryQueue(TaskQueueProtocol[DeliverySchedule]):
+    """A queue that never yields work.
+
+    `DeliverContentsJob` builds its queue on construction, and the database
+    queue claims a batch of due schedules the moment it is built. Sending one
+    delivery must not take rows away from a running job, so the job instance
+    used here is given a queue with nothing in it: only `process_delivery` is
+    called, and that takes its schedule as an argument.
+    """
+
+    def next_task(self) -> DeliverySchedule | None:
+        return None
+
+
+def send_delivery_schedule_now(delivery_schedule: DeliverySchedule) -> ManualDeliveryResult:
+    """Deliver `delivery_schedule` as if the job had just picked it up.
+
+    Returns the outcome together with the schedule's resulting status. A
+    schedule that is not `SCHEDULED` when the claim runs is left untouched and
+    reported as `NOT_SCHEDULED` - it is already delivered, canceled, blocked,
+    or in the hands of a running job.
+    """
+    claimed = DeliverySchedule.objects.filter(id=delivery_schedule.id, status=DeliveryStatus.SCHEDULED).update(
+        status=DeliveryStatus.PROCESSING,
+        # The delivery is happening now, so the schedule should say so: it keeps
+        # `delivered_at` consistent with `time`, and a retry after a failed
+        # attempt is then measured from now rather than from a future date.
+        time=timezone.now(),
+    )
+    if not claimed:
+        delivery_schedule.refresh_from_db()
+        logger.info(
+            f"Manual delivery skipped for DeliverySchedule ID {delivery_schedule.id}: "
+            f"status is {delivery_schedule.status}, not {DeliveryStatus.SCHEDULED}."
+        )
+        return ManualDeliveryResult(
+            outcome=ManualDeliveryOutcome.NOT_SCHEDULED,
+            delivery_status=delivery_schedule.status,
+        )
+
+    claimed_schedule = DeliverySchedule.objects.select_related(
+        "delivery__enrollment__learner",
+        "delivery__course_content__course__organization",
+        "delivery__course_content__course__imap_connection",
+        "delivery__course_content__lesson",
+        "delivery__course_content__quiz",
+        "delivery__course_content__assignment",
+    ).get(id=delivery_schedule.id)
+
+    job = DeliverContentsJob(delivery_queue=_EmptyDeliveryQueue())
+    try:
+        job.process_delivery(claimed_schedule)
+    except Exception as e:
+        job.block_delivery(claimed_schedule, e)
+        return ManualDeliveryResult(outcome=ManualDeliveryOutcome.FAILED, delivery_status=DeliveryStatus.BLOCKED)
+
+    claimed_schedule.refresh_from_db()
+    if claimed_schedule.status == DeliveryStatus.PROCESSING:
+        # `process_delivery` recognised no content to send - a content row whose
+        # type has no matching lesson/quiz/assignment. Leaving it PROCESSING
+        # would hide it from the job forever, so hand it back to the schedule.
+        logger.error(
+            f"Manual delivery for DeliverySchedule ID {claimed_schedule.id} produced no delivery. "
+            f"Returning it to {DeliveryStatus.SCHEDULED}."
+        )
+        claimed_schedule.status = DeliveryStatus.SCHEDULED
+        claimed_schedule.save()
+        metric_service.delivery_schedule_blocked(claimed_schedule.delivery.course_content.id)
+        return ManualDeliveryResult(outcome=ManualDeliveryOutcome.FAILED, delivery_status=claimed_schedule.status)
+
+    if claimed_schedule.status != DeliveryStatus.DELIVERED:
+        logger.warning(
+            f"Manual delivery for DeliverySchedule ID {claimed_schedule.id} ended as {claimed_schedule.status}."
+        )
+        return ManualDeliveryResult(outcome=ManualDeliveryOutcome.FAILED, delivery_status=claimed_schedule.status)
+
+    logger.info(f"Manual delivery completed for DeliverySchedule ID {claimed_schedule.id}.")
+    return ManualDeliveryResult(outcome=ManualDeliveryOutcome.DELIVERED, delivery_status=claimed_schedule.status)