basic campaign support

core/data/alembic/versions/1b78db5898f4_add_campaigns.py

@@ -0,0 +1,63 @@
+"""add_campaigns
+
+Revision ID: 1b78db5898f4
+Revises: ada3e2d02483
+Create Date: 2026-07-05 16:46:01.929705
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+# revision identifiers, used by Alembic.
+revision = '1b78db5898f4'
+down_revision = 'ada3e2d02483'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('campaign',
+    sa.Column('id', sa.INTEGER(), nullable=False),
+    sa.Column('name', sa.TEXT(length=127), nullable=False),
+    sa.Column('announce_date', sa.TIMESTAMP(), nullable=True),
+    sa.Column('start_date', sa.TIMESTAMP(), nullable=True),
+    sa.Column('end_date', sa.TIMESTAMP(), nullable=True),
+    sa.Column('distrib_start_date', sa.TIMESTAMP(), nullable=True),
+    sa.Column('distrib_end_date', sa.TIMESTAMP(), nullable=True),
+    sa.Column('info0', sa.INTEGER(), nullable=True),
+    sa.Column('info1', sa.INTEGER(), nullable=True),
+    sa.Column('info2', sa.INTEGER(), nullable=True),
+    sa.Column('info3', sa.INTEGER(), nullable=True),
+    sa.PrimaryKeyConstraint('id'),
+    mysql_charset='utf8mb4'
+    )
+    op.create_table('campaign_game',
+    sa.Column('campaign_id', sa.INTEGER(), nullable=False),
+    sa.Column('game_id', sa.TEXT(length=5), nullable=False),
+    sa.ForeignKeyConstraint(['campaign_id'], ['campaign.id'], onupdate='cascade', ondelete='cascade'),
+    sa.UniqueConstraint('campaign_id', 'game_id', name='campaign_game_uk'),
+    mysql_charset='utf8mb4'
+    )
+    op.create_table('campaign_progress',
+    sa.Column('id', sa.BIGINT(), autoincrement=True, nullable=False),
+    sa.Column('user_id', sa.Integer(), nullable=False),
+    sa.Column('campaign_id', sa.INTEGER(), nullable=False),
+    sa.Column('is_participating', sa.INTEGER(), server_default='0', nullable=False),
+    sa.Column('progress', sa.INTEGER(), server_default='0', nullable=False),
+    sa.ForeignKeyConstraint(['campaign_id'], ['campaign.id'], onupdate='cascade', ondelete='cascade'),
+    sa.ForeignKeyConstraint(['user_id'], ['aime_user.id'], onupdate='cascade', ondelete='cascade'),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('campaign_id', 'user_id', name='campaign_progress_uk'),
+    mysql_charset='utf8mb4'
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('campaign_progress')
+    op.drop_table('campaign_game')
+    op.drop_table('campaign')
+    # ### end Alembic commands ###

core/data/schema/base.py

@@ -3,15 +3,17 @@ import json
 import logging
 from random import randrange
 from typing import Any, Dict, List, Optional
+from datetime import datetime
 
-from sqlalchemy import Column, MetaData, Table
+from sqlalchemy import Column, MetaData, Table, UniqueConstraint
 from sqlalchemy.engine import Row
 from sqlalchemy.engine.cursor import CursorResult
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.schema import ForeignKey
-from sqlalchemy.sql import func, text
+from sqlalchemy.sql import func, text, and_, or_
+from sqlalchemy.dialects.mysql import insert
 from sqlalchemy.types import INTEGER, JSON, TEXT, TIMESTAMP, Integer, String
 
 from core.config import CoreConfig
@@ -37,6 +39,33 @@ event_log: Table = Table(
     mysql_charset="utf8mb4",
 )
 
+campaign: Table = Table(
+    "campaign",
+    metadata,
+    Column("id", INTEGER, primary_key=True, nullable=False),
+    Column("name", TEXT(127), nullable=False),
+    Column("announce_date", TIMESTAMP), # None = start announcing now
+    Column("start_date", TIMESTAMP), # None = start now
+    Column("end_date", TIMESTAMP), # None = never end
+    Column("distrib_start_date", TIMESTAMP), # None = start distributing now
+    Column("distrib_end_date", TIMESTAMP), # None = never stop distributing
+    # These are for legacy campaign requests prior to ADB v3.03.0.
+    Column("info0", INTEGER),
+    Column("info1", INTEGER),
+    Column("info2", INTEGER),
+    Column("info3", INTEGER),
+    mysql_charset="utf8mb4",
+)
+
+# Unfortunatly without a machine table change there's no real way to gate by game version...
+campaign_game: Table = Table(
+    "campaign_game",
+    metadata,
+    Column("campaign_id", INTEGER, ForeignKey("campaign.id", ondelete="cascade", onupdate="cascade"), nullable=False),
+    Column("game_id", TEXT(5), nullable=False),
+    UniqueConstraint("campaign_id", "game_id", name="campaign_game_uk"),
+    mysql_charset="utf8mb4",
+)
 
 class BaseData:
     def __init__(self, cfg: CoreConfig, conn: "sessionmaker[AsyncSession]") -> None:
@@ -119,6 +148,127 @@ class BaseData:
             return None
         return result.fetchall()
 
+    async def create_campaign(self, campaign_id: int, name: str, announce_date: Optional[datetime] = None, 
+                              start_date: Optional[datetime] = None, end_date: Optional[datetime] = None,
+                              distribute_start_date: Optional[datetime] = None, distribute_end_date: Optional[datetime] = None) -> Optional[int]:
+        sql = insert(campaign).values(
+            id = campaign_id,
+            name = name,
+            announce_date = announce_date,
+            start_date = start_date,
+            end_date = end_date,
+            distribute_start_date = distribute_start_date,
+            distribute_end_date = distribute_end_date,
+        )
+
+        conflict = sql.on_duplicate_key_update(
+            announce_date = announce_date,
+            start_date = start_date,
+            end_date = end_date,
+            distribute_start_date = distribute_start_date,
+            distribute_end_date = distribute_end_date,
+        )
+
+        result = await self.execute(conflict)
+        if result is None:
+            self.logger.error(f"Failed to create campaign {name} ID {campaign_id}!")
+            return None
+        return result.lastrowid
+
+    async def create_campaign_old(self, campaign_id: int, info0: int = 0, info1: int = 0, info2: int = 0, info3: int = 0) -> Optional[int]:
+        sql = insert(campaign).values(
+            id = campaign_id,
+            info0 = info0,
+            info1 = info1,
+            info2 = info2,
+            info3 = info3
+        )
+
+        conflict = sql.on_duplicate_key_update(
+            info0 = info0,
+            info1 = info1,
+            info2 = info2,
+            info3 = info3
+        )
+
+        result = await self.execute(conflict)
+        if result is None:
+            self.logger.error(f"Failed to create legacy campaign with ID {campaign_id}!")
+            return None
+        return result.lastrowid
+
+    async def add_game_to_campaign(self, campaign_id: int, game_id: str) -> bool:
+        sql = insert(campaign_game).values(
+            id = campaign_id,
+            game_id = game_id
+        )
+
+        conflict = sql.on_duplicate_key_do_nothing()
+
+        result = await self.execute(conflict)
+        if result is None:
+            self.logger.error(f"Failed to add game {game_id} to campaign {campaign_id}!")
+            return False
+        return True
+
+    async def get_campaigns_by_game(self, game_id: str) -> Optional[List[Row]]:
+        result = await self.execute(campaign
+            .join(campaign_game, campaign.c.id == campaign_game.c.campaign_id)
+            .select(campaign_game.c.game_id == game_id))
+        
+        if result is not None:
+            return result.fetchall()
+
+    async def get_active_campaigns_by_game(self, game_id: str) -> Optional[List[Row]]:
+        result = await self.execute(campaign
+            .join(campaign_game, campaign.c.id == campaign_game.c.campaign_id)
+            .select(and_(campaign_game.c.game_id == game_id, and_(
+                # TODO: Maybe add a RANK statement to prioritize campaigns that have set end dates?
+                or_(campaign.c.announce_date >= datetime.now(), campaign.c.announce_date == None),
+                or_(campaign.c.distrib_end_date < datetime.now(), campaign.c.distrib_end_date == None),
+            ))))
+        
+        if result is not None:
+            return result.fetchall()
+
+    async def get_campaigns(self) -> Optional[List[Row]]:
+        result = await self.execute(campaign.select())
+
+        if result is not None:
+            return result.fetchall()
+
+    async def get_active_campaigns(self) -> Optional[List[Row]]:
+        result = await self.execute(campaign
+            .select(and_(
+                or_(campaign.c.announce_date >= datetime.now(), campaign.c.announce_date == None),
+                or_(campaign.c.distrib_end_date < datetime.now(), campaign.c.distrib_end_date == None),
+            )))
+        
+        if result is not None:
+            return result.fetchall()
+
+    async def get_campaign_by_id(self, campaign_id: int) -> Optional[Row]:
+        result = await self.execute(campaign.select(campaign.c.id == campaign_id))
+
+        if result is not None:
+            return result.fetchone()
+
+    async def get_games_in_campaign(self, campaign_id: int) -> Optional[List[Row]]:
+        result = await self.execute(campaign_game.select(campaign_game.c.campaign_id == campaign_id))
+
+        if result is not None:
+            return result.fetchall()
+
+    async def is_game_in_campaign(self, game_id: str, campaign_id: int) -> bool:
+        result = await self.execute(campaign_game.select(and_(
+            campaign_game.c.campaign_id == campaign_id,
+            campaign_game.c.game_id == game_id
+        )))
+        
+        if result is not None:
+            return result.fetchone() is not None
+        return False
+
     def fix_bools(self, data: Dict) -> Dict:
         for k, v in data.items():
             if k == "userName" or k == "teamName":

core/data/schema/user.py

@@ -1,13 +1,14 @@
 from typing import List, Optional
 
 import bcrypt
-from sqlalchemy import Column, Table
+from sqlalchemy import Column, Table, ForeignKey, UniqueConstraint
 from sqlalchemy.dialects.mysql import insert
 from sqlalchemy.engine import Row
-from sqlalchemy.sql import func, select
-from sqlalchemy.types import TIMESTAMP, Integer, String
+from sqlalchemy.sql import func, select, and_, or_
+from sqlalchemy.types import TIMESTAMP, Integer, String, BIGINT, INTEGER
+from datetime import datetime
 
-from core.data.schema.base import BaseData, metadata
+from core.data.schema.base import BaseData, metadata, campaign, campaign_game
 
 aime_user: Table = Table(
     "aime_user",
@@ -23,6 +24,18 @@ aime_user: Table = Table(
     mysql_charset="utf8mb4",
 )
 
+campaign_progress: Table = Table(
+    "campaign_progress",
+    metadata,
+    Column("id", BIGINT, nullable=False, primary_key=True, autoincrement=True),
+    Column("user_id", Integer, ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False),
+    Column("campaign_id", INTEGER, ForeignKey("campaign.id", ondelete="cascade", onupdate="cascade"), nullable=False),
+    Column("is_participating", INTEGER, nullable=False, server_default="0"),
+    Column("progress", INTEGER, nullable=False, server_default="0"),
+    UniqueConstraint("campaign_id", "user_id", name="campaign_progress_uk"),
+    mysql_charset="utf8mb4",
+)
+
 class UserData(BaseData):
     async def create_user(
         self,
@@ -136,3 +149,81 @@ class UserData(BaseData):
 
         result = await self.execute(sql)
         return result is not None
+
+    async def get_user_campaign_progress_by_campaign(self, user_id: int, campaign_id: int) -> Optional[Row]:
+        result = await self.execute(campaign_progress
+            .join(campaign, campaign_progress.c.campaign_id == campaign.c.id)
+            .select(and_(campaign_progress.c.user_id == user_id, campaign_progress.c.campaign_id == campaign_id)))
+        
+        if result is not None:
+            return result.fetchone()
+
+    async def get_user_all_campaign_progress(self, user_id: int) -> Optional[List[Row]]:
+        result = await self.execute(campaign_progress
+            .join(campaign, campaign_progress.c.campaign_id == campaign.c.id)
+            .select(campaign_progress.c.user_id == user_id))
+        
+        if result is not None:
+            return result.fetchall()
+
+    async def get_user_active_campaign_progress(self, user_id: int) -> Optional[List[Row]]:
+        # distrib_end_date instead of end_date to give the games time to distribute the rewards
+        result = await self.execute(campaign_progress
+            .join(campaign, campaign_progress.c.campaign_id == campaign.c.id)
+            .select(and_(campaign_progress.c.user_id == user_id, and_(
+                or_(campaign.c.start_date >= datetime.now(), campaign.c.start_date == None),
+                or_(campaign.c.distrib_end_date < datetime.now(), campaign.c.distrib_end_date == None),
+            ))))
+        
+        if result is not None:
+            return result.fetchall()
+
+    async def get_user_all_campaign_progress_by_game(self, user_id: int, game_id: str) -> Optional[List[Row]]:
+        result = await self.execute(campaign_progress
+            .join(campaign, campaign_progress.c.campaign_id == campaign.c.id)
+            .join(campaign_game, campaign_progress.c.campaign_id == campaign_game.c.campaign_id)
+            .select(and_(campaign_progress.c.user_id == user_id, campaign_game.c.game_id == game_id)))
+        
+        if result is not None:
+            return result.fetchall()
+
+    async def get_user_active_campaign_progress_by_game(self, user_id: int, game_id: str) -> Optional[List[Row]]:
+        result = await self.execute(campaign_progress
+            .join(campaign, campaign_progress.c.campaign_id == campaign.c.id)
+            .join(campaign_game, campaign_progress.c.campaign_id == campaign_game.c.campaign_id)
+            .select(and_(and_(campaign_progress.c.user_id == user_id, campaign_game.c.game_id == game_id), and_(
+                or_(campaign.c.start_date >= datetime.now(), campaign.c.start_date == None),
+                or_(campaign.c.distrib_end_date < datetime.now(), campaign.c.distrib_end_date == None),
+            ))))
+        
+        if result is not None:
+            return result.fetchall()
+
+    async def set_user_campaign_is_participating(self, user_id: int, campaign_id: int, is_participating: bool) -> Optional[bool]:
+        sql = insert(campaign_progress).values(
+            user_id = user_id,
+            campaign_id = campaign_id,
+            is_participating = is_participating
+        )
+
+        conflict = sql.on_duplicate_key_update(is_participating = is_participating)
+        result = await self.execute(conflict)
+        if result is None:
+            self.logger.error(f"Failed to update campaign participation status for user {user_id} on campaign {campaign_id}!")
+            return None
+        return result.lastrowid
+
+    async def set_user_campaign_progress(self, user_id: int, campaign_id: int, progress: int) -> Optional[bool]:
+        sql = insert(campaign_progress).values(
+            user_id = user_id,
+            campaign_id = campaign_id,
+            is_participating = 1,
+            progress = progress
+        )
+
+        conflict = sql.on_duplicate_key_update(progress = progress)
+        result = await self.execute(conflict)
+        if result is None:
+            self.logger.error(f"Failed to update campaign progress for user {user_id} on campaign {campaign_id} to {progress}!")
+            return None
+        return result.lastrowid