diff --git a/app/main.py b/app/main.py index 69c88dd..1988981 100644 --- a/app/main.py +++ b/app/main.py @@ -1,5 +1,6 @@ """manages the job and pubsub message handling""" +import json import logging import time from typing import List @@ -11,7 +12,33 @@ from app.process import process_messages -logging.basicConfig(level=settings.LOG_LEVEL) +class _GCPFormatter(logging.Formatter): + """Emit structured JSON so GCP Cloud Logging picks up the correct severity.""" + + _SEVERITY = { + logging.DEBUG: "DEBUG", + logging.INFO: "INFO", + logging.WARNING: "WARNING", + logging.ERROR: "ERROR", + logging.CRITICAL: "CRITICAL", + } + + def format(self, record: logging.LogRecord) -> str: + message = record.getMessage() + if record.exc_info: + message += "\n" + self.formatException(record.exc_info) + return json.dumps({ + "severity": self._SEVERITY.get(record.levelno, "DEFAULT"), + "message": message, + "logger": record.name, + }) + + +_handler = logging.StreamHandler() +_handler.setFormatter(_GCPFormatter()) +logging.root.setLevel(settings.LOG_LEVEL) +logging.root.addHandler(_handler) + logger = logging.getLogger(__name__) @@ -52,6 +79,8 @@ def main(): if not settings.REDIRECT_EMAILS and settings.ENV != "PRODUCTION": logger.error("SEND_EMAILS=True and REDIRECT_EMAILS=False but ENV is not PRODUCTION — exiting") return + elif settings.ENV == "PRODUCTION": + logger.warning("SEND_EMAILS=False in PRODUCTION — messages will be acked without sending email") #get messages subscriber = pubsub_v1.SubscriberClient() diff --git a/app/process.py b/app/process.py index 6f2b23b..cfe6a22 100644 --- a/app/process.py +++ b/app/process.py @@ -134,10 +134,6 @@ def _send_email_tasks( ) -> None: """render and send emails, acking each submission only after its email sends""" - if not settings.SEND_EMAILS: - logger.info(f"Email sending disabled — {len(email_tasks)} email task(s) skipped") - return - if settings.REDIRECT_EMAILS: logger.info(f"REDIRECT_EMAILS active — all emails → {settings.REDIRECT_RECIPIENT}") @@ -167,8 +163,8 @@ def _send_email_tasks( submission_id=task.submission_id, reply_to_emails=task.reply_to_emails, ) + ack_fn(task.notifications.ack_ids) # ack on success or all-refused (terminal) if accepted: - ack_fn(task.notifications.ack_ids) # ack only after successful send sent += 1 except Exception: logger.exception(f"Failed to send email for submission {task.submission_id}, will redeliver") @@ -182,16 +178,35 @@ def process_messages(messages: list[ReceivedMessage], ack_fn: Callable[[list[str #turn messages into data — parse failures acked immediately via ack_fn all_notifications = _convert_messages(messages, ack_fn) + #if sending is disabled, ack everything and stop — no DB calls, no rendering + if not settings.SEND_EMAILS: + for notifications in all_notifications.values(): + ack_fn(notifications.ack_ids) + logger.info(f"Email sending disabled — acked {len(all_notifications)} notification group(s)") + return + if not all_notifications: logger.info("No valid notifications after parsing, nothing to send") return #determine who to email what email_tasks, ids_to_contact = _build_email_tasks(all_notifications) + + #ack any submissions skipped due to no valid recipients (terminal — retrying won't help) + task_sub_ids = {t.submission_id for t in email_tasks} + skipped_sub_ids = [] + for sub_id, notifications in all_notifications.items(): + if sub_id not in task_sub_ids: + ack_fn(notifications.ack_ids) + skipped_sub_ids.append(sub_id) + if not email_tasks: - logger.info("No emails to send") + logger.warning(f"No emails to send — submissions with no recipients: {sorted(skipped_sub_ids)}") return + if skipped_sub_ids: + logger.info(f"Skipped submissions (no recipients): {sorted(skipped_sub_ids)}") + #fetch submission data — if batch query fails, skip all sends (will redeliver) try: sub_infos = get_submission_info({t.submission_id for t in email_tasks}) diff --git a/tests/test_process_message.py b/tests/test_process_message.py index ca0097a..3e263e6 100644 --- a/tests/test_process_message.py +++ b/tests/test_process_message.py @@ -324,6 +324,44 @@ def test_send_error_does_not_abort(): assert "ack-2" in acked # successful send — sub 124 acked assert mock_send.call_count == 2 +@pytest.mark.usefixtures("db_session") +def test_no_recipients_is_acked(): + # sub 126 uses math-ph; no math-ph moderators in test DB → no recipients → terminal, ack it + note = {**GOOD_COMMENT, "submission_id": 126, "categories": ["math-ph"]} + msg = _make_pubsub_message("ack-1", note) + + mock_ack = Mock() + process_messages([msg], ack_fn=mock_ack) + + acked = [id for call in mock_ack.call_args_list for id in call.args[0]] + assert "ack-1" in acked + +@pytest.mark.usefixtures("db_session") +def test_all_recipients_refused_is_acked(): + # relay explicitly rejects all addresses — terminal, ack it + msg = _make_pubsub_message("ack-1", GOOD_COMMENT) + + mock_ack = Mock() + mock_send = Mock(return_value=False) + with patch("app.process.settings.SEND_EMAILS", True), patch("app.process.send_email", mock_send): + process_messages([msg], ack_fn=mock_ack) + + acked = [id for call in mock_ack.call_args_list for id in call.args[0]] + assert "ack-1" in acked + +@pytest.mark.usefixtures("db_session") +def test_transient_smtp_failure_is_not_acked(): + # connection/server error — transient, do not ack so it redelivers + msg = _make_pubsub_message("ack-1", GOOD_COMMENT) + + mock_ack = Mock() + mock_send = Mock(side_effect=RuntimeError("smtp error")) + with patch("app.process.settings.SEND_EMAILS", True), patch("app.process.send_email", mock_send): + process_messages([msg], ack_fn=mock_ack) + + acked = [id for call in mock_ack.call_args_list for id in call.args[0]] + assert "ack-1" not in acked + @pytest.mark.usefixtures("db_session") def test_all_successful_sends_all_acked(): msg1 = _make_pubsub_message("ack-1", GOOD_COMMENT) # sub 123