fix: Handle no timecap.

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,10 +265,52 @@ 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)
 );
 -- }}}
 
+-- Subathon timer data {{{
+
+CREATE TABLE subathons(
+  subathon_id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+  communityid INTEGER NOT NULL REFERENCES communities(communityid),
+  started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  initial_time INTEGER NOT NULL,
+  sub1_score NUMERIC NOT NULL DEFAULT 1,
+  sub2_score NUMERIC NOT NULL DEFAULT 2,
+  sub3_score NUMERIC NOT NULL DEFAULT 6,
+  bit_score NUMERIC NOT NULL,
+  score_time NUMERIC NOT NULL,
+  duration INTEGER NOT NULL DEFAULT 0,
+  ended_at TIMESTAMPTZ
+);
+
+CREATE TABLE running_subathons(
+  subathon_id INTEGER PRIMARY KEY REFERENCES subathons(subathon_id),
+  last_started TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE TABLE subathon_contributions(
+  contribution_id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+  subathon_id INTEGER NOT NULL REFERENCES subathons(subathon_id),
+  profileid INTEGER REFERENCES user_profiles(profileid),
+  score NUMERIC NOT NULL,
+  event_id INTEGER REFERENCES events(event_id)
+);
+
+CREATE TABLE subathon_goals(
+  goal_id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+  subathon_id INTEGER NOT NULL REFERENCES subathons(subathon_id),
+  required_score NUMERIC NOT NULL,
+  description TEXT NOT NULL,
+  notified BOOLEAN DEFAULT false,
+  created_at TIMESTAMPTZ DEFAULT NOW()
+);
+
+
+-- }}}
+
+
 -- vim: set fdm=marker:

src/bot.py

@@ -1,11 +1,13 @@
 import asyncio
 import logging
+import websockets
 
 from twitchio.web import AiohttpAdapter
 
 from meta import CrocBot, conf, setup_main_logger, args
 from data import Database
 from constants import DATA_VERSION
+import sockets
 
 from modules import setup
 
@@ -26,8 +28,10 @@ async def main():
             host=conf.bot.get('wshost', None),
             port=conf.bot.getint('wsport', None),
             domain=conf.bot.get('wsdomain', None),
+            eventsub_secret=conf.bot.get('eventsub_secret', None)
         )
 
