Ryan Kegel
d147bd6f27
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 3m9s
- Added tutorial controller to manage tutorial state and progress. - Introduced HelpButton component for contextual help throughout the application. - Created various tutorial steps for onboarding and feature guidance. - Integrated tutorial prompts in multiple components (ChildrenListView, LoginButton, ScheduleModal, etc.) to enhance user experience. - Implemented logic to show tutorials based on user actions and state. - Added functionality to dismiss and skip tutorial sessions. - Established a mechanism to hydrate tutorial state from user profile.
312 lines
12 KiB
Python
312 lines
12 KiB
Python
from flask import Blueprint, request, jsonify, current_app
|
|
from events.types.user_modified import UserModified
|
|
from models.user import User
|
|
from tinydb import Query
|
|
from db.db import users_db
|
|
import jwt
|
|
import random
|
|
import string
|
|
import utils.email_sender as email_sender
|
|
from datetime import datetime, timedelta, timezone
|
|
from api.utils import get_validated_user_id, normalize_email, send_event_for_current_user
|
|
from events.sse import send_event_to_user
|
|
from events.types.payload import Payload
|
|
from api.error_codes import ACCOUNT_MARKED_FOR_DELETION, ALREADY_MARKED
|
|
from events.types.event_types import EventType
|
|
from events.types.event import Event
|
|
from events.types.profile_updated import ProfileUpdated
|
|
from utils.tracking_logger import log_tracking_event
|
|
from models.tracking_event import TrackingEvent
|
|
from db.tracking import insert_tracking_event
|
|
|
|
user_api = Blueprint('user_api', __name__)
|
|
UserQuery = Query()
|
|
|
|
def get_current_user():
|
|
token = request.cookies.get('access_token')
|
|
if not token:
|
|
return None
|
|
try:
|
|
payload = jwt.decode(token, current_app.config['SECRET_KEY'], algorithms=['HS256'])
|
|
user_id = payload.get('user_id')
|
|
user_dict = users_db.get(UserQuery.id == user_id)
|
|
return User.from_dict(user_dict) if user_dict else None
|
|
except Exception:
|
|
return None
|
|
|
|
@user_api.route('/user/profile', methods=['GET'])
|
|
def get_profile():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
return jsonify({
|
|
'first_name': user.first_name,
|
|
'last_name': user.last_name,
|
|
'email': user.email,
|
|
'image_id': user.image_id,
|
|
'email_digest_enabled': user.email_digest_enabled,
|
|
'push_notifications_enabled': user.push_notifications_enabled,
|
|
'tutorial_enabled': user.tutorial_enabled,
|
|
'tutorial_progress': user.tutorial_progress or {},
|
|
}), 200
|
|
|
|
@user_api.route('/user/profile', methods=['PUT'])
|
|
def update_profile():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json()
|
|
# Only allow first_name, last_name, image_id, email_digest_enabled, push_notifications_enabled to be updated
|
|
first_name = data.get('first_name')
|
|
last_name = data.get('last_name')
|
|
image_id = data.get('image_id')
|
|
email_digest_enabled = data.get('email_digest_enabled')
|
|
push_notifications_enabled = data.get('push_notifications_enabled')
|
|
if first_name is not None:
|
|
user.first_name = first_name
|
|
if last_name is not None:
|
|
user.last_name = last_name
|
|
if image_id is not None:
|
|
user.image_id = image_id
|
|
if email_digest_enabled is not None:
|
|
user.email_digest_enabled = bool(email_digest_enabled)
|
|
if push_notifications_enabled is not None:
|
|
user.push_notifications_enabled = bool(push_notifications_enabled)
|
|
users_db.update(user.to_dict(), UserQuery.email == user.email)
|
|
|
|
# Create tracking event
|
|
metadata = {}
|
|
if first_name is not None:
|
|
metadata['first_name_updated'] = True
|
|
if last_name is not None:
|
|
metadata['last_name_updated'] = True
|
|
if image_id is not None:
|
|
metadata['image_updated'] = True
|
|
if email_digest_enabled is not None:
|
|
metadata['email_digest_enabled_updated'] = True
|
|
if push_notifications_enabled is not None:
|
|
metadata['push_notifications_enabled_updated'] = True
|
|
|
|
tracking_event = TrackingEvent.create_event(
|
|
user_id=user_id,
|
|
child_id=None, # No child for user profile
|
|
entity_type='user',
|
|
entity_id=user.id,
|
|
action='updated',
|
|
points_before=0, # Not relevant
|
|
points_after=0,
|
|
metadata=metadata
|
|
)
|
|
insert_tracking_event(tracking_event)
|
|
log_tracking_event(tracking_event)
|
|
|
|
# Send SSE event
|
|
send_event_for_current_user(Event(EventType.PROFILE_UPDATED.value, ProfileUpdated(user.id)))
|
|
|
|
return jsonify({'message': 'Profile updated'}), 200
|
|
|
|
@user_api.route('/user/tutorial-progress', methods=['PATCH'])
|
|
def update_tutorial_progress():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json() or {}
|
|
|
|
if data.get('reset') is True:
|
|
user.tutorial_progress = {}
|
|
elif 'enabled' in data:
|
|
user.tutorial_enabled = bool(data.get('enabled'))
|
|
elif 'step_id' in data:
|
|
step_id = str(data.get('step_id') or '').strip()
|
|
if not step_id:
|
|
return jsonify({'error': 'Missing step_id'}), 400
|
|
progress = dict(user.tutorial_progress or {})
|
|
progress[step_id] = bool(data.get('seen', True))
|
|
user.tutorial_progress = progress
|
|
else:
|
|
return jsonify({'error': 'No-op'}), 400
|
|
|
|
users_db.update(user.to_dict(), UserQuery.email == user.email)
|
|
send_event_for_current_user(Event(EventType.PROFILE_UPDATED.value, ProfileUpdated(user.id)))
|
|
return jsonify({
|
|
'tutorial_enabled': user.tutorial_enabled,
|
|
'tutorial_progress': user.tutorial_progress,
|
|
}), 200
|
|
|
|
@user_api.route('/user/image', methods=['PUT'])
|
|
def update_image():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json()
|
|
image_id = data.get('image_id')
|
|
if not image_id:
|
|
return jsonify({'error': 'Missing image_id'}), 400
|
|
user.image_id = image_id
|
|
users_db.update(user.to_dict(), UserQuery.email == user.email)
|
|
return jsonify({'message': 'Image updated', 'image_id': image_id}), 200
|
|
|
|
@user_api.route('/user/check-pin', methods=['POST'])
|
|
def check_pin():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json()
|
|
pin = data.get('pin')
|
|
if not pin:
|
|
return jsonify({'error': 'Missing pin'}), 400
|
|
if user.pin and pin == user.pin:
|
|
return jsonify({'valid': True}), 200
|
|
return jsonify({'valid': False}), 200
|
|
|
|
@user_api.route('/user/has-pin', methods=['GET'])
|
|
def has_pin():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
return jsonify({'has_pin': bool(user.pin)}), 200
|
|
|
|
@user_api.route('/user/request-pin-setup', methods=['POST'])
|
|
def request_pin_setup():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user or not user.verified:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
# Generate 6-digit/character code
|
|
code = ''.join(random.choices(string.ascii_uppercase + string.digits, k=6))
|
|
user.pin_setup_code = code
|
|
user.pin_setup_code_created = datetime.utcnow().isoformat()
|
|
users_db.update(user.to_dict(), UserQuery.email == user.email)
|
|
# Send email
|
|
send_pin_setup_email(user.email, code)
|
|
return jsonify({'message': 'Verification code sent to your email.'}), 200
|
|
|
|
def send_pin_setup_email(email, code):
|
|
# Use the reusable email sender
|
|
email_sender.send_pin_setup_email(email, code)
|
|
|
|
@user_api.route('/user/verify-pin-setup', methods=['POST'])
|
|
def verify_pin_setup():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user or not user.verified:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json()
|
|
code = data.get('code')
|
|
if not code:
|
|
return jsonify({'error': 'Missing code'}), 400
|
|
if not user.pin_setup_code or not user.pin_setup_code_created:
|
|
return jsonify({'error': 'No code requested'}), 400
|
|
# Check expiry (10 min)
|
|
created = datetime.fromisoformat(user.pin_setup_code_created)
|
|
if datetime.utcnow() > created + timedelta(minutes=10):
|
|
return jsonify({'error': 'Code expired'}), 400
|
|
if code.strip().upper() != user.pin_setup_code.upper():
|
|
return jsonify({'error': 'Invalid code'}), 400
|
|
return jsonify({'message': 'Code verified'}), 200
|
|
|
|
@user_api.route('/user/set-pin', methods=['POST'])
|
|
def set_pin():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user or not user.verified:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json()
|
|
pin = data.get('pin')
|
|
if not pin or not pin.isdigit() or not (4 <= len(pin) <= 6):
|
|
return jsonify({'error': 'PIN must be 4-6 digits'}), 400
|
|
# Only allow if code was recently verified
|
|
if not user.pin_setup_code or not user.pin_setup_code_created:
|
|
return jsonify({'error': 'No code verified'}), 400
|
|
created = datetime.fromisoformat(user.pin_setup_code_created)
|
|
if datetime.utcnow() > created + timedelta(minutes=10):
|
|
return jsonify({'error': 'Code expired'}), 400
|
|
# Set pin, clear code
|
|
user.pin = pin
|
|
user.pin_setup_code = ''
|
|
user.pin_setup_code_created = None
|
|
users_db.update(user.to_dict(), UserQuery.email == user.email)
|
|
return jsonify({'message': 'Parent PIN set'}), 200
|
|
|
|
@user_api.route('/user/mark-for-deletion', methods=['POST'])
|
|
def mark_for_deletion():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
|
|
# Validate email from request body
|
|
data = request.get_json()
|
|
email = data.get('email', '').strip()
|
|
if not email:
|
|
return jsonify({'error': 'Email is required', 'code': 'EMAIL_REQUIRED'}), 400
|
|
|
|
# Verify email matches the logged-in user - make sure to normalize the email address first
|
|
if normalize_email(email) != normalize_email(user.email):
|
|
return jsonify({'error': 'Email does not match your account', 'code': 'EMAIL_MISMATCH'}), 400
|
|
|
|
# Check if already marked
|
|
if user.marked_for_deletion:
|
|
return jsonify({'error': 'Account already marked for deletion', 'code': ALREADY_MARKED}), 400
|
|
|
|
# Mark for deletion
|
|
user.marked_for_deletion = True
|
|
user.marked_for_deletion_at = datetime.now(timezone.utc).isoformat()
|
|
|
|
# Invalidate any outstanding verification/reset tokens so they cannot be used after marking
|
|
user.verify_token = None
|
|
user.verify_token_created = None
|
|
user.reset_token = None
|
|
user.reset_token_created = None
|
|
|
|
users_db.update(user.to_dict(), UserQuery.id == user.id)
|
|
|
|
# Trigger SSE event
|
|
send_event_for_current_user(Event(EventType.USER_MARKED_FOR_DELETION.value, UserModified(user.id, UserModified.OPERATION_DELETE)))
|
|
|
|
# Notify all other active sessions to sign out and go to landing page
|
|
send_event_to_user(user.id, Event(EventType.FORCE_LOGOUT.value, Payload({'reason': 'account_deleted'})))
|
|
|
|
return jsonify({'success': True}), 200
|
|
|
|
|
|
@user_api.route('/user/e2e-get-pin-code', methods=['GET'])
|
|
def e2e_get_pin_code():
|
|
"""Return the current user's pin_setup_code for e2e testing. Never available in production."""
|
|
import os
|
|
if os.environ.get('DB_ENV', 'prod') == 'prod':
|
|
return jsonify({'error': 'Not available in production'}), 403
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
return jsonify({'code': user.pin_setup_code or ''}), 200
|