calendar/backend/email_service.py

import hashlib
import secrets
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path

import os
import smtplib
from email.message import EmailMessage

from dotenv import load_dotenv

load_dotenv()

DB_NAME = Path(__file__).parent.parent / "database" / "calendar.db"


def get_connection():
    conn = sqlite3.connect(DB_NAME)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON;")
    return conn


def hash_token(token: str) -> str:
    return hashlib.sha256(
        token.encode("utf-8")
    ).hexdigest()


def update_email(user_id: int, new_email: str):
    new_email = new_email.strip().lower()

    conn = get_connection()
    cur = conn.cursor()

    try:
        cur.execute(
            """
            SELECT id
            FROM users
            WHERE id = ?
            """,
            (user_id,),
        )

        if cur.fetchone() is None:
            return {
                "status": "failed",
                "reason": "user_not_found",
            }

        cur.execute(
            """
            SELECT id
            FROM users
            WHERE email = ?
              AND id != ?
            """,
            (new_email, user_id),
        )

        if cur.fetchone() is not None:
            return {
                "status": "failed",
                "reason": "email_already_in_use",
            }

        cur.execute(
            """
            UPDATE users
            SET email = ?,
                email_verified = 0
            WHERE id = ?
            """,
            (new_email, user_id),
        )

        cur.execute(
            """
            DELETE FROM email_verifications
            WHERE user_id = ?
            """,
            (user_id,),
        )

        conn.commit()

        return {
            "status": "updated",
            "email": new_email,
            "email_verified": False,
        }

    except sqlite3.IntegrityError:
        conn.rollback()

        return {
            "status": "failed",
            "reason": "email_already_in_use",
        }

    finally:
        conn.close()


def create_verification(user_id: int):
    token = secrets.token_urlsafe(32)
    token_hash = hash_token(token)

    expires_at = (
        datetime.now(timezone.utc)
        + timedelta(hours=24)
    ).isoformat()

    conn = get_connection()
    cur = conn.cursor()

    try:
        cur.execute(
            """
            DELETE FROM email_verifications
            WHERE user_id = ?
            """,
            (user_id,),
        )

        cur.execute(
            """
            INSERT INTO email_verifications (
                user_id,
                token_hash,
                expires_at
            )
            VALUES (?, ?, ?)
            """,
            (
                user_id,
                token_hash,
                expires_at,
            ),
        )

        conn.commit()

        return token

    finally:
        conn.close()


def verify_email(token: str):
    token_hash = hash_token(token)

    conn = get_connection()
    cur = conn.cursor()

    try:
        cur.execute(
            """
            SELECT
                id,
                user_id,
                expires_at
            FROM email_verifications
            WHERE token_hash = ?
            """,
            (token_hash,),
        )

        verification = cur.fetchone()

        if verification is None:
            return {
                "status": "failed",
                "reason": "invalid_token",
            }

        expires_at = datetime.fromisoformat(
            verification["expires_at"]
        )

        if datetime.now(timezone.utc) >= expires_at:
            cur.execute(
                """
                DELETE FROM email_verifications
                WHERE id = ?
                """,
                (verification["id"],),
            )

            conn.commit()

            return {
                "status": "failed",
                "reason": "token_expired",
            }

        cur.execute(
            """
            UPDATE users
            SET email_verified = 1
            WHERE id = ?
            """,
            (verification["user_id"],),
        )

        cur.execute(
            """
            DELETE FROM email_verifications
            WHERE id = ?
            """,
            (verification["id"],),
        )

        conn.commit()

        return {
            "status": "verified",
        }

    finally:
        conn.close()

def send_verification_email(email: str, token: str):
    smtp_host = os.getenv("SMTP_HOST")
    smtp_port = int(os.getenv("SMTP_PORT", "587"))
    smtp_username = os.getenv("SMTP_USERNAME")
    smtp_password = os.getenv("SMTP_PASSWORD")
    smtp_from = os.getenv("SMTP_FROM")

    if not all([
        smtp_host,
        smtp_username,
        smtp_password,
        smtp_from,
    ]):
        raise RuntimeError("SMTP configuration is incomplete")

    verification_url = (
        f"https://ben.de-roo.org/api/verify-email?token={token}"
    )

    message = EmailMessage()

    message["From"] = smtp_from
    message["To"] = email
    message["Subject"] = "Verify your Calendar email address"

    message.set_content(
        f"""Hello,

You requested to use this email address for your Calendar account.

Verify your email address using this link:

{verification_url}

This link expires after 24 hours.

If you did not request this, you can ignore this email.
"""
    )

    with smtplib.SMTP(smtp_host, smtp_port) as smtp:
        smtp.starttls()
        smtp.login(
            smtp_username,
            smtp_password,
        )
        smtp.send_message(message)