+        async with websockets.serve(sockets.root_handler, '', conf.wserver.getint('port')):
             bot = CrocBot(
                 config=conf,
                 dbconn=db,

src/meta/crocbot.py

@@ -6,7 +6,7 @@ from twitchio.ext import commands
 from twitchio import Scopes, eventsub
 
 from data import Database
-from datamodels import BotData, UserAuth, BotChannel
+from datamodels import BotData, Communities, UserAuth, BotChannel, UserProfile
 
 from .config import Conf
 
@@ -39,6 +39,22 @@ class CrocBot(commands.Bot):
         # logger.info(f"Logged in as {self.nick}. User id is {self.user_id}")
         logger.info("Logged in as %s", self.bot_id)
 
+    async def profile_fetch(self, twitchid, name):
+        profiles = await UserProfile.fetch_where(twitchid=twitchid)
+        if not profiles:
+            profile = await UserProfile.create(twitchid=twitchid, name=name)
+        else:
+            profile = profiles[0]
+        return profile
+
+    async def community_fetch(self, twitchid, name):
+        profiles = await Communities.fetch_where(twitchid=twitchid)
+        if not profiles:
+            profile = await Communities.create(twitchid=twitchid, name=name)
+        else:
+            profile = profiles[0]
+        return profile
+
     async def setup_hook(self):
         await self.data.init()
 
@@ -159,4 +175,7 @@ class CrocBot(commands.Bot):
 
     async def load_tokens(self, path: str | None = None):
         for row in await UserAuth.fetch_where():
+            try:
                 await self.add_token(row.token, row.refresh_token)
+            except Exception:
+                logger.exception(f"Failed to add token for {row}")

src/modules/__init__.py

@@ -1,3 +1,5 @@
 async def setup(bot):
     from . import tracker
+    from . import subathons
     await tracker.setup(bot)
+    await subathons.setup(bot)

src/modules/subathons/__init__.py

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

src/modules/subathons/component.py

@@ -0,0 +1,493 @@
+from datetime import datetime, timedelta
+from typing import Optional
+import random
+import twitchio
+from twitchio import PartialUser, 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, strfdelta
+from sockets import Channel, register_channel
+
+from . import logger
+from .data import SubathonData, Subathon, RunningSubathon, SubathonContribution, SubathonGoal
+
+
+class TimerChannel(Channel):
+    name = 'Timer'
+
+    def __init__(self, cog: 'SubathonComponent', **kwargs):
+        super().__init__(**kwargs)
+        self.cog = cog
+
+        self.communityid = 1
+
+    async def on_connection(self, websocket, event):
+        await super().on_connection(websocket, event)
+        await self.send_set(
+            **await self.get_args_for(self.communityid),
+            websocket=websocket,
+        )
+
+    async def send_updates(self):
+        await self.send_set(
+            **await self.get_args_for(self.communityid),
+        )
+
+    async def get_args_for(self, channelid):
+        active = await self.cog.get_active_subathon(channelid)
+        if active is not None:
+            ending = utc_now() + timedelta(seconds=await active.get_remaining())
+            return {
+                'end_at': ending,
+                'running': active.running
+            }
+        else:
+            return {
+                'end_at': utc_now,
+                'running': False,
+            }
+
+    async def send_set(self, end_at, running, websocket=None):
+        await self.send_event({
+            'type': "DO",
+            'method': 'setTimer',
+            'args': {
+                'end_at': end_at.isoformat(),
+                'running': running,
+            }
+        }, websocket=websocket)
+
+
+class ActiveSubathon:
+    def __init__(self, subathondata: Subathon, runningdata: RunningSubathon | None):
+        self.subathondata = subathondata
+        self.runningdata = runningdata
+
+    @property
+    def running(self):
+        return self.runningdata is not None
+
+    async def check_cap(self):
+        if not (cap := self.subathondata.timecap):
+            return False
+        else:
+            score = await self.get_score()
+            time_earned = self.get_score_time(score)
+            total_time = self.subathondata.initial_time + time_earned
+            return total_time >= cap
+
+    async def pause(self):
+        if not self.running:
+            raise ValueError("This subathon is not running!")
+        assert self.runningdata is not None
+        new_duration = self.get_duration()
+        await self.subathondata.update(duration=new_duration)
+        await self.runningdata.delete()
+
+    async def resume(self):
+        if self.running:
+            raise ValueError("This subathon is already running!")
+        self.runningdata = await RunningSubathon.create(subathon_id=self.subathondata.subathon_id)
+
+    async def get_score(self) -> float:
+        rows = await SubathonContribution.fetch_where(subathon_id=self.subathondata.subathon_id)
+        return sum(row.score for row in rows)
+
+    def get_score_time(self, score: float) -> int:
+        # Get time contributed by this score
+        return int(score * self.subathondata.score_time)
+
+    def get_duration(self) -> int:
+        # Get the duration of this subathon so far
+        duration = self.subathondata.duration
+        if self.runningdata:
+            now = utc_now()
+            added = int( (now - self.runningdata.last_started).total_seconds() )
+            duration += added
+        return duration
+
+    async def get_remaining(self) -> int:
+        # Get the remaining time
+        score = await self.get_score()
+        time_earned = self.get_score_time(score)
+        total_time = self.subathondata.initial_time + time_earned
+        if cap := self.subathondata.timecap:
+            total_time = min(total_time, cap)
+
+        return total_time - self.get_duration()
+
+    async def add_contribution(self, profileid: int | None, score: float, event_id: int | None) -> SubathonContribution:
+        return await SubathonContribution.create(
+            subathon_id=self.subathondata.subathon_id,
+            profileid=profileid, score=score, event_id=event_id
+        )
+
+    async def get_goals(self) -> list[SubathonGoal]:
+        goals = await SubathonGoal.fetch_where(subathon_id=self.subathondata.subathon_id).order_by('required_score')
+        return goals
+
+
+class SubathonComponent(cmds.Component):
+    def __init__(self, bot: CrocBot):
+        self.bot = bot
+        self.data = bot.dbconn.load_registry(SubathonData())
+        self.channel = TimerChannel(self)
+        register_channel('SubTimer', self.channel)
+
+
+    # ----- API -----
+    async def component_load(self):
+        # TODO: Setup the websocket
+        await self.data.init()
+
+    async def component_teardown(self):
+        pass
+
+    # ----- Methods -----
+    async def get_community(self, twitchid: str, name: str | None) -> Communities:
+        return await self.bot.community_fetch(twitchid=twitchid, name=name)
+
+    async def get_profile(self, twitchid: str, name: str | None) -> UserProfile:
+        return await self.bot.profile_fetch(twitchid=twitchid, name=name)
+
+    async def get_active_subathon(self, communityid: int) -> ActiveSubathon | None:
+        rows = await Subathon.fetch_where(communityid=communityid, ended_at=None)
+        if rows:
+            subathondata = rows[0]
+            running = await RunningSubathon.fetch(subathondata.subathon_id)
+            subba = ActiveSubathon(subathondata, running)
+            return subba
+
+    async def goalcheck(self, active: ActiveSubathon, channel: PartialUser):
+        goals = await active.get_goals()
+        score = await active.get_score()
+        for i, goal in enumerate(goals):
+            if not goal.notified and goal.required_score <= score:
+                # Goal completed, notify channel
+                await channel.send_message(
+                        f"We have reached Goal #{i+1}: {goal.description} !! Thank you everyone for your support <3",
+                        sender=self.bot.bot_id,
+                )
+                await goal.update(notified=True)
+
+    # ----- Event Handlers -----
+    @cmds.Component.listener()
+    async def event_safe_bits_use(self, payload):
+        event_row, detail_row, bits_payload = payload
+        if (active := await self.get_active_subathon(event_row['communityid'])) is not None:
+            # In an active subathon
+            pid = event_row['profileid']
+            uid = event_row['user_id']
+            score = detail_row['bits'] * active.subathondata.bit_score
+            await active.add_contribution(pid, score, event_row['event_id'])
+
+            # Send message to channel
+            sec = active.get_score_time(score)
+            added_min = int(sec // 60)
+            if added_min > 0:
+                added = f"{added_min} minutes"
+            else:
+                added = f"{sec} seconds"
+            name = bits_payload.user.name
+            pl = 's' if bits_payload.bits != 1 else ''
+
+            contrib_str = f"{name} contributed {score} bit{pl}"
+            if not await active.check_cap():
+                contrib_str += f" and added {added} to the timer! Thank you holono1Heart"
+            else:
+                contrib_str += " towards our studython! Thank you holono1Heart"
+
+            await bits_payload.broadcaster.send_message(
+                contrib_str,
+                sender=self.bot.bot_id
+            )
+            await self.channel.send_updates()
+            await self.goalcheck(active, bits_payload.broadcaster)
+            # Check goals
+
+    @cmds.Component.listener()
+    async def event_safe_subscription(self, payload):
+        event_row, detail_row, sub_payload = payload
+        if sub_payload.gift:
+            # Ignore gifted here
+            return
+
+        if (active := await self.get_active_subathon(event_row['communityid'])) is not None:
+            data = active.subathondata
+            # In an active subathon
+            pid = event_row['profileid']
+            tier = int(sub_payload.tier)
+
+            if tier == 1000:
+                mult = data.sub1_score 
+            elif tier == 2000:
+                mult = data.sub2_score 
+            elif tier == 3000:
+                mult = data.sub3_score
+            else:
+                raise ValueError(f"Unknown sub tier {sub_payload.tier}")
+
+            score = mult * 1
+
+            await active.add_contribution(pid, score, event_row['event_id'])
+
+            # Send message to channel
+            added_min = int(active.get_score_time(score) // 60)
+            added = f"{added_min} minutes"
+            name = sub_payload.user.name
+            pl = 's' if score > 1 else ''
+
+            contrib_str = f"{name} contributed {score} sub{pl}"
+            if not await active.check_cap():
+                contrib_str += f" and added {added} to the timer! Thank you holono1Heart"
+            else:
+                contrib_str += " towards our studython! Thank you holono1Heart"
+
+            await sub_payload.broadcaster.send_message(
+                contrib_str,
+                sender=self.bot.bot_id
+            )
+            await self.channel.send_updates()
+            # Check goals
+            await self.goalcheck(active, sub_payload.broadcaster)
+
+    @cmds.Component.listener()
+    async def event_safe_subscription_gift(self, payload):
+        event_row, detail_row, gift_payload = payload
+
+        if (active := await self.get_active_subathon(event_row['communityid'])) is not None:
+            data = active.subathondata
+            # In an active subathon
+            pid = event_row['profileid']
+
+            tier = int(gift_payload.tier)
+            if tier == 1000:
+                mult = data.sub1_score 
+            elif tier == 2000:
+                mult = data.sub2_score 
+            elif tier == 3000:
+                mult = data.sub3_score
+            else:
+                raise ValueError(f"Unknown sub tier {gift_payload.tier}")
+
+            score = mult * gift_payload.total
+
+            await active.add_contribution(pid, score, event_row['event_id'])
+
+            # Send message to channel
+            added_min = int(active.get_score_time(score) // 60)
+            added = f"{added_min} minutes"
+            name = gift_payload.user.name if gift_payload.user else 'Anonymous'
+
+            contrib_str = f"{name} contributed {score} subs"
+            if not await active.check_cap():
+                contrib_str += f" and added {added} to the timer! Thank you holono1Heart"
+            else:
+                contrib_str += " towards our studython! Thank you holono1Heart"
+
+            await gift_payload.broadcaster.send_message(
+                contrib_str,
+                sender=self.bot.bot_id
+            )
+            await self.channel.send_updates()
+            # Check goals
+            await self.goalcheck(active, gift_payload.broadcaster)
+
+    @cmds.Component.listener()
+    async def event_safe_subscription_message(self, payload):
+        event_row, detail_row, sub_payload = payload
+
+        if (active := await self.get_active_subathon(event_row['communityid'])) is not None:
+            data = active.subathondata
+            # In an active subathon
+            pid = event_row['profileid']
+            tier = int(sub_payload.tier)
+
+            if tier == 1000:
+                mult = data.sub1_score 
+            elif tier == 2000:
+                mult = data.sub2_score 
+            elif tier == 3000:
+                mult = data.sub3_score
+            else:
+                raise ValueError(f"Unknown sub tier {sub_payload.tier}")
+
+            score = mult * 1
+
+            await active.add_contribution(pid, score, event_row['event_id'])
+
+            # Send message to channel
+            added_min = int(active.get_score_time(score) // 60)
+            added = f"{added_min} minutes"
+            name = sub_payload.user.name
+            pl = 's' if score > 1 else ''
+
+            contrib_str = f"{name} contributed {score} sub{pl}"
+            if not await active.check_cap():
+                contrib_str += f" and added {added} to the timer! Thank you holono1Heart"
+            else:
+                contrib_str += " towards our studython! Thank you holono1Heart"
+
+            await sub_payload.broadcaster.send_message(
+                contrib_str,
+                sender=self.bot.bot_id
+            )
+            await self.channel.send_updates()
+            # Check goals
+            await self.goalcheck(active, sub_payload.broadcaster)
+
+    # end stream => Automatically pause the timer
+    @cmds.Component.listener()
+    async def event_stream_offline(self, payload: twitchio.StreamOffline):
+        community = await self.bot.community_fetch(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
+        cid = community.communityid
+        if (active := await self.get_active_subathon(cid)) is not None:
+            if active.running:
+                await active.pause()
+                await payload.broadcaster.send_message(
+                        "Paused the subathon timer because the stream went offline!",
+                        sender=self.bot.bot_id
+                    )
+                await self.channel.send_updates()
+
+    # ----- Commands -----
+
+    @cmds.group(name='subathon', aliases=['studython'], invoke_fallback=True)
+    async def group_subathon(self, ctx: cmds.Context):
+        # TODO: Status
+        community = await self.bot.community_fetch(twitchid=ctx.broadcaster.id, name=ctx.broadcaster.name)
+        cid = community.communityid
+        if (active := await self.get_active_subathon(cid)) is not None:
+            score = await active.get_score()
+            goals = await active.get_goals()
+            total_goals = len(goals)
+            donegoals = len([goal for goal in goals if score >= goal.required_score])
+            goalstr = f"{donegoals}/{total_goals} goals achieved"
+
+            secs = await active.get_remaining()
+            remaining = strfdelta(timedelta(seconds=secs))
+
+            secs = active.get_duration()
+            duration = strfdelta(timedelta(seconds=secs))
+
+            text = (
+                f"Subathon running for {duration}! {score} (equivalent) subscriptions recieved, {goalstr}, and {remaining} left on the timer"
+            )
+            await ctx.reply(text)
+        else:
+            await ctx.reply("No active subathon running!")
+
+    # subathon start 
+    @group_subathon.command(name='setup')
+    async def cmd_setup(self, ctx: cmds.Context, initial_hours: float, sub1: float, sub2: float, sub3: float, bit: float, timescore: int, timecap: Optional[int]=None):
+        if ctx.broadcaster:
+            community = await self.bot.community_fetch(twitchid=ctx.broadcaster.id, name=ctx.broadcaster.name)
+            cid = community.communityid
+            if (active := await self.get_active_subathon(cid)) is not None:
+                await ctx.reply("There is already an active subathon running! Use !subathon stop to stop it!")
+                return
+            initial_time = initial_hours * 60 * 60
+            
+            active = await Subathon.create(
+                communityid=cid,
+                initial_time=initial_time,
+                sub1_score=sub1,
+                sub2_score=sub2,
+                sub3_score=sub3,
+                bit_score=bit,
+                score_time=timescore,
+                timecap=timecap
+            )
+            await ctx.reply("Setup a new subathon! Use !subathon resume to get the timer running.")
+            await self.channel.send_updates()
+
+    # subathon stop
+    @group_subathon.command(name='stop')
+    async def cmd_stop(self, ctx: cmds.Context):
+        if ctx.broadcaster:
+            community = await self.bot.community_fetch(twitchid=ctx.broadcaster.id, name=ctx.broadcaster.name)
+            cid = community.communityid
+            if (active := await self.get_active_subathon(cid)) is not None:
+                if active.running:
+                    await active.pause()
+                    await self.channel.send_updates()
+                await active.subathondata.update(ended_at=utc_now())
+                total = await active.get_score()
+                dursecs = active.get_duration()
+                dur = strfdelta(timedelta(seconds=dursecs))
+
+                await ctx.reply(
+                    f"Subathon complete after {dur} with a total of {total} subs, congratulations!"
+                )
+            else:
+                await ctx.reply("No active subathon to stop.")
+
+    # subathon pause 
+    @group_subathon.command(name='pause')
+    async def cmd_pause(self, ctx: cmds.Context):
+        if ctx.broadcaster or ctx.author.moderator:
+            community = await self.bot.community_fetch(twitchid=ctx.broadcaster.id, name=ctx.broadcaster.name)
+            cid = community.communityid
+            if (active := await self.get_active_subathon(cid)) is not None:
+                if active.running:
+                    await active.pause()
+                    await ctx.reply("Subathon timer paused!")
+                    await self.channel.send_updates()
+                else:
+                    await ctx.reply("Subathon timer already paused!")
+            else:
+                await ctx.reply("No active subathon to pause")
+
+    # subathon resume 
+    @group_subathon.command(name='resume')
+    async def cmd_resume(self, ctx: cmds.Context):
+        if ctx.broadcaster or ctx.author.moderator:
+            community = await self.bot.community_fetch(twitchid=ctx.broadcaster.id, name=ctx.broadcaster.name)
+            cid = community.communityid
+            if (active := await self.get_active_subathon(cid)) is not None:
+                if not active.running:
+                    await active.resume()
+                    await ctx.reply("Subathon timer resumed!")
+                    await self.channel.send_updates()
+                else:
+                    await ctx.reply("Subathon timer already running!")
+            else:
+                await ctx.reply("No active subathon to resume")
+
+    @cmds.group(name='goals', invoke_fallback=True)
+    async def group_goals(self, ctx: cmds.Context):
+        # List the goals
+        community = await self.bot.community_fetch(twitchid=ctx.broadcaster.id, name=ctx.broadcaster.name)
+        cid = community.communityid
+        if (active := await self.get_active_subathon(cid)) is not None:
+            goals = await active.get_goals()
+            goalstrs = []
+            for i, goal in enumerate(goals, start=1):
+                line = f"{goal.required_score} subs: {goal.description}"
+                goalstrs.append(line)
+
+            if goals:
+                text = ', '.join(goalstrs)
+                await ctx.reply(f"Subathon Goals! -- {text}")
+            else:
+                await ctx.reply("No goals have been configured!")
+        else:
+            await ctx.reply("No active subathon running!")
+
+    @group_goals.command(name='add')
+    async def cmd_add(self, ctx: cmds.Context, required: int, *, description: str):
+        if ctx.broadcaster or ctx.author.moderator:
+            community = await self.bot.community_fetch(twitchid=ctx.broadcaster.id, name=ctx.broadcaster.name)
+            cid = community.communityid
+            if (active := await self.get_active_subathon(cid)) is not None:
+                await SubathonGoal.create(
+                    subathon_id=active.subathondata.subathon_id,
+                    required_score=required,
+                    description=description,
+                )
+                await ctx.reply("Goal added!")
+            else:
+                await ctx.reply("No active subathon to goal!")
+

src/modules/subathons/data.py

@@ -0,0 +1,58 @@
+from data import Registry, RowModel, Table
+from data.columns import String, Timestamp, Integer, Bool
+
+# User contributed {} subs and added {}:{} to the timer! Thank you :love:
+# We have reached goal #5: Karaoke !! Thank you everyone for your support <3
+
+class Subathon(RowModel):
+    _tablename_ = 'subathons'
+    _cache_ = {}
+
+    subathon_id = Integer(primary=True)
+    communityid = Integer()
+    started_at = Timestamp()
+    initial_time = Integer()  # Initial number of seconds
+    sub1_score = Integer()
+    sub2_score = Integer()
+    sub3_score = Integer()
+    bit_score = Integer()
+    score_time = Integer()  # Conversion factor score to seconds
+    duration = Integer()
+    ended_at = Timestamp()
+    timecap = Integer()  # Maximum subathon duration in seconds
+
+class RunningSubathon(RowModel):
+    _tablename_ = 'running_subathons'
+    _cache_ = {}
+
+    subathon_id = Integer(primary=True)
+    last_started = Timestamp()
+
+class SubathonContribution(RowModel):
+    _tablename_ = 'subathon_contributions'
+    _cache_ = {}
+
+    contribution_id = Integer(primary=True)
+    subathon_id = Integer()
+    profileid = Integer()
+    score = Integer()
+    event_id = Integer()
+    # TODO: Should add a created timestamp here, since not all contributions have event ids
+
+class SubathonGoal(RowModel):
+    _tablename_ = 'subathon_goals'
+    _cache_ = {}
+
+    goal_id = Integer(primary=True)
+    subathon_id = Integer()
+    required_score = Integer()
+    description = String()
+    notified = Bool()
+    created_at = Timestamp()
+
+
+class SubathonData(Registry):
+    subathons = Subathon.table
+    running_subathons = RunningSubathon.table
+    subathon_contributions = SubathonContribution.table
+    subathon_goals = SubathonGoal.table

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)))
 
@@ -118,9 +131,9 @@ class TrackerComponent(cmds.Component):
     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)
+            community = await self.bot.community_fetch(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
             cid = community.communityid
-            profile = await UserProfile.fetch_or_create(
+            profile = await self.bot.profile_fetch(
                 twitchid=payload.user.id, name=payload.user.name,
             )
             pid = profile.profileid
@@ -147,9 +160,9 @@ class TrackerComponent(cmds.Component):
     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)
+            community = await self.bot.community_fetch(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
             cid = community.communityid
-            profile = await UserProfile.fetch_or_create(
+            profile = await self.bot.profile_fetch(
                 twitchid=payload.user.id, name=payload.user.name,
             )
             pid = profile.profileid
@@ -175,9 +188,9 @@ class TrackerComponent(cmds.Component):
     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)
+            community = await self.bot.community_fetch(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
             cid = community.communityid
-            profile = await UserProfile.fetch_or_create(
+            profile = await self.bot.profile_fetch(
                 twitchid=payload.user.id, name=payload.user.name,
             )
             pid = profile.profileid
@@ -201,9 +214,9 @@ class TrackerComponent(cmds.Component):
     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)
+            community = await self.bot.community_fetch(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
             cid = community.communityid
-            profile = await UserProfile.fetch_or_create(
+            profile = await self.bot.profile_fetch(
                 twitchid=payload.user.id, name=payload.user.name,
             )
             pid = profile.profileid
@@ -222,14 +235,15 @@ class TrackerComponent(cmds.Component):
                 message=payload.text,
                 powerup_type=payload.power_up.type if payload.power_up else None
             )
+            self.bot.safe_dispatch('bits_use', payload=(event_row, detail_row, payload))
 
     @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)
+            community = await self.bot.community_fetch(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
             cid = community.communityid
-            profile = await UserProfile.fetch_or_create(
+            profile = await self.bot.profile_fetch(
                 twitchid=payload.user.id, name=payload.user.name,
             )
             pid = profile.profileid
@@ -243,18 +257,19 @@ class TrackerComponent(cmds.Component):
             )
             detail_row = await self.data.subscribe_events.insert(
                 event_id=event_row['event_id'],
-                tier=payload.tier,
+                tier=int(payload.tier),
                 gifted=payload.gift,
             )
+            self.bot.safe_dispatch('subscription', payload=(event_row, detail_row, payload))
 
     @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)
+            community = await self.bot.community_fetch(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
             cid = community.communityid
             if payload.user is not None:
-                profile = await UserProfile.fetch_or_create(
+                profile = await self.bot.profile_fetch(
                     twitchid=payload.user.id, name=payload.user.name,
                 )
                 pid = profile.profileid
@@ -270,17 +285,18 @@ class TrackerComponent(cmds.Component):
             )
             detail_row = await self.data.gift_events.insert(
                 event_id=event_row['event_id'],
-                tier=payload.tier,
+                tier=int(payload.tier),
                 gifted_count=payload.total,
             )
+            self.bot.safe_dispatch('subscription_gift', payload=(event_row, detail_row, payload))
 
     @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)
+            community = await self.bot.community_fetch(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
             cid = community.communityid
-            profile = await UserProfile.fetch_or_create(
+            profile = await self.bot.profile_fetch(
                 twitchid=payload.user.id, name=payload.user.name,
             )
             pid = profile.profileid
@@ -294,12 +310,123 @@ class TrackerComponent(cmds.Component):
             )
             detail_row = await self.data.subscribe_message_events.insert(
                 event_id=event_row['event_id'],
-                tier=payload.tier,
+                tier=int(payload.tier),
                 duration_months=payload.months,
                 cumulative_months=payload.cumulative_months,
                 streak_months=payload.streak_months,
                 message=payload.text,
             )
+            self.bot.safe_dispatch('subscription_message', payload=(event_row, detail_row, payload))
+
+    @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 self.bot.community_fetch(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_stream_offline(self, payload: twitchio.StreamOffline):
+        tracked = await TrackingChannel.fetch(payload.broadcaster.id)
+        if tracked and tracked.joined:
+            community = await self.bot.community_fetch(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
+            cid = community.communityid
+
+            event_row = await self.data.events.insert(
+                event_type='stream_offline',
+                communityid=cid,
+                channel_id=payload.broadcaster.id,
+            )
+            detail_row = await self.data.stream_offline_events.insert(
+                event_id=event_row['event_id'],
+            )
+
+    @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 self.bot.community_fetch(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 self.bot.community_fetch(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 self.bot.community_fetch(twitchid=payload.broadcaster.id, name=payload.broadcaster.name)
+            cid = community.communityid
+            profile = await self.bot.profile_fetch(
+                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')
@@ -319,6 +446,7 @@ class TrackerComponent(cmds.Component):
                     Scopes.bits_read,
                     Scopes.channel_read_polls,
                     Scopes.channel_read_vips,
+                    Scopes.moderator_read_followers,
                     *scopes
                 })
             )
@@ -342,3 +470,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')