Spaces:

NurseCitizenDeveloper
/

PNA-Assistant

Sleeping

File size: 8,898 Bytes

b7e98d4

"""
PNA Assistant — Stripe Webhook Handler
POST /webhook  →  Stripe sends checkout.session.completed events here
                 →  We generate a PNA-PRO-XXXXX code and email it to the customer

Run standalone:  uvicorn webhook:app --host 0.0.0.0 --port 8080
In Docker:       separate service in docker-compose.yml
"""

import hashlib
import hmac
import json
import logging
import os
import random
import string
from datetime import datetime, timezone
from pathlib import Path

import boto3
from botocore.exceptions import ClientError
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

# ─── Config ──────────────────────────────────────────────────────────────────

STRIPE_WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET", "")
AWS_REGION            = os.getenv("AWS_REGION", "eu-west-2")
SES_FROM_EMAIL        = os.getenv("SES_FROM_EMAIL", "lincoln@clinyqai.com")

# Where we persist issued codes (append-only JSONL)
CODES_LOG = Path("/app/codes.jsonl")

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("pna-webhook")

app = FastAPI(title="PNA Assistant Webhook", docs_url=None, redoc_url=None)

# ─── Helpers ─────────────────────────────────────────────────────────────────

def _generate_code(tier: str = "PRO") -> str:
    """Generate a unique activation code e.g. PNA-PRO-A7K2M9"""
    suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=6))
    return f"PNA-{tier}-{suffix}"


def _verify_stripe_signature(payload: bytes, sig_header: str, secret: str) -> bool:
    """
    Validate Stripe-Signature header.
    Stripe uses HMAC-SHA256 with the raw request body.
    """
    if not secret:
        log.warning("STRIPE_WEBHOOK_SECRET not set — skipping signature verification")
        return True  # dev mode only; set the secret in production

    try:
        parts = {k: v for part in sig_header.split(",") for k, v in [part.split("=", 1)]}
        timestamp = parts.get("t", "")
        v1_sig    = parts.get("v1", "")
        signed_payload = f"{timestamp}.".encode() + payload
        expected = hmac.new(
            secret.encode(),
            signed_payload,
            hashlib.sha256,
        ).hexdigest()
        return hmac.compare_digest(expected, v1_sig)
    except Exception as exc:
        log.error("Signature verification error: %s", exc)
        return False


def _log_code(email: str, code: str, plan: str, session_id: str) -> None:
    """Append issued code to local JSONL log for admin audit trail."""
    CODES_LOG.parent.mkdir(parents=True, exist_ok=True)
    record = {
        "ts":         datetime.now(timezone.utc).isoformat(),
        "email":      email,
        "code":       code,
        "plan":       plan,
        "session_id": session_id,
    }
    with CODES_LOG.open("a") as f:
        f.write(json.dumps(record) + "\n")
    log.info("Code issued: %s → %s (%s)", email, code, plan)


