merged

auth/jwtcodec.py

@@ -36,7 +36,7 @@ class JWTCodec:
                 issuer="discours",
             )
             r = TokenPayload(**payload)
-            print("[auth.jwtcodec] debug token %r" % r)
+            # print('[auth.jwtcodec] debug token %r' % r)
             return r
         except jwt.InvalidIssuedAtError:
             print("[auth.jwtcodec] invalid issued at: %r" % payload)

main.py

@@ -13,8 +13,6 @@ from orm import init_tables
 
 from auth.authenticate import JWTAuthenticate
 from auth.oauth import oauth_login, oauth_authorize
-from services.redis import redis
-from services.schema import resolvers
 from resolvers.auth import confirm_email_handler
 from resolvers.upload import upload_handler
 from settings import DEV_SERVER_PID_FILE_NAME, SENTRY_DSN
@@ -26,7 +24,7 @@ import_module("resolvers")
 schema = make_executable_schema(load_schema_from_path("schemas/core.graphql"), resolvers)  # type: ignore
 middleware = [
     Middleware(AuthenticationMiddleware, backend=JWTAuthenticate()),
-    Middleware(SessionMiddleware, secret_key="!secret"),
+    Middleware(SessionMiddleware, secret_key=SESSION_SECRET_KEY),
 ]
 
 
@@ -39,7 +37,6 @@ async def start_up():
     _views_stat_task = asyncio.create_task(ViewedStorage().worker())
     try:
         import sentry_sdk
-
         sentry_sdk.init(SENTRY_DSN)
         print("[sentry] started")
     except Exception as e:
@@ -78,14 +75,12 @@ app = Starlette(
     middleware=middleware,
     routes=routes,
 )
-app.mount(
-    "/",
-    GraphQL(schema, debug=True),
-)
+app.mount("/", GraphQL(
+    schema,
+    debug=True
+))
 
