Poll dispatch_due in the worker so scheduled posts fire.

Co-authored-by: Cursor <cursoragent@cursor.com>

.env.example

@@ -92,4 +92,7 @@ TIANJI_ENABLED=false
 TIANJI_TRACKER_URL=https://tianji.aimloperations.com/tracker.js
 TIANJI_WEBSITE_ID=cmshzhxdf6gee10qzkzrfn9iw
 
+# Worker poll interval for due scheduled campaigns / social posts (seconds).
+# DISPATCH_DUE_INTERVAL_SECONDS=15
+
 GUNICORN_WORKERS=2

README.md

@@ -25,13 +25,14 @@ uv sync
 cd site
 uv run python manage.py migrate
 uv run python manage.py runserver
-# optional: docker compose up   # web + postgres (hot reload)
+# optional: docker compose up   # web + postgres + worker (hot reload)
-# optional worker: docker compose --profile worker up worker
 ```
 
 Docker compose mounts `./site` into the container and runs Django `runserver` when
-`DJANGO_ENV=dev` (the default). Edit Python/templates → auto-reload; no image rebuild.
+`DJANGO_ENV=dev` (the default). The **worker** service polls `dispatch_due` every 15s so
-Rebuild only when Dockerfile / deps (`pyproject.toml`, `uv.lock`) change:
+scheduled campaigns and social posts fire without a manual command. Edit Python/templates
+→ auto-reload; no image rebuild.
+Rebuild only when Dockerfile, entrypoint scripts, or deps (`pyproject.toml`, `uv.lock`) change:
 
 ```bash
 docker compose up --build

docker-compose.prod.yml

@@ -4,6 +4,7 @@
 # Web runs on every app host (active/active). Start the dj-queue worker on
 # exactly ONE host via:
 #   docker compose -f docker-compose.prod.yml --profile worker up -d
+# Worker also polls dispatch_due so scheduled campaigns/social posts fire.
 services:
   web:
     build: .

docker-compose.yml

@@ -67,7 +67,6 @@ services:
 
   worker:
     build: .
-    profiles: ["worker"]
     entrypoint: ["/worker-entrypoint.sh"]
     volumes:
       - ./site:/app/site
