push notifications

genesis_notification/common/constants.py

@@ -29,3 +29,9 @@ class EventStatus(str, enum.Enum):
     IN_PROGRESS = "IN_PROGRESS"
     ACTIVE = "ACTIVE"
     ERROR = "ERROR"
+
+
+class PushDeliveryStatus(str, enum.Enum):
+    SUCCESS = "success"
+    PERMANENT_FAILURE = "permanent_failure"
+    RETRYABLE_FAILURE = "retryable_failure"

genesis_notification/dm/models.py

@@ -16,11 +16,13 @@
 
 import datetime
 import logging
+from dataclasses import dataclass
 from email.mime import text
 from email.mime import multipart
 import smtplib
 
 import jinja2
+import requests
 from restalchemy.dm import filters
 from restalchemy.dm import models
 from restalchemy.dm import properties
@@ -31,7 +33,7 @@
 import zulip
 
 from genesis_notification.common import constants as c
-
+from genesis_notification.common.constants import PushDeliveryStatus
 
 LOG = logging.getLogger(__name__)
 
@@ -56,6 +58,211 @@ class ModelWithAlwaysActiveStatus(models.Model):
     )
 
 
+class Installation(
+    models.ModelWithUUID,
+    models.ModelWithProject,
+    ModelWithAlwaysActiveStatus,
+    models.ModelWithTimestamp,
+    orm.SQLStorableMixin,
+):
+    __tablename__ = "installations"
+
+    installation_id = properties.property(
+        types.String(min_length=8, max_length=128),
+        required=True,
+    )
+
+    user_id = properties.property(
+        types.UUID(),
+        required=True,
+    )
+
+    push_token = properties.property(
+        types.String(min_length=16, max_length=512),
+        required=True,
+    )
+
+    platform = properties.property(
+        types.Enum(["ios", "android", "web"]),
+        required=True,
+    )
+
+    last_seen_at = properties.property(
+        types.UTCDateTimeZ(),
+        default=next_time(0),
+    )
+
+
+FCM_PERMANENT_ERRORS = {
+    "UNREGISTERED",
+    "INVALID_ARGUMENT",
+    "NOT_FOUND",
+}
+
+FCM_RETRYABLE_ERRORS = {
+    "UNAVAILABLE",
+    "INTERNAL",
+    "QUOTA_EXCEEDED",
+}
+
+
+@dataclass
+class PushDeliveryResult:
+
+    installation_id: str
+    token: str
+
+    status: PushDeliveryStatus
+
+    error_code: str | None = None
+    error_message: str | None = None
+
+    provider_response: dict | None = None
+
+
+@dataclass
+class PushBatchResult:
+
+    results: list[PushDeliveryResult]
+
+    def success_count(self):
+        return sum(1 for r in self.results if r.status == PushDeliveryStatus.SUCCESS)
+
+    def permanent_failures(self):
+        return [
+            r for r in self.results
+            if r.status == PushDeliveryStatus.PERMANENT_FAILURE
+        ]
+
+    def retryable_failures(self):
+        return [
+            r for r in self.results
+            if r.status == PushDeliveryStatus.RETRYABLE_FAILURE
+        ]
+
+    def total_failure(self):
+        return self.success_count() == 0
+
+
+class FCMProtocol(types_dynamic.AbstractKindModel):
+    KIND = "fcm"
+
+    project_id = properties.property(
+        types.String(),
+        required=True,
+    )
+
+    service_account_json = properties.property(
+        types.String(),
+        required=True,
+    )
+
+    def _send_to_token(self, installation, content):
+
+        try:
+            resp = requests.post(...)
+
+            if resp.status_code == 200:
+                return PushDeliveryResult(
+                    installation_id=installation.installation_id,
+                    token=installation.push_token,
+                    status=PushDeliveryStatus.SUCCESS,
+                    provider_response=resp.json(),
+                )
+
+            error = self._extract_error(resp)
+
+            if error in FCM_PERMANENT_ERRORS:
+                return PushDeliveryResult(
+                    installation_id=installation.installation_id,
+                    token=installation.push_token,
+                    status=PushDeliveryStatus.PERMANENT_FAILURE,
+                    error_code=error,
+                    error_message=resp.text,
+                )
+
+            return PushDeliveryResult(
+                installation_id=installation.installation_id,
+                token=installation.push_token,
+                status=PushDeliveryStatus.RETRYABLE_FAILURE,
+                error_code=error,
+                error_message=resp.text,
+            )
+
+        except Exception as e:
+            return PushDeliveryResult(
+                installation_id=installation.installation_id,
+                token=installation.push_token,
+                status=PushDeliveryStatus.RETRYABLE_FAILURE,
+                error_code="CLIENT_EXCEPTION",
+                error_message=str(e),
+            )
+
+    def _send_batch(self, installations, content):
+
+        results = []
+
+        for inst in installations:
+            result = self._send_to_token(inst, content)
+            results.append(result)
+
+        return PushBatchResult(results)
+
+    def _is_total_failure(self, results):
+
+        if not results:
+            return False
+
+        success_count = sum(1 for r in results if r["success"])
+        return success_count == 0
+
+    def _build_payload(self, token, content):
+
+        return {
+            "message": {
+                "token": token,
+                "notification": {
+                    "title": content.title,
+                    "body": content.body,
+                },
+                "data": content.data or {},
+            }
+        }
+
+    def _process_batch_result(self, batch_result):
+
+        for r in batch_result.permanent_failures():
+
+            inst = Installation.objects.get_one(
+                filters={"installation_id": r.installation_id}
+            )
+
+            if inst:
+                inst.status = c.AlwaysActiveStatus.INACTIVE.value
+                inst.save()
+
+    def send(self, content, user_context):
+
+        user_id = user_context["user"]["id"]
+
+        installations = Installation.objects.get_all(
+            filters={
+                "user_id": filters.EQ(user_id),
+                "status": filters.EQ(c.AlwaysActiveStatus.ACTIVE.value),
+            }
+        )
+
+        if not installations:
+            return
+
+        batch_result = self._send_batch(installations, content)
+
+        self._process_batch_result(batch_result)
+
+        if batch_result.total_failure():
+            raise RuntimeError("Push delivery totally failed")
+
+
 class SimpleSmtpProtocol(types_dynamic.AbstractKindModel):
     KIND = "SimpleSMTP"
 
