133 lines
4.1 KiB
Python
133 lines
4.1 KiB
Python
import logging
|
|
from functools import singledispatch
|
|
from typing import Any
|
|
from typing import Dict
|
|
|
|
from little_boxes import activitypub as ap
|
|
|
|
from config import BASE_URL
|
|
from config import DB
|
|
from config import MetaKey
|
|
from config import _meta
|
|
from tasks import Tasks
|
|
from utils.meta import by_actor
|
|
from utils.meta import by_type
|
|
from utils.meta import in_inbox
|
|
from utils.meta import not_undo
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
_NewMeta = Dict[str, Any]
|
|
|
|
|
|
def _is_from_outbox(activity: ap.BaseActivity) -> bool:
|
|
return activity.id.startswith(BASE_URL)
|
|
|
|
|
|
def _flag_as_notification(activity: ap.BaseActivity, new_meta: _NewMeta) -> None:
|
|
new_meta.update(
|
|
{_meta(MetaKey.NOTIFICATION): True, _meta(MetaKey.NOTIFICATION_UNREAD): True}
|
|
)
|
|
return None
|
|
|
|
|
|
@singledispatch
|
|
def set_inbox_flags(activity: ap.BaseActivity, new_meta: _NewMeta) -> None:
|
|
return None
|
|
|
|
|
|
@set_inbox_flags.register
|
|
def _accept_set_inbox_flags(activity: ap.Accept, new_meta: _NewMeta) -> None:
|
|
"""Handle notifications for "accepted" following requests."""
|
|
_logger.info(f"set_inbox_flags activity={activity!r}")
|
|
# Check if this actor already follow us back
|
|
follows_back = False
|
|
follow_query = {
|
|
**in_inbox(),
|
|
**by_type(ap.ActivityType.FOLLOW),
|
|
**by_actor(activity.get_actor()),
|
|
**not_undo(),
|
|
}
|
|
raw_follow = DB.activities.find_one(follow_query)
|
|
if raw_follow:
|
|
follows_back = True
|
|
|
|
DB.activities.update_many(
|
|
follow_query, {"$set": {_meta(MetaKey.NOTIFICATION_FOLLOWS_BACK): True}}
|
|
)
|
|
|
|
# This Accept will be a "You started following $actor" notification
|
|
_flag_as_notification(activity, new_meta)
|
|
new_meta.update({_meta(MetaKey.NOTIFICATION_FOLLOWS_BACK): follows_back})
|
|
return None
|
|
|
|
|
|
@set_inbox_flags.register
|
|
def _follow_set_inbox_flags(activity: ap.Follow, new_meta: _NewMeta) -> None:
|
|
"""Handle notification for new followers."""
|
|
# Check if we're already following this actor
|
|
follows_back = False
|
|
accept_query = {
|
|
**in_inbox(),
|
|
**by_type(ap.ActivityType.ACCEPT),
|
|
**by_actor(activity.get_actor()),
|
|
**not_undo(),
|
|
}
|
|
raw_accept = DB.activities.find_one(accept_query)
|
|
if raw_accept:
|
|
follows_back = True
|
|
|
|
DB.activities.update_many(
|
|
accept_query, {"$set": {_meta(MetaKey.NOTIFICATION_FOLLOWS_BACK): True}}
|
|
)
|
|
|
|
# This Follow will be a "$actor started following you" notification
|
|
_flag_as_notification(activity, new_meta)
|
|
new_meta.update({_meta(MetaKey.NOTIFICATION_FOLLOWS_BACK): follows_back})
|
|
return None
|
|
|
|
|
|
@set_inbox_flags.register
|
|
def _like_set_inbox_flags(activity: ap.Like, new_meta: _NewMeta) -> None:
|
|
# Is it a Like of local acitivty/from the outbox
|
|
if _is_from_outbox(activity.get_object()):
|
|
# Flag it as a notification
|
|
_flag_as_notification(activity, new_meta)
|
|
|
|
# Cache the object (for display on the notifcation page)
|
|
Tasks.cache_object(activity.id)
|
|
|
|
# Also set the "keep mark" for the GC (as we want to keep it forever)
|
|
new_meta.update({_meta(MetaKey.GC_KEEP): True})
|
|
|
|
return None
|
|
|
|
|
|
@set_inbox_flags.register
|
|
def _announce_set_inbox_flags(activity: ap.Announce, new_meta: _NewMeta) -> None:
|
|
# Is it a Like of local acitivty/from the outbox
|
|
if _is_from_outbox(activity.get_object()):
|
|
# Flag it as a notification
|
|
_flag_as_notification(activity, new_meta)
|
|
|
|
# Also set the "keep mark" for the GC (as we want to keep it forever)
|
|
new_meta.update({_meta(MetaKey.GC_KEEP): True})
|
|
|
|
# Cache the object in all case (for display on the notifcation page **and** the stream page)
|
|
Tasks.cache_object(activity.id)
|
|
|
|
return None
|
|
|
|
|
|
@set_inbox_flags.register
|
|
def _undo_set_inbox_flags(activity: ap.Undo, new_meta: _NewMeta) -> None:
|
|
obj = activity.get_object()
|
|
|
|
if obj.has_type(ap.ActivityType.FOLLOW):
|
|
# Flag it as a noticiation (for the "$actor unfollowed you"
|
|
_flag_as_notification(activity, new_meta)
|
|
|
|
# Also set the "keep mark" for the GC (as we want to keep it forever)
|
|
new_meta.update({_meta(MetaKey.GC_KEEP): True})
|
|
|
|
return None
|