[PAY-3848] Add ListenStreakReminder notification (#11389)

Co-authored-by: Dharit Tantiviramanond <dharit.tan@gmail.com>

packages/common/src/adapters/notification.ts

@@ -619,5 +619,13 @@ export const notificationFromSDK = (
         ...formatBaseNotification(notification)
       }
     }
+    case 'listen_streak_reminder': {
+      const data = notification.actions[0].data
+      return {
+        type: NotificationType.ListenStreakReminder,
+        streak: data.streak,
+        ...formatBaseNotification(notification)
+      }
+    }
   }
 }

packages/common/src/messages/index.ts

@@ -6,3 +6,4 @@ export * from './edit'
 export * from './comments'
 export * from './settings'
 export * from './trackPage'
+export * from './notifications'

packages/common/src/messages/notifications.ts

@@ -0,0 +1,5 @@
+export const listenStreakReminderMessages = {
+  title: 'Keep Your Streak Going!',
+  body: (streak: number) =>
+    `Your ${streak} day listening streak will end in 6 hours! Keep listening to earn daily rewards!`
+}

packages/common/src/store/notifications/types.ts

@@ -43,7 +43,8 @@ export enum NotificationType {
   Comment = 'Comment',
   CommentThread = 'CommentThread',
   CommentMention = 'CommentMention',
-  CommentReaction = 'CommentReaction'
+  CommentReaction = 'CommentReaction',
+  ListenStreakReminder = 'ListenStreakReminder'
 }
 
 export enum PushNotificationType {
@@ -91,7 +92,8 @@ export enum PushNotificationType {
   Comment = 'Comment',
   CommentThread = 'CommentThread',
   CommentMention = 'CommentMention',
-  CommentReaction = 'CommentReaction'
+  CommentReaction = 'CommentReaction',
+  ListenStreakReminder = 'ListenStreakReminder'
 }
 
 export enum Entity {
@@ -688,6 +690,11 @@ export type CommentReactionNotification = BaseNotification & {
   entityType: Entity.Playlist | Entity.Album | Entity.Track
 }
 
+export type ListenStreakReminderNotification = BaseNotification & {
+  type: NotificationType.ListenStreakReminder
+  streak: number
+}
+
 export type Notification =
   | AnnouncementNotification
   | UserSubscriptionNotification
@@ -722,6 +729,7 @@ export type Notification =
   | CommentThreadNotification
   | CommentMentionNotification
   | CommentReactionNotification
+  | ListenStreakReminderNotification
 
 export type IdentityNotification = Omit<Notification, 'timestamp'> & {
   timestamp: string

packages/discovery-provider/integration_tests/tasks/test_create_listen_streak_reminder_notifications.py

@@ -0,0 +1,156 @@
+import logging
+from datetime import datetime, timedelta
+
+from integration_tests.utils import populate_mock_db
+from src.models.notifications.notification import Notification
+from src.tasks.create_listen_streak_reminder_notifications import (
+    LISTEN_STREAK_REMINDER,
+    _create_listen_streak_reminder_notifications,
+    get_listen_streak_notification_group_id,
+)
+from src.utils.db_session import get_db
+
+logger = logging.getLogger(__name__)
+
+TEST_USER_ID = 1
+TEST_STREAK = 3
+TEST_DATE = (datetime.now() - timedelta(hours=18, minutes=30)).date()
+TEST_GROUP_ID = get_listen_streak_notification_group_id(TEST_USER_ID, TEST_DATE)
+
+
+def test_create_listen_streak_reminder_notification(app):
+    """Test basic creation of listen streak notification"""
+    with app.app_context():
+        db = get_db()
+
+    now = datetime.now()
+    # Place listen time clearly within the window (between 18-19 hours ago)
+    listen_time = now - timedelta(hours=18, minutes=30)
+
+    entities = {
+        "challenge_listen_streaks": [
+            {
+                "user_id": TEST_USER_ID,
+                "listen_streak": TEST_STREAK,
+                "last_listen_date": listen_time,
+            }
+        ]
+    }
+
+    populate_mock_db(db, entities)
+    with db.scoped_session() as session:
+        _create_listen_streak_reminder_notifications(session)
+        all_notifications = session.query(Notification).all()
+
+        assert len(all_notifications) == 1
+        notification = all_notifications[0]
+        assert notification.specifier == str(TEST_USER_ID)
+        assert notification.group_id == TEST_GROUP_ID
+        assert notification.data == {"streak": TEST_STREAK}
+        assert notification.type == LISTEN_STREAK_REMINDER
+        assert notification.user_ids == [TEST_USER_ID]
+
+
+def test_ignore_outside_time_window(app):
+    """Test that streaks outside the notification window are ignored"""
+    with app.app_context():
+        db = get_db()
+
+    now = datetime.now()
+    too_recent = now - timedelta(hours=17)  # Too recent (< 18 hours ago)
+    too_old = now - timedelta(hours=20)  # Too old (> 19 hours ago)
+
+    entities = {
+        "challenge_listen_streaks": [
+            {
+                "user_id": TEST_USER_ID,
+                "listen_streak": TEST_STREAK,
+                "last_listen_date": too_recent,
+            },
+            {
+                "user_id": TEST_USER_ID + 1,
+                "listen_streak": TEST_STREAK,
+                "last_listen_date": too_old,
+            },
+        ]
+    }
+
+    populate_mock_db(db, entities)
+    with db.scoped_session() as session:
+        _create_listen_streak_reminder_notifications(session)
+        all_notifications = session.query(Notification).all()
+        assert len(all_notifications) == 0
+
+
+def test_ignore_existing_notification(app):
+    """Test that duplicate notifications are not created"""
+    with app.app_context():
+        db = get_db()
+
+    now = datetime.now()
+    listen_time = now - timedelta(hours=18, minutes=30)
+
+    entities = {
+        "challenge_listen_streaks": [
+            {
+                "user_id": TEST_USER_ID,
+                "listen_streak": TEST_STREAK,
+                "last_listen_date": listen_time,
+            }
+        ],
+        "notifications": [
+            {
+                "specifier": str(TEST_USER_ID),
+                "group_id": TEST_GROUP_ID,
+                "type": LISTEN_STREAK_REMINDER,
+                "user_ids": [TEST_USER_ID],
+                "data": {"streak": TEST_STREAK},
+                "timestamp": listen_time,
+            }
+        ],
+    }
+
+    populate_mock_db(db, entities)
+    with db.scoped_session() as session:
+        _create_listen_streak_reminder_notifications(session)
+        all_notifications = session.query(Notification).all()
+        assert len(all_notifications) == 1  # Only the existing notification
+
+
+def test_multiple_streaks(app):
+    """Test handling multiple eligible listen streaks"""
+    with app.app_context():
+        db = get_db()
+
+    now = datetime.now()
+    listen_time = now - timedelta(hours=18, minutes=30)
+
+    entities = {
+        "challenge_listen_streaks": [
+            {
+                "user_id": TEST_USER_ID,
+                "listen_streak": TEST_STREAK,
+                "last_listen_date": listen_time,
+            },
+            {
+                "user_id": TEST_USER_ID + 1,
+                "listen_streak": TEST_STREAK + 1,
+                "last_listen_date": listen_time,
+            },
+        ]
+    }
+
+    populate_mock_db(db, entities)
+    with db.scoped_session() as session:
+        _create_listen_streak_reminder_notifications(session)
+        all_notifications = session.query(Notification).all()
+
+        assert len(all_notifications) == 2
+        for notification in all_notifications:
+            assert notification.type == LISTEN_STREAK_REMINDER
+            user_id = int(notification.specifier)
+            assert notification.user_ids == [user_id]
+            expected_streak = (
+                TEST_STREAK if user_id == TEST_USER_ID else TEST_STREAK + 1
+            )
+            assert notification.data == {"streak": expected_streak}

packages/discovery-provider/integration_tests/utils.py

@@ -24,6 +24,7 @@
 from src.models.playlists.playlist_track import PlaylistTrack
 from src.models.rewards.challenge import Challenge
 from src.models.rewards.challenge_disbursement import ChallengeDisbursement
+from src.models.rewards.listen_streak_challenge import ChallengeListenStreak
 from src.models.rewards.reward_manager import RewardManagerTransaction
 from src.models.rewards.user_challenge import UserChallenge
 from src.models.social.aggregate_monthly_plays import AggregateMonthlyPlay
@@ -154,6 +155,7 @@ def populate_mock_db(db, entities, block_offset=None):
         stems = entities.get("stems", [])
         challenges = entities.get("challenges", [])
         user_challenges = entities.get("user_challenges", [])
+        challenge_listen_streaks = entities.get("challenge_listen_streaks", [])
         plays = entities.get("plays", [])
         aggregate_plays = entities.get("aggregate_plays", [])
         aggregate_track = entities.get("aggregate_track", [])
@@ -932,4 +934,14 @@ def populate_mock_db(db, entities, block_offset=None):
             )
             session.add(collectible_data_record)
 
+        for i, challenge_listen_streak in enumerate(challenge_listen_streaks):
+            streak = ChallengeListenStreak(
+                user_id=challenge_listen_streak.get("user_id", i),
+                last_listen_date=challenge_listen_streak.get(
+                    "last_listen_date", datetime.now()
+                ),
+                listen_streak=challenge_listen_streak.get("listen_streak", 1),
+            )
+            session.add(streak)
+
         session.commit()

packages/discovery-provider/scripts/lint.sh

@@ -1,4 +1,4 @@
-isort --diff . --skip ./src/tasks/core/gen --skip ./plugins
+isort . --skip ./src/tasks/core/gen --skip ./plugins
 flake8 . --exclude=./src/tasks/core/gen,./plugins
-black --diff . --exclude './src/tasks/core/gen|./plugins'
+black . --exclude './src/tasks/core/gen|./plugins'
 mypy . --exclude './src/tasks/core/gen|./plugins'

packages/discovery-provider/src/api/v1/models/notifications.py

@@ -910,6 +910,31 @@ def format(self, value):
     },
 )
 
