import os from datetime import datetime, timezone from typing import Any, Callable, Dict, List, Optional from modules.logger_tool import initialise_logger from modules.database.supabase.utils.client import SupabaseServiceRoleClient import modules.database.tools.neo4j_driver_tools as driver_tools logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True) MembershipRow = Dict[str, Any] GraphProbe = Callable[..., Dict[str, Any]] ROLE_PERMISSIONS: Dict[str, Dict[str, bool]] = { "school_admin": { "can_manage_school": True, "can_manage_calendar": True, "can_manage_timetable": True, "can_invite_staff": True, "can_manage_classes": True, "can_view_student_data": True, }, "department_head": { "can_manage_school": False, "can_manage_calendar": False, "can_manage_timetable": True, "can_invite_staff": False, "can_manage_classes": True, "can_view_student_data": True, }, "teacher": { "can_manage_school": False, "can_manage_calendar": False, "can_manage_timetable": False, "can_invite_staff": False, "can_manage_classes": True, "can_view_student_data": False, }, "student": { "can_manage_school": False, "can_manage_calendar": False, "can_manage_timetable": False, "can_invite_staff": False, "can_manage_classes": False, "can_view_student_data": False, }, } BASE_PERMISSIONS: Dict[str, bool] = { "platform_admin": False, "platform_super_admin": False, "can_create_school": True, "can_manage_school": False, "can_manage_calendar": False, "can_manage_timetable": False, "can_invite_staff": False, "can_manage_classes": False, "can_view_student_data": False, "can_use_canvas": True, } def utc_now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def _safe_list(result: Any) -> List[Dict[str, Any]]: data = getattr(result, "data", None) if not data: return [] if isinstance(data, list): return data return [data] def _safe_one(result: Any) -> Optional[Dict[str, Any]]: data = getattr(result, "data", None) return data if isinstance(data, dict) else None def default_graph_probe(user_id: str, user_email: str, active_institute: Optional[Dict[str, Any]], institute: Optional[Dict[str, Any]]) -> Dict[str, Any]: """Best-effort derived graph status; never authoritative for bootstrap state.""" user_db = f"cc.users.teacher.{user_id.replace('-', '')}" if user_id else None neo4j_uuid = (institute or {}).get("neo4j_uuid_string") institute_db = f"cc.institutes.{neo4j_uuid}" if neo4j_uuid else None status = { "available": False, "user_db": user_db, "institute_db": institute_db, "projection_state": "unknown", "needs_rebuild": True, "last_checked_at": utc_now_iso(), "error_code": None, } try: if not user_db and not institute_db: status["projection_state"] = "missing" return status db_to_check = institute_db or user_db with driver_tools.get_session(database=db_to_check) as session: session.run("RETURN 1 AS ok").single() status.update({"available": True, "projection_state": "ready", "needs_rebuild": False}) return status except Exception: status.update({"available": False, "projection_state": "error", "needs_rebuild": True, "error_code": "neo4j_unavailable"}) return status class BootstrapService: def __init__(self, supabase: Any, graph_probe: Optional[GraphProbe] = None): self.supabase = supabase self.graph_probe = graph_probe or default_graph_probe def build(self, credentials: Dict[str, Any]) -> Dict[str, Any]: user_id = credentials.get("sub") or "" user_email = credentials.get("email") or "" if not user_id: raise ValueError("Missing authenticated user id") profile = self._get_profile(user_id, user_email) admin_profile = self._get_admin_profile(user_id) memberships = self._get_memberships(user_id) active = self._select_active_institute(profile, memberships) active_institute = self._membership_institute(active, memberships) permissions = self._permissions(active, bool(admin_profile), bool((admin_profile or {}).get("is_super_admin"))) profile_user_type = "platform_admin" if admin_profile else (profile.get("user_type") or active.get("role") or "unknown") school_status = self._school_status(active, memberships, admin_profile) calendar_status = self._calendar_status(active.get("institute_id")) timetable_status = self._timetable_status(user_id, active.get("institute_id")) graph_status = self._graph_status(user_id, user_email, active, active_institute) onboarding = self._onboarding(school_status, permissions, calendar_status, timetable_status, graph_status) return { "profile": { "id": user_id, "email": profile.get("email") or user_email, "display_name": profile.get("display_name") or profile.get("full_name") or profile.get("username") or user_email, "user_type": profile_user_type, "school_id": profile.get("school_id"), }, "memberships": [self._serialize_membership(row) for row in memberships], "active_institute": { "id": active.get("institute_id"), "source": active.get("source", "none"), "membership_role": active.get("role"), }, "permissions": permissions, "school_status": school_status, "onboarding": onboarding, "calendar_status": calendar_status, "timetable_status": timetable_status, "graph_status": graph_status, } def _get_profile(self, user_id: str, user_email: str) -> Dict[str, Any]: try: result = ( self.supabase.table("profiles") .select("id,email,user_type,username,full_name,display_name,user_db_name,school_db_name,school_id") .eq("id", user_id) .single() .execute() ) profile = _safe_one(result) or {} except Exception as exc: logger.warning("Bootstrap profile lookup failed for authenticated user: %s", exc) profile = {} profile.setdefault("id", user_id) profile.setdefault("email", user_email) return profile def _get_admin_profile(self, user_id: str) -> Optional[Dict[str, Any]]: try: result = ( self.supabase.table("admin_profiles") .select("id,admin_role,is_super_admin") .eq("id", user_id) .single() .execute() ) return _safe_one(result) except Exception: return None def _get_memberships(self, user_id: str) -> List[MembershipRow]: try: member_rows = _safe_list( self.supabase.table("institute_memberships") .select("profile_id,institute_id,role") .eq("profile_id", user_id) .execute() ) except Exception as exc: logger.warning("Bootstrap membership lookup failed: %s", exc) return [] institute_ids = [str(row.get("institute_id")) for row in member_rows if row.get("institute_id")] institutes_by_id: Dict[str, Dict[str, Any]] = {} if institute_ids: try: inst_rows = _safe_list( self.supabase.table("institutes") .select("id,name,urn,website,address,metadata,status,neo4j_uuid_string") .in_("id", institute_ids) .execute() ) institutes_by_id = {str(row.get("id")): row for row in inst_rows} except Exception as exc: logger.warning("Bootstrap institute lookup failed: %s", exc) rows: List[MembershipRow] = [] for member in member_rows: inst = institutes_by_id.get(str(member.get("institute_id")), {}) if inst and inst.get("status") not in (None, "active"): continue rows.append({**member, "institute": inst, "status": member.get("status") or "active", "is_active": True}) return rows def _select_active_institute(self, profile: Dict[str, Any], memberships: List[MembershipRow]) -> MembershipRow: active_memberships = [m for m in memberships if m.get("is_active")] profile_school_id = profile.get("school_id") if profile_school_id: for member in active_memberships: if str(member.get("institute_id")) == str(profile_school_id): return {**member, "source": "profile.school_id"} if len(active_memberships) == 1: return {**active_memberships[0], "source": "single_membership"} return {"institute_id": None, "role": None, "source": "none"} def _membership_institute(self, active: MembershipRow, memberships: List[MembershipRow]) -> Optional[Dict[str, Any]]: active_id = active.get("institute_id") if not active_id: return None for member in memberships: if str(member.get("institute_id")) == str(active_id): return member.get("institute") or None return None def _serialize_membership(self, row: MembershipRow) -> Dict[str, Any]: inst = row.get("institute") or {} return { "institute_id": row.get("institute_id"), "role": row.get("role") or "teacher", "status": row.get("status") or "active", "is_active": bool(row.get("is_active", True)), "institute": { "id": inst.get("id") or row.get("institute_id"), "name": inst.get("name"), "urn": inst.get("urn"), "website": inst.get("website"), "address": inst.get("address") or {}, "metadata": inst.get("metadata") or {}, }, } def _permissions(self, active: MembershipRow, platform_admin: bool, super_admin: bool) -> Dict[str, bool]: permissions = dict(BASE_PERMISSIONS) permissions["platform_admin"] = platform_admin permissions["platform_super_admin"] = super_admin role_permissions = ROLE_PERMISSIONS.get(active.get("role") or "", {}) permissions.update(role_permissions) if platform_admin: # Platform authority is additive and must not be reduced by a user's # school membership role (for example a platform admin who also has # a teacher/student membership). permissions.update({ "can_create_school": True, "can_manage_school": True, "can_manage_calendar": True, "can_manage_timetable": True, "can_invite_staff": True, "can_manage_classes": True, "can_view_student_data": True, }) return permissions def _school_status(self, active: MembershipRow, memberships: List[MembershipRow], admin_profile: Optional[Dict[str, Any]]) -> str: if admin_profile: return "platform_admin" if active.get("institute_id"): return "school_admin" if active.get("role") == "school_admin" else "member" if len([m for m in memberships if m.get("is_active")]) > 1: return "multi_school_needs_selection" return "no_school" def _calendar_status(self, institute_id: Optional[str]) -> Dict[str, Any]: status = {"available": False, "academic_year_count": 0, "term_count": 0, "current_academic_year_id": None, "needs_setup": True} if not institute_id: return status try: years = _safe_list(self.supabase.table("academic_years").select("id,institute_id").eq("institute_id", institute_id).execute()) terms = _safe_list(self.supabase.table("academic_terms").select("id,institute_id").eq("institute_id", institute_id).execute()) status["academic_year_count"] = len(years) status["term_count"] = len(terms) status["current_academic_year_id"] = years[0].get("id") if years else None status["available"] = bool(years and terms) status["needs_setup"] = not status["available"] except Exception as exc: logger.warning("Bootstrap calendar status lookup failed: %s", exc) return status def _timetable_status(self, user_id: str, institute_id: Optional[str]) -> Dict[str, Any]: status = {"available": False, "teacher_timetable_id": None, "slot_count": 0, "needs_setup": True} if not institute_id: return status try: timetables = _safe_list( self.supabase.table("teacher_timetables") .select("id,profile_id,institute_id") .eq("institute_id", institute_id) .eq("profile_id", user_id) .limit(1) .execute() ) if not timetables: # Test/future compatibility for teacher_profile_id naming. timetables = _safe_list( self.supabase.table("teacher_timetables") .select("id,teacher_profile_id,institute_id") .eq("institute_id", institute_id) .eq("teacher_profile_id", user_id) .limit(1) .execute() ) if timetables: timetable_id = timetables[0].get("id") slots = _safe_list( self.supabase.table("teacher_timetable_slots") .select("id,teacher_timetable_id") .eq("teacher_timetable_id", timetable_id) .execute() ) status["teacher_timetable_id"] = timetable_id status["slot_count"] = len(slots) status["available"] = bool(slots) status["needs_setup"] = not status["available"] except Exception as exc: logger.warning("Bootstrap timetable status lookup failed: %s", exc) return status def _graph_status(self, user_id: str, user_email: str, active: MembershipRow, institute: Optional[Dict[str, Any]]) -> Dict[str, Any]: base = {"available": False, "user_db": None, "institute_db": None, "projection_state": "unknown", "needs_rebuild": True, "last_checked_at": utc_now_iso(), "error_code": None} try: result = self.graph_probe(user_id=user_id, user_email=user_email, active_institute=active, institute=institute) return {**base, **(result or {}), "last_checked_at": (result or {}).get("last_checked_at") or base["last_checked_at"]} except Exception: return {**base, "projection_state": "error", "needs_rebuild": True, "error_code": "neo4j_unavailable"} def _onboarding(self, school_status: str, permissions: Dict[str, bool], calendar: Dict[str, Any], timetable: Dict[str, Any], graph: Dict[str, Any]) -> Dict[str, Any]: optional = [] if graph.get("available") else ["graph_rebuild"] if school_status == "no_school": return {"next_step": "create_or_join_school", "required": ["school_membership"], "optional": optional, "message": "Create or join a school to finish setup."} if school_status == "multi_school_needs_selection": return {"next_step": "select_school", "required": ["active_school_selection"], "optional": optional, "message": "Select which school to use for this session."} if permissions.get("can_manage_calendar") and calendar.get("needs_setup"): return {"next_step": "setup_calendar", "required": ["calendar"], "optional": optional, "message": "Set up the school calendar."} if permissions.get("can_manage_timetable") and timetable.get("needs_setup"): return {"next_step": "setup_timetable", "required": ["timetable"], "optional": optional, "message": "Set up the timetable."} if school_status == "school_admin" and permissions.get("can_invite_staff"): return {"next_step": "invite_staff", "required": [], "optional": optional, "message": "Invite staff or continue to your workspace."} return {"next_step": "ready", "required": [], "optional": optional, "message": "Your workspace is ready."} def build_bootstrap_response(credentials: Dict[str, Any]) -> Dict[str, Any]: sb = SupabaseServiceRoleClient() return BootstrapService(sb.supabase).build(credentials)