-print("[main] app mounted")
-
-dev_app = app = Starlette(
+dev_app = Starlette(
     debug=True,
     on_startup=[dev_start_up],
     on_shutdown=[shutdown],

orm/__init__.py

@@ -7,7 +7,18 @@ from orm.shout import Shout
 from orm.topic import Topic, TopicFollower
 from orm.user import User, UserRating
 
-# NOTE: keep orm module isolated
+
+def init_tables():
+    Base.metadata.create_all(engine)
+    Operation.init_table()
+    Resource.init_table()
+    User.init_table()
+    Community.init_table()
+    Role.init_table()
+    UserRating.init_table()
+    Shout.init_table()
+    print("[orm] tables initialized")
+
 
 __all__ = [
     "User",
@@ -21,16 +32,5 @@ __all__ = [
     "Notification",
     "Reaction",
     "UserRating",
+    "init_tables"
 ]
-
-
-def init_tables():
-    Base.metadata.create_all(engine)
-    Operation.init_table()
-    Resource.init_table()
-    User.init_table()
-    Community.init_table()
-    Role.init_table()
-    UserRating.init_table()
-    Shout.init_table()
-    print("[orm] tables initialized")

orm/notification.py

@@ -1,13 +1,25 @@
 from datetime import datetime
-from sqlalchemy import Column, String, JSON, ForeignKey, DateTime, Boolean
-from services.db import Base
+from sqlalchemy import Column, Enum, ForeignKey, DateTime, Boolean, Integer
+from sqlalchemy.dialects.postgresql import JSONB
+
+from base.orm import Base
+from enum import Enum as Enumeration
+
+
+class NotificationType(Enumeration):
+    NEW_REACTION = 1
+    NEW_SHOUT = 2
+    NEW_FOLLOWER = 3
 
 
 class Notification(Base):
     __tablename__ = "notification"
 
+    shout = Column(ForeignKey("shout.id"), index=True)
+    reaction = Column(ForeignKey("reaction.id"), index=True)
     user = Column(ForeignKey("user.id"), index=True)
     createdAt = Column(DateTime, nullable=False, default=datetime.now, index=True)
     seen = Column(Boolean, nullable=False, default=False, index=True)
-    type = Column(String, nullable=False)
-    data = Column(JSON, nullable=True)
+    type = Column(Enum(NotificationType), nullable=False)
+    data = Column(JSONB, nullable=True)
+    occurrences = Column(Integer, default=1)

resolvers/load.py

@@ -137,7 +137,7 @@ async def load_shouts_by(_, info, options):
     """
     :param options: {
         filters: {
-            layout: 'audio',
+            layout: 'music',
             excludeLayout: 'article',
             visibility: "public",
             author: 'discours',
@@ -208,6 +208,7 @@ async def load_shouts_by(_, info, options):
 
 
 @query.field("loadDrafts")
+@login_required
 async def get_drafts(_, info):
     auth: AuthCredentials = info.context["request"].auth
     user_id = auth.user_id

resolvers/notifications.py

@@ -0,0 +1,84 @@
+from sqlalchemy import select, desc, and_, update
+
+from auth.credentials import AuthCredentials
+from base.resolvers import query, mutation
+from auth.authenticate import login_required
+from base.orm import local_session
+from orm import Notification
+
+
+@query.field("loadNotifications")
+@login_required
+async def load_notifications(_, info, params=None):
+    if params is None:
+        params = {}
+
+    auth: AuthCredentials = info.context["request"].auth
+    user_id = auth.user_id
+
+    limit = params.get('limit', 50)
+    offset = params.get('offset', 0)
+
+    q = select(Notification).where(
+        Notification.user == user_id
+    ).order_by(desc(Notification.createdAt)).limit(limit).offset(offset)
+
+    with local_session() as session:
+        total_count = session.query(Notification).where(
+            Notification.user == user_id
+        ).count()
+
+        total_unread_count = session.query(Notification).where(
+            and_(
+                Notification.user == user_id,
+                Notification.seen is False
+            )
+        ).count()
+
+        notifications = session.execute(q).fetchall()
+
+    return {
+        "notifications": notifications,
+        "totalCount": total_count,
+        "totalUnreadCount": total_unread_count
+    }
+
+
+@mutation.field("markNotificationAsRead")
+@login_required
+async def mark_notification_as_read(_, info, notification_id: int):
+    auth: AuthCredentials = info.context["request"].auth
+    user_id = auth.user_id
+
+    with local_session() as session:
+        notification = session.query(Notification).where(
+            and_(Notification.id == notification_id, Notification.user == user_id)
+        ).one()
+        notification.seen = True
+        session.commit()
+
+    return {}
+
+
+@mutation.field("markAllNotificationsAsRead")
+@login_required
+async def mark_all_notifications_as_read(_, info):
+    auth: AuthCredentials = info.context["request"].auth
+    user_id = auth.user_id
+
+    statement = update(Notification).where(
+        and_(
+            Notification.user == user_id,
+            Notification.seen == False
+        )
+    ).values(seen=True)
+
+    with local_session() as session:
+        try:
+            session.execute(statement)
+            session.commit()
+        except Exception as e:
+            session.rollback()
+            print(f"[mark_all_notifications_as_read] error: {str(e)}")
+
+    return {}

resolvers/profile.py

@@ -266,10 +266,20 @@ async def get_authors_all(_, _info):
 @query.field("getAuthor")
 async def get_author(_, _info, slug):
     q = select(User).where(User.slug == slug)
-    q = add_author_stat_columns(q, True)
+    q = add_author_stat_columns(q)
 
-    authors = get_authors_from_query(q)
-    return authors[0]
+    [author] = get_authors_from_query(q)
+
+    with local_session() as session:
+        comments_count = session.query(Reaction).where(
+            and_(
+                Reaction.createdBy == author.id,
+                Reaction.kind == ReactionKind.COMMENT
+            )
+        ).count()
+        author.stat["commented"] = comments_count
+
+    return author
 
 
 @query.field("loadAuthorsBy")

resolvers/reactions.py

@@ -10,6 +10,7 @@ from services.schema import mutation, query
 from orm.reaction import Reaction, ReactionKind
 from orm.shout import Shout, ShoutReactionsFollower
 from orm.user import User
+from services.notifications.notification_service import notification_service
 
 
 def add_reaction_stat_columns(q):
@@ -217,6 +218,8 @@ async def create_reaction(_, info, reaction):
         r = Reaction.create(**reaction)
 
         # Proposal accepting logix
+        # FIXME: will break if there will be 2 proposals
+        # FIXME: will break if shout will be changed
         if (
             r.replyTo is not None
             and r.kind == ReactionKind.ACCEPT
@@ -237,12 +240,14 @@ async def create_reaction(_, info, reaction):
 
         session.add(r)
         session.commit()
+
+        await notification_service.handle_new_reaction(r.id)
+
         rdict = r.dict()
         rdict["shout"] = shout.dict()
         rdict["createdBy"] = author.dict()
 
         # self-regulation mechanics
-
         if check_to_hide(session, auth.user_id, r):
             set_hidden(session, r.shout)
         elif check_to_publish(session, auth.user_id, r):

schemas/core.graphql

@@ -131,6 +131,15 @@ enum FollowingEntity {
 ################################### Mutation
 
 type Mutation {
+    # inbox
+    createChat(title: String, members: [Int]!): Result!
+    updateChat(chat: ChatInput!): Result!
+    deleteChat(chatId: String!): Result!
+
+    createMessage(chat: String!, body: String!, replyTo: Int): Result!
+    updateMessage(chatId: String!, id: Int!, body: String!): Result!
+    deleteMessage(chatId: String!, id: Int!): Result!
+    markAsRead(chatId: String!, ids: [Int]!): Result!
 
     # auth
     getSession: AuthResult!
@@ -145,7 +154,6 @@ type Mutation {
 
     # user profile
     rateUser(slug: String!, value: Int!): Result!
-  updateOnlineStatus: Result!
     updateProfile(profile: ProfileInput!): Result!
 
     # topics
@@ -162,6 +170,9 @@ type Mutation {
     # following
     follow(what: FollowingEntity!, slug: String!): Result!
     unfollow(what: FollowingEntity!, slug: String!): Result!
+
+    markNotificationAsRead(notification_id: Int!): Result!
+    markAllNotificationsAsRead: Result!
 }
 
 input AuthorsBy {
@@ -206,7 +217,17 @@ input ReactionBy {
     days: Int # before
     sort: String # how to sort, default createdAt
 }
-################################### Query
+
+input NotificationsQueryParams {
+    limit: Int
+    offset: Int
+}
+
+type NotificationsQueryResult {
+    notifications: [Notification]!
+    totalCount: Int!
+    totalUnreadCount: Int!
+}
 
 type Query {
 
@@ -350,7 +371,7 @@ type Shout {
     lang: String
     community: String
     cover: String
-  layout: String # audio video literature image
+    layout: String # music video literature image
     versionOf: String # for translations and re-telling the same story
     visibility: String # owner authors community public
     updatedAt: DateTime
@@ -420,3 +441,19 @@ type Token {
     usedAt: DateTime
     value: String!
 }
+
+enum NotificationType {
+    NEW_COMMENT,
+    NEW_REPLY
+}
+
+type Notification {
+    id: Int!
+    shout: Int
+    reaction: Int
+    type: NotificationType
+    createdAt: DateTime!
+    seen: Boolean!
+    data: String # JSON
+    occurrences: Int!
+}

server.py

@@ -44,7 +44,7 @@ log_settings = {
 
 local_headers = [
     ("Access-Control-Allow-Methods", "GET, POST, OPTIONS, HEAD"),
-    ("Access-Control-Allow-Origin", "http://localhost:3000"),
+    ("Access-Control-Allow-Origin", "https://localhost:3000"),
     (
         "Access-Control-Allow-Headers",
         "DNT,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Range,Authorization",

services/notifications/notification_service.py

@@ -0,0 +1,137 @@
+import asyncio
+import json
+from datetime import datetime, timezone
+
+from sqlalchemy import and_
+
+from base.orm import local_session
+from orm import Reaction, Shout, Notification, User
+from orm.notification import NotificationType
+from orm.reaction import ReactionKind
+from services.notifications.sse import connection_manager
+
+
+def update_prev_notification(notification, user):
+    notification_data = json.loads(notification.data)
+
+    notification_data["users"] = [
+        user for user in notification_data["users"] if user['id'] != user.id
+    ]
+    notification_data["users"].append({
+        "id": user.id,
+        "name": user.name
+    })
+
+    notification.data = json.dumps(notification_data, ensure_ascii=False)
+    notification.seen = False
+    notification.occurrences = notification.occurrences + 1
+    notification.createdAt = datetime.now(tz=timezone.utc)
+
+
+class NewReactionNotificator:
+    def __init__(self, reaction_id):
+        self.reaction_id = reaction_id
+
+    async def run(self):
+        with local_session() as session:
+            reaction = session.query(Reaction).where(Reaction.id == self.reaction_id).one()
+            shout = session.query(Shout).where(Shout.id == reaction.shout).one()
+            user = session.query(User).where(User.id == reaction.createdBy).one()
+            notify_user_ids = []
+
+            if reaction.kind == ReactionKind.COMMENT:
+                parent_reaction = None
+                if reaction.replyTo:
+                    parent_reaction = session.query(Reaction).where(Reaction.id == reaction.replyTo).one()
+                    if parent_reaction.createdBy != reaction.createdBy:
+                        prev_new_reply_notification = session.query(Notification).where(
+                            and_(
+                                Notification.user == shout.createdBy,
+                                Notification.type == NotificationType.NEW_REPLY,
+                                Notification.shout == shout.id,
+                                Notification.reaction == parent_reaction.id
+                            )
+                        ).first()
+
+                        if prev_new_reply_notification:
+                            update_prev_notification(prev_new_reply_notification, user)
+                        else:
+                            reply_notification_data = json.dumps({
+                                "shout": {
+                                    "title": shout.title
+                                },
+                                "users": [
+                                    {"id": user.id, "name": user.name}
+                                ]
+                            }, ensure_ascii=False)
+
+                            reply_notification = Notification.create(**{
+                                "user": parent_reaction.createdBy,
+                                "type": NotificationType.NEW_REPLY.name,
+                                "shout": shout.id,
+                                "reaction": parent_reaction.id,
+                                "data": reply_notification_data
+                            })
+
+                            session.add(reply_notification)
+
+                        notify_user_ids.append(parent_reaction.createdBy)
+
+                if reaction.createdBy != shout.createdBy and (
+                    parent_reaction is None or parent_reaction.createdBy != shout.createdBy
+                ):
+                    prev_new_comment_notification = session.query(Notification).where(
+                        and_(
+                            Notification.user == shout.createdBy,
+                            Notification.type == NotificationType.NEW_COMMENT,
+                            Notification.shout == shout.id
+                        )
+                    ).first()
+
+                    if prev_new_comment_notification:
+                        update_prev_notification(prev_new_comment_notification, user)
+                    else:
+                        notification_data_string = json.dumps({
+                            "shout": {
+                                "title": shout.title
+                            },
+                            "users": [
+                                {"id": user.id, "name": user.name}
+                            ]
+                        }, ensure_ascii=False)
+
+                        author_notification = Notification.create(**{
+                            "user": shout.createdBy,
+                            "type": NotificationType.NEW_COMMENT.name,
+                            "shout": shout.id,
+                            "data": notification_data_string
+                        })
+
+                        session.add(author_notification)
+
+                    notify_user_ids.append(shout.createdBy)
+
+            session.commit()
+
+            for user_id in notify_user_ids:
+                await connection_manager.notify_user(user_id)
+
+
+class NotificationService:
+    def __init__(self):
+        self._queue = asyncio.Queue()
+
+    async def handle_new_reaction(self, reaction_id):
+        notificator = NewReactionNotificator(reaction_id)
+        await self._queue.put(notificator)
+
+    async def worker(self):
+        while True:
+            notificator = await self._queue.get()
+            try:
+                await notificator.run()
+            except Exception as e:
+                print(f'[NotificationService.worker] error: {str(e)}')
+
+
+notification_service = NotificationService()

services/notifications/sse.py

@@ -0,0 +1,72 @@
+import json
+
+from sse_starlette.sse import EventSourceResponse
+from starlette.requests import Request
+import asyncio
+
+
+class ConnectionManager:
+    def __init__(self):
+        self.connections_by_user_id = {}
+
+    def add_connection(self, user_id, connection):
+        if user_id not in self.connections_by_user_id:
+            self.connections_by_user_id[user_id] = []
+        self.connections_by_user_id[user_id].append(connection)
+
+    def remove_connection(self, user_id, connection):
+        if user_id not in self.connections_by_user_id:
+            return
+
+        self.connections_by_user_id[user_id].remove(connection)
+
+        if len(self.connections_by_user_id[user_id]) == 0:
+            del self.connections_by_user_id[user_id]
+
+    async def notify_user(self, user_id):
+        if user_id not in self.connections_by_user_id:
+            return
+
+        for connection in self.connections_by_user_id[user_id]:
+            data = {
+                "type": "newNotifications"
+            }
+            data_string = json.dumps(data, ensure_ascii=False)
+            await connection.put(data_string)
+
+    async def broadcast(self, data: str):
+        for user_id in self.connections_by_user_id:
+            for connection in self.connections_by_user_id[user_id]:
+                await connection.put(data)
+
+
+class Connection:
+    def __init__(self):
+        self._queue = asyncio.Queue()
+
+    async def put(self, data: str):
+        await self._queue.put(data)
+
+    async def listen(self):
+        data = await self._queue.get()
+        return data
+
+
+connection_manager = ConnectionManager()
+
+
+async def sse_subscribe_handler(request: Request):
+    user_id = int(request.path_params["user_id"])
+    connection = Connection()
+    connection_manager.add_connection(user_id, connection)
+
+    async def event_publisher():
+        try:
+            while True:
+                data = await connection.listen()
+                yield data
+        except asyncio.CancelledError as e:
+            connection_manager.remove_connection(user_id, connection)
+            raise e
+
+    return EventSourceResponse(event_publisher())

settings.py

@@ -27,6 +27,7 @@ SHOUTS_REPO = "content"
 SESSION_TOKEN_HEADER = "Authorization"
 
 SENTRY_DSN = environ.get("SENTRY_DSN")
+SESSION_SECRET_KEY = environ.get("SESSION_SECRET_KEY") or "!secret"
 
 # for local development
 DEV_SERVER_PID_FILE_NAME = 'dev-server.pid'