+listen_streak_reminder_notification_action_data = ns.model(
+    "listen_streak_reminder_notification_action_data",
+    {
+        "streak": fields.Integer(required=True),
+    },
+)
+listen_streak_reminder_notification_action = ns.clone(
+    "listen_streak_reminder_notification_action",
+    notification_action_base,
+    {
+        "data": fields.Nested(
+            listen_streak_reminder_notification_action_data, required=True
+        )
+    },
+)
+listen_streak_reminder_notification = ns.clone(
+    "listen_streak_reminder_notification",
+    notification_base,
+    {
+        "actions": fields.List(
+            fields.Nested(listen_streak_reminder_notification_action, required=True),
+            required=True,
+        )
+    },
+)
 
 notification = ns.add_model(
     "notification",
@@ -949,6 +974,7 @@ def format(self, value):
             "comment_thread": comment_thread_notification,
             "comment_mention": comment_mention_notification,
             "comment_reaction": comment_reaction_notification,
+            "listen_streak_reminder": listen_streak_reminder_notification,
         },
         discriminator="type",
     ),

packages/discovery-provider/src/api/v1/utils/extend_notification.py

@@ -16,6 +16,7 @@
     CreateTrackNotification,
     FollowerMilestoneNotification,
     FollowNotification,
+    ListenStreakReminderNotification,
     Notification,
     NotificationAction,
     NotificationData,
