906: update user streaks (#1016)

* fix: update user streaks
* Moved  streak_last_day_date & streak_days from UserStats to User
* Updated Alembic version
* update last_activity after tm.handle_interaction()
* periodically executed function in main.py to update the user_streak every 4hrs

* fix:removed log messages

* fix: pre commit issues

* refactor: incorporated review comments

* fix: removed the managed -scope.

* refactor: managed_tx to Outer REST Db Ops in /interaction API

* fix: pre-commit changes

* fix: added streak and last_activity_date to protocol.FrontEndUser

* fix: pre-commit fixes after merge

* fix: added streak info to user_stat_repository, leaderboard and review comments

* fix: proper 4h delay and simpler startup_time initialisation

---------

Co-authored-by: James Melvin <melvin@gameface.ai>

backend/alembic/versions/2023_01_29_1207-7b8f0011e0b0_move_user_streak_from_user_stats_to_.py

@@ -0,0 +1,41 @@
+"""move user_streak from user_stats to user table
+
+Revision ID: 7b8f0011e0b0
+Revises: 8a5feed819ee
+Create Date: 2023-01-29 12:07:29.379326
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "7b8f0011e0b0"
+down_revision = "49d8445b4c90"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "user",
+        sa.Column(
+            "streak_last_day_date",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            autoincrement=False,
+            nullable=True,
+        ),
+    )
+    op.add_column("user", sa.Column("streak_days", sa.INTEGER(), autoincrement=False, nullable=True))
+    op.add_column(
+        "user", sa.Column("last_activity_date", sa.DateTime(timezone=True), autoincrement=False, nullable=True)
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("user_stats", "streak_days")
+    op.drop_column("user_stats", "streak_last_day_date")
+    # ### end Alembic commands ###

backend/main.py

@@ -1,4 +1,5 @@
 import json
+from datetime import datetime
 from http import HTTPStatus
 from math import ceil
 from pathlib import Path
@@ -19,15 +20,18 @@ from oasst_backend.database import engine
 from oasst_backend.models import message_tree_state
 from oasst_backend.prompt_repository import PromptRepository, TaskRepository, UserRepository
 from oasst_backend.tree_manager import TreeManager
+from oasst_backend.user_repository import User
 from oasst_backend.user_stats_repository import UserStatsRepository, UserStatsTimeFrame
 from oasst_backend.utils.database_utils import CommitMode, managed_tx_function
 from oasst_shared.exceptions import OasstError, OasstErrorCode
 from oasst_shared.schemas import protocol as protocol_schema
+from oasst_shared.utils import utcnow
 from pydantic import BaseModel
-from sqlmodel import Session
+from sqlmodel import Session, select
 from starlette.middleware.cors import CORSMiddleware
 
 app = fastapi.FastAPI(title=settings.PROJECT_NAME, openapi_url=f"{settings.API_V1_STR}/openapi.json")
+startup_time: datetime = utcnow()
 
 
 @app.exception_handler(OasstError)
@@ -268,6 +272,51 @@ def update_leader_board_total(session: Session) -> None:
         logger.exception("Error during user states update (total)")
 
 
+@app.on_event("startup")
+@repeat_every(seconds=60 * 60 * settings.USER_STREAK_UPDATE_INTERVAL, wait_first=False)
+def update_user_streak(session: Session) -> None:
+    try:
+        current_time = utcnow()
+        timedelta = current_time - startup_time
+        if timedelta.days > 0:
+            # Update only greater than 24 hours . Do nothing
+            logger.debug("Process timedelta greater than 24h")
+            statement = select(User)
+            result = session.exec(statement).all()
+            if result is not None:
+                for user in result:
+                    last_activity_date = user.last_activity_date
+                    streak_last_day_date = user.streak_last_day_date
+                    # set NULL streak_days to 0
+                    if user.streak_days is None:
+                        user.streak_days = 0
+                    # if the user had completed a task
+                    if last_activity_date is not None:
+                        lastactitvitydelta = current_time - last_activity_date
+                        # if the user missed consecutive days of completing a task
+                        # reset the streak_days to 0 and set streak_last_day_date to the current_time
+                        if lastactitvitydelta.days > 1 or user.streak_days is None:
+                            user.streak_days = 0
+                            user.streak_last_day_date = current_time
+                    # streak_last_day_date has a current timestamp in DB. Idealy should not be NULL.
+                    if streak_last_day_date is not None:
+                        streak_delta = current_time - streak_last_day_date
+                        # if user completed tasks on consecutive days then increment the streak days
+                        # update the streak_last_day_date to current time for the next calculation
+                        if streak_delta.days > 0:
+                            user.streak_days += 1
+                            user.streak_last_day_date = current_time
+                    session.add(user)
+                    session.commit()
+
+        else:
+            logger.debug("Not yet 24hours since the process started! ...")
+
+    except Exception as e:
+        logger.error(str(e))
+    return
+
+
 app.include_router(api_router, prefix=settings.API_V1_STR)
 
 

backend/oasst_backend/api/v1/tasks.py

@@ -7,6 +7,8 @@ from loguru import logger
 from oasst_backend.api import deps
 from oasst_backend.prompt_repository import PromptRepository, TaskRepository
 from oasst_backend.tree_manager import TreeManager
+from oasst_backend.user_repository import UserRepository
+from oasst_backend.utils.database_utils import CommitMode, managed_tx_function
 from oasst_shared.exceptions import OasstError, OasstErrorCode
 from oasst_shared.schemas import protocol as protocol_schema
 from sqlmodel import Session
@@ -134,13 +136,21 @@ async def tasks_interaction(
     """
     The frontend reports an interaction.
     """
-    api_client = deps.api_auth(api_key, db)
 
-    try:
+    @managed_tx_function(CommitMode.COMMIT)
+    async def interaction_tx(session: deps.Session):
+        api_client = deps.api_auth(api_key, db)
         pr = PromptRepository(db, api_client, client_user=interaction.user)
         tm = TreeManager(db, pr)
-        return await tm.handle_interaction(interaction)
+        ur = UserRepository(db, api_client)
+        task = await tm.handle_interaction(interaction)
+        match (type(task)):
+            case protocol_schema.TaskDone:
+                ur.update_user_last_activity(client_user=interaction.user)
+        return task
 
+    try:
+        return await interaction_tx()
     except OasstError:
         raise
     except Exception:

backend/oasst_backend/config.py

@@ -204,6 +204,8 @@ class Settings(BaseSettings):
             raise ValueError(v)
         return v
 
+    USER_STREAK_UPDATE_INTERVAL: int = 4  # Hours
+
     class Config:
         env_file = ".env"
         env_file_encoding = "utf-8"

backend/oasst_backend/models/user.py

@@ -32,6 +32,15 @@ class User(SQLModel, table=True):
     deleted: bool = Field(sa_column=sa.Column(sa.Boolean, nullable=False, server_default=sa.false()))
     show_on_leaderboard: bool = Field(sa_column=sa.Column(sa.Boolean, nullable=False, server_default=sa.true()))
 
+    # only used for time span "total"
+    streak_last_day_date: Optional[datetime] = Field(
+        sa_column=sa.Column(sa.DateTime(timezone=True), nullable=True, server_default=sa.func.current_timestamp())
+    )
+    streak_days: Optional[int] = Field(nullable=True)
+    last_activity_date: Optional[datetime] = Field(
+        sa_column=sa.Column(sa.DateTime(timezone=True), nullable=True, server_default=sa.func.current_timestamp())
+    )
+
     def to_protocol_frontend_user(self):
         return protocol.FrontEndUser(
             user_id=self.id,
@@ -43,4 +52,7 @@ class User(SQLModel, table=True):
             notes=self.notes,
             created_date=self.created_date,
             show_on_leaderboard=self.show_on_leaderboard,
+            streak_days=self.streak_days,
+            streak_last_day_date=self.streak_last_day_date,
+            last_activity_date=self.last_activity_date,
         )

backend/oasst_backend/models/user_stats.py

@@ -51,10 +51,6 @@ class UserStats(SQLModel, table=True):
     reply_ranked_2: int = 0
     reply_ranked_3: int = 0
 
-    # only used for time span "total"
-    streak_last_day_date: Optional[datetime] = Field(nullable=True)
-    streak_days: Optional[int] = Field(nullable=True)
-
     def compute_leader_score(self) -> int:
         return (
             self.prompts

backend/oasst_backend/tree_manager.py

@@ -522,7 +522,7 @@ class TreeManager:
 
         return task, message_tree_id, parent_message_id
 
-    @async_managed_tx_method(CommitMode.COMMIT)
+    @async_managed_tx_method(CommitMode.FLUSH)
     async def handle_interaction(self, interaction: protocol_schema.AnyInteraction) -> protocol_schema.Task:
         pr = self.pr
         pr.ensure_user_is_enabled()

backend/oasst_backend/user_repository.py

@@ -6,6 +6,7 @@ from oasst_backend.models import ApiClient, User
 from oasst_backend.utils.database_utils import CommitMode, managed_tx_method
 from oasst_shared.exceptions import OasstError, OasstErrorCode
 from oasst_shared.schemas import protocol as protocol_schema
+from oasst_shared.utils import utcnow
 from sqlalchemy.exc import IntegrityError
 from sqlmodel import Session, and_, or_
 from starlette.status import HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND
@@ -309,3 +310,12 @@ class UserRepository:
             qry = qry.limit(limit)
 
         return qry.all()
+
+    @managed_tx_method(CommitMode.FLUSH)
+    def update_user_last_activity(self, client_user: protocol_schema.User) -> None:
+        user = self.lookup_client_user(client_user=client_user, create_missing=False)
+        if user is None:
+            raise OasstError("User not found", OasstErrorCode.USER_NOT_FOUND, HTTP_404_NOT_FOUND)
+
+        user.last_activity_date = utcnow()
+        self.db.add(user)

backend/oasst_backend/user_stats_repository.py

@@ -22,7 +22,15 @@ def _create_user_score(r):
         d = r["UserStats"].dict()
     else:
         d = {"modified_date": utcnow()}
-    for k in ["user_id", "username", "auth_method", "display_name"]:
+    for k in [
+        "user_id",
+        "username",
+        "auth_method",
+        "display_name",
+        "streak_days",
+        "streak_last_day_date",
+        "last_activity_date",
+    ]:
         d[k] = r[k]
     return UserScore(**d)
 
@@ -37,7 +45,16 @@ class UserStatsRepository:
         """
 
         qry = (
-            self.session.query(User.id.label("user_id"), User.username, User.auth_method, User.display_name, UserStats)
+            self.session.query(
+                User.id.label("user_id"),
+                User.username,
+                User.auth_method,
+                User.display_name,
+                User.streak_days,
+                User.streak_last_day_date,
+                User.last_activity_date,
+                UserStats,
+            )
             .join(UserStats, User.id == UserStats.user_id)
             .filter(UserStats.time_frame == time_frame.value, User.show_on_leaderboard)
             .order_by(UserStats.rank)

oasst-shared/oasst_shared/schemas/protocol.py

@@ -36,6 +36,9 @@ class FrontEndUser(User):
     notes: str
     created_date: Optional[datetime] = None
     show_on_leaderboard: bool
+    streak_days: Optional[int] = None
+    streak_last_day_date: Optional[datetime] = None
+    last_activity_date: Optional[datetime] = None
 
 
 class PageResult(BaseModel):
@@ -454,9 +457,9 @@ class UserScore(BaseModel):
     reply_ranked_2: int = 0
     reply_ranked_3: int = 0
 
-    # only used for time frame "total"
     streak_last_day_date: Optional[datetime]
     streak_days: Optional[int]
+    last_activity_date: Optional[datetime]
 
 
 class LeaderboardStats(BaseModel):