@@ -77,6 +76,8 @@ services:
       DJANGO_DEBUG: ${DJANGO_DEBUG:-true}
       DJANGO_ALLOWED_HOSTS: ${DJANGO_ALLOWED_HOSTS:-localhost,127.0.0.1,0.0.0.0}
       DATABASE_URL: ${COMPOSE_DATABASE_URL:-postgres://monica_site:monica_site@db:5432/monica_site}
+      PUBLIC_SITE_URL: ${PUBLIC_SITE_URL:-http://127.0.0.1:8000}
+      DISPATCH_DUE_INTERVAL_SECONDS: ${DISPATCH_DUE_INTERVAL_SECONDS:-15}
       EMAIL_HOST: ${EMAIL_HOST:-mail.smtp2go.com}
       EMAIL_HOST_USER: ${EMAIL_HOST_USER:-}
       EMAIL_HOST_PASSWORD: ${EMAIL_HOST_PASSWORD:-}

scripts/worker-entrypoint.sh

@@ -47,5 +47,24 @@ uv run python manage.py migrate --noinput
 # worker must too or hashed lookups raise ValueError.
 uv run python manage.py collectstatic --noinput
 
+interval="${DISPATCH_DUE_INTERVAL_SECONDS:-15}"
+echo "Starting dispatch_due poller (every ${interval}s)."
+uv run python manage.py dispatch_due --loop --interval "${interval}" &
+dispatch_pid=$!
+
+cleanup() {
+  echo "Stopping dispatch_due poller..."
+  kill "${dispatch_pid}" 2>/dev/null || true
+  wait "${dispatch_pid}" 2>/dev/null || true
+}
+trap cleanup EXIT INT TERM
+
+# Dev uses ImmediateBackend: enqueue runs inline in dispatch_due. No dj-queue.
+if [[ "${DJANGO_ENV:-}" == "dev" ]]; then
+  echo "Dev worker: dispatch_due loop publishes scheduled work."
+  wait "${dispatch_pid}"
+  exit $?
+fi
+
 # dj-queue supervisor (workers + dispatcher + scheduler). Run on ONE host only.
-exec uv run python manage.py dj_queue
+uv run python manage.py dj_queue

site/core/management/commands/dispatch_due.py

@@ -1,3 +1,5 @@
+import time
+
 from django.core.management.base import BaseCommand
 from django.utils import timezone
 
@@ -10,10 +12,37 @@ from social.tasks import publish_social_post
 class Command(BaseCommand):
     help = (
         "Enqueue due scheduled campaign messages and social posts. "
-        "Optional when the task backend supports run_after defer; useful as a safety net."
+        "Pass --loop to poll until stopped (worker entrypoint)."
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--loop",
+            action="store_true",
+            help="Poll forever until SIGINT/SIGTERM.",
+        )
+        parser.add_argument(
+            "--interval",
+            type=int,
+            default=15,
+            help="Seconds between polls when --loop is set (default 15).",
         )
 
     def handle(self, *args, **options):
+        self.verbosity = int(options.get("verbosity", 1))
+        if options["loop"]:
+            interval = max(1, int(options["interval"] or 15))
+            self.stdout.write(f"Polling due scheduled work every {interval}s.")
+            try:
+                while True:
+                    self._dispatch_once()
+                    time.sleep(interval)
+            except KeyboardInterrupt:
+                self.stdout.write("dispatch_due loop stopped.")
+                return
+        self._dispatch_once()
+
+    def _dispatch_once(self):
         now = timezone.now()
         enqueued = 0
 
@@ -35,4 +64,7 @@ class Command(BaseCommand):
             publish_social_post.enqueue(post_id=str(post.pk))
             enqueued += 1
 
+        if enqueued:
             self.stdout.write(self.style.SUCCESS(f"Enqueued {enqueued} due item(s)."))
+        elif self.verbosity >= 2:
+            self.stdout.write("No due items.")

site/core/tests.py

@@ -1,5 +1,13 @@
+from datetime import timedelta
+from io import StringIO
+from unittest.mock import patch
+
+from django.core.management import call_command
 from django.test import Client, TestCase, override_settings
 from django.urls import reverse
+from django.utils import timezone
+
+from social.models import SocialPost
 
 
 class HealthzTests(TestCase):
@@ -27,3 +35,43 @@ class PublicSmokeTests(TestCase):
         client = Client()
         self.assertEqual(client.get(reverse("public:about")).status_code, 200)
         self.assertEqual(client.get(reverse("public:contact")).status_code, 200)
+
+
+class DispatchDueTests(TestCase):
+    def test_enqueues_past_scheduled_social_post(self):
+        post = SocialPost.objects.create(
+            body="Open house",
+            status=SocialPost.Status.SCHEDULED,
+            scheduled_for=timezone.now() - timedelta(minutes=5),
+        )
+        mock_task = patch(
+            "core.management.commands.dispatch_due.publish_social_post"
+        ).start()
+        self.addCleanup(patch.stopall)
+        out = StringIO()
+        call_command("dispatch_due", stdout=out)
+        post.refresh_from_db()
+        self.assertEqual(post.status, SocialPost.Status.QUEUED)
+        mock_task.enqueue.assert_called_once_with(post_id=str(post.pk))
+        self.assertIn("Enqueued 1 due item(s).", out.getvalue())
+
+    def test_skips_future_scheduled_social_post(self):
+        post = SocialPost.objects.create(
+            body="Later",
+            status=SocialPost.Status.SCHEDULED,
+            scheduled_for=timezone.now() + timedelta(hours=1),
+        )
+        mock_task = patch(
+            "core.management.commands.dispatch_due.publish_social_post"
+        ).start()
+        self.addCleanup(patch.stopall)
+        call_command("dispatch_due", stdout=StringIO())
+        post.refresh_from_db()
+        self.assertEqual(post.status, SocialPost.Status.SCHEDULED)
+        mock_task.enqueue.assert_not_called()
+
+    def test_quiet_when_nothing_due(self):
+        out = StringIO()
+        call_command("dispatch_due", stdout=out)
+        self.assertEqual(out.getvalue(), "")
+