email template

backend/app/services/member_invitations.py

@@ -0,0 +1,79 @@
+"""Single-use invitation tokens for employee onboarding."""
+
+from __future__ import annotations
+
+import hashlib
+import secrets
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.models import Account, MemberInvitation
+
+INVITE_EXPIRY_HOURS = 48
+INVITE_FROM_LOCAL = "invites"
+
+
+def hash_invite_token(raw_token: str) -> str:
+    return hashlib.sha256(raw_token.encode("utf-8")).hexdigest()
+
+
+def generate_invite_token() -> str:
+    return secrets.token_urlsafe(32)
+
+
+async def invalidate_pending_invitations(
+    session: AsyncSession,
+    account_id,
+) -> None:
+    invitations = await session.scalars(
+        select(MemberInvitation).where(
+            MemberInvitation.account_id == account_id,
+            MemberInvitation.used_at.is_(None),
+        )
+    )
+    now = datetime.now(timezone.utc)
+    for invitation in invitations.all():
+        invitation.used_at = now
+
+
+async def create_member_invitation(
+    session: AsyncSession,
+    member: Account,
+) -> str:
+    await invalidate_pending_invitations(session, member.id)
+    raw_token = generate_invite_token()
+    invitation = MemberInvitation(
+        account_id=member.id,
+        token_hash=hash_invite_token(raw_token),
+        expires_at=datetime.now(timezone.utc)
+        + timedelta(hours=INVITE_EXPIRY_HOURS),
+    )
+    session.add(invitation)
+    await session.flush()
+    return raw_token
+
+
+async def get_valid_invitation(
+    session: AsyncSession,
+    raw_token: str,
+) -> tuple[MemberInvitation, Account]:
+    token_hash = hash_invite_token(raw_token.strip())
+    invitation = await session.scalar(
+        select(MemberInvitation)
+        .options(selectinload(MemberInvitation.account))
+        .where(MemberInvitation.token_hash == token_hash)
+    )
+    if invitation is None:
+        raise ValueError("This invitation link is invalid.")
+    if invitation.used_at is not None:
+        raise ValueError("This invitation has already been used.")
+    now = datetime.now(timezone.utc)
+    if invitation.expires_at <= now:
+        raise ValueError("This invitation has expired.")
+    account = invitation.account
+    if account is None or account.member_status != "invited":
+        raise ValueError("This invitation is no longer valid.")
+    return invitation, account