zilencer: Notify when paid plan attached to now-deleted remote realm.

When a server doesn't submit a remote realm info which was
previously submitted, we mark it as locally deleted.

If such a realm has paid plan attached to it, we should investigate.

This commit adds logic to send an email to sales@zulip.com for
investigation.

templates/zerver/emails/internal_billing_notice.html

@@ -6,6 +6,8 @@
 <p>Reminder to re-evaluate the pricing and configure a new fixed-price plan accordingly.</p>
 {% elif notice_reason == "invoice_overdue" %}
 <b>Last data upload</b>: {{ last_audit_log_update }}
+{% elif notice_reason == "locally_deleted_realm_on_paid_plan" %}
+<p>Investigate the reason for {{billing_entity}} on paid plan marked as locally deleted.</p>
 {% endif %}
 
 <br /><br />

templates/zerver/emails/internal_billing_notice.subject.txt

@@ -2,4 +2,6 @@
 Fixed-price plan for {{billing_entity}} ends on {{end_date}}
 {% elif notice_reason == "invoice_overdue" %}
 Invoice overdue due to stale data
+{% elif notice_reason == "locally_deleted_realm_on_paid_plan" %}
+{{billing_entity}} on paid plan marked as locally deleted
 {% endif %}

templates/zerver/emails/internal_billing_notice.txt

@@ -2,6 +2,8 @@
 Reminder to re-evaluate the pricing and configure a new fixed-price plan accordingly.
 {% elif notice_reason == "invoice_overdue" %}
 Last data upload: {{ last_audit_log_update }}
+{% elif notice_reason == "locally_deleted_realm_on_paid_plan" %}
+Investigate the reason for {{billing_entity}} on paid plan marked as locally deleted.
 {% endif %}
 
 Support URL: {{ support_url }}

zerver/tests/test_push_notifications.py

@@ -25,6 +25,7 @@ from typing_extensions import override
 
 from analytics.lib.counts import CountStat, LoggingCountStat
 from analytics.models import InstallationCount, RealmCount, UserCount
+from corporate.lib.stripe import RemoteRealmBillingSession
 from corporate.models import CustomerPlan
 from version import ZULIP_VERSION
 from zerver.actions.create_realm import do_create_realm
@@ -2566,7 +2567,9 @@ class AnalyticsBouncerTest(BouncerTestCase):
 
         # Now we want to test the other side of this - bouncer's handling
         # of a deleted realm.
-        with self.assertLogs(logger, level="WARNING") as analytics_logger:
+        with self.assertLogs(logger, level="WARNING") as analytics_logger, mock.patch(
+            "zilencer.views.RemoteRealmBillingSession.on_paid_plan", return_value=True
+        ):
             # This time the logger shouldn't get triggered - because the bouncer doesn't
             # include .realm_locally_deleted realms in its response.
             # Note: This is hacky, because until Python 3.10 we don't have access to
@@ -2585,6 +2588,22 @@ class AnalyticsBouncerTest(BouncerTestCase):
         self.assertEqual(audit_log.event_type, RemoteRealmAuditLog.REMOTE_REALM_LOCALLY_DELETED)
         self.assertEqual(audit_log.remote_realm, remote_realm_for_deleted_realm)
 
+        from django.core.mail import outbox
+
+        email = outbox[-1]
+        self.assert_length(email.to, 1)
+        self.assertEqual(email.to[0], "sales@zulip.com")
+
+        billing_session = RemoteRealmBillingSession(remote_realm=remote_realm_for_deleted_realm)
+        self.assertIn(
+            f"Support URL: {billing_session.support_url()}",
+            email.body,
+        )
+        self.assertEqual(
+            f"{billing_session.billing_entity_display_name} on paid plan marked as locally deleted",
+            email.subject,
+        )
+
         # Restore the deleted realm to verify that the bouncer correctly handles that
         # by togglin off .realm_locally_deleted.
         restored_zephyr_realm = do_create_realm("zephyr", "Zephyr")

zilencer/views.py

@@ -27,6 +27,7 @@ from analytics.lib.counts import (
     do_increment_logging_stat,
 )
 from corporate.lib.stripe import (
+    BILLING_SUPPORT_EMAIL,
     RemoteRealmBillingSession,
     RemoteServerBillingSession,
     do_deactivate_remote_server,
@@ -52,6 +53,7 @@ from zerver.lib.push_notifications import (
     send_apple_push_notification,
     send_test_push_notification_directly_to_devices,
 )
+from zerver.lib.queue import queue_json_publish
 from zerver.lib.remote_server import (
     InstallationCountDataForAnalytics,
     RealmAuditLogDataForAnalytics,
@@ -897,6 +899,7 @@ def update_remote_realm_data_for_server(
 
     remote_realms_to_update = []
     remote_realm_audit_logs = []
+    new_locally_deleted_remote_realms_on_paid_plan_contexts = []
     for remote_realm in remote_realms_missing_from_server_data:
         if not remote_realm.realm_locally_deleted:
             # Otherwise we already knew about this, so nothing to do.
@@ -918,12 +921,30 @@ def update_remote_realm_data_for_server(
             )
             remote_realms_to_update.append(remote_realm)
 
+            billing_session = RemoteRealmBillingSession(remote_realm=remote_realm)
+            if billing_session.on_paid_plan():
+                context = {
+                    "billing_entity": billing_session.billing_entity_display_name,
+                    "support_url": billing_session.support_url(),
+                    "notice_reason": "locally_deleted_realm_on_paid_plan",
+                }
+                new_locally_deleted_remote_realms_on_paid_plan_contexts.append(context)
+
     RemoteRealm.objects.bulk_update(
         remote_realms_to_update,
         ["realm_locally_deleted"],
     )
     RemoteRealmAuditLog.objects.bulk_create(remote_realm_audit_logs)
 
+    email_dict: Dict[str, Any] = {
+        "template_prefix": "zerver/emails/internal_billing_notice",
+        "to_emails": [BILLING_SUPPORT_EMAIL],
+        "from_address": FromAddress.tokenized_no_reply_address(),
+    }
+    for context in new_locally_deleted_remote_realms_on_paid_plan_contexts:
+        email_dict["context"] = context
+        queue_json_publish("email_senders", email_dict)
+
 
 def get_human_user_realm_uuids(
     server: RemoteZulipServer,