85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import logging
|
||
|
|
from datetime import UTC, datetime, timedelta
|
||
|
|
from typing import Callable
|
||
|
|
|
||
|
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
||
|
|
from sqlalchemy import delete, select
|
||
|
|
from sqlalchemy.exc import SQLAlchemyError
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||
|
|
|
||
|
|
from app.config import RuntimeSettings, Settings
|
||
|
|
from app.models.token_binding import TokenBinding
|
||
|
|
from app.services.binding_service import BindingService
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
class ArchiveService:
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
settings: Settings,
|
||
|
|
session_factory: async_sessionmaker[AsyncSession],
|
||
|
|
binding_service: BindingService,
|
||
|
|
runtime_settings_getter: Callable[[], RuntimeSettings],
|
||
|
|
) -> None:
|
||
|
|
self.settings = settings
|
||
|
|
self.session_factory = session_factory
|
||
|
|
self.binding_service = binding_service
|
||
|
|
self.runtime_settings_getter = runtime_settings_getter
|
||
|
|
self.scheduler = AsyncIOScheduler(timezone="UTC")
|
||
|
|
|
||
|
|
async def start(self) -> None:
|
||
|
|
if self.scheduler.running:
|
||
|
|
return
|
||
|
|
self.scheduler.add_job(
|
||
|
|
self.archive_inactive_bindings,
|
||
|
|
trigger="interval",
|
||
|
|
minutes=self.settings.archive_job_interval_minutes,
|
||
|
|
id="archive-inactive-bindings",
|
||
|
|
replace_existing=True,
|
||
|
|
max_instances=1,
|
||
|
|
coalesce=True,
|
||
|
|
)
|
||
|
|
self.scheduler.start()
|
||
|
|
|
||
|
|
async def stop(self) -> None:
|
||
|
|
if self.scheduler.running:
|
||
|
|
self.scheduler.shutdown(wait=False)
|
||
|
|
|
||
|
|
async def archive_inactive_bindings(self) -> int:
|
||
|
|
runtime_settings = self.runtime_settings_getter()
|
||
|
|
cutoff = datetime.now(UTC) - timedelta(days=runtime_settings.archive_days)
|
||
|
|
total_archived = 0
|
||
|
|
|
||
|
|
while True:
|
||
|
|
async with self.session_factory() as session:
|
||
|
|
try:
|
||
|
|
result = await session.execute(
|
||
|
|
select(TokenBinding.token_hash)
|
||
|
|
.where(TokenBinding.last_used_at < cutoff)
|
||
|
|
.order_by(TokenBinding.last_used_at.asc())
|
||
|
|
.limit(self.settings.archive_batch_size)
|
||
|
|
)
|
||
|
|
token_hashes = list(result.scalars())
|
||
|
|
if not token_hashes:
|
||
|
|
break
|
||
|
|
|
||
|
|
await session.execute(delete(TokenBinding).where(TokenBinding.token_hash.in_(token_hashes)))
|
||
|
|
await session.commit()
|
||
|
|
except SQLAlchemyError:
|
||
|
|
await session.rollback()
|
||
|
|
logger.exception("Failed to archive inactive bindings.")
|
||
|
|
break
|
||
|
|
|
||
|
|
await self.binding_service.invalidate_many(token_hashes)
|
||
|
|
total_archived += len(token_hashes)
|
||
|
|
|
||
|
|
if len(token_hashes) < self.settings.archive_batch_size:
|
||
|
|
break
|
||
|
|
|
||
|
|
if total_archived:
|
||
|
|
logger.info("Archived inactive bindings.", extra={"count": total_archived})
|
||
|
|
return total_archived
|