def _send_activation_email(email: str, code: str, plan: str) -> bool:
    """
    Send activation code to customer via Amazon SES.
    Returns True if sent, False on error.
    """
    label     = "Pro" if plan == "PRO" else "Institution"
    price     = "£9.99/month" if plan == "PRO" else "£99/month"
    body_html = f"""
<!DOCTYPE html>
<html>
<body style="font-family:Inter,sans-serif;max-width:600px;margin:0 auto;padding:2rem;color:#1e293b">
  <div style="background:linear-gradient(135deg,#1a2460,#0d9488);padding:2rem;border-radius:12px;margin-bottom:2rem">
    <h1 style="color:white;font-size:1.5rem;margin:0">👨🏾‍⚕️ PNA Assistant</h1>
    <p style="color:rgba(255,255,255,0.85);margin:0.5rem 0 0">
      Professional Nurse Advocate AI — {label} Plan
    </p>
  </div>

  <p>Thank you for subscribing to PNA Assistant <strong>{label}</strong> ({price}).</p>

  <p>Your activation code is:</p>

  <div style="background:#f0fdf4;border:2px solid #0d9488;border-radius:12px;padding:1.5rem;text-align:center;margin:1.5rem 0">
    <span style="font-family:monospace;font-size:1.75rem;font-weight:700;color:#0d9488;letter-spacing:0.1em">
      {code}
    </span>
  </div>

  <p><strong>How to activate:</strong></p>
  <ol>
    <li>Go to <a href="https://pna.nursingcitizendevelopment.com">pna.nursingcitizendevelopment.com</a></li>
    <li>Open the sidebar and find <em>Subscription</em></li>
    <li>Paste your code in the <em>Activation code</em> box</li>
    <li>Click Enter — your {label} features will unlock immediately</li>
  </ol>

  <p>Your {label} plan includes:</p>
  <ul>
    <li>📄 Download supervision notes (.docx)</li>
    <li>📋 NMC revalidation CPD log export (.docx)</li>
    <li>💾 Save and resume sessions</li>
    <li>📧 Monthly PNA newsletter</li>
    {"<li>🏫 Multi-user access</li><li>🎨 Custom branding</li>" if plan == "INST" else ""}
  </ul>

  <p style="font-size:0.875rem;color:#64748b">
    If you have any questions, reply to this email or contact
    <a href="mailto:lincoln@clinyqai.com">lincoln@clinyqai.com</a>.<br><br>
    This tool is for educational purposes only and does not replace clinical supervision
    or your employer's formal governance processes.<br><br>
    Contains public sector information licensed under the
    <a href="https://www.nationalarchives.gov.uk/doc/open-government-licence/version/3/">
      Open Government Licence v3.0
    </a> — NHS England.
  </p>
</body>
</html>
"""
    body_text = (
        f"Thank you for subscribing to PNA Assistant {label} ({price}).\n\n"
        f"Your activation code is: {code}\n\n"
        "To activate:\n"
        "1. Go to pna.nursingcitizendevelopment.com\n"
        "2. Open the sidebar → Subscription\n"
        "3. Paste your code in the Activation code box\n\n"
        "Questions? Email lincoln@clinyqai.com"
    )

    try:
        ses = boto3.client("ses", region_name=AWS_REGION)
        ses.send_email(
            Source      = f"PNA Assistant <{SES_FROM_EMAIL}>",
            Destination = {"ToAddresses": [email]},
            Message     = {
                "Subject": {"Data": f"Your PNA Assistant {label} activation code 🎉"},
                "Body": {
                    "Text": {"Data": body_text},
                    "Html": {"Data": body_html},
                },
            },
        )
        log.info("Activation email sent to %s", email)
        return True
    except ClientError as exc:
        log.error("SES send failed: %s", exc.response["Error"]["Message"])
        return False
    except Exception as exc:
        log.error("Email error: %s", exc)
        return False


# ─── Routes ──────────────────────────────────────────────────────────────────

@app.get("/health")
async def health():
    return {"status": "ok"}


@app.post("/webhook")
async def stripe_webhook(request: Request):
    payload    = await request.body()
    sig_header = request.headers.get("Stripe-Signature", "")

    # 1. Verify Stripe signature
    if not _verify_stripe_signature(payload, sig_header, STRIPE_WEBHOOK_SECRET):
        log.warning("Invalid Stripe signature — rejected")
        raise HTTPException(status_code=400, detail="Invalid signature")

    # 2. Parse event
    try:
        event = json.loads(payload)
    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="Invalid JSON")

    event_type = event.get("type", "")
    log.info("Received Stripe event: %s", event_type)

    # 3. Handle checkout completion
    if event_type == "checkout.session.completed":
        session    = event["data"]["object"]
        email      = session.get("customer_details", {}).get("email") or session.get("customer_email")
        amount     = session.get("amount_total", 0)      # pence/cents
        session_id = session.get("id", "unknown")

        if not email:
            log.warning("No email in session %s — cannot send code", session_id)
            return JSONResponse({"received": True, "warning": "no_email"})

        # Determine tier from amount (999 = £9.99, 9900 = £99.00)
        tier  = "INST" if amount >= 9900 else "PRO"
        code  = _generate_code(tier)

        # 4. Log the code (always)
        _log_code(email, code, tier, session_id)

        # 5. Email the customer
        sent = _send_activation_email(email, code, tier)
        if not sent:
            log.warning(
                "SES send failed for %s — code %s logged to %s",
                email, code, CODES_LOG,
            )

    return JSONResponse({"received": True})