saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.base import ensure_aware_utc
|
|
from app.models.email_code import EmailCode
|
|
from app.models.enums import EmailCodePurpose
|
|
|
|
|
|
class EmailCodeRepository:
|
|
def __init__(self, db: AsyncSession) -> None:
|
|
self.db = db
|
|
|
|
async def create(
|
|
self, *, user_id: uuid.UUID, purpose: EmailCodePurpose, code_hash: str, expires_at: datetime
|
|
) -> EmailCode:
|
|
record = EmailCode(
|
|
user_id=user_id, purpose=purpose, code_hash=code_hash, expires_at=expires_at
|
|
)
|
|
self.db.add(record)
|
|
await self.db.flush()
|
|
return record
|
|
|
|
async def get_latest_valid(
|
|
self, user_id: uuid.UUID, purpose: EmailCodePurpose, code_hash: str
|
|
) -> EmailCode | None:
|
|
"""Most recent unused, unexpired code for this user/purpose whose
|
|
hash matches. `invalidate_unused` is what actually guarantees only
|
|
the most recently issued code can ever satisfy this - this method
|
|
alone doesn't enforce that."""
|
|
result = await self.db.execute(
|
|
select(EmailCode)
|
|
.where(
|
|
EmailCode.user_id == user_id,
|
|
EmailCode.purpose == purpose,
|
|
EmailCode.code_hash == code_hash,
|
|
EmailCode.used_at.is_(None),
|
|
)
|
|
.order_by(EmailCode.created_at.desc())
|
|
)
|
|
for record in result.scalars().all():
|
|
if ensure_aware_utc(record.expires_at) > datetime.now(UTC):
|
|
return record
|
|
return None
|
|
|
|
async def mark_used(self, record: EmailCode) -> None:
|
|
record.used_at = datetime.now(UTC)
|
|
await self.db.flush()
|
|
|
|
async def invalidate_unused(self, user_id: uuid.UUID, purpose: EmailCodePurpose) -> None:
|
|
"""Called right before issuing a fresh code - a resend must fully
|
|
supersede every prior unused code for this purpose, not just make
|
|
them harder to guess. Without this, an old code (e.g. still sitting
|
|
in an old email) stays valid until it naturally expires, even after
|
|
the user has explicitly asked for a new one."""
|
|
await self.db.execute(
|
|
update(EmailCode)
|
|
.where(
|
|
EmailCode.user_id == user_id,
|
|
EmailCode.purpose == purpose,
|
|
EmailCode.used_at.is_(None),
|
|
)
|
|
.values(used_at=datetime.now(UTC))
|
|
)
|
|
await self.db.flush()
|