@@ -172,6 +379,7 @@ class Provider(
             types_dynamic.KindModelType(SimpleSmtpProtocol),
             types_dynamic.KindModelType(StartTlsSmtpProtocol),
             types_dynamic.KindModelType(ZulipProtocol),
+            types_dynamic.KindModelType(FCMProtocol),
         ),
         required=True,
     )
@@ -311,6 +519,28 @@ def render(self, params):
 }
 
 
+class RenderedPushContent(AbstractContent):
+    KIND = "rendered_push"
+
+    title = properties.property(types.String(), default="")
+    body = properties.property(types.String(), default="")
+    data = properties.property(types.Dict(), default=dict)
+
+
+class PushContent(RenderedPushContent):
+    KIND = "push"
+
+    def render(self, params):
+        return RenderedPushContent(
+            title=jinja2.Template(self.title).render(**params),
+            body=jinja2.Template(self.body).render(**params),
+            data={
+                k: jinja2.Template(str(v)).render(**params)
+                for k, v in self.data.items()
+            },
+        )
+
+
 class Template(
     models.ModelWithUUID,
     models.ModelWithRequiredNameDesc,
@@ -326,6 +556,7 @@ class Template(
             types_dynamic.KindModelType(EmailContent),
             types_dynamic.KindModelType(ZulipStreamMessageContent),
             types_dynamic.KindModelType(ZulipDirectMessageContent),
+            types_dynamic.KindModelType(PushContent),
         ),
         required=True,
     )

genesis_notification/user_api/api/controllers.py

@@ -13,10 +13,12 @@
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    License for the specific language governing permissions and limitations
 #    under the License.
+import datetime
 
 from restalchemy.api import controllers as ra_controllers
 from restalchemy.api import resources
 
+from genesis_notification.common import constants as c
 from genesis_notification.dm import models
 from genesis_notification.user_api.api import versions
 
@@ -56,3 +58,40 @@ class EventController(ra_controllers.BaseResourceController):
     __resource__ = resources.ResourceByRAModel(
         models.Event, convert_underscore=False
     )
+
+
+class InstallationController(ra_controllers.BaseResourceController):
+    __resource__ = resources.ResourceByRAModel(
+        models.Installation,
+        convert_underscore=False,
+    )
+
+    def _update_existing(self, existing, resource):
+        existing.push_token = resource["push_token"]
+        existing.platform = resource["platform"]
+
+        existing.app_version = resource.get("app_version", "")
+        existing.os_version = resource.get("os_version", "")
+        existing.device_model = resource.get("device_model", "")
+
+        existing.status = c.AlwaysActiveStatus.ACTIVE.value
+        existing.last_seen_at = datetime.datetime.now(datetime.timezone.utc)
+
+        existing.save()
+
+        return existing
+
+    def create(self, **kwargs):
+        installation_id = kwargs.get("installation_id")
+
+        existing = models.Installation.objects.get_one(
+            filters={
+                "installation_id": installation_id,
+                "project_id": kwargs.get("project_id")
+            }
+        )
+
+        if existing:
+            return self._update_existing(existing, kwargs)
+
+        return super().create(**kwargs)