feat(stats): Initial stats tracker.

data/schema.sql

@@ -96,21 +96,180 @@ CREATE TRIGGER communities_timestamp BEFORE UPDATE ON communities
 -- }}}
 
 
--- Koan data {{{
+-- Twitch tracked event data {{{
 
----- !koans lists koans. !koan gives a random koan. !koans add name ... !koans del name ...
-
-CREATE TABLE koans(
-  koanid INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
-  communityid INTEGER NOT NULL REFERENCES communities ON UPDATE CASCADE ON DELETE CASCADE,
-  name TEXT NOT NULL,
-  message TEXT NOT NULL,
-  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+CREATE TABLE tracking_channels(
+  userid TEXT PRIMARY KEY REFERENCES bot_channels(userid) ON DELETE CASCADE,
+  joined BOOLEAN,
+  joined_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
   _timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW()
 );
-CREATE TRIGGER koans_timestamp BEFORE UPDATE ON koans
+CREATE TRIGGER tracking_channels_timestamp BEFORE UPDATE ON tracking_channels
   FOR EACH ROW EXECUTE FUNCTION update_timestamp_column();
 
+CREATE TABLE events(
+  event_id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+  event_type TEXT NOT NULL,
+  communityid INTEGER NOT NULL REFERENCES communities (communityid),
+  channel_id TEXT NOT NULL,
+  profileid INTEGER REFERENCES user_profiles (profileid),
+  user_id TEXT NOT NULL,
+  occurred_at TIMESTAMPTZ,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  UNIQUE (event_id, event_type)
+);
+
+CREATE TABLE follow_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'follow' CHECK (event_type = 'follow'),
+  follower_count INTEGER NOT NULL,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+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,
+  powerup_type TEXT,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE subscribe_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'subscribe' CHECK (event_type = 'subscribe'),
+  tier INTEGER NOT NULL,
+  gifted BOOLEAN NOT NULL,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE gift_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'gift' CHECK (event_type = 'gift'),
+  tier INTEGER NOT NULL,
+  gifted_count INTEGER NOT NULL,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE subscribe_message_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'subscribe_message' CHECK (event_type = 'subscribe_message'),
+  tier INTEGER NOT NULL,
+  duration_months INTEGER NOT NULL,
+  cumulative_months INTEGER NOT NULL,
+  streak_months INTEGER NOT NULL,
+  message TEXT,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE cheer_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'cheer' CHECK (event_type = 'cheer'),
+  amount INTEGER NOT NULL,
+  message TEXT,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE redemption_add_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'redemption_add' CHECK (event_type = 'redemption_add'),
+  redeem_id TEXT NOT NULL,
+  redeem_title TEXT NOT NULL,
+  redeem_cost INTEGER NOT NULL,
+  redemption_id TEXT NOT NULL,
+  redemption_status TEXT NOT NULL,
+  message TEXT,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE redemption_update_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'redemption_update' CHECK (event_type = 'redemption_update'),
+  redeem_id TEXT NOT NULL,
+  redeem_title TEXT NOT NULL,
+  redeem_cost INTEGER NOT NULL,
+  redemption_id TEXT NOT NULL,
+  redemption_status TEXT NOT NULL,
+  redeemed_at TIMESTAMPTZ NOT NULL,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE poll_end_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'poll_end' CHECK (event_type = 'poll_end'),
+  poll_id TEXT NOT NULL,
+  poll_title TEXT NOT NULL,
+  poll_choices TEXT NOT NULL,
+  poll_started TIMESTAMPTZ NOT NULL,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE stream_online_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'stream_online' CHECK (event_type = 'stream_online'),
+  stream_id TEXT NOT NULL,
+  stream_type TEXT NOT NULL,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE stream_offline_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'stream_offline' CHECK (event_type = 'stream_offline'),
+  stream_id TEXT,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE channel_update_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'channel_update' CHECK (event_type = 'channel_update'),
+  title TEXT,
+  language TEXT,
+  category_id TEXT,
+  category_name TEXT,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE vip_add_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'vip_add' CHECK (event_type = 'vip_add'),
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE vip_remove_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'vip_remove' CHECK (event_type = 'vip_remove'),
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+CREATE TABLE message_events(
+  event_id INTEGER PRIMARY KEY REFERENCES events (event_id),
+  event_type TEXT NOT NULL DEFAULT 'message' CHECK (event_type = 'message'),
+  message_id TEXT NOT NULL,
+  message_type TEXT NOT NULL,
+  content TEXT,
+  source_channel_id TEXT,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+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,
+  viewer_count INTEGER NOT NULL,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
+
+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,
+  viewer_count INTEGER NOT NULL,
+  FOREIGN KEY (event_id, event_type) REFERENCES events (event_id, event_type)
+);
 -- }}}
 
 -- vim: set fdm=marker:

src/bot.py

@@ -7,6 +7,8 @@ from meta import CrocBot, conf, setup_main_logger, args
 from data import Database
 from constants import DATA_VERSION
 
+from modules import setup
+
 logger = logging.getLogger(__name__)
 
 
@@ -30,10 +32,9 @@ async def main():
             config=conf,
             dbconn=db,
             adapter=adapter,
+            setup=setup,
         )
 
-        # await bot.load_module('modules')
-
         try:
             await bot.start()
         finally:

src/datamodels.py

@@ -90,31 +90,6 @@ class Communities(RowModel):
     _timestamp = Timestamp()
 
 
-class Koan(RowModel):
-    """
-    Schema
-    ======
-    CREATE TABLE koans(
-      koanid INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
-      communityid INTEGER NOT NULL REFERENCES communities ON UPDATE CASCADE ON DELETE CASCADE,
-      name TEXT NOT NULL,
-      message TEXT NOT NULL,
-      created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-      _timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW()
-    );
-    """
-    _tablename_ = 'koans'
-    _cache_ = {}
-
-    koanid = Integer(primary=True)
-    communityid = Integer()
-    name = String()
-    message = String()
-    created_at = Timestamp()
-    _timestamp = Timestamp()
-
-
-
 class BotData(Registry):
     user_auth = UserAuth.table
 
@@ -129,5 +104,3 @@ class BotData(Registry):
     bot_channels = BotChannel.table
     user_profiles = UserProfile.table 
     communities = Communities.table
-
-    koans = Koan.table

src/meta/crocbot.py

@@ -15,7 +15,7 @@ logger = logging.getLogger(__name__)
 
 
 class CrocBot(commands.Bot):
-    def __init__(self, *args, config: Conf, dbconn: Database, **kwargs):
+    def __init__(self, *args, config: Conf, dbconn: Database, setup=None, **kwargs):
         kwargs.setdefault('client_id', config.bot['client_id'])
         kwargs.setdefault('client_secret', config.bot['client_secret'])
         kwargs.setdefault('bot_id', config.bot['bot_id'])
@@ -23,9 +23,15 @@ class CrocBot(commands.Bot):
 
         super().__init__(*args, **kwargs)
 
+        if config.bot.get('eventsub_secret', None):
+            self.using_webhooks = True
+        else:
+            self.using_webhooks = False
+
         self.config = config
         self.dbconn = dbconn
         self.data: BotData = dbconn.load_registry(BotData())
+        self._setup_hook = setup
 
         self.joined: dict[str, BotChannel] = {}
 
@@ -35,6 +41,9 @@ class CrocBot(commands.Bot):
 
     async def setup_hook(self):
         await self.data.init()
+
+        if self._setup_hook is not None:
+            await self._setup_hook(self)
         
         # Get all current bot channels 
         channels = await BotChannel.fetch_where(autojoin=True)
@@ -77,7 +86,6 @@ class CrocBot(commands.Bot):
         Register webhook subscriptions to the given channel(s).
         """
         # TODO: If channels are already joined, unsubscribe
-        # TODO: Determine (or switch?) whether to use webhook or websocket
         for channel in channels:
             sub = None
             try:
@@ -85,19 +93,16 @@ class CrocBot(commands.Bot):
                     broadcaster_user_id=channel.userid,
                     user_id=self.bot_id,
                 )
-                resp = await self.subscribe_websocket(sub)
+                if self.using_webhooks:
+                    resp = await self.subscribe_webhook(sub)
+                else:
+                    resp = await self.subscribe_websocket(sub)
                 logger.info("Subscribed to %s with %s response %s", channel.userid, sub, resp)
-                if channel.listen_redeems:
-                    sub = eventsub.ChannelPointsRedeemAddSubscription(
-                        broadcaster_user_id=channel.userid 
-                    )
-                    resp = await self.subscribe_websocket(sub, as_bot=False, token_for=channel.userid)
-                    logger.info("Subscribed to %s with %s response %s", channel.userid, sub, resp)
                 self.joined[channel.userid] = channel
+                self.safe_dispatch('channel_joined', payload=channel)
             except Exception:
                 logger.exception("Failed to subscribe to %s with %s", channel.userid, sub)
 
-
     async def event_oauth_authorized(self, payload: UserTokenPayload):
         logger.debug("Oauth flow authorization with payload %s", repr(payload))
         # Save the token and scopes and update internal authorisations

src/modules/__init__.py

@@ -1,15 +1,3 @@
-
-this_package = 'modules'
-
-active = [
-    '.vstudio',
-]
-
-
-def prepare(bot):
-    for ext in active:
-        bot.load_module(this_package + ext)
-
 async def setup(bot):
-    for ext in active:
-        await bot.load_module(this_package + ext)
+    from . import tracker
+    await tracker.setup(bot)

src/modules/tracker/__init__.py

@@ -0,0 +1,7 @@
+import logging
+
+logger = logging.getLogger()
+
+async def setup(bot):
+    from .component import TrackerComponent 
+    await bot.add_component(TrackerComponent(bot))

src/modules/tracker/component.py

@@ -0,0 +1,344 @@
+from typing import Optional
+import random
+import twitchio
+from twitchio import Scopes, eventsub
+from twitchio.ext import commands as cmds
+
+from datamodels import BotChannel, Communities, UserProfile
+from meta import CrocBot
+from utils.lib import utc_now
+
+from . import logger
+from .data import EventData, TrackingChannel
+
+
+class TrackerComponent(cmds.Component):
+    def __init__(self, bot: CrocBot):
+        self.bot = bot
+        self.data = bot.dbconn.load_registry(EventData())
+        print(self.__all_listeners__)
+
+    # One command, which sends join link to start tracking
+    # Utility for fetch_or_create community and profiles and setting names
+    # Then just a stack of event listeners.
+
+    # ----- API -----
+    async def component_load(self):
+        await self.data.init()
+
+    async def component_teardown(self):
+        pass
+
+    # ----- Methods -----
+    async def start_tracking(self, channel: TrackingChannel):
+        # TODO: Make sure that we aren't trying to make duplicate subscriptions here
+        logger.debug(
+            "Initialising event tracking for %s",
+            channel.userid
+        )
+        # Get associated auth scopes
+        rows = await self.bot.data.user_auth_scopes.select_where(userid=channel.userid)
+        scopes = Scopes([row['scope'] for row in rows])
+
+        # Build subscription payloads based on available scopes
+        subs = []
+        subcls = []
+        if Scopes.channel_read_redemptions in scopes or Scopes.channel_manage_redemptions in scopes:
+            subcls.append(eventsub.ChannelPointsRedeemAddSubscription)
+            subcls.append(eventsub.ChannelPointsRedeemUpdateSubscription)
+        if Scopes.bits_read in scopes:
+            subcls.append(eventsub.ChannelBitsUseSubscription)
+            subcls.append(eventsub.ChannelCheerSubscription)
+        if Scopes.channel_read_subscriptions in scopes:
+            subcls.extend((
+                eventsub.ChannelSubscribeSubscription,
+                eventsub.ChannelSubscribeMessageSubscription,
+                eventsub.ChannelSubscriptionGiftSubscription,
+            ))
+        if Scopes.channel_read_polls in scopes or Scopes.channel_manage_polls in scopes:
+            subcls.append(eventsub.ChannelPollEndSubscription)
+        if Scopes.channel_read_vips in scopes or Scopes.channel_manage_vips in scopes:
+            subcls.extend((
+                eventsub.ChannelVIPAddSubscription,
+                eventsub.ChannelVIPRemoveSubscription,
+            ))
+
+        subcls.extend((
+            eventsub.StreamOnlineSubscription,
+            eventsub.StreamOfflineSubscription,
+            eventsub.ChannelUpdateSubscription,
+        ))
+
+        for subbr in subcls:
+            subs.append(subbr(broadcaster_user_id=channel.userid))
+
+        # This isn't needed because joining the channel means we have these
+        # Including them for completeness though
+        # if Scopes.chat_read in scopes or Scopes.channel_bot in scopes:
+        #     subs.append(
+        #         eventsub.ChatMessageSubscription(
+        #             broadcaster_user_id=channel.userid,
+        #             user_id=self.bot.bot_id,
+        #         )
+        #     )
+        if Scopes.moderator_read_followers in scopes:
+            subs.append(
+                eventsub.ChannelFollowSubscription(
+                    broadcaser_user_id=channel.userid,
+                    moderator_user_id=channel.userid,
+                )
+            )
+
+        subs.extend((
+            eventsub.ChannelRaidSubscription(to_broadcaster_user_id=channel.userid),
+            eventsub.ChannelRaidSubscription(from_broadcaster_user_id=channel.userid),
+        ))
+
+        responses = []
+        for sub in subs:
+            if self.bot.using_webhooks:
+                resp = await self.bot.subscribe_webhook(sub)
+            else:
+                resp = await self.bot.subscribe_websocket(sub)
+            responses.append(resp)
+
+        logger.info("Finished tracker subscription to %s: %s", channel.userid, ', '.join(map(str, responses)))
+
+    # ----- Events -----
+
+    @cmds.Component.listener()
+    async def event_safe_channel_joined(self, payload: BotChannel):
+        # Check if the channel is tracked 
+        # If it is, call start_tracking
+        tracked = await TrackingChannel.fetch(payload.userid)
+        if tracked and tracked.joined:
+            await self.start_tracking(tracked)
+
+    @cmds.Component.listener()
+    async def event_custom_redemption_add(self, payload: twitchio.ChannelPointsRedemptionAdd):
+        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.user.id, name=payload.user.name,
+            )
+            pid = profile.profileid
+
+            event_row = await self.data.events.insert(
+                event_type='redemption_add',
+                communityid=cid,
+                channel_id=payload.broadcaster.id,
+                profileid=pid,
+                user_id=payload.user.id,
+                occurred_at=payload.redeemed_at,
+            )
+            detail_row = await self.data.redemption_add_events.insert(
+                event_id=event_row['event_id'],
+                redeem_id=payload.reward.id,
+                redeem_title=payload.reward.title,
+                redeem_cost=payload.reward.cost,
+                redemption_id=payload.id,
+                redemption_status=payload.status,
+                message=payload.user_input,
+            )
+
+    @cmds.Component.listener()
+    async def event_custom_redemption_update(self, payload: twitchio.ChannelPointsRedemptionUpdate):
+        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.user.id, name=payload.user.name,
+            )
+            pid = profile.profileid
+
+            event_row = await self.data.events.insert(
+                event_type='redemption_update',
+                communityid=cid,
+                channel_id=payload.broadcaster.id,
+                profileid=pid,
+                user_id=payload.user.id,
+            )
+            detail_row = await self.data.redemption_update_events.insert(
+                event_id=event_row['event_id'],
+                redeem_id=payload.reward.id,
+                redeem_title=payload.reward.title,
+                redeem_cost=payload.reward.cost,
+                redemption_id=payload.id,
+                redemption_status=payload.status,
+                redeemed_at=utc_now()
+            )
+
+    @cmds.Component.listener()
+    async def event_follow(self, payload: twitchio.ChannelFollow):
+        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.user.id, name=payload.user.name,
+            )
+            pid = profile.profileid
+
+            # Computer follower count 
+            followers = await payload.broadcaster.fetch_followers()
+
+            event_row = await self.data.events.insert(
+                event_type='follow',
+                communityid=cid,
+                channel_id=payload.broadcaster.id,
+                profileid=pid,
+                user_id=payload.user.id,
+            )
+            detail_row = await self.data.follow_events.insert(
+                event_id=event_row['event_id'],
+                follower_count=followers.total
+            )
+
+    @cmds.Component.listener()
+    async def event_bits_use(self, payload: twitchio.ChannelBitsUse):
+        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.user.id, name=payload.user.name,
+            )
+            pid = profile.profileid
+
+            event_row = await self.data.events.insert(
+                event_type='bits',
+                communityid=cid,
+                channel_id=payload.broadcaster.id,
+                profileid=pid,
+                user_id=payload.user.id,
+            )
+            detail_row = await self.data.bits_events.insert(
+                event_id=event_row['event_id'],
+                bits=payload.bits,
+                bits_type=payload.type,
+                message=payload.text,
+                powerup_type=payload.power_up.type if payload.power_up else None
+            )
+
+    @cmds.Component.listener()
+    async def event_subscription(self, payload: twitchio.ChannelSubscribe):
+        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.user.id, name=payload.user.name,
+            )
+            pid = profile.profileid
+
+            event_row = await self.data.events.insert(
+                event_type='subscribe',
+                communityid=cid,
+                channel_id=payload.broadcaster.id,
+                profileid=pid,
+                user_id=payload.user.id,
+            )
+            detail_row = await self.data.subscribe_events.insert(
+                event_id=event_row['event_id'],
+                tier=payload.tier,
+                gifted=payload.gift,
+            )
+
+    @cmds.Component.listener()
+    async def event_subscription_gift(self, payload: twitchio.ChannelSubscriptionGift):
+        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
+            if payload.user is not None:
+                profile = await UserProfile.fetch_or_create(
+                    twitchid=payload.user.id, name=payload.user.name,
+                )
+                pid = profile.profileid
+            else:
+                pid = None
+
+            event_row = await self.data.events.insert(
+                event_type='gift',
+                communityid=cid,
+                channel_id=payload.broadcaster.id,
+                profileid=pid,
+                user_id=payload.user.id if payload.user else None,
+            )
+            detail_row = await self.data.gift_events.insert(
+                event_id=event_row['event_id'],
+                tier=payload.tier,
+                gifted_count=payload.total,
+            )
+
+    @cmds.Component.listener()
+    async def event_subscription_message(self, payload: twitchio.ChannelSubscriptionMessage):
+        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.user.id, name=payload.user.name,
+            )
+            pid = profile.profileid
+
+            event_row = await self.data.events.insert(
+                event_type='subscribe_message',
+                communityid=cid,
+                channel_id=payload.broadcaster.id,
+                profileid=pid,
+                user_id=payload.user.id,
+            )
+            detail_row = await self.data.subscribe_message_events.insert(
+                event_id=event_row['event_id'],
+                tier=payload.tier,
+                duration_months=payload.months,
+                cumulative_months=payload.cumulative_months,
+                streak_months=payload.streak_months,
+                message=payload.text,
+            )
+
+    # ----- Commands -----
+    @cmds.command(name='starttracking')
+    async def cmd_starttracking(self, ctx: cmds.Context):
+        if ctx.broadcaster:
+            tracking = await TrackingChannel.fetch_or_create(ctx.channel.id, joined=True)
+            if not tracking.joined:
+                await tracking.update(joined=True)
+
+            rows = await self.bot.data.user_auth_scopes.select_where(userid=ctx.channel.id)
+            scopes = Scopes([row['scope'] for row in rows])
+
+            url = self.bot.get_auth_url(
+                Scopes({
+                    Scopes.channel_read_subscriptions,
+                    Scopes.channel_read_redemptions,
+                    Scopes.bits_read,
+                    Scopes.channel_read_polls,
+                    Scopes.channel_read_vips,
+                    *scopes
+                })
+            )
+            await ctx.reply(f"Tracking enabled! Please authorise me to track events in this channel: {url}")
+        else:
+            await ctx.reply("Only the broadcaster can enable tracking.")
+
+    @cmds.command(name='stoptracking')
+    async def cmd_stoptracking(self, ctx: cmds.Context):
+        if ctx.broadcaster:
+            tracking = await TrackingChannel.fetch(ctx.channel.id)
+            if tracking and tracking.joined:
+                await tracking.update(joined=False)
+                # TODO: Actually disable the subscriptions instead of just on the next restart 
+                # This is tricky because some of the subscriptions may have been requested by other modules
+                # Requires keeping track of the source of subscriptions, and having a central manager disable them when no-one is listening anymore.
+                pass
+                await ctx.reply("Event tracking has been disabled.")
+            else:
+                await ctx.reply("Event tracking is not enabled!")
+            
+        else:
+            await ctx.reply("Only the broadcaster can enable tracking.")

src/modules/tracker/data.py

@@ -0,0 +1,35 @@
+from data import Registry, RowModel, Table
+from data.columns import String, Timestamp, Integer, Bool
+
+
+class TrackingChannel(RowModel):
+    _tablename_ = 'tracking_channels'
+    _cache_ = {}
+
+    userid = String(primary=True)
+    joined = Bool 
+    joined_at = Timestamp()
+    _timestamp = Timestamp()
+
+
+class EventData(Registry):
+    tracking_channels = TrackingChannel.table
+
+    events = Table('events')
+    follow_events = Table('follow_events')
+    bits_events = Table('bit_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')
+    poll_end_events = Table('poll_end_events')
+    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')