@@ -742,6 +743,22 @@ def extend_comment_reaction(action: NotificationAction):
     }
 
 
+def extend_listen_streak_reminder(action: NotificationAction):
+    data: ListenStreakReminderNotification = action["data"]  # type: ignore
+    return {
+        "specifier": encode_int_id(int(action["specifier"])),
+        "type": action["type"],
+        "timestamp": (
+            datetime.timestamp(action["timestamp"])
+            if action["timestamp"]
+            else action["timestamp"]
+        ),
+        "data": {
+            "streak": data["streak"],
+        },
+    }
+
+
 notification_action_handler = {
     "follow": extend_follow,
     "repost": extend_repost,
@@ -776,4 +793,5 @@ def extend_comment_reaction(action: NotificationAction):
     "comment_thread": extend_comment_thread,
     "comment_mention": extend_comment_mention,
     "comment_reaction": extend_comment_reaction,
+    "listen_streak_reminder": extend_listen_streak_reminder,
 }

packages/discovery-provider/src/app.py

@@ -326,6 +326,7 @@ def configure_celery(celery, test_config=None):
             "src.tasks.cache_entity_counts",
             "src.tasks.publish_scheduled_releases",
             "src.tasks.create_engagement_notifications",
+            "src.tasks.create_listen_streak_reminder_notifications",
             "src.tasks.index_core",
         ],
         beat_schedule={
@@ -425,6 +426,10 @@ def configure_celery(celery, test_config=None):
                 "task": "create_engagement_notifications",
                 "schedule": timedelta(minutes=10),
             },
+            "create_listen_streak_reminder_notifications": {
+                "task": "create_listen_streak_reminder_notifications",
+                "schedule": timedelta(seconds=10),
+            },
             "repair_audio_analyses": {
                 "task": "repair_audio_analyses",
                 "schedule": timedelta(minutes=3),

packages/discovery-provider/src/queries/get_notifications.py

@@ -173,6 +173,7 @@ class NotificationType(str, Enum):
     COMMENT_THREAD = "comment_thread"
     COMMENT_MENTION = "comment_mention"
     COMMENT_REACTION = "comment_reaction"
+    LISTEN_STREAK_REMINDER = "listen_streak_reminder"
 
     def __str__(self) -> str:
         return str.__str__(self)
@@ -477,6 +478,10 @@ class CommentReactionNotification(TypedDict):
     reacter_user_id: int
 
 
+class ListenStreakReminderNotification(TypedDict):
+    streak: int
+
+
 NotificationData = Union[
     AnnouncementNotification,
     FollowNotification,
@@ -508,6 +513,7 @@ class CommentReactionNotification(TypedDict):
     CommentThreadNotification,
     CommentMentionNotification,
     CommentReactionNotification,
+    ListenStreakReminderNotification,
 ]