Add join command.

data/schema.sql

@@ -113,7 +113,7 @@ CREATE TABLE events(
   communityid INTEGER NOT NULL REFERENCES communities (communityid),
   channel_id TEXT NOT NULL,
   profileid INTEGER REFERENCES user_profiles (profileid),
-  user_id TEXT NOT NULL,
+  user_id TEXT,
   occurred_at TIMESTAMPTZ,
   created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
   UNIQUE (event_id, event_type)
@@ -129,7 +129,6 @@ CREATE TABLE follow_events(
 CREATE TABLE bits_events(
   event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
   event_type TEXT NOT NULL DEFAULT 'bits' CHECK (event_type = 'bits'),
-  user_id TEXT NOT NULL,
   bits INTEGER NOT NULL,
   bits_type TEXT NOT NULL,
   message TEXT,
@@ -159,7 +158,7 @@ CREATE TABLE subscribe_message_events(
   tier INTEGER NOT NULL,
   duration_months INTEGER NOT NULL,
   cumulative_months INTEGER NOT NULL,
-  streak_months INTEGER NOT NULL,
+  streak_months INTEGER,
   message TEXT,
   FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
 );
@@ -257,7 +256,7 @@ CREATE TABLE raid_out_events(
   event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
   event_type TEXT NOT NULL DEFAULT 'raidout' CHECK (event_type = 'raidout'),
   target_id TEXT NOT NULL,
-  target_name TEXT NOT NULL,
+  target_name TEXT,
   viewer_count INTEGER NOT NULL,
   FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
 );
@@ -266,7 +265,7 @@ CREATE TABLE raid_in_events(
   event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
   event_type TEXT NOT NULL DEFAULT 'raidin' CHECK (event_type = 'raidin'),
   source_id TEXT NOT NULL,
-  source_name TEXT NOT NULL,
+  source_name TEXT,
   viewer_count INTEGER NOT NULL,
   FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
 );

src/modules/tracker/component.py

@@ -1,7 +1,7 @@
 from typing import Optional
 import random
 import twitchio
-from twitchio import Scopes, eventsub
+from twitchio import PartialUser, Scopes, eventsub
 from twitchio.ext import commands as cmds
 
 from datamodels import BotChannel, Communities, UserProfile
@@ -42,6 +42,7 @@ class TrackerComponent(cmds.Component):
 
         # Build subscription payloads based on available scopes
         subs = []
+        usersubs = []
         subcls = []
         if Scopes.channel_read_redemptions in scopes or Scopes.channel_manage_redemptions in scopes:
             subcls.append(eventsub.ChannelPointsRedeemAddSubscription)
@@ -82,9 +83,9 @@ class TrackerComponent(cmds.Component):
         #         )
         #     )
         if Scopes.moderator_read_followers in scopes:
-            subs.append(
+            usersubs.append(
                 eventsub.ChannelFollowSubscription(
-                    broadcaser_user_id=channel.userid,
+                    broadcaster_user_id=channel.userid,
                     moderator_user_id=channel.userid,
                 )
             )
@@ -96,11 +97,23 @@ class TrackerComponent(cmds.Component):
 
         responses = []
         for sub in subs:
+            try:
                 if self.bot.using_webhooks:
                     resp = await self.bot.subscribe_webhook(sub)
                 else:
                     resp = await self.bot.subscribe_websocket(sub)
                 responses.append(resp)
+            except Exception:
+                logger.exception("Failed to subscribe to %s", str(sub))
+        for sub in usersubs:
+            try:
+                if self.bot.using_webhooks:
+                    resp = await self.bot.subscribe_webhook(sub)
+                else:
+                    resp = await self.bot.subscribe_websocket(sub, token_for=channel.userid, as_bot=False)
+                responses.append(resp)
+            except Exception:
+                logger.exception("Failed to subscribe to %s", str(sub))
 
         logger.info("Finished tracker subscription to %s: %s", channel.userid, ', '.join(map(str, responses)))
 
@@ -301,6 +314,100 @@ class TrackerComponent(cmds.Component):
                 message=payload.text,
             )
 
+    @cmds.Component.listener()
+    async def event_stream_online(self, payload: twitchio.StreamOnline):
+        tracked = await TrackingChannel.fetch(payload.broadcaster.id)
+        if tracked and tracked.joined:
+            community = await Communities.fetch_or_create(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
+            cid = community.communityid
+
+            event_row = await self.data.events.insert(
+                event_type='stream_online',
+                communityid=cid,
+                channel_id=payload.broadcaster.id,
+                occurred_at=payload.started_at,
+            )
+            detail_row = await self.data.stream_online_events.insert(
+                event_id=event_row['event_id'],
+                stream_id=payload.id,
+                stream_type=payload.type,
+            )
+
+    @cmds.Component.listener()
+    async def event_raid(self, payload: twitchio.ChannelRaid):
+        await self._event_raid_out(
+            payload.from_broadcaster,
+            payload.to_broadcaster,
+            payload.viewer_count,
+        )
+        await self._event_raid_in(
+            payload.to_broadcaster,
+            payload.from_broadcaster,
+            payload.viewer_count,
+        )
+
+    async def _event_raid_out(self, broadcaster: PartialUser, to_broadcaster: PartialUser, viewer_count: int):
+        tracked = await TrackingChannel.fetch(broadcaster.id)
+        if tracked and tracked.joined:
+            community = await Communities.fetch_or_create(twitchid=broadcaster.id, name=broadcaster.name)
+            cid = community.communityid
+
+            event_row = await self.data.events.insert(
+                event_type='raidout',
+                communityid=cid,
+                channel_id=broadcaster.id,
+            )
+            detail_row = await self.data.raid_out_events.insert(
+                event_id=event_row['event_id'],
+                target_id=to_broadcaster.id,
+                target_name=to_broadcaster.name,
+                viewer_count=viewer_count
+            )
+
+    async def _event_raid_in(self, broadcaster: PartialUser, from_broadcaster: PartialUser, viewer_count: int):
+        tracked = await TrackingChannel.fetch(broadcaster.id)
+        if tracked and tracked.joined:
+            community = await Communities.fetch_or_create(twitchid=broadcaster.id, name=broadcaster.name)
+            cid = community.communityid
+
+            event_row = await self.data.events.insert(
+                event_type='raidin',
+                communityid=cid,
+                channel_id=broadcaster.id,
+            )
+            detail_row = await self.data.raid_in_events.insert(
+                event_id=event_row['event_id'],
+                source_id=from_broadcaster.id,
+                source_name=from_broadcaster.name,
+                viewer_count=viewer_count
+            )
+
+    @cmds.Component.listener()
+    async def event_message(self, payload: twitchio.ChatMessage):
+        tracked = await TrackingChannel.fetch(payload.broadcaster.id)
+        if tracked and tracked.joined:
+            community = await Communities.fetch_or_create(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
+            cid = community.communityid
+            profile = await UserProfile.fetch_or_create(
+                twitchid=payload.chatter.id, name=payload.chatter.name,
+            )
+            pid = profile.profileid
+
+            event_row = await self.data.events.insert(
+                event_type='message',
+                communityid=cid,
+                channel_id=payload.broadcaster.id,
+                profileid=pid,
+                user_id=payload.chatter.id,
+            )
+            detail_row = await self.data.message_events.insert(
+                event_id=event_row['event_id'],
+                message_id=payload.id,
+                message_type=payload.type,
+                content=payload.text,
+                source_channel_id=payload.source_id
+            )
+
     # ----- Commands -----
     @cmds.command(name='starttracking')
     async def cmd_starttracking(self, ctx: cmds.Context):
@@ -319,6 +426,7 @@ class TrackerComponent(cmds.Component):
                     Scopes.bits_read,
                     Scopes.channel_read_polls,
                     Scopes.channel_read_vips,
+                    Scopes.moderator_read_followers,
                     *scopes
                 })
             )
@@ -342,3 +450,8 @@ class TrackerComponent(cmds.Component):
             
         else:
             await ctx.reply("Only the broadcaster can enable tracking.")
+
+    @cmds.command(name='join')
+    async def cmd_join(self, ctx: cmds.Context):
+        url = self.bot.get_auth_url()
+        await ctx.reply(f"Invite me to your channel with: {url}")

src/modules/tracker/data.py

@@ -18,9 +18,11 @@ class EventData(Registry):
     events = Table('events')
     follow_events = Table('follow_events')
     bits_events = Table('bits_events')
+
     subscribe_events = Table('subscribe_events')
     gift_events = Table('gift_events')
     subscribe_message_events = Table('subscribe_message_events')
+
     cheer_events = Table('cheer_events')
     redemption_add_events = Table('redemption_add_events')
     redemption_update_events = Table('redemption_update_events')
@@ -28,8 +30,10 @@ class EventData(Registry):
     stream_online_events = Table('stream_online_events')
     stream_offline_events = Table('stream_offline_events')
     channel_update_events = Table('channel_update_events')
+
     vip_add_events = Table('vip_add_events')
     vip_remove_events = Table('vip_remove_events')
+
     raid_out_events = Table('raid_out_events')
     raid_in_events = Table('raid_in_events')
     message_events = Table('message_events')