# retoor <retoor@molodetz.nl>
from .core import TTLCache, bump_cache_version, datetime, db, get_table, sync_local_cache, timezone
from .settings import get_int_setting, set_setting
from .soft_delete import soft_delete
NOTIFICATION_TYPES = [
{"key": "comment", "label": "Comments", "description": "Someone comments on your post"},
{"key": "reply", "label": "Replies", "description": "Someone replies to your comment"},
{"key": "mention", "label": "Mentions", "description": "Someone mentions you with @username"},
{"key": "vote", "label": "Upvotes", "description": "Someone ++'d your content"},
{"key": "follow", "label": "Followers", "description": "Someone starts following you"},
{"key": "message", "label": "Direct messages", "description": "Someone sends you a message"},
{"key": "badge", "label": "Badges", "description": "You earn a badge"},
{"key": "level", "label": "Level-ups", "description": "You reach a new level"},
{"key": "issue", "label": "Issue tracker", "description": "Updates on issue reports you filed"},
{"key": "reminder", "label": "Reminders", "description": "A reminder or scheduled task you asked Devii to run fires"},
{"key": "harvest_stolen", "label": "Farm raids", "description": "Someone steals a ready build from your Code Farm"},
{"key": "award", "label": "Awards", "description": "Someone gives you an award on your profile"},
{"key": "system", "label": "System alerts", "description": "Platform infrastructure alerts (e.g. the AI gateway going down)"},
]
NOTIFICATION_CHANNELS = ("in_app", "push", "telegram")
_NOTIFICATION_CHANNEL_COLUMNS = {
"in_app": "in_app_enabled",
"push": "push_enabled",
"telegram": "telegram_enabled",
}
_NOTIFICATION_CHANNEL_DEFAULTS = {"in_app": 1, "push": 1, "telegram": 0}
_NOTIFICATION_TYPE_KEYS = {entry["key"] for entry in NOTIFICATION_TYPES}
_notification_prefs_cache = TTLCache(ttl=300, max_size=500)
def _notification_default(notification_type: str, channel: str) -> bool:
fallback = _NOTIFICATION_CHANNEL_DEFAULTS.get(channel, 1)
return get_int_setting(f"notif_default_{notification_type}_{channel}", fallback) != 0
def get_notification_default(notification_type: str, channel: str) -> bool:
return _notification_default(notification_type, channel)
def set_notification_default(
notification_type: str, channel: str, enabled: bool
) -> None:
if channel not in _NOTIFICATION_CHANNEL_COLUMNS:
raise ValueError(f"Unknown notification channel: {channel}")
set_setting(f"notif_default_{notification_type}_{channel}", "1" if enabled else "0")
def _notification_overrides(user_uid: str) -> dict:
sync_local_cache("notif_prefs", _notification_prefs_cache)
cached = _notification_prefs_cache.get(user_uid)
if cached is not None:
return cached
overrides: dict = {}
if "notification_preferences" in db.tables:
for row in db["notification_preferences"].find(
user_uid=user_uid, deleted_at=None
):
overrides[row["notification_type"]] = {
channel: bool(
row.get(column, _NOTIFICATION_CHANNEL_DEFAULTS.get(channel, 1))
)
for channel, column in _NOTIFICATION_CHANNEL_COLUMNS.items()
}
_notification_prefs_cache.set(user_uid, overrides)
return overrides
def notification_enabled(user_uid: str, notification_type: str, channel: str) -> bool:
if channel not in _NOTIFICATION_CHANNEL_COLUMNS:
return True
override = _notification_overrides(user_uid).get(notification_type)
if override is not None:
return bool(override[channel])
return _notification_default(notification_type, channel)
def get_notification_prefs(user_uid: str) -> list:
overrides = _notification_overrides(user_uid)
result = []
for entry in NOTIFICATION_TYPES:
key = entry["key"]
override = overrides.get(key)
channels = {
channel: bool(override[channel])
if override
else _notification_default(key, channel)
for channel in _NOTIFICATION_CHANNEL_COLUMNS
}
result.append(
{
"key": key,
"label": entry["label"],
"description": entry["description"],
**channels,
"customized": override is not None,
}
)
return result
def set_notification_pref(
user_uid: str, notification_type: str, channel: str, enabled: bool
) -> dict:
if notification_type not in _NOTIFICATION_TYPE_KEYS:
raise ValueError(f"Unknown notification type: {notification_type}")
if channel not in _NOTIFICATION_CHANNEL_COLUMNS:
raise ValueError(f"Unknown notification channel: {channel}")
from devplacepy.utils import generate_uid
table = get_table("notification_preferences")
now = datetime.now(timezone.utc).isoformat()
existing = table.find_one(user_uid=user_uid, notification_type=notification_type)
if existing:
values = {
name: bool(
existing.get(column, _NOTIFICATION_CHANNEL_DEFAULTS.get(name, 1))
)
for name, column in _NOTIFICATION_CHANNEL_COLUMNS.items()
}
else:
values = {
name: _notification_default(notification_type, name)
for name in _NOTIFICATION_CHANNEL_COLUMNS
}
values[channel] = enabled
columns = {
column: 1 if values[name] else 0
for name, column in _NOTIFICATION_CHANNEL_COLUMNS.items()
}
if existing:
record = {
"id": existing["id"],
**columns,
"updated_at": now,
"deleted_at": None,
"deleted_by": None,
}
table.update(record, ["id"])
result = {**existing, **record}
else:
result = {
"uid": generate_uid(),
"user_uid": user_uid,
"notification_type": notification_type,
**columns,
"created_at": now,
"updated_at": now,
"deleted_at": None,
"deleted_by": None,
}
table.insert(result)
bump_cache_version("notif_prefs")
return result
def reset_notification_prefs(user_uid: str, deleted_by: str | None = None) -> int:
if "notification_preferences" not in db.tables:
return 0
count = soft_delete(
"notification_preferences", deleted_by or f"user:{user_uid}", user_uid=user_uid
)
bump_cache_version("notif_prefs")
return int(count)
def mark_notifications_read_by_target(user_uid: str, target_url: str) -> int:
if not user_uid or not target_url or "notifications" not in db.tables:
return 0
notifications_table = get_table("notifications")
ids = [
n["id"]
for n in notifications_table.find(user_uid=user_uid, read=False)
if n.get("target_url")
and (
n["target_url"] == target_url
or n["target_url"].startswith(f"{target_url}#")
)
]
if not ids:
return 0
with db:
for notification_id in ids:
notifications_table.update({"id": notification_id, "read": True}, ["id"])
from devplacepy.templating import clear_unread_cache
clear_unread_cache(user_uid)
return len(ids)
Reference in New Issue View Git Blame Copy Permalink