email integration

6 files changed, 155 insertions, 51 deletions

diff --git a/jb/api/magic_token.py b/jb/api/magic_token.py
index e0a1cca..562ba81 100644
--- a/jb/api/magic_token.py
+++ b/jb/api/magic_token.py
@@ -43,10 +43,10 @@ def consume_magic_token(token: str) -> str:
     return user_email
 
 
-def create_amt_account_link_token(user: User, amt_worker_id: str) -> str:
+def create_amt_account_link_token(email: str, amt_worker_id: str) -> str:
     """Bind an email and AMT worker ID to an opaque, short-lived token."""
     data = AmtAccountLink(
-        email=user.email,
+        email=email,
         amt_worker_id=amt_worker_id,
     )
     token = secrets.token_urlsafe(32)
diff --git a/jb/flow/events.py b/jb/flow/events.py
index 04c4bb6..0eb91fd 100644
--- a/jb/flow/events.py
+++ b/jb/flow/events.py
@@ -2,7 +2,7 @@ import logging
 import time
 from concurrent import futures
 from concurrent.futures import Executor, ThreadPoolExecutor
-from typing import TypedDict, cast
+from typing import cast
 
 import redis
 
@@ -19,13 +19,6 @@ from jb.models.event import MTurkEvent
 StreamMessages = list[tuple[str, list[tuple[bytes, dict[bytes, bytes]]]]]
 
 
-class PendingEntry(TypedDict):
-    message_id: bytes
-    consumer: bytes
-    time_since_delivered: int
-    times_delivered: int
-
-
 def process_mturk_events_task():
     executor = ThreadPoolExecutor(max_workers=5)
     create_consumer_group()
diff --git a/jb/managers/email_manager.py b/jb/managers/email_manager.py
new file mode 100644
index 0000000..dcbe167
--- /dev/null
+++ b/jb/managers/email_manager.py
@@ -0,0 +1,57 @@
+import requests
+
+from jb.config import settings
+
+MAUTIC_BASE_URL = "https://mail.jamesbillings67.com"
+EMAIL_TEMPLATE_ID = 1
+auth_headers = {"Authorization": f"Basic {settings.mautic_api_key.get_secret_value()}"}
+
+
+def get_or_create_contact(email: str, amt_worker_id: str | None = None):
+    # amt_worker_id = "A2Z2FRA128FNW"
+    body = {"email": email}
+    if amt_worker_id:
+        body["amt_worker_id"] = amt_worker_id
+    res = requests.post(
+        url=f"{MAUTIC_BASE_URL}/api/contacts/new",
+        json=body,
+        headers=auth_headers,
+    ).json()
+    contact_id = res["contact"]["id"]
+    return contact_id
+
+
+def send_login_email_from_url(mautic_url, magic_link) -> None:
+    email_tokens = {
+        "magic_link": magic_link,
+    }
+    body = {"tokens": email_tokens}
+    response = requests.post(url=mautic_url, json=body, headers=auth_headers)
+    try:
+        response.raise_for_status()
+    except requests.exceptions.HTTPError:
+        print(f"Failed to send email. Status code: {response.status_code}")
+        print(response.text)
+        raise
+    d = response.json()
+    assert d.get("success"), f"Failed to send email: {d.get('failed')}"
+    print("Email sent successfully")
+
+
+def send_login_email(email: str, magic_token: str):
+    contact_id = get_or_create_contact(email=email)
+    mautic_url = (
+        f"{MAUTIC_BASE_URL}/api/emails/{EMAIL_TEMPLATE_ID}/contact/{contact_id}/send"
+    )
+    magic_link = f"{settings.base_url}auth/magic-link/?token={magic_token}"
+    return send_login_email_from_url(mautic_url, magic_link)
+
+
+def send_amt_link_email(email: str, magic_token: str):
+    # don't actually associate the email with the worker ID until they click the link
+    contact_id = get_or_create_contact(email=email)
+    mautic_url = (
+        f"{MAUTIC_BASE_URL}/api/emails/{EMAIL_TEMPLATE_ID}/contact/{contact_id}/send"
+    )
+    magic_link = f"{settings.base_url}auth/link-amt/?token={magic_token}"
+    return send_login_email_from_url(mautic_url, magic_link)
diff --git a/jb/settings.py b/jb/settings.py
index 7fe2a5a..86c8a36 100644
--- a/jb/settings.py
+++ b/jb/settings.py
@@ -41,6 +41,7 @@ class Settings(AmtJbBaseSettings):
     )
     debug: bool = False
     app_name: str = "AMT JB API"
