1from collections import deque
  2from datetime import datetime as dt
  3from enum import Enum
  4
  5from pydantic import BaseModel
  6
  7from app.config.env import WEBHOOK_ADDRESS
  8from app.models.admin import Admin
  9from app.models.user import UserResponse
 10
 11queue = deque()
 12
 13
 14class Notification(BaseModel):
 15    class Type(str, Enum):
 16        user_created = "user_created"
 17        user_updated = "user_updated"
 18        user_deleted = "user_deleted"
 19        user_limited = "user_limited"
 20        user_expired = "user_expired"
 21        user_enabled = "user_enabled"
 22        user_disabled = "user_disabled"
 23        data_usage_reset = "data_usage_reset"
 24        subscription_revoked = "subscription_revoked"
 25
 26        reached_usage_percent = "reached_usage_percent"
 27        reached_days_left = "reached_days_left"
 28
 29    enqueued_at: float = dt.utcnow().timestamp()
 30    send_at: float = dt.utcnow().timestamp()
 31    tries: int = 0
 32
 33
 34class UserNotification(Notification):
 35    username: str
 36
 37
 38class ReachedUsagePercent(UserNotification):
 39    action: Notification.Type = Notification.Type.reached_usage_percent
 40    user: UserResponse
 41    used_percent: float
 42
 43
 44class ReachedDaysLeft(UserNotification):
 45    action: Notification.Type = Notification.Type.reached_days_left
 46    user: UserResponse
 47    days_left: int
 48
 49
 50class UserCreated(UserNotification):
 51    action: Notification.Type = Notification.Type.user_created
 52    by: Admin
 53    user: UserResponse
 54
 55
 56class UserUpdated(UserNotification):
 57    action: Notification.Type = Notification.Type.user_updated
 58    by: Admin
 59    user: UserResponse
 60
 61
 62class UserDeleted(UserNotification):
 63    action: Notification.Type = Notification.Type.user_deleted
 64    by: Admin
 65
 66
 67class UserLimited(UserNotification):
 68    action: Notification.Type = Notification.Type.user_limited
 69    user: UserResponse
 70
 71
 72class UserExpired(UserNotification):
 73    action: Notification.Type = Notification.Type.user_expired
 74    user: UserResponse
 75
 76
 77class UserEnabled(UserNotification):
 78    action: Notification.Type = Notification.Type.user_enabled
 79    by: Admin = None
 80    user: UserResponse
 81
 82
 83class UserDisabled(UserNotification):
 84    action: Notification.Type = Notification.Type.user_disabled
 85    by: Admin
 86    user: UserResponse
 87    reason: str = None
 88
 89
 90class UserDataUsageReset(UserNotification):
 91    action: Notification.Type = Notification.Type.data_usage_reset
 92    by: Admin
 93    user: UserResponse
 94
 95
 96class UserSubscriptionRevoked(UserNotification):
 97    action: Notification.Type = Notification.Type.subscription_revoked
 98    by: Admin
 99    user: UserResponse
100
101
102async def notify(message: Notification) -> None:
103    if WEBHOOK_ADDRESS:
104        queue.append(message)