Magic link sign-in: your token, our delivery.

# Your app has already generated the token, stored sha256(token)
# with an expiry, and built the link. Notix only carries the email.
curl -X POST https://app.usenotix.dev/api/v1/emails \
  -H "Authorization: Bearer $NOTIX_API_KEY" \
  -H "Idempotency-Key: login-$REQUEST_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "from": "Acme <sign-in@acme.com>",
    "to": "user@example.com",
    "templateId": "tmpl_magic_link",
    "variables": {
      "link": "https://acme.com/auth/callback?token=Wm9vZC1yYW5kb20tdG9rZW4",
      "expiresMinutes": "15"
    }
  }'

# 201 { "emailId": "em_abc123" }

import { randomBytes, createHash } from "node:crypto";
import { Notix } from "notix-js";

const notix = new Notix(process.env.NOTIX_API_KEY);
const EXPIRES_MINUTES = 15;

export async function requestMagicLink(email: string, requestId: string) {
  const token = randomBytes(32).toString("base64url");
  const tokenHash = createHash("sha256").update(token).digest("hex");

  // Store the hash, never the token. The row is what the callback consumes.
  await db.loginToken.create({
    data: {
      email,
      tokenHash,
      expiresAt: new Date(Date.now() + EXPIRES_MINUTES * 60_000),
    },
  });

  const link = `https://acme.com/auth/callback?token=${token}`;

  const { error } = await notix.emails.send(
    {
      from: "Acme <sign-in@acme.com>",
      to: email,
      templateId: "tmpl_magic_link",
      variables: { link, expiresMinutes: String(EXPIRES_MINUTES) },
    },
    // A retried request sends one email, not two.
    { idempotencyKey: `login-${requestId}` },
  );

  if (error) {
    // Log it; the user sees the same "check your inbox" screen either way.
    logger.error("magic link send failed", error);
  }
}

import hashlib
import os
import secrets
from datetime import datetime, timedelta, timezone

from notix import Notix

notix = Notix(os.environ["NOTIX_API_KEY"])
EXPIRES_MINUTES = 15


def request_magic_link(email: str, request_id: str) -> None:
    token = secrets.token_urlsafe(32)
    token_hash = hashlib.sha256(token.encode()).hexdigest()

    # Store the hash, never the token.
    db.login_tokens.insert(
        email=email,
        token_hash=token_hash,
        expires_at=datetime.now(timezone.utc) + timedelta(minutes=EXPIRES_MINUTES),
    )

    link = f"https://acme.com/auth/callback?token={token}"

    _, error = notix.emails.send(
        {
            "from": "Acme <sign-in@acme.com>",
            "to": email,
            "templateId": "tmpl_magic_link",
            "variables": {"link": link, "expiresMinutes": str(EXPIRES_MINUTES)},
        },
        {"idempotency_key": f"login-{request_id}"},
    )

    if error:
        # Log it; the user sees the same "check your inbox" screen either way.
        logger.error("magic link send failed: %s", error)

import { createHash } from "node:crypto";

// GET only renders a page with a button. Scanners and link previews
// fetch this URL before the user does; nothing may change here.
export async function GET(request: Request) {
  const token = new URL(request.url).searchParams.get("token") ?? "";
  return renderConfirmPage({ token });
}

// The button posts back. This is the only place the token is consumed.
export async function POST(request: Request) {
  const { token } = await request.json();
  const tokenHash = createHash("sha256").update(token).digest("hex");

  const consumed = await db.loginToken.updateMany({
    where: { tokenHash, usedAt: null, expiresAt: { gt: new Date() } },
    data: { usedAt: new Date() },
  });

  if (consumed.count !== 1) {
    // Expired, unknown or already used: offer a fresh link, say nothing else.
    return renderExpired();
  }

  const row = await db.loginToken.findUniqueOrThrow({ where: { tokenHash } });
  const user = await db.user.upsert({
    where: { email: row.email },
    update: {},
    create: { email: row.email },
  });
  return startSession(user);
}

import hashlib
from datetime import datetime, timezone

from flask import request, render_template


@app.get("/auth/callback")
def callback_page():
    # Render only. Scanners fetch this before the user does.
    return render_template("confirm.html", token=request.args.get("token", ""))


@app.post("/auth/callback")
def callback_consume():
    token = request.form["token"]
    token_hash = hashlib.sha256(token.encode()).hexdigest()

    row = db.login_tokens.consume(
        token_hash=token_hash, now=datetime.now(timezone.utc)
    )  # UPDATE ... WHERE token_hash = ? AND used_at IS NULL AND expires_at > ?
    if row is None:
        return render_template("expired.html")

    user = db.users.upsert(email=row.email)
    return start_session(user)