+    base_url: HttpUrl = Field(default=HttpUrl("https://jamesbillings67.com/"))
 
     fsb_host: HttpUrl = Field(default=HttpUrl("https://fsb.generalresearch.com/"))
     # Needed for admin function on fsb w/o authentication
@@ -62,6 +63,8 @@ class Settings(AmtJbBaseSettings):
     )
     gr_api_token: SecretStr = Field(min_length=1)
 
+    mautic_api_key: SecretStr = Field(min_length=32)
+
 
 class TestSettings(Settings):
     model_config = SettingsConfigDict(
@@ -73,6 +76,7 @@ class TestSettings(Settings):
     )
     debug: bool = True
     app_name: str = "AMT JB API Test"
+    base_url: HttpUrl = Field(default=HttpUrl("http://127.0.0.1:8081/"))
 
 
 @lru_cache
diff --git a/jb/views/auth.py b/jb/views/auth.py
index dc63ca3..6f8a3e2 100644
--- a/jb/views/auth.py
+++ b/jb/views/auth.py
@@ -1,8 +1,9 @@
+import logging
 from typing import Annotated
 from urllib.parse import urlencode
 
 from fastapi import APIRouter, Depends, HTTPException, Response, status
-from fastapi.responses import HTMLResponse
+from fastapi.responses import HTMLResponse, RedirectResponse
 
 from jb.api.auth import (
     SESSION_COOKIE_NAME,
@@ -17,6 +18,11 @@ from jb.api.magic_token import (
 )
 from jb.config import settings
 from jb.dependencies import get_gr_api_manager
+from jb.managers.email_manager import (
+    get_or_create_contact,
+    send_amt_link_email,
+    send_login_email,
+)
 from jb.managers.gr_api import GRApiManager
 from jb.models.auth import (
     AccountLogin,
@@ -30,36 +36,34 @@ auth_router = APIRouter(prefix="/auth", tags=["Auth"])
 
 
 @auth_router.post("/magic-link/request")
-def request_mock_magic_link(body: AccountLogin) -> dict[str, str]:
-    """Create a magic link without sending email in development."""
-    if not settings.debug:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
-
-    # todo: send email here
-
-    user = User(email=body.email)
-    token = create_magic_token(str(user.email))
-    query = urlencode({"token": token})
-    return {"magic_link": f"/auth/magic-link/?{query}"}
+def request_magic_link(body: AccountLogin) -> dict[str, str]:
+    """Create a magic link."""
+    email = str(body.email)
+    token = create_magic_token(user_email=email)
 
+    if settings.debug:
+        query = urlencode({"token": token})
+        return {"magic_link": f"{settings.base_url}auth/magic-link/?{query}"}
 
-@auth_router.post("/link-amt/request")
-def link_amt_account(body: AmtAccountLink) -> dict[str, str]:
-    """Create a mock AMT account-link email in development."""
-    if not settings.debug:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
-
-    # TODO: Derive amt_worker_id from a server-validated AMT assignment and
-    # send this link by email instead of returning it.
-    user = User(email=body.email)
-    token = create_amt_account_link_token(user, body.amt_worker_id)
-    query = urlencode({"token": token})
-    return {"magic_link": f"/auth/link-amt/?{query}"}
+    send_login_email(email=email, magic_token=token)
+    return {"detail": "Link sent. Check your inbox and follow the link to log in."}
 
 
 @auth_router.get("/magic-link/", response_class=HTMLResponse, include_in_schema=False)
-def magic_link_landing_page() -> HTMLResponse:
+def magic_link_landing_page(
+    gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)],
+    token: str | None = None,
+) -> Response:
     """Serve the SPA without redeeming the token; email prefetches are harmless."""
+    if settings.debug:
+        if token is None:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="token is required",
+            )
+        response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
+        _exchange_magic_link(token, response, gr_api)
+        return response
     return HTMLResponse(
         BASE_HTML,
         headers={
@@ -70,12 +74,6 @@ def magic_link_landing_page() -> HTMLResponse:
     )
 
 
-@auth_router.get("/link-amt/", response_class=HTMLResponse, include_in_schema=False)
-def link_amt_account_landing_page() -> HTMLResponse:
-    """Serve the account-link SPA without consuming the one-time token."""
-    return magic_link_landing_page()
-
-
 @auth_router.post("/magic-link/exchange", status_code=status.HTTP_204_NO_CONTENT)
 def exchange_magic_link(
     body: MagicLinkExchangeRequest,
@@ -83,16 +81,15 @@ def exchange_magic_link(
     gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)],
 ) -> None:
     """Exchange a magic link only after its landing page makes an explicit POST."""
-    user_email = consume_magic_token(body.token)
+    _exchange_magic_link(body.token, response, gr_api)
 
-    user = User.model_validate({"email": user_email})
-    # hit thl to make sure this user exists
-    user = gr_api.ensure_user_exists(user)
 
-    session_token = create_session(user.product_user_id)
+def _exchange_magic_link(token: str, response: Response, gr_api: GRApiManager) -> None:
+    user_email = consume_magic_token(token)
+    user = gr_api.ensure_user_exists(User.model_validate({"email": user_email}))
     response.set_cookie(
         key=SESSION_COOKIE_NAME,
-        value=session_token,
+        value=create_session(user.product_user_id),
         max_age=settings.session_token_ttl_seconds,
         httponly=True,
         secure=not settings.debug,
@@ -101,6 +98,45 @@ def exchange_magic_link(
     )
 
 
+@auth_router.post("/link-amt/request")
+def link_amt_account(body: AmtAccountLink) -> dict[str, str]:
+    """Link an AMT account and login."""
+    email = str(body.email)
+    amt_worker_id = body.amt_worker_id
+    token = create_amt_account_link_token(email=email, amt_worker_id=amt_worker_id)
+
+    if settings.debug:
+        query = urlencode({"token": token})
+        return {"magic_link": f"{settings.base_url}auth/link-amt/?{query}"}
+
+    send_amt_link_email(email=email, magic_token=token)
+    return {"detail": "Link sent. Check your inbox and follow the link to log in."}
+
+
+@auth_router.get("/link-amt/", response_class=HTMLResponse, include_in_schema=False)
+def link_amt_account_landing_page(
+    gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)],
+    token: str | None = None,
+) -> HTMLResponse:
+    """Serve the account-link SPA without consuming the one-time token."""
+    if settings.debug:
+        if token is None:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="token is required",
+            )
+        response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
+        _exchange_amt_account_link(token, response, gr_api)
+    return HTMLResponse(
+        BASE_HTML,
+        headers={
+            "Cache-Control": "no-store",
+            "Referrer-Policy": "no-referrer",
+            "X-Robots-Tag": "noindex, nofollow",
+        },
+    )
+
+
 @auth_router.post("/link-amt/exchange", status_code=status.HTTP_204_NO_CONTENT)
 def exchange_amt_account_link(
     body: MagicLinkExchangeRequest,
@@ -108,9 +144,23 @@ def exchange_amt_account_link(
     gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)],
 ) -> None:
     """Validate the email link, then transition the bound AMT account."""
-    token_data = consume_amt_account_link_token(body.token)
-    user = User(email=token_data.email)
-    user = gr_api.transition_user_from_amt(user, token_data.amt_worker_id)
+    try:
+        _exchange_amt_account_link(body.token, response, gr_api)
+    except ValueError as e:
+        logging.error(f"Failed to exchange AMT account link: {e}")
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+
+
+def _exchange_amt_account_link(token: str, response: Response, gr_api: GRApiManager):
+    token_data = consume_amt_account_link_token(token)
+    email = token_data.email
+    amt_worker_id = token_data.amt_worker_id
+
+    user = User(email=email)
+    user = gr_api.transition_user_from_amt(user=user, amt_worker_id=amt_worker_id)
+
+    # In Mautic, associate the email with the worker ID (AFTER the user has transitioned)
+    get_or_create_contact(email=email, amt_worker_id=amt_worker_id)
 
     session_token = create_session(user.product_user_id)
     response.set_cookie(
diff --git a/tests/fixtures/flow.py b/tests/fixtures/flow.py
index 08ec49e..3fcca81 100644
--- a/tests/fixtures/flow.py
+++ b/tests/fixtures/flow.py
@@ -5,7 +5,7 @@ from uuid import uuid4
 import pytest
 import requests
 from generalresearch.models.thl.payout import UserPayoutEvent
-from generalresearch.models.thl.wallet import PayoutType
+from generalresearch.models.thl.wallet.definitions import PayoutType
 from generalresearch.models.thl.wallet.cashout_method import (
     CashoutRequestResponse,
     CashoutRequestInfo,