feat(ops): add AUTH_KEY endpoints to trigger feed sync tasks

Expose POST/GET /api/v1/ops/feeds/sync for inline debug runs or async Celery
queueing of news, seismic, and trails sync jobs.

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

Author: hc-sousa

src/shared/feed_syncs.py

@@ -0,0 +1,117 @@
+"""Feed sync registry — shared by bootstrap command and ops HTTP triggers."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable
+
+logger = logging.getLogger(__name__)
+
+FeedRunner = Callable[[str | None], dict[str, Any]]
+
+FEED_LABELS: tuple[str, ...] = ('news', 'seismic', 'trails')
+
+FEED_TASK_NAMES: dict[str, str] = {
+    'news': 'news.poll_sources',
+    'seismic': 'seismic.sync_events',
+    'trails': 'trails.sync_open_data',
+}
+
+
+def _run_news(island_key: str | None) -> dict[str, Any]:
+    from news.services import poll_all_sources
+
+    return {'status': 'ok', **poll_all_sources(island_key=island_key)}
+
+
+def _run_seismic(island_key: str | None) -> dict[str, Any]:
+    from seismic.services import sync_all_events
+
+    return {'status': 'ok', **sync_all_events(island_key=island_key)}
+
+
+def _run_trails(island_key: str | None) -> dict[str, Any]:
+    from trails.services import sync_all_open_data
+
+    return {'status': 'ok', **sync_all_open_data(island_key=island_key)}
+
+
+FEED_RUNNERS: dict[str, FeedRunner] = {
+    'news': _run_news,
+    'seismic': _run_seismic,
+    'trails': _run_trails,
+}
+
+
+def _queue_news(island_key: str | None):
+    from news.tasks import poll_sources_task
+
+    return poll_sources_task.delay(island_key=island_key)
+
+
+def _queue_seismic(island_key: str | None):
+    from seismic.tasks import sync_events_task
+
+    return sync_events_task.delay(island_key=island_key)
+
+
+def _queue_trails(island_key: str | None):
+    from trails.tasks import sync_open_data_task
+
+    return sync_open_data_task.delay(island_key=island_key)
+
+
+FEED_QUEUERS = {
+    'news': _queue_news,
+    'seismic': _queue_seismic,
+    'trails': _queue_trails,
+}
+
+
+def normalize_feed_param(feed: str) -> list[str]:
+    value = (feed or 'all').strip().lower()
+    if value == 'all':
+        return list(FEED_LABELS)
+    if value in FEED_LABELS:
+        return [value]
+    raise ValueError(f'Unknown feed {feed!r}; use all or one of {", ".join(FEED_LABELS)}')
+
+
+def run_feed_sync(label: str, *, island_key: str | None = None) -> dict[str, Any]:
+    runner = FEED_RUNNERS[label]
+    return runner(island_key)
+
+
+def queue_feed_sync(label: str, *, island_key: str | None = None) -> dict[str, Any]:
+    async_result = FEED_QUEUERS[label](island_key)
+    return {
+        'task': FEED_TASK_NAMES[label],
+        'celery_task_id': async_result.id,
+    }
+
+
+def trigger_feed_syncs(
+    labels: list[str],
+    *,
+    island_key: str | None = None,
+    run_async: bool = False,
+) -> dict[str, Any]:
+    results: dict[str, Any] = {}
+    for label in labels:
+        try:
+            if run_async:
+                queued = queue_feed_sync(label, island_key=island_key)
+                results[label] = {'ok': True, 'mode': 'async', **queued}
+            else:
+                payload = run_feed_sync(label, island_key=island_key)
+                results[label] = {'ok': True, 'mode': 'sync', **payload}
+        except Exception as exc:
+            logger.exception('feed sync failed label=%s island=%s', label, island_key)
+            results[label] = {
+                'ok': False,
+                'mode': 'async' if run_async else 'sync',
+                'task': FEED_TASK_NAMES.get(label),
+                'error': str(exc),
+                'error_type': type(exc).__name__,
+            }
+    return results

src/shared/management/commands/bootstrap_feed_syncs.py

@@ -6,37 +6,22 @@
 
 from django.core.management.base import BaseCommand
 
-logger = logging.getLogger(__name__)
+from shared.feed_syncs import FEED_LABELS, queue_feed_sync
 
-FEED_TASKS: tuple[tuple[str, str], ...] = (
-    ('news', 'news.poll_sources'),
-    ('seismic', 'seismic.sync_events'),
-    ('trails', 'trails.sync_open_data'),
-)
+logger = logging.getLogger(__name__)
 
 
 class Command(BaseCommand):
     help = 'Queue initial news, seismic, and trails feed sync tasks (runs after migrate on deploy).'
 
     def handle(self, *args: object, **options: object) -> None:
         queued: list[str] = []
-        for label, task_name in FEED_TASKS:
+        for label in FEED_LABELS:
             try:
-                if label == 'news':
-                    from news.tasks import poll_sources_task
-
-                    poll_sources_task.delay()
-                elif label == 'seismic':
-                    from seismic.tasks import sync_events_task
-
-                    sync_events_task.delay()
-                elif label == 'trails':
-                    from trails.tasks import sync_open_data_task
-
-                    sync_open_data_task.delay()
-                queued.append(task_name)
+                info = queue_feed_sync(label)
+                queued.append(info['task'])
             except Exception:
-                logger.exception('bootstrap_feed_syncs failed to queue %s', task_name)
+                logger.exception('bootstrap_feed_syncs failed to queue %s', label)
 
         if queued:
             self.stdout.write(f'Queued feed sync tasks: {", ".join(queued)}')

src/tenancy/tests/test_ops_feeds.py

@@ -0,0 +1,74 @@
+"""Ops feed sync trigger tests."""
+
+from unittest.mock import patch
+
+from django.conf import settings
+from django.test import TestCase, override_settings
+from rest_framework.test import APIClient
+
+
+@override_settings(AUTH_KEY='test-auth-key')
+class OpsFeedSyncTestCase(TestCase):
+    def setUp(self):
+        self.client = APIClient()
+        self.url = '/api/v1/ops/feeds/sync'
+        self.auth = {'QUERY_STRING': f'key={settings.AUTH_KEY}'}
+
+    def test_requires_auth_key(self):
+        response = self.client.post(f'{self.url}?feed=news')
+        self.assertEqual(response.status_code, 401)
+
+    def test_invalid_feed(self):
+        response = self.client.post(f'{self.url}?key={settings.AUTH_KEY}&feed=bad')
+        self.assertEqual(response.status_code, 400)
+
+    @patch('shared.feed_syncs.run_feed_sync')
+    def test_sync_trails_inline(self, mock_run):
+        mock_run.return_value = {
+            'status': 'ok',
+            'islands': 1,
+            'trails_created': 2,
+            'trails_updated': 0,
+            'pois_created': 0,
+            'pois_updated': 0,
+            'skipped': 0,
+        }
+        response = self.client.post(
+            f'{self.url}?key={settings.AUTH_KEY}&feed=trails&island=sao-miguel',
+        )
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertTrue(body['ok'])
+        self.assertFalse(body['async'])
+        self.assertEqual(body['feeds']['trails']['mode'], 'sync')
+        self.assertEqual(body['feeds']['trails']['trails_created'], 2)
+        mock_run.assert_called_once_with('trails', island_key='sao-miguel')
+
+    @patch('shared.feed_syncs.queue_feed_sync')
+    def test_async_queues_celery(self, mock_queue):
+        mock_queue.return_value = {
+            'task': 'trails.sync_open_data',
+            'celery_task_id': 'abc-123',
+        }
+        response = self.client.post(
+            f'{self.url}?key={settings.AUTH_KEY}&feed=trails&async=true',
+        )
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertTrue(body['async'])
+        self.assertEqual(body['feeds']['trails']['celery_task_id'], 'abc-123')
+
+    @patch('shared.feed_syncs.run_feed_sync')
+    def test_sync_all_feeds(self, mock_run):
+        mock_run.return_value = {'status': 'ok', 'created': 0}
+        response = self.client.get(f'{self.url}?key={settings.AUTH_KEY}&feed=all')
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(mock_run.call_count, 3)
+
+    @patch('shared.feed_syncs.run_feed_sync', side_effect=RuntimeError('boom'))
+    def test_sync_error_returns_502(self, _mock_run):
+        response = self.client.post(f'{self.url}?key={settings.AUTH_KEY}&feed=news')
+        self.assertEqual(response.status_code, 502)
+        body = response.json()
+        self.assertFalse(body['ok'])
+        self.assertEqual(body['feeds']['news']['error_type'], 'RuntimeError')

src/tenancy/urls.py

@@ -8,4 +8,9 @@
         views.cancel_all_celery_jobs,
         name='ops_celery_cancel_all',
     ),
+    path(
+        'feeds/sync',
+        views.trigger_feed_sync,
+        name='ops_feed_sync',
+    ),
 ]

src/tenancy/views.py

@@ -51,3 +51,49 @@ def cancel_all_celery_jobs(request: Request) -> Response:
             **result,
         }
     )
+
+
+@api_view(['GET', 'POST'])
+@permission_classes([AllowAny])
+def trigger_feed_sync(request: Request) -> Response:
+    """
+    Run or queue feed sync tasks (news, seismic, trails) for debugging.
+
+    Query params:
+      - key / X-Auth-Key: AUTH_KEY (required)
+      - feed: all | news | seismic | trails (default all)
+      - island: island key slug, e.g. sao-miguel (optional)
+      - async: true to queue on Celery; false to run inline and return counts (default false)
+    """
+    denied = _require_auth_key(request)
+    if denied:
+        return denied
+
+    from shared.feed_syncs import FEED_LABELS, normalize_feed_param, trigger_feed_syncs
+
+    feed_param = request.query_params.get('feed', 'all')
+    island_key = (request.query_params.get('island') or '').strip() or None
+    run_async = request.query_params.get('async', 'false').lower() in ('1', 'true', 'yes')
+
+    try:
+        labels = normalize_feed_param(feed_param)
+    except ValueError as exc:
+        return Response(
+            {
+                'error': str(exc),
+                'allowed_feeds': ['all', *FEED_LABELS],
+            },
+            status=400,
+        )
+
+    results = trigger_feed_syncs(labels, island_key=island_key, run_async=run_async)
+    all_ok = all(item.get('ok') for item in results.values())
+    return Response(
+        {
+            'ok': all_ok,
+            'island_key': island_key,
+            'async': run_async,
+            'feeds': results,
+        },
+        status=200 if all_ok else 502,
+    )