""" app.py — B24 Messenger & Database backend. """ import os from flask import Flask, request, jsonify, send_from_directory from flask_cors import CORS from flask_socketio import SocketIO, emit, join_room, leave_room from werkzeug.utils import secure_filename import database as db import auth from joy_ai import ask_joy, describe_image, transcribe_audio from media_storage import upload_media from hf_video_storage import upload_large_file, download_and_delete import base64 import time import psutil import requests as http_requests import json import uuid import hashlib import zipfile import io from datetime import datetime, timezone app = Flask(__name__) CORS(app, resources={r"/*": {"origins": "*"}}, supports_credentials=True) app.config["SECRET_KEY"] = os.environ.get("FLASK_SECRET", "dev-secret") socketio = SocketIO(app, cors_allowed_origins="*", async_mode="eventlet") db.init_db() @app.route("/debug/jwt-secret-check") def debug_jwt(): s = os.environ.get("JWT_SECRET", "NOT SET") return jsonify({"secret_length": len(s), "first_6": s[:6], "last_6": s[-6:]}) # Status media (photo/video/voice/song attached to a 24h status) lives on the # Space's own persistent disk rather than a HF dataset repo, since it's # short-lived by design -- cleanup_expired_status_media() below deletes the # files as soon as their status row expires, so it never grows unbounded. STATUS_MEDIA_DIR = os.environ.get("STATUS_MEDIA_DIR", "/data/status_media") os.makedirs(STATUS_MEDIA_DIR, exist_ok=True) def save_status_media(file_bytes, filename, user_id): """Saves status media to local disk and returns its full public URL.""" ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "bin" safe_name = f"user{user_id}_{int(time.time())}_{os.urandom(4).hex()}.{ext}" with open(os.path.join(STATUS_MEDIA_DIR, safe_name), "wb") as f: f.write(file_bytes) return f"{request.host_url.rstrip('/')}/status/media/{safe_name}" def _remove_local_status_media(media_url): """Deletes the on-disk file for a status media_url, if it's a local one.""" if not media_url: return prefix = "/status/media/" if prefix not in media_url: return fname = os.path.basename(media_url.split(prefix, 1)[1]) fpath = os.path.join(STATUS_MEDIA_DIR, fname) if os.path.isfile(fpath): try: os.remove(fpath) except OSError: pass def cleanup_expired_status_media(): """Deletes DB rows + files for any status past its 24h expiry.""" try: for media_url in db.delete_expired_statuses(): _remove_local_status_media(media_url) except Exception as e: print(f"Status cleanup failed: {e}") online_users = {} def get_auth_user(): header = request.headers.get("Authorization", "") if not header.startswith("Bearer "): return None token = header.split(" ", 1)[1] payload = auth.verify_token(token) if not payload: return None return payload def require_auth(fn): from functools import wraps @wraps(fn) def wrapper(*args, **kwargs): payload = get_auth_user() if not payload: return jsonify({"error": "Unauthorized"}), 401 request.user_id = payload["sub"] request.username = payload["username"] return fn(*args, **kwargs) return wrapper @app.route("/auth/signup", methods=["POST"]) def signup(): data = request.get_json(force=True, silent=True) or {} username = (data.get("username") or "").strip().lower() password = data.get("password") or "" if not auth.is_valid_username(username): return jsonify({"error": "Username must be 3-20 characters, letters/numbers/underscore only"}), 400 if len(password) < 8: return jsonify({"error": "Password must be at least 8 characters"}), 400 if db.get_user_by_username(username): return jsonify({"error": "Username already taken"}), 409 handle = f"{username}@b24.me" avatar_letter = username[0].upper() password_hash = auth.hash_password(password) user_id = db.create_user(username, password_hash, handle, avatar_letter) if user_id is None: return jsonify({"error": "Username already taken"}), 409 token = auth.issue_token(user_id, username) return jsonify({ "token": token, "user": { "id": user_id, "username": username, "handle": handle, "avatar_letter": avatar_letter, "avatar_url": None, "bio": None, "banner": {"type": "color", "value": None}, "verified": None, "account_type": "personal" } }), 201 @app.route("/auth/login", methods=["POST"]) def login(): data = request.get_json(force=True, silent=True) or {} username = (data.get("username") or "").strip().lower() password = data.get("password") or "" user = db.get_user_by_username(username) if not user or not auth.verify_password(password, user["password_hash"]): return jsonify({"error": "Invalid username or password"}), 401 token = auth.issue_token(user["id"], user["username"]) return jsonify({ "token": token, "user": { "id": user["id"], "username": user["username"], "handle": user["handle"], "avatar_letter": user["avatar_letter"], "avatar_url": user["avatar_url"], "bio": user["bio"], "banner": {"type": user["banner_type"], "value": user["banner_value"]}, "verified": user["verified"], "account_type": user["account_type"] } }) @app.route("/users/", methods=["GET"]) @require_auth def get_user_public(username): """Public lookup by username — used by Add Friend search.""" user = db.get_user_by_username(username.lower().strip()) if not user: return jsonify({"error": "User not found"}), 404 return jsonify({ "id": user["id"], "username": user["username"], "handle": user["handle"], "avatar_letter": user["avatar_letter"], "account_type": user.get("account_type", "personal"), "verified": user.get("verified"), "avatar_url": db._avatar_or_hidden(user["id"], user.get("avatar_url"), request.user_id), "bio": user.get("bio"), }) @app.route("/friends/add", methods=["POST"]) @require_auth def add_friend(): data = request.get_json(force=True, silent=True) or {} target_username = (data.get("username") or "").strip().lower() friend_id, error = db.send_friend_request(request.user_id, target_username) if error: return jsonify({"error": error}), 400 return jsonify({"status": "requested", "friend_id": friend_id}) @app.route("/friends/respond", methods=["POST"]) @require_auth def respond_friend(): data = request.get_json(force=True, silent=True) or {} requester_id = data.get("requester_id") action = data.get("action", "accept") # 'accept', 'decline', or 'block' ok = db.respond_friend_request(request.user_id, requester_id, action) if not ok: return jsonify({"error": "Invalid action"}), 400 if action == "accept": db.maybe_award_friend_count_tasks(request.user_id) db.maybe_award_friend_count_tasks(requester_id) return jsonify({"status": action}) @app.route("/friends/block", methods=["POST"]) @require_auth def friends_block(): data = request.get_json(force=True, silent=True) or {} target_id = data.get("user_id") if not target_id: return jsonify({"error": "user_id is required"}), 400 db.block_user(request.user_id, target_id) return jsonify({"status": "blocked"}) @app.route("/friends/blocked", methods=["GET"]) @require_auth def friends_blocked_list(): return jsonify({"blocked": db.list_blocked_users(request.user_id)}) @app.route("/friends/remove", methods=["POST"]) @require_auth def friends_remove(): data = request.get_json(force=True, silent=True) or {} target_id = data.get("user_id") if not target_id: return jsonify({"error": "user_id is required"}), 400 db.remove_friend(request.user_id, target_id) return jsonify({"status": "removed"}) @app.route("/friends/list", methods=["GET"]) @require_auth def friends_list(): return jsonify({"friends": db.list_friends(request.user_id)}) @app.route("/friends/pending", methods=["GET"]) @require_auth def friends_pending(): return jsonify({"pending": db.list_pending_requests(request.user_id)}) @app.route("/friends/sent", methods=["GET"]) @require_auth def friends_sent(): return jsonify(db.list_sent_requests(request.user_id)) @app.route("/friends/cancel", methods=["POST"]) @require_auth def friends_cancel(): data = request.get_json(force=True, silent=True) or {} target_id = data.get("user_id") if not target_id: return jsonify({"error": "user_id is required"}), 400 ok = db.cancel_sent_request(request.user_id, target_id) if not ok: return jsonify({"error": "No pending request to cancel"}), 400 return jsonify({"status": "canceled"}) @app.route("/messages/", methods=["GET"]) @require_auth def get_messages(other_id): history = db.get_conversation(request.user_id, other_id) settings = db.get_privacy_settings(request.user_id) # Ghost mode: viewing does NOT mark read - the blue tick only appears once # this user actually sends a reply (see handle_send_message). if settings.get("read_receipts", True) and not settings.get("ghost_mode"): db.mark_read(request.user_id, other_id) return jsonify({"messages": history}) @app.route("/messages/pending", methods=["GET"]) @require_auth def messages_pending(): return jsonify({"messages": db.get_pending_messages(request.user_id)}) @app.route("/messages/ack", methods=["POST"]) @require_auth def messages_ack(): data = request.get_json(force=True, silent=True) or {} ids = data.get("ids") or [] deleted = db.ack_messages(request.user_id, ids) return jsonify({"deleted": deleted}) @app.route("/groups/create", methods=["POST"]) @require_auth def groups_create(): data = request.get_json(force=True, silent=True) or {} name = (data.get("name") or "").strip() description = (data.get("description") or "").strip() group_type = data.get("type", "group") member_ids = data.get("member_ids") or [] if not name: return jsonify({"error": "Group name required"}), 400 if group_type not in ("group", "channel", "broadcast"): group_type = "group" group_id, invite_code = db.create_group( name, request.user_id, member_ids, group_type=group_type, description=description ) return jsonify({ "group": { "id": group_id, "name": name, "description": description, "type": group_type, "invite_code": invite_code, } }), 201 @app.route("/groups//members", methods=["POST"]) @require_auth def groups_add_members(group_id): data = request.get_json(force=True, silent=True) or {} member_ids = data.get("member_ids") or [] if not member_ids: return jsonify({"error": "member_ids is required"}), 400 added = db.add_group_members(group_id, member_ids, request.user_id) if added is None: return jsonify({"error": "Not a member of this group"}), 403 return jsonify({"added": added}) @app.route("/groups/join/", methods=["POST"]) @require_auth def groups_join(invite_code): group = db.get_group_by_invite_code(invite_code) if not group: return jsonify({"error": "Invalid invite link"}), 404 added = db.add_group_members(group["id"], [request.user_id], request.user_id) if added is None: # requester isn't a member yet (expected) — add them directly via a member insert added = db.add_group_members(group["id"], [request.user_id], group["created_by"]) db.award_points(request.user_id, "first_group") return jsonify({"group_id": group["id"], "name": group["name"]}) @app.route("/groups/list", methods=["GET"]) @require_auth def groups_list(): return jsonify({"groups": db.list_user_groups(request.user_id)}) @app.route("/groups//messages", methods=["GET"]) @require_auth def group_messages(group_id): members = [m["user_id"] for m in db.get_group_members(group_id)] if request.user_id not in members: return jsonify({"error": "Not a member of this group"}), 403 return jsonify({"messages": db.get_group_conversation(group_id)}) @app.route("/groups/messages/pending", methods=["GET"]) @require_auth def group_messages_pending(): return jsonify({"messages": db.get_pending_group_messages(request.user_id)}) @app.route("/groups/messages/ack", methods=["POST"]) @require_auth def group_messages_ack(): data = request.get_json(force=True, silent=True) or {} ids = data.get("ids") or [] db.ack_group_messages(request.user_id, ids) return jsonify({"status": "ok"}) def send_push_notification(user_id, title, body, data=None): """Best-effort push via Expo's push service. Never raises - a failed push should never break message sending.""" token = db.get_push_token(user_id) if not token: return try: http_requests.post( "https://exp.host/--/api/v2/push/send", json={ "to": token, "title": title, "body": body, "sound": "default", "priority": "high", "channelId": "default", "data": data or {}, }, headers={"Content-Type": "application/json"}, timeout=5, ) except Exception: pass @app.route("/push/register", methods=["POST"]) @require_auth def push_register(): data = request.get_json(force=True, silent=True) or {} token = (data.get("push_token") or "").strip() if not token: return jsonify({"error": "push_token required"}), 400 db.set_push_token(request.user_id, token) return jsonify({"status": "ok"}) @socketio.on("connect") def on_connect(): token = request.args.get("token") payload = auth.verify_token(token) if token else None if not payload: return False online_users[payload["sub"]] = request.sid join_room(f"user_{payload['sub']}") @socketio.on("disconnect") def on_disconnect(): for uid, sid in list(online_users.items()): if sid == request.sid: db.touch_last_seen(uid) del online_users[uid] @socketio.on("send_message") def handle_send_message(data): token = data.get("token") payload = auth.verify_token(token) if token else None if not payload: emit("error", {"error": "Unauthorized"}) return sender_id = payload["sub"] recipient_id = data.get("recipient_id") content = (data.get("content") or "").strip() if not content: return media_type = data.get("media_type", "text") media_url = data.get("media_url") is_request = 0 if db.can_message_directly(sender_id, recipient_id) else 1 disappear_seconds = db.get_disappearing_duration(sender_id, recipient_id) expires_at = (int(time.time()) + disappear_seconds) if disappear_seconds else None msg_id = db.save_message(sender_id, recipient_id, content, media_type, media_url, is_request, expires_at) db.award_points(sender_id, "first_message") sender_settings = db.get_privacy_settings(sender_id) if sender_settings.get("ghost_mode") and sender_settings.get("read_receipts", True): # Ghost mode: the recipient's messages only get ticked read now, on reply. db.mark_read(sender_id, recipient_id) message_payload = { "id": msg_id, "sender_id": sender_id, "recipient_id": recipient_id, "content": content, "media_type": media_type, "media_url": media_url, "is_request": is_request } emit("new_message", message_payload, room=f"user_{recipient_id}") emit("new_message", message_payload, room=f"user_{sender_id}") if recipient_id != db.AI_USER_ID: sender_user = db.get_user_by_id(sender_id) sender_name = sender_user["username"] if sender_user else "Someone" push_body = content if media_type == "text" else f"Sent a {media_type}" send_push_notification( recipient_id, sender_name, push_body, data={"type": "message", "chatId": str(sender_id)} ) if recipient_id == db.AI_USER_ID: reply_text = ask_joy(sender_id, content) reply_id = db.save_message(db.AI_USER_ID, sender_id, reply_text) emit("new_message", { "id": reply_id, "sender_id": db.AI_USER_ID, "recipient_id": sender_id, "content": reply_text }, room=f"user_{sender_id}") @socketio.on("typing") def handle_typing(data): """data: { token, recipient_id }""" payload = auth.verify_token(data.get("token") or "") if not payload: return sender_id = payload["sub"] if db.get_privacy_settings(sender_id).get("ghost_mode"): return recipient_id = data.get("recipient_id") emit("typing", {"sender_id": sender_id}, room=f"user_{recipient_id}") @socketio.on("stop_typing") def handle_stop_typing(data): """data: { token, recipient_id }""" payload = auth.verify_token(data.get("token") or "") if not payload: return sender_id = payload["sub"] if db.get_privacy_settings(sender_id).get("ghost_mode"): return recipient_id = data.get("recipient_id") emit("stop_typing", {"sender_id": sender_id}, room=f"user_{recipient_id}") @socketio.on("send_group_message") def handle_send_group_message(data): token = data.get("token") payload = auth.verify_token(token) if token else None if not payload: emit("error", {"error": "Unauthorized"}) return sender_id = payload["sub"] group_id = data.get("group_id") content = (data.get("content") or "").strip() if not content: return members = [m["user_id"] for m in db.get_group_members(group_id)] if sender_id not in members: emit("error", {"error": "Not a member of this group"}) return media_type = data.get("media_type", "text") media_url = data.get("media_url") msg_id = db.save_group_message(group_id, sender_id, content, media_type, media_url) sender_user = db.get_user_by_id(sender_id) message_payload = { "id": msg_id, "group_id": group_id, "sender_id": sender_id, "content": content, "media_type": media_type, "media_url": media_url, "sender_username": sender_user["username"] if sender_user else None, "sender_name": sender_user["username"] if sender_user else None, "sender_color": db._color_for_id(sender_id), "sender_verified": sender_user["verified"] if sender_user else None, } for uid in members: emit("new_group_message", message_payload, room=f"user_{uid}") group_push_body = content if media_type == "text" else f"Sent a {media_type}" for uid in members: if uid != sender_id and uid != db.AI_USER_ID: send_push_notification( uid, message_payload["sender_name"] or "New message", group_push_body, data={"type": "group_message", "groupId": str(group_id)} ) if "@joy" in content.lower() and db.AI_USER_ID in members: reply_text = ask_joy(f"group_{group_id}", content) reply_id = db.save_group_message(group_id, db.AI_USER_ID, reply_text) for uid in members: emit("new_group_message", { "id": reply_id, "group_id": group_id, "sender_id": db.AI_USER_ID, "content": reply_text }, room=f"user_{uid}") @socketio.on("join_group") def handle_join_group(data): token = data.get("token") payload = auth.verify_token(token) if token else None if not payload: emit("error", {"error": "Unauthorized"}) return user_id = payload["sub"] group_id = data.get("group_id") members = [m["user_id"] for m in db.get_group_members(group_id)] if user_id not in members: emit("error", {"error": "Not a member of this group"}) return join_room(f"group_{group_id}") @socketio.on("leave_group") def handle_leave_group(data): token = data.get("token") payload = auth.verify_token(token) if token else None if not payload: return group_id = data.get("group_id") leave_room(f"group_{group_id}") @socketio.on("group_typing") def handle_group_typing(data): token = data.get("token") payload = auth.verify_token(token) if token else None if not payload: return user_id = payload["sub"] group_id = data.get("group_id") members = [m["user_id"] for m in db.get_group_members(group_id)] if user_id not in members: return user = db.get_user_by_id(user_id) for uid in members: if uid != user_id: emit("group_typing", { "group_id": group_id, "user_id": user_id, "username": user["username"] if user else None, }, room=f"user_{uid}") @socketio.on("group_stop_typing") def handle_group_stop_typing(data): token = data.get("token") payload = auth.verify_token(token) if token else None if not payload: return user_id = payload["sub"] group_id = data.get("group_id") members = [m["user_id"] for m in db.get_group_members(group_id)] for uid in members: if uid != user_id: emit("group_stop_typing", {"group_id": group_id, "user_id": user_id}, room=f"user_{uid}") # --------------------------------------------------------------------- # Admin (separate passkey — NOT the user JWT) # --------------------------------------------------------------------- ADMIN_PASSKEY = os.environ.get("ADMIN_PASSKEY") import time as _time _START_TIME = _time.time() def require_admin(fn): from functools import wraps @wraps(fn) def wrapper(*args, **kwargs): key = request.headers.get("X-Admin-Key") if not ADMIN_PASSKEY or key != ADMIN_PASSKEY: return jsonify({"error": "Forbidden"}), 403 return fn(*args, **kwargs) return wrapper @app.route("/admin/status", methods=["GET"]) @require_admin def admin_status(): return jsonify({ "uptime_seconds": int(_time.time() - _START_TIME), "online_users": len(online_users), "online_user_ids": list(online_users.keys()), "total_users": db.count_users(), "total_messages": db.count_messages(), "total_groups": db.count_groups(), "db_size_bytes": db.db_file_size_bytes(), "cpu_percent": psutil.cpu_percent(interval=0.3), "ram_percent": psutil.virtual_memory().percent, "ram_used_mb": round(psutil.virtual_memory().used / (1024 * 1024), 1), "ram_total_mb": round(psutil.virtual_memory().total / (1024 * 1024), 1), }) @app.route("/admin/purge-old-messages", methods=["POST"]) @require_admin def admin_purge_old_messages(): days = int(request.args.get("days", 30)) deleted_1on1 = db.purge_stale_messages(days) deleted_group = db.purge_stale_group_messages(days) return jsonify({"deleted_1on1": deleted_1on1, "deleted_group": deleted_group}) @app.route("/admin/users", methods=["GET"]) @require_admin def admin_users(): limit = int(request.args.get("limit", 100)) offset = int(request.args.get("offset", 0)) users = db.list_all_users(limit, offset) counts = db.get_report_counts() for u in users: u["report_count"] = counts.get(u["id"], 0) return jsonify({"users": users}) @app.route("/admin/set-account-type", methods=["POST"]) @require_admin def admin_set_account_type(): data = request.get_json(force=True, silent=True) or {} user_id = data.get("user_id") account_type = data.get("account_type") # 'personal', 'business', 'business_noreply' ok = db.set_account_type(user_id, account_type) if not ok: return jsonify({"error": "Invalid account_type"}), 400 return jsonify({"status": "updated", "user_id": user_id, "account_type": account_type}) @app.route("/admin/verify", methods=["POST"]) @require_admin def admin_verify(): data = request.get_json(force=True, silent=True) or {} user_id = data.get("user_id") tick = data.get("tick") # None to remove, 'purple' or 'cyan' to grant ok = db.set_verified(user_id, tick) if not ok: return jsonify({"error": "Invalid tick value — use null, 'purple', or 'cyan'"}), 400 return jsonify({"status": "updated", "user_id": user_id, "verified": tick}) @app.route("/report/user", methods=["POST"]) @require_auth def report_user(): data = request.get_json(force=True, silent=True) or {} reported_user_id = data.get("reported_user_id") reason = (data.get("reason") or "").strip() if not reported_user_id: return jsonify({"error": "reported_user_id is required"}), 400 db.create_report(request.user_id, reported_user_id, reason) return jsonify({"status": "reported"}) @app.route("/admin/reports/", methods=["GET"]) @require_admin def admin_reports(user_id): reports = db.get_reports_for_user(user_id) return jsonify({"reports": reports}) @app.route("/joy/ask", methods=["POST"]) @require_auth def joy_ask(): """Stateless JOY call for one-off use (e.g. search summaries), not tied to the chat history.""" data = request.get_json(force=True, silent=True) or {} prompt = (data.get("prompt") or "").strip() if not prompt: return jsonify({"error": "prompt is required"}), 400 reply = ask_joy(f"oneoff_{request.user_id}", prompt) return jsonify({"reply": reply}) @app.route("/files/upload", methods=["POST"]) @require_auth def files_upload(): """Large file (up to ~2GB) upload — stored in the shared Google Drive folder. Returns a lazy-bubble payload; the actual bytes are only pulled by the recipient when they tap to download.""" if "file" not in request.files: return jsonify({"error": "No file provided"}), 400 file = request.files["file"] filename = file.filename or "file" mime_type = file.mimetype or "application/octet-stream" try: file_bytes = file.read() drive_file = upload_large_file(file_bytes, filename, mime_type) except Exception as e: return jsonify({"error": f"Upload failed: {e}"}), 500 return jsonify({ "file_id": drive_file["id"], "filename": drive_file.get("name", filename), "size": len(file_bytes), "media_type": "file", }) @app.route("/files/download/", methods=["GET"]) @require_auth def files_download(file_id): """Recipient taps the lazy bubble -> this pulls the file from Drive, streams it back, then permanently deletes it from Drive.""" try: file_bytes, filename = download_and_delete(file_id) except Exception as e: return jsonify({"error": f"Download failed: {e}"}), 500 from flask import Response return Response( file_bytes, mimetype="application/octet-stream", headers={"Content-Disposition": f"attachment; filename=\"{filename}\""} ) @app.route("/media/upload", methods=["POST"]) @require_auth def media_upload(): """Uploads an image or voice note to the media dataset repo, returns its URL. If it's audio, also returns a transcription so the message has readable text.""" if "file" not in request.files: return jsonify({"error": "No file provided"}), 400 file = request.files["file"] media_kind = request.form.get("type", "image") # 'image' or 'voice' try: file_bytes = file.read() url = upload_media(file_bytes, file.filename or "upload.bin", request.user_id) except Exception as e: return jsonify({"error": f"Upload failed: {e}"}), 500 result = {"url": url, "media_type": media_kind} if media_kind == "voice": result["transcription"] = transcribe_audio(file_bytes, file.filename or "voice.m4a") return jsonify(result) @app.route("/joy/vision", methods=["POST"]) @require_auth def joy_vision(): """Accepts an uploaded image (multipart 'file'), returns JOY's confident description.""" if "file" not in request.files: return jsonify({"error": "No file provided"}), 400 file = request.files["file"] prompt = request.form.get("prompt") try: image_bytes = file.read() image_b64 = base64.b64encode(image_bytes).decode("utf-8") except Exception as e: return jsonify({"error": f"Could not read image: {e}"}), 400 description = describe_image(image_b64, prompt) return jsonify({"description": description}) # --------------------------------------------------------------------- # Voice calls — WebRTC signaling (audio only for now) # --------------------------------------------------------------------- ICE_SERVERS = [ {"urls": "stun:stun.relay.metered.ca:80"}, {"urls": "turn:openrelay.metered.ca:80", "username": "openrelayproject", "credential": "openrelayproject"}, {"urls": "turn:openrelay.metered.ca:443", "username": "openrelayproject", "credential": "openrelayproject"}, {"urls": "turn:openrelay.metered.ca:443?transport=tcp", "username": "openrelayproject", "credential": "openrelayproject"}, ] @app.route("/webrtc/ice-servers", methods=["GET"]) @require_auth def webrtc_ice_servers(): return jsonify({"iceServers": ICE_SERVERS}) @socketio.on("call_offer") def handle_call_offer(data): """data: { token, callee_id, call_type, offer }""" payload = auth.verify_token(data.get("token") or "") if not payload: emit("error", {"error": "Unauthorized"}) return caller_id = payload["sub"] caller = db.get_user_by_id(caller_id) callee_id = data.get("callee_id") call_type = data.get("call_type", "voice") call_id = db.create_call(caller_id, callee_id, call_type) emit("incoming_call", { "call_id": call_id, "caller_id": caller_id, "caller_username": caller["username"] if caller else "Unknown", "call_type": call_type, "offer": data.get("offer"), }, room=f"user_{callee_id}") @socketio.on("call_answer") def handle_call_answer(data): """data: { token, call_id, caller_id, answer }""" payload = auth.verify_token(data.get("token") or "") if not payload: emit("error", {"error": "Unauthorized"}) return db.update_call_status(data.get("call_id"), "answered") db.award_points(payload["sub"], "first_call") emit("call_answered", { "call_id": data.get("call_id"), "answer": data.get("answer"), }, room=f"user_{data.get('caller_id')}") @socketio.on("ice_candidate") def handle_ice_candidate(data): """data: { token, target_id, candidate }""" payload = auth.verify_token(data.get("token") or "") if not payload: return sender_id = payload["sub"] target_id = data.get("target_id") emit("ice_candidate", { "from_id": sender_id, "candidate": data.get("candidate"), }, room=f"user_{target_id}") @socketio.on("call_reject") def handle_call_reject(data): """data: { token, call_id, caller_id, reason }""" payload = auth.verify_token(data.get("token") or "") if not payload: return status = "missed" if data.get("reason") == "timeout" else "declined" db.update_call_status(data.get("call_id"), status) emit("call_rejected", { "call_id": data.get("call_id"), "status": status, }, room=f"user_{data.get('caller_id')}") @socketio.on("call_end") def handle_call_end(data): """data: { token, call_id, other_id }""" payload = auth.verify_token(data.get("token") or "") if not payload: return db.end_call(data.get("call_id")) emit("call_ended", { "call_id": data.get("call_id"), }, room=f"user_{data.get('other_id')}") @app.route("/calls/history", methods=["GET"]) @require_auth def call_history(): calls = db.get_call_history(request.user_id) return jsonify({"calls": calls}) # --------------------------------------------------------------------- # Stickers / GIFs (self-hosted via HF dataset repo — no external API needed) + Favorites # --------------------------------------------------------------------- STICKER_DATASET_REPO = os.environ.get("STICKER_DATASET_REPO", "Brighton233j/Media-image-voice") _sticker_list_cache = {"time": 0, "data": []} STICKER_LIST_CACHE_SECONDS = 600 @app.route("/stickers/packs", methods=["GET"]) @require_auth def stickers_packs(): """Lists all sticker files stored under stickers/ in the media dataset repo.""" now = time.time() if now - _sticker_list_cache["time"] < STICKER_LIST_CACHE_SECONDS: return jsonify({"cached": True, "packs": _sticker_list_cache["data"]}) try: from huggingface_hub import HfApi api = HfApi(token=os.environ.get("HF_TOKEN") or os.environ.get("Hf_token")) files = api.list_repo_files(STICKER_DATASET_REPO, repo_type="dataset") sticker_files = [f for f in files if f.startswith("stickers/")] packs = {} for path in sticker_files: parts = path.split("/") if len(parts) < 3: continue pack_name = parts[1] url = f"https://huggingface.co/datasets/{STICKER_DATASET_REPO}/resolve/main/{path}" packs.setdefault(pack_name, []).append(url) result = [{"pack": name, "stickers": urls} for name, urls in packs.items()] _sticker_list_cache["time"] = now _sticker_list_cache["data"] = result return jsonify({"cached": False, "packs": result}) except Exception as e: return jsonify({"error": f"Could not list stickers: {e}"}), 500 @app.route("/favorites/add", methods=["POST"]) @require_auth def favorites_add(): data = request.get_json(force=True, silent=True) or {} url = data.get("url") kind = data.get("kind", "gif") if not url: return jsonify({"error": "url is required"}), 400 db.add_favorite(request.user_id, url, kind) return jsonify({"status": "added"}) @app.route("/favorites/remove", methods=["POST"]) @require_auth def favorites_remove(): data = request.get_json(force=True, silent=True) or {} url = data.get("url") db.remove_favorite(request.user_id, url) return jsonify({"status": "removed"}) @app.route("/favorites/list", methods=["GET"]) @require_auth def favorites_list(): return jsonify({"favorites": db.list_favorites(request.user_id)}) @app.route("/status/post", methods=["POST"]) @require_auth def status_post(): """Accepts multipart form for photo/video/voice/song (file field optional), or just form fields for a text status. privacy_user_ids is a comma-separated list of user ids, used when privacy is 'except' or 'only'.""" content_type = request.form.get("content_type", "text") text_content = request.form.get("text_content") bg_color = request.form.get("bg_color") privacy = request.form.get("privacy", "all") raw_ids = request.form.get("privacy_user_ids") privacy_user_ids = [int(x) for x in raw_ids.split(",") if x] if raw_ids else [] if content_type not in ("photo", "video", "text", "voice", "song"): return jsonify({"error": "Invalid content_type"}), 400 if privacy not in ("all", "except", "only"): return jsonify({"error": "Invalid privacy"}), 400 cleanup_expired_status_media() media_url = None if "file" in request.files: file = request.files["file"] try: file_bytes = file.read() media_url = save_status_media(file_bytes, file.filename or "status.bin", request.user_id) except Exception as e: return jsonify({"error": f"Upload failed: {e}"}), 500 status_id = db.create_status( request.user_id, content_type, media_url=media_url, text_content=text_content, bg_color=bg_color, privacy=privacy, privacy_user_ids=privacy_user_ids ) db.award_points(request.user_id, "first_status") return jsonify({"status_id": status_id}), 201 @app.route("/status/feed", methods=["GET"]) @require_auth def status_feed(): cleanup_expired_status_media() return jsonify({ "my_statuses": db.get_my_statuses(request.user_id), "feed": db.get_status_feed(request.user_id), }) @app.route("/status/media/", methods=["GET"]) def status_media_serve(filename): safe_name = secure_filename(filename) return send_from_directory(STATUS_MEDIA_DIR, safe_name, conditional=True) @app.route("/status/view", methods=["POST"]) @require_auth def status_view(): data = request.get_json(force=True, silent=True) or {} status_id = data.get("status_id") if not status_id: return jsonify({"error": "status_id is required"}), 400 db.record_status_view(status_id, request.user_id) return jsonify({"status": "viewed"}) @app.route("/status//viewers", methods=["GET"]) @require_auth def status_viewers(status_id): viewers = db.get_status_viewers(status_id, request.user_id) if viewers is None: return jsonify({"error": "Not found or not yours"}), 404 return jsonify({"viewers": viewers, "count": len(viewers)}) @app.route("/status/", methods=["DELETE"]) @require_auth def status_delete(status_id): result = db.delete_status(status_id, request.user_id) if not result: return jsonify({"error": "Not found or not yours"}), 404 _remove_local_status_media(result.get("media_url")) return jsonify({"status": "deleted"}) @app.route("/chats", methods=["GET"]) @require_auth def get_chats(): return jsonify({"chats": db.get_chat_list(request.user_id)}) @app.route("/search", methods=["GET"]) @require_auth def search(): q = (request.args.get("q") or "").strip() if not q: return jsonify({"friends": [], "messages": []}) friends, messages = db.search_content(request.user_id, q) return jsonify({"friends": friends, "messages": messages}) @app.route("/") def status(): return jsonify({"status": "B24 Messenger & Database running", "version": "1.0.0"}) @app.route("/chats//mute", methods=["POST"]) @require_auth def chat_mute(chat_id): data = request.get_json(force=True, silent=True) or {} muted = bool(data.get("muted", True)) is_group = bool(data.get("is_group", False)) db.set_chat_mute(request.user_id, chat_id, muted, is_group) return jsonify({"status": "muted" if muted else "unmuted"}) @app.route("/chats//disappearing", methods=["POST"]) @require_auth def chat_disappearing(chat_id): data = request.get_json(force=True, silent=True) or {} seconds = data.get("seconds") seconds = int(seconds) if seconds else 0 db.set_disappearing_duration(request.user_id, chat_id, seconds) return jsonify({"status": "ok", "seconds": seconds}) @app.route("/chats//disappearing", methods=["GET"]) @require_auth def chat_disappearing_get(chat_id): seconds = db.get_disappearing_duration(request.user_id, chat_id) return jsonify({"seconds": seconds or 0}) @app.route("/groups//members/detailed", methods=["GET"]) @require_auth def group_members_detailed(group_id): members = [m["user_id"] for m in db.get_group_members(group_id)] if request.user_id not in members: return jsonify({"error": "Not a member of this group"}), 403 return jsonify({ "members": db.get_group_members_detailed(group_id, request.user_id), "permissions": db.get_group_permissions(group_id), }) @app.route("/groups//members//promote", methods=["POST"]) @require_auth def group_member_promote(group_id, member_id): ok, error = db.promote_member(group_id, member_id, request.user_id) if not ok: return jsonify({"error": error}), 403 return jsonify({"status": "promoted"}) @app.route("/groups//members//demote", methods=["POST"]) @require_auth def group_member_demote(group_id, member_id): ok, error = db.demote_member(group_id, member_id, request.user_id) if not ok: return jsonify({"error": error}), 403 return jsonify({"status": "demoted"}) @app.route("/groups//members/", methods=["DELETE"]) @require_auth def group_member_remove(group_id, member_id): ok, error = db.remove_member(group_id, member_id, request.user_id) if not ok: return jsonify({"error": error}), 403 return jsonify({"status": "removed"}) @app.route("/groups//leave", methods=["POST"]) @require_auth def group_leave(group_id): db.leave_group(group_id, request.user_id) return jsonify({"status": "left"}) @app.route("/groups/", methods=["DELETE"]) @require_auth def group_delete(group_id): ok = db.delete_group(group_id, request.user_id) if not ok: return jsonify({"error": "Only the owner can delete this group"}), 403 return jsonify({"status": "deleted"}) @app.route("/groups//permissions", methods=["POST"]) @require_auth def group_permissions(group_id): data = request.get_json(force=True, silent=True) or {} ok = db.update_group_permissions( group_id, request.user_id, send_perm=data.get("send_perm"), edit_perm=data.get("edit_perm") ) if not ok: return jsonify({"error": "Not authorized"}), 403 return jsonify({"status": "updated"}) @app.route("/report/group", methods=["POST"]) @require_auth def report_group(): data = request.get_json(force=True, silent=True) or {} group_id = data.get("group_id") reason = (data.get("reason") or "").strip() if not group_id: return jsonify({"error": "group_id is required"}), 400 db.create_group_report(request.user_id, group_id, reason) return jsonify({"status": "reported"}) @app.route("/groups//invite", methods=["GET"]) @require_auth def group_invite(group_id): info = db.get_group_invite(group_id, request.user_id) if not info: return jsonify({"error": "Not a member of this group"}), 403 return jsonify(info) @app.route("/groups//messages", methods=["DELETE"]) @require_auth def group_clear_messages(group_id): ok = db.clear_group_messages(group_id, request.user_id) if not ok: return jsonify({"error": "Only admins can clear group history"}), 403 return jsonify({"status": "cleared"}) @app.route("/points/tasks", methods=["GET"]) @require_auth def points_tasks(): return jsonify(db.get_points_summary(request.user_id)) @app.route("/admin/tasks", methods=["GET"]) @require_admin def admin_tasks_list(): return jsonify({"tasks": db.list_all_tasks()}) @app.route("/admin/tasks", methods=["POST"]) @require_admin def admin_tasks_create(): data = request.get_json(force=True, silent=True) or {} key = (data.get("key") or "").strip() label = (data.get("label") or "").strip() points = data.get("points") category = (data.get("category") or "General").strip() if not key or not label or not isinstance(points, int) or points <= 0: return jsonify({"error": "key, label, and a positive integer points are required"}), 400 task_id = db.create_task(key, label, points, category) if task_id is None: return jsonify({"error": "A task with that key already exists"}), 409 return jsonify({"id": task_id, "key": key, "label": label, "points": points, "category": category}), 201 @app.route("/admin/tasks//toggle", methods=["POST"]) @require_admin def admin_tasks_toggle(task_id): new_state = db.toggle_task_active(task_id) if new_state is None: return jsonify({"error": "Task not found"}), 404 return jsonify({"id": task_id, "active": new_state}) @app.route("/admin/tasks//award", methods=["POST"]) @require_admin def admin_tasks_award(task_id): data = request.get_json(force=True, silent=True) or {} user_id = data.get("user_id") if not user_id: return jsonify({"error": "user_id is required"}), 400 ok, error = db.admin_award_task(task_id, user_id) if ok is None: return jsonify({"error": error}), 404 if ok is False: return jsonify({"status": "already_awarded", "detail": error}) return jsonify({"status": "awarded"}) @app.route("/admin/announcements", methods=["POST"]) @require_admin def admin_announcements_create(): data = request.get_json(force=True, silent=True) or {} title = (data.get("title") or "").strip() body = (data.get("body") or "").strip() or None expires_in_hours = data.get("expires_in_hours", 72) if not title: return jsonify({"error": "title is required"}), 400 announcement_id = db.create_announcement(title, body, expires_in_hours) return jsonify({"id": announcement_id, "title": title, "body": body}), 201 @app.route("/admin/announcements", methods=["GET"]) @require_admin def admin_announcements_list(): return jsonify({"announcements": db.list_announcements()}) @app.route("/admin/announcements//deactivate", methods=["POST"]) @require_admin def admin_announcements_deactivate(announcement_id): ok = db.deactivate_announcement(announcement_id) if not ok: return jsonify({"error": "Announcement not found"}), 404 return jsonify({"status": "deactivated"}) @app.route("/announcements/active", methods=["GET"]) @require_auth def announcements_active(): announcement = db.get_active_announcement() return jsonify({"announcement": announcement}) @app.route("/messages/", methods=["DELETE"]) @require_auth def messages_delete(other_id): db.clear_direct_messages(request.user_id, other_id) return jsonify({"status": "deleted"}) def _parse_version(v): """'1.2.10' -> (1, 2, 10). Non-numeric parts are treated as 0 so odd strings don't crash the comparison.""" parts = [] for p in (v or "").strip().split("."): digits = "".join(ch for ch in p if ch.isdigit()) parts.append(int(digits) if digits else 0) return tuple(parts) or (0,) def _version_is_newer(latest, current): if not latest: return False if not current: return True a, b = _parse_version(latest), _parse_version(current) length = max(len(a), len(b)) a = a + (0,) * (length - len(a)) b = b + (0,) * (length - len(b)) return a > b @app.route("/admin/app-version", methods=["POST"]) @require_admin def admin_set_app_version(): data = request.get_json(force=True, silent=True) or {} version = (data.get("version") or "").strip() apk_url = (data.get("apk_url") or "").strip() notes = data.get("notes") if not version or not apk_url: return jsonify({"error": "version and apk_url are required"}), 400 db.set_latest_app_version(version, apk_url, notes) return jsonify({"status": "updated", "version": version, "apk_url": apk_url}) APP_UPDATE_DATASET_REPO = os.environ.get("APP_UPDATE_DATASET_REPO", "Brighton233j/APP_UPDATES") @app.route("/admin/upload-apk", methods=["POST"]) @require_admin def admin_upload_apk(): """Admin uploads a new .apk (multipart 'file' + form fields 'version', optional 'notes'). Pushes it to the HF dataset repo under releases//, then records it as the latest version -- same one-step flow as the sticker/media dataset repos.""" if "file" not in request.files: return jsonify({"error": "No file provided"}), 400 file = request.files["file"] version = (request.form.get("version") or "").strip() notes = request.form.get("notes") if not version: return jsonify({"error": "version is required"}), 400 filename = file.filename or "app-release.apk" try: file_bytes = file.read() from huggingface_hub import HfApi api = HfApi(token=os.environ.get("HF_TOKEN") or os.environ.get("Hf_token")) path_in_repo = f"releases/{version}/{filename}" api.upload_file( path_or_fileobj=file_bytes, path_in_repo=path_in_repo, repo_id=APP_UPDATE_DATASET_REPO, repo_type="dataset", ) apk_url = f"https://huggingface.co/datasets/{APP_UPDATE_DATASET_REPO}/resolve/main/{path_in_repo}" except Exception as e: return jsonify({"error": f"Upload failed: {e}"}), 500 db.set_latest_app_version(version, apk_url, notes) return jsonify({ "status": "updated", "version": version, "apk_url": apk_url, "size": len(file_bytes), }) def _content_type_for_ext(ext): ext = (ext or "").lower().lstrip(".") return { "png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg", "gif": "image/gif", "webp": "image/webp", "svg": "image/svg+xml", "ttf": "font/ttf", "otf": "font/otf", "woff": "font/woff", "woff2": "font/woff2", "json": "application/json", }.get(ext, "application/octet-stream") def _asset_key(file_bytes): """Stable per-asset id (md5 of content) -- mirrors the 'key' field Expo's own export tooling uses for on-device asset caching.""" return hashlib.md5(file_bytes).hexdigest() def _sha256_base64url(file_bytes): digest = hashlib.sha256(file_bytes).digest() return base64.urlsafe_b64encode(digest).decode("utf-8").rstrip("=") def _uuid_from_update_id(update_id): """Deterministic UUID derived from our own update_id, in the same spirit as Expo's reference server's convertSHA256HashToUUID().""" hex_digest = hashlib.sha256(update_id.encode("utf-8")).hexdigest() return str(uuid.UUID(hex_digest[:32])) @app.route("/admin/upload-ota", methods=["POST"]) @require_admin def admin_upload_ota(): """Admin uploads a zip of the dist/ folder from `npx expo export --platform android --output-dir dist` (multipart 'file' + form fields 'runtime_version', optional 'notes'). Every file inside gets pushed to the APP_UPDATES dataset repo under ota///..., then recorded as the latest OTA update -- same one-step upload-and-record flow as admin_upload_apk.""" if "file" not in request.files: return jsonify({"error": "No file provided"}), 400 file = request.files["file"] runtime_version = (request.form.get("runtime_version") or "").strip() notes = request.form.get("notes") if not runtime_version: return jsonify({"error": "runtime_version is required"}), 400 update_id = str(uuid.uuid4()) try: zf = zipfile.ZipFile(io.BytesIO(file.read())) names = zf.namelist() meta_name = next((n for n in names if n.endswith("metadata.json")), None) if not meta_name: return jsonify({"error": "metadata.json not found -- zip the dist/ folder produced by expo export"}), 400 metadata = json.loads(zf.read(meta_name)) base_dir = meta_name[: -len("metadata.json")] from huggingface_hub import HfApi api = HfApi(token=os.environ.get("HF_TOKEN") or os.environ.get("Hf_token")) def upload_asset(rel_path): data = zf.read(base_dir + rel_path) path_in_repo = f"ota/{runtime_version}/{update_id}/{rel_path}" api.upload_file( path_or_fileobj=data, path_in_repo=path_in_repo, repo_id=APP_UPDATE_DATASET_REPO, repo_type="dataset", ) url = f"https://huggingface.co/datasets/{APP_UPDATE_DATASET_REPO}/resolve/main/{path_in_repo}" return url, data android_meta = metadata["fileMetadata"]["android"] bundle_url, bundle_bytes = upload_asset(android_meta["bundle"]) launch_asset = { "hash": _sha256_base64url(bundle_bytes), "key": _asset_key(bundle_bytes), "contentType": "application/javascript", "url": bundle_url, } assets = [] for a in android_meta.get("assets", []): url, data = upload_asset(a["path"]) assets.append({ "hash": _sha256_base64url(data), "key": _asset_key(data), "contentType": _content_type_for_ext(a.get("ext")), "url": url, }) db.set_latest_ota_update(update_id, runtime_version, json.dumps(launch_asset), json.dumps(assets), notes) except KeyError as e: return jsonify({"error": f"Unexpected metadata.json shape, missing {e}"}), 400 except Exception as e: return jsonify({"error": f"OTA upload failed: {e}"}), 500 return jsonify({ "status": "updated", "update_id": update_id, "runtime_version": runtime_version, "asset_count": len(assets) + 1, }) @app.route("/api/manifest", methods=["GET"]) def api_manifest(): """The endpoint expo-updates itself calls (set as `updates.url` in app.json) -- separate from /app/version below, which only powers the manual 'Check for updates' button's native-vs-OTA decision. Implements the Expo Updates protocol's plain-JSON response path (no code signing; protocol version 1's 'no update' case is a 406, per spec).""" protocol_version = request.headers.get("expo-protocol-version", "0") platform = request.headers.get("expo-platform") or request.args.get("platform") runtime_version = request.headers.get("expo-runtime-version") or request.args.get("runtime-version") if platform != "android": return jsonify({"error": "Unsupported platform. Expected android."}), 400 if not runtime_version: return jsonify({"error": "No runtimeVersion provided."}), 400 update = db.get_latest_ota_update() if not update or update.get("runtime_version") != runtime_version: return jsonify({"error": "No update available for this runtime version."}), 404 manifest_id = _uuid_from_update_id(update["update_id"]) current_update_id = request.headers.get("expo-current-update-id") if protocol_version == "1" and current_update_id == manifest_id: return jsonify({"error": "No update available."}), 406 created_at = datetime.fromtimestamp(update["created_at"], tz=timezone.utc).isoformat() manifest = { "id": manifest_id, "createdAt": created_at, "runtimeVersion": runtime_version, "launchAsset": json.loads(update["launch_asset_json"]), "assets": json.loads(update["assets_json"]), "metadata": {}, "extra": {}, } resp = jsonify(manifest) resp.headers["expo-protocol-version"] = protocol_version resp.headers["expo-sfv-version"] = "0" resp.headers["cache-control"] = "private, max-age=0" return resp @app.route("/app/version", methods=["GET"]) @require_auth def app_version(): """Pass ?current_version=1.2.0 (native app version) and ?current_runtime_version=... (expo-updates runtime version) and the backend tells you which kind of update -- if any -- is waiting: 'native' (needs the full-APK flow) or 'ota' (safe to hot-load via expo-updates' own checkForUpdateAsync, which talks to /api/manifest).""" current_version = request.args.get("current_version") current_runtime_version = request.args.get("current_runtime_version") native_info = db.get_latest_app_version() if native_info and _version_is_newer(native_info.get("version"), current_version): info = dict(native_info) info["update_available"] = True info["update_type"] = "native" return jsonify(info) ota_info = db.get_latest_ota_update() if ota_info and current_runtime_version and ota_info.get("runtime_version") == current_runtime_version: return jsonify({ "version": None, "apk_url": None, "notes": ota_info.get("notes"), "update_available": True, "update_type": "ota", }) return jsonify({"version": None, "apk_url": None, "notes": None, "update_available": False, "update_type": "none"}) @app.route("/profile/avatar", methods=["POST"]) @require_auth def profile_avatar(): data = request.get_json(force=True, silent=True) or {} avatar_url = data.get("avatar_url") if not avatar_url: return jsonify({"error": "avatar_url is required"}), 400 db.set_avatar(request.user_id, avatar_url) return jsonify({"status": "ok", "avatar_url": avatar_url}) @app.route("/profile/username", methods=["POST"]) @require_auth def profile_username(): data = request.get_json(force=True, silent=True) or {} ok, error = db.set_username(request.user_id, data.get("username")) if not ok: return jsonify({"error": error}), 400 return jsonify({"status": "ok"}) @app.route("/profile/bio", methods=["POST"]) @require_auth def profile_bio(): data = request.get_json(force=True, silent=True) or {} db.set_bio(request.user_id, data.get("bio")) return jsonify({"status": "ok"}) @app.route("/profile/banner", methods=["POST"]) @require_auth def profile_banner(): data = request.get_json(force=True, silent=True) or {} banner_type = data.get("type", "color") banner_value = data.get("value") if not db.set_banner(request.user_id, banner_type, banner_value): return jsonify({"error": "invalid banner type"}), 400 return jsonify({"status": "ok", "type": banner_type, "value": banner_value}) @app.route("/auth/logout", methods=["POST"]) @require_auth def auth_logout(): return jsonify({"status": "ok"}) @app.route("/privacy/settings", methods=["GET"]) @require_auth def get_privacy_settings_route(): return jsonify(db.get_privacy_settings(request.user_id)) @app.route("/privacy/settings", methods=["POST"]) @require_auth def set_privacy_settings_route(): data = request.get_json(force=True, silent=True) or {} updated = db.set_privacy_settings(request.user_id, data) return jsonify(updated) @app.route("/presence/", methods=["GET"]) @require_auth def get_presence(target_id): target = db.get_user_by_id(target_id) if not target: return jsonify({"error": "User not found"}), 404 settings = db.get_privacy_settings(target_id) if settings.get("ghost_mode") or not settings.get("last_seen", True): return jsonify({"online": False, "last_seen_at": None, "hidden": True}) is_online = target_id in online_users last_seen_at = target.get("last_seen_at") if settings.get("freeze_last_seen") and target.get("frozen_last_seen_at"): last_seen_at = target.get("frozen_last_seen_at") return jsonify({"online": is_online, "last_seen_at": last_seen_at, "hidden": False}) @app.route("/privacy/frozen-last-seen", methods=["POST"]) @require_auth def set_frozen_last_seen_route(): """Body: { frozen_at: } to pick a fixed date, or { frozen_at: null } to clear it.""" data = request.get_json(force=True, silent=True) or {} frozen_at = data.get("frozen_at") db.set_frozen_last_seen(request.user_id, frozen_at) return jsonify({"status": "ok", "frozen_at": frozen_at}) if __name__ == "__main__": port = int(os.environ.get("PORT", 7860)) socketio.run(app, host="0.0.0.0", port=port)