Background services

# Start journey checker loop
from app.vbb.service.checker import check_journeys
asyncio.create_task(check_journeys())

# Start APScheduler
from app.utils import add_scheduler_jobs
await add_scheduler_jobs()

await app.dp.start_polling(app.bot)

async def check_journeys():
    while True:
        async with sessionmanager() as session:
            users: list[User] = await session.get_registered_users()

            for user in users:
                logging.debug("checking user: %s" % user)
                if user.is_notified:
                    continue

                if not is_time_matching(user.check_time):
                    continue

                # Start journey dialog for user
                await start_dialog(states.JourneysSG.MAIN, user.id, now=False)

                # Mark user as notified
                await session.update_notified(user, True)

        await asyncio.sleep(15)

def is_time_matching(check_time: time):
    now = datetime.now().time()
    return all([now.hour == check_time.hour,
                now.minute == check_time.minute])

async def start_dialog(state: State, user_id, **kwargs):
    from aiogram_dialog.manager.bg_manager import BgManager
    from aiogram.types import Chat, User as TgUser
    from aiogram_dialog import StartMode, ShowMode
    from app import bot, dp
    
    user = TgUser(id=user_id, is_bot=False, first_name="test")
    chat = Chat(id=user_id, type="private")
    manager = BgManager(user=user, chat=chat, bot=bot, router=dp, intent_id=None, stack_id="")
    await manager.start(state, mode=StartMode.RESET_STACK, show_mode=ShowMode.SEND, data=kwargs)

async def paging_getter(dialog_manager: DialogManager, **_kwargs):
    user_id = dialog_manager.event.from_user.id
    f: FMT = dialog_manager.middleware_data.get("f")
    
    if f is None:
        async with sessionmanager() as session:
            user = await session.get_user(user_id)
            home_address = await session.get_address(user.home_address_id)
            destination_address = await session.get_address(user.default_destination_address_id)
    else:
        user = await f.db.get_user(user_id)

    start_data = dialog_manager.start_data
    now = start_data.get("now", True) if start_data else True

    # Fetch journeys with arrival time
    data = await get_journeys(user, session, now=now)

    journey = data[current_page]

    return {
        "pages": len(data),
        "current_page": current_page + 1,
        "journey_information": await message_builder.build_journey_text(journey)
    }

async def add_scheduler_jobs():
    from app import scheduler
    from app.vbb.service import remove_is_notified

    from apscheduler.triggers.cron import CronTrigger

    scheduler.add_job(remove_is_notified, CronTrigger(hour=0, minute=0))

    scheduler.start()

from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()

async def remove_is_notified():
    async with sessionmanager() as session:
        users: list[User] = await session.get_registered_users()
        for user in users:
            await session.update_notified(user, False)

CronTrigger(hour=0, minute=0)

06:30 - User's check_time reached, is_notified = False
      → Send notification
      → Set is_notified = True
06:30 - check_time still matches, but is_notified = True
      → Skip (no duplicate notification)
00:00 - Daily reset job runs
      → Set is_notified = False for all users
06:30 - Next day: User receives notification again

async with sessionmanager() as session:
    try:
        users = await session.get_registered_users()
    except Exception as e:
        logging.error(f"Database error: {e}")
        # Session rolled back automatically

try:
    journeys = await get_journeys(user, session, now=False)
except aiohttp.ClientError as e:
    logging.error(f"VBB API error for user {user.id}: {e}")
    # User not marked as notified, will retry in 15 seconds

# Instead of individual updates
await session.update_notified(user, True)

# Batch update all notified users
notified_user_ids = [user.id for user in notified_users]
await session.execute(
    update(User)
    .where(User.id.in_(notified_user_ids))
    .values(is_notified=True)
)

CREATE INDEX idx_users_check_time ON users(check_time) WHERE is_notified = false;

for i, user in enumerate(users):
    await start_dialog(states.JourneysSG.MAIN, user.id, now=False)
    if i % 10 == 0:
        await asyncio.sleep(1)  # Pause every 10 users

logging.debug("checking user: %s" % user)

@app.router.get("/health")
async def health_check():
    return {
        "status": "ok",
        "journey_checker_running": check_journeys_task.done() == False,
        "scheduler_running": scheduler.running
    }

import pytest
from datetime import time
from app.vbb.service.checker import is_time_matching

def test_time_matching():
    # Mock current time to 08:30
    with mock_time("08:30:00"):
        assert is_time_matching(time(8, 30)) == True
        assert is_time_matching(time(8, 31)) == False
        assert is_time_matching(time(9, 30)) == False

async def monitor_journeys():
    while True:
        users = await session.get_users_with_saved_journeys()
        for user in users:
            saved_journey = await session.get_saved_journey(user.id)
            current_journey = await refresh_journey(saved_journey.refresh_token)
            if journey_changed(saved_journey, current_journey):
                await notify_user(user.id, "Your journey has changed!")
        await asyncio.sleep(60)

class UserNotificationTime(Base):
    user_id = Column(BigInteger, ForeignKey('users.id'))
    check_time = Column(Time)
    day_of_week = Column(Integer)  # 0=Monday, 6=Sunday
    is_notified = Column(Boolean, default=False)

class UserPreferences(Base):
    user_id = Column(BigInteger, ForeignKey('users.id'))
    notify_on_delay = Column(Boolean, default=True)
    notify_on_cancellation = Column(Boolean, default=True)
    min_delay_minutes = Column(Integer, default=5)