feat: database-first refactor, unified architecture and visual developer manual

Summary of changes:
- Consolidated Studio+CMS and Experience+LMS into unified services.
- Moved core business logic (enrollment, grading, auth) to PostgreSQL functions.
- Implemented advanced auditing via DB triggers and session context.
- Added gamification (XP/Levels/Leaderboards) and logic encapsulation.
- Updated installation/diagnostic scripts for the new architecture.
- Created a comprehensive Visual Developer Manual in README.md with hardware scaling.

services/cms-service/Cargo.toml

@@ -20,3 +20,6 @@ tower-http.workspace = true
 reqwest.workspace = true
 bcrypt.workspace = true
 jsonwebtoken.workspace = true
+hmac.workspace = true
+sha2.workspace = true
+hex.workspace = true

services/cms-service/Dockerfile

@@ -1,23 +0,0 @@
-# Build stage
-FROM rustlang/rust:nightly AS builder
-
-WORKDIR /usr/src/app
-COPY . .
-
-# Install necessary build dependencies
-RUN apt-get update && apt-get install -y pkg-config libssl-dev && rm -rf /var/lib/apt/lists/*
-
-# Build the specific service
-RUN cargo build --release -p cms-service
-
-# Final stage
-FROM debian:bookworm-slim
-RUN apt-get update && apt-get install -y libssl3 ca-certificates && rm -rf /var/lib/apt/lists/*
-
-WORKDIR /usr/local/bin
-COPY --from=builder /usr/src/app/target/release/cms-service .
-
-ENV RUST_LOG=info
-EXPOSE 3001
-
-CMD ["./cms-service"]

services/cms-service/migrations/20260111000000_gamification_users.sql

@@ -0,0 +1,4 @@
+-- Add XP and Level to users for gamification
+ALTER TABLE users
+ADD COLUMN xp INTEGER NOT NULL DEFAULT 0,
+ADD COLUMN level INTEGER NOT NULL DEFAULT 1;

services/cms-service/migrations/20260111000001_webhooks.sql

@@ -0,0 +1,14 @@
+-- Migration: Create Webhooks Table
+CREATE TABLE webhooks (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    organization_id UUID NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
+    url VARCHAR(500) NOT NULL,
+    events VARCHAR(50)[] NOT NULL, -- e.g., ['course.published', 'lesson.completed']
+    secret VARCHAR(255), -- For HMAC-SHA256 signatures
+    is_active BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Index for organization_id
+CREATE INDEX idx_webhooks_organization_id ON webhooks(organization_id);

services/cms-service/migrations/20260111000002_advanced_auditing.sql

@@ -0,0 +1,130 @@
+-- Migration: Advanced Auditing and Automatic Triggers
+-- Upgrade audit_logs table and implement automated change tracking
+
+-- 1. Upgrade audit_logs table
+ALTER TABLE audit_logs 
+ADD COLUMN IF NOT EXISTS event_type VARCHAR(50) DEFAULT 'USER_EVENT',
+ADD COLUMN IF NOT EXISTS old_data JSONB,
+ADD COLUMN IF NOT EXISTS new_data JSONB,
+ADD COLUMN IF NOT EXISTS ip_address INET,
+ADD COLUMN IF NOT EXISTS public_ip INET,
+ADD COLUMN IF NOT EXISTS user_agent TEXT,
+ADD COLUMN IF NOT EXISTS metadata JSONB DEFAULT '{}';
+
+-- 2. Create Audit Trigger Function
+CREATE OR REPLACE FUNCTION fn_trigger_audit_log()
+RETURNS TRIGGER AS $$
+DECLARE
+    v_user_id UUID;
+    v_org_id UUID;
+    v_ip INET;
+    v_user_agent TEXT;
+    v_event_type VARCHAR(50);
+    v_old_data JSONB := NULL;
+    v_new_data JSONB := NULL;
+    v_action VARCHAR(50);
+BEGIN
+    -- Try to get context from session variables
+    BEGIN
+        v_user_id := current_setting('app.current_user_id', true)::UUID;
+    EXCEPTION WHEN OTHERS THEN
+        v_user_id := NULL;
+    END;
+
+    BEGIN
+        v_org_id := current_setting('app.current_org_id', true)::UUID;
+    EXCEPTION WHEN OTHERS THEN
+        v_org_id := NULL;
+    END;
+
+    BEGIN
+        v_ip := current_setting('app.client_ip', true)::INET;
+    EXCEPTION WHEN OTHERS THEN
+        v_ip := NULL;
+    END;
+
+    BEGIN
+        v_user_agent := current_setting('app.user_agent', true);
+    EXCEPTION WHEN OTHERS THEN
+        v_user_agent := NULL;
+    END;
+
+    BEGIN
+        v_event_type := current_setting('app.event_type', true);
+    EXCEPTION WHEN OTHERS THEN
+        v_event_type := 'USER_EVENT';
+    END;
+
+    -- Handle different operations
+    IF (TG_OP = 'DELETE') THEN
+        v_old_data := to_jsonb(OLD);
+        v_action := 'DELETE';
+    ELSIF (TG_OP = 'UPDATE') THEN
+        v_old_data := to_jsonb(OLD);
+        v_new_data := to_jsonb(NEW);
+        v_action := 'UPDATE';
+    ELSIF (TG_OP = 'INSERT') THEN
+        v_new_data := to_jsonb(NEW);
+        v_action := 'INSERT';
+    END IF;
+
+    -- Insert into audit_logs
+    INSERT INTO audit_logs (
+        organization_id, 
+        user_id, 
+        action, 
+        entity_type, 
+        entity_id, 
+        event_type, 
+        old_data, 
+        new_data, 
+        ip_address, 
+        user_agent
+    )
+    VALUES (
+        COALESCE(v_org_id, (CASE WHEN TG_OP = 'DELETE' THEN OLD.organization_id ELSE NEW.organization_id END)),
+        v_user_id,
+        v_action,
+        TG_TABLE_NAME,
+        CASE WHEN TG_OP = 'DELETE' THEN OLD.id ELSE NEW.id END,
+        COALESCE(v_event_type, 'USER_EVENT'),
+        v_old_data,
+        v_new_data,
+        v_ip,
+        v_user_agent
+    );
+
+    IF (TG_OP = 'DELETE') THEN
+        RETURN OLD;
+    END IF;
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+-- 3. Attach triggers to core tables
+DO $$
+BEGIN
+    -- Courses
+    IF EXISTS (SELECT 1 FROM pg_tables WHERE tablename = 'courses') THEN
+        DROP TRIGGER IF EXISTS trg_audit_courses ON courses;
+        CREATE TRIGGER trg_audit_courses
+        AFTER INSERT OR UPDATE OR DELETE ON courses
+        FOR EACH ROW EXECUTE FUNCTION fn_trigger_audit_log();
+    END IF;
+
+    -- Lessons
+    IF EXISTS (SELECT 1 FROM pg_tables WHERE tablename = 'lessons') THEN
+        DROP TRIGGER IF EXISTS trg_audit_lessons ON lessons;
+        CREATE TRIGGER trg_audit_lessons
+        AFTER INSERT OR UPDATE OR DELETE ON lessons
+        FOR EACH ROW EXECUTE FUNCTION fn_trigger_audit_log();
+    END IF;
+
+    -- Users
+    IF EXISTS (SELECT 1 FROM pg_tables WHERE tablename = 'users') THEN
+        DROP TRIGGER IF EXISTS trg_audit_users ON users;
+        CREATE TRIGGER trg_audit_users
+        AFTER INSERT OR UPDATE OR DELETE ON users
+        FOR EACH ROW EXECUTE FUNCTION fn_trigger_audit_log();
+    END IF;
+END $$;

services/cms-service/migrations/20260111000005_crud_functions.sql

@@ -0,0 +1,239 @@
+-- Migration: CMS CRUD Functions
+-- Encapsulate all data mutations in stored functions
+
+-- 1. Course Management
+CREATE OR REPLACE FUNCTION fn_create_course(
+    p_organization_id UUID,
+    p_instructor_id UUID,
+    p_title VARCHAR(255),
+    p_pacing_mode VARCHAR(50) DEFAULT 'self_paced'
+) RETURNS SETOF courses AS $$
+BEGIN
+    RETURN QUERY
+    INSERT INTO courses (organization_id, instructor_id, title, pacing_mode)
+    VALUES (p_organization_id, p_instructor_id, p_title, p_pacing_mode)
+    RETURNING *;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION fn_update_course(
+    p_id UUID,
+    p_organization_id UUID,
+    p_title VARCHAR(255),
+    p_description TEXT,
+    p_passing_percentage INTEGER,
+    p_pacing_mode VARCHAR(50),
+    p_start_date TIMESTAMPTZ,
+    p_end_date TIMESTAMPTZ,
+    p_certificate_template VARCHAR(255) DEFAULT NULL
+) RETURNS SETOF courses AS $$
+BEGIN
+    RETURN QUERY
+    UPDATE courses 
+    SET title = p_title,
+        description = p_description,
+        passing_percentage = p_passing_percentage,
+        pacing_mode = p_pacing_mode,
+        start_date = p_start_date,
+        end_date = p_end_date,
+        certificate_template = p_certificate_template,
+        updated_at = NOW()
+    WHERE id = p_id AND organization_id = p_organization_id
+    RETURNING *;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION fn_delete_course(
+    p_id UUID,
+    p_organization_id UUID
+) RETURNS BOOLEAN AS $$
+DECLARE
+    v_deleted_count INTEGER;
+BEGIN
+    DELETE FROM courses 
+    WHERE id = p_id AND organization_id = p_organization_id;
+    
+    GET DIAGNOSTICS v_deleted_count = ROW_COUNT;
+    RETURN v_deleted_count > 0;
+END;
+$$ LANGUAGE plpgsql;
+
+-- 2. Module Management
+CREATE OR REPLACE FUNCTION fn_create_module(
+    p_organization_id UUID,
+    p_course_id UUID,
+    p_title VARCHAR(255),
+    p_position INTEGER
+) RETURNS SETOF modules AS $$
+BEGIN
+    RETURN QUERY
+    INSERT INTO modules (organization_id, course_id, title, position)
+    VALUES (p_organization_id, p_course_id, p_title, p_position)
+    RETURNING *;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION fn_update_module(
+    p_id UUID,
+    p_organization_id UUID,
+    p_title VARCHAR(255) DEFAULT NULL,
+    p_position INTEGER DEFAULT NULL
+) RETURNS SETOF modules AS $$
+BEGIN
+    RETURN QUERY
+    UPDATE modules 
+    SET title = COALESCE(p_title, title),
+        position = COALESCE(p_position, position),
+        updated_at = NOW()
+    WHERE id = p_id AND organization_id = p_organization_id
+    RETURNING *;
+END;
+$$ LANGUAGE plpgsql;
+
+-- 3. Lesson Management
+CREATE OR REPLACE FUNCTION fn_create_lesson(
+    p_organization_id UUID,
+    p_module_id UUID,
+    p_title VARCHAR(255),
+    p_content_type VARCHAR(50),
+    p_content_url VARCHAR(500) DEFAULT NULL,
+    p_position INTEGER DEFAULT 0,
+    p_transcription JSONB DEFAULT NULL,
+    p_metadata JSONB DEFAULT NULL,
+    p_is_graded BOOLEAN DEFAULT FALSE,
+    p_grading_category_id UUID DEFAULT NULL,
+    p_max_attempts INTEGER DEFAULT NULL,
+    p_allow_retry BOOLEAN DEFAULT TRUE,
+    p_due_date TIMESTAMPTZ DEFAULT NULL,
+    p_important_date_type VARCHAR(50) DEFAULT NULL
+) RETURNS SETOF lessons AS $$
+BEGIN
+    RETURN QUERY
+    INSERT INTO lessons (
+        organization_id, module_id, title, content_type, content_url, 
+        position, transcription, metadata, is_graded, grading_category_id, 
+        max_attempts, allow_retry, due_date, important_date_type
+    )
+    VALUES (
+        p_organization_id, p_module_id, p_title, p_content_type, p_content_url, 
+        p_position, p_transcription, p_metadata, p_is_graded, p_grading_category_id, 
+        p_max_attempts, p_allow_retry, p_due_date, p_important_date_type
+    )
+    RETURNING *;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION fn_update_lesson(
+    p_id UUID,
+    p_organization_id UUID,
+    p_title VARCHAR(255) DEFAULT NULL,
+    p_content_type VARCHAR(50) DEFAULT NULL,
+    p_content_url VARCHAR(500) DEFAULT NULL,
+    p_content_blocks JSONB DEFAULT NULL,
+    p_transcription JSONB DEFAULT NULL,
+    p_metadata JSONB DEFAULT NULL,
+    p_is_graded BOOLEAN DEFAULT NULL,
+    p_grading_category_id UUID DEFAULT NULL,
+    p_max_attempts INTEGER DEFAULT NULL,
+    p_allow_retry BOOLEAN DEFAULT NULL,
+    p_position INTEGER DEFAULT NULL,
+    p_due_date TIMESTAMPTZ DEFAULT NULL,
+    p_important_date_type VARCHAR(50) DEFAULT NULL,
+    p_summary TEXT DEFAULT NULL,
+    p_clear_due_date BOOLEAN DEFAULT FALSE,
+    p_clear_grading_category BOOLEAN DEFAULT FALSE
+) RETURNS SETOF lessons AS $$
+BEGIN
+    RETURN QUERY
+    UPDATE lessons 
+    SET title = COALESCE(p_title, title),
+        content_type = COALESCE(p_content_type, content_type),
+        content_url = COALESCE(p_content_url, content_url),
+        content_blocks = COALESCE(p_content_blocks, content_blocks),
+        transcription = COALESCE(p_transcription, transcription),
+        metadata = COALESCE(p_metadata, metadata),
+        is_graded = COALESCE(p_is_graded, is_graded),
+        grading_category_id = CASE 
+            WHEN p_clear_grading_category THEN NULL 
+            ELSE COALESCE(p_grading_category_id, grading_category_id) 
+        END,
+        max_attempts = COALESCE(p_max_attempts, max_attempts),
+        allow_retry = COALESCE(p_allow_retry, allow_retry),
+        position = COALESCE(p_position, position),
+        due_date = CASE 
+            WHEN p_clear_due_date THEN NULL 
+            ELSE COALESCE(p_due_date, due_date) 
+        END,
+        important_date_type = COALESCE(p_important_date_type, important_date_type),
+        summary = COALESCE(p_summary, summary),
+        updated_at = NOW()
+    WHERE id = p_id AND organization_id = p_organization_id
+    RETURNING *;
+END;
+$$ LANGUAGE plpgsql;
+
+-- 4. Content Reordering
+CREATE OR REPLACE PROCEDURE pr_reorder_modules(
+    p_organization_id UUID,
+    p_updates JSONB -- Array of {id, position}
+) AS $$
+DECLARE
+    v_update JSONB;
+BEGIN
+    FOR v_update IN SELECT * FROM jsonb_array_elements(p_updates)
+    LOOP
+        UPDATE modules 
+        SET position = (v_update->>'position')::INTEGER 
+        WHERE id = (v_update->>'id')::UUID AND organization_id = p_organization_id;
+    END LOOP;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE PROCEDURE pr_reorder_lessons(
+    p_organization_id UUID,
+    p_updates JSONB -- Array of {id, position}
+) AS $$
+DECLARE
+    v_update JSONB;
+BEGIN
+    FOR v_update IN SELECT * FROM jsonb_array_elements(p_updates)
+    LOOP
+        UPDATE lessons 
+        SET position = (v_update->>'position')::INTEGER 
+        WHERE id = (v_update->>'id')::UUID AND organization_id = p_organization_id;
+    END LOOP;
+END;
+$$ LANGUAGE plpgsql;
+
+-- 5. User & Auth Management
+CREATE OR REPLACE FUNCTION fn_register_user(
+    p_email VARCHAR(255),
+    p_password_hash VARCHAR(255),
+    p_full_name VARCHAR(255),
+    p_role VARCHAR(50),
+    p_org_name VARCHAR(255)
+) RETURNS SETOF users AS $$
+DECLARE
+    v_org_id UUID;
+BEGIN
+    -- Find or create organization
+    INSERT INTO organizations (name) 
+    VALUES (p_org_name) 
+    ON CONFLICT (name) DO UPDATE SET name = EXCLUDED.name 
+    RETURNING id INTO v_org_id;
+
+    -- Create user
+    RETURN QUERY
+    INSERT INTO users (email, password_hash, full_name, role, organization_id)
+    VALUES (p_email, p_password_hash, p_full_name, p_role, v_org_id)
+    RETURNING *;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION fn_get_user_by_email(
+    p_email VARCHAR(255)
+) RETURNS SETOF users AS $$
+BEGIN
+    RETURN QUERY SELECT * FROM users WHERE email = p_email;
+END;
+$$ LANGUAGE plpgsql;

services/cms-service/src/db_util.rs

@@ -0,0 +1,43 @@
+use sqlx::{Postgres, Transaction};
+use uuid::Uuid;
+
+pub async fn set_session_context(
+    tx: &mut Transaction<'_, Postgres>,
+    user_id: Option<Uuid>,
+    org_id: Option<Uuid>,
+    ip_address: Option<String>,
+    user_agent: Option<String>,
+    event_type: Option<String>,
+) -> Result<(), sqlx::Error> {
+    if let Some(uid) = user_id {
+        let _ = sqlx::query("SELECT set_config('app.current_user_id', $1, true)")
+            .bind(uid.to_string())
+            .execute(&mut **tx)
+            .await?;
+    }
+    if let Some(oid) = org_id {
+        let _ = sqlx::query("SELECT set_config('app.current_org_id', $1, true)")
+            .bind(oid.to_string())
+            .execute(&mut **tx)
+            .await?;
+    }
+    if let Some(ip) = ip_address {
+        let _ = sqlx::query("SELECT set_config('app.client_ip', $1, true)")
+            .bind(ip)
+            .execute(&mut **tx)
+            .await?;
+    }
+    if let Some(ua) = user_agent {
+        let _ = sqlx::query("SELECT set_config('app.user_agent', $1, true)")
+            .bind(ua)
+            .execute(&mut **tx)
+            .await?;
+    }
+    if let Some(et) = event_type {
+        let _ = sqlx::query("SELECT set_config('app.event_type', $1, true)")
+            .bind(et)
+            .execute(&mut **tx)
+            .await?;
+    }
+    Ok(())
+}

services/cms-service/src/handlers.rs

@@ -1,3 +1,4 @@
+use crate::webhooks::WebhookService;
 use axum::{
     Json,
     extract::{Path, Query, State},
@@ -89,6 +90,21 @@ pub async fn publish_course(
 
     log_action(&pool, Uuid::new_v4(), "PUBLISH", "Course", id, json!({})).await;
 
+    // 5. Trigger Webhook
+    let webhook_service = WebhookService::new(pool.clone());
+    webhook_service
+        .dispatch(
+            org_ctx.id,
+            "course.published",
+            &json!({
+                "course_id": id,
+                "title": payload.course.title,
+                "pacing_mode": payload.course.pacing_mode,
+                "published_at": Utc::now()
+            }),
+        )
+        .await;
+
     Ok(StatusCode::OK)
 }
 
@@ -114,6 +130,7 @@ pub async fn create_course(
     Org(org_ctx): Org,
     claims: common::auth::Claims,
     State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Json(payload): Json<serde_json::Value>,
 ) -> Result<Json<Course>, StatusCode> {
     let title = payload
@@ -127,29 +144,53 @@ pub async fn create_course(
         .and_then(|v| v.as_str())
         .unwrap_or("self_paced");
 
-    let course = sqlx::query_as::<_, Course>(
-        "INSERT INTO courses (title, instructor_id, organization_id, pacing_mode) VALUES ($1, $2, $3, $4) RETURNING *"
-    )
-    .bind(title)
-    .bind(instructor_id)
-    .bind(org_ctx.id)
-    .bind(pacing_mode)
-    .fetch_one(&pool)
-    .await
-    .map_err(|e| {
-        tracing::error!("Create course failed: {}", e);
+    let mut tx = pool.begin().await.map_err(|e| {
+        tracing::error!("Failed to start transaction: {}", e);
         StatusCode::INTERNAL_SERVER_ERROR
     })?;
 
-    log_action(
-        &pool,
-        instructor_id,
-        "CREATE",
-        "Course",
-        course.id,
-        json!({ "title": title, "pacing_mode": pacing_mode }),
+    // Set session context for the DB trigger to find
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(instructor_id),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("USER_EVENT".to_string()),
     )
-    .await;
+    .await
+    .map_err(|e| {
+        tracing::error!("Failed to set session context: {}", e);
+        StatusCode::INTERNAL_SERVER_ERROR
+    })?;
+
+    let course = sqlx::query_as::<_, Course>("SELECT * FROM fn_create_course($1, $2, $3, $4)")
+        .bind(org_ctx.id)
+        .bind(instructor_id)
+        .bind(title)
+        .bind(pacing_mode)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|e| {
+            tracing::error!("Create course failed: {}", e);
+            StatusCode::INTERNAL_SERVER_ERROR
+        })?;
+
+    tx.commit().await.map_err(|e| {
+        tracing::error!("Failed to commit transaction: {}", e);
+        StatusCode::INTERNAL_SERVER_ERROR
+    })?;
 
     Ok(Json(course))
 }
@@ -222,27 +263,34 @@ pub async fn update_course(
         .or(existing.end_date);
 
     let course = sqlx::query_as::<_, Course>(
-        "UPDATE courses SET title = $1, description = $2, passing_percentage = $3, certificate_template = $4, pacing_mode = $5, start_date = $6, end_date = $7, updated_at = NOW() WHERE id = $8 AND organization_id = $9 RETURNING *"
+        "SELECT * FROM fn_update_course($1, $2, $3, $4, $5, $6, $7, $8, $9)",
     )
+    .bind(id)
+    .bind(org_ctx.id)
     .bind(title)
     .bind(description)
     .bind(passing_percentage)
-    .bind(certificate_template)
     .bind(pacing_mode)
     .bind(start_date)
     .bind(end_date)
-    .bind(id)
-    .bind(org_ctx.id)
+    .bind(certificate_template)
     .fetch_one(&pool)
     .await
-    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to update course: {}", e)))?;
+    .map_err(|e| {
+        (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            format!("Failed to update course: {}", e),
+        )
+    })?;
 
     Ok(Json(course))
 }
 
 pub async fn create_module(
+    Org(org_ctx): Org,
     claims: common::auth::Claims,
     State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Json(payload): Json<serde_json::Value>,
 ) -> Result<Json<Module>, StatusCode> {
     let title = payload
@@ -259,32 +307,57 @@ pub async fn create_module(
         .and_then(|v| v.as_i64())
         .unwrap_or(0) as i32;
 
-    let module = sqlx::query_as::<_, Module>(
-        "INSERT INTO modules (course_id, title, position) VALUES ($1, $2, $3) RETURNING *",
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(claims.sub),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("USER_EVENT".to_string()),
     )
-    .bind(course_id)
-    .bind(title)
-    .bind(position)
-    .fetch_one(&pool)
     .await
     .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
-    log_action(
-        &pool,
-        claims.sub,
-        "CREATE",
-        "Module",
-        module.id,
-        json!({ "title": title, "course_id": course_id }),
-    )
-    .await;
+    let module = sqlx::query_as::<_, Module>("SELECT * FROM fn_create_module($1, $2, $3, $4)")
+        .bind(org_ctx.id)
+        .bind(course_id)
+        .bind(title)
+        .bind(position)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|e| {
+            tracing::error!("Create module failed: {}", e);
+            StatusCode::INTERNAL_SERVER_ERROR
+        })?;
+
+    tx.commit()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
     Ok(Json(module))
 }
 
 pub async fn create_lesson(
+    Org(org_ctx): Org,
     claims: common::auth::Claims,
     State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Json(payload): Json<serde_json::Value>,
 ) -> Result<Json<Lesson>, StatusCode> {
     let title = payload
@@ -332,10 +405,37 @@ pub async fn create_lesson(
 
     let important_date_type = payload.get("important_date_type").and_then(|v| v.as_str());
 
-    let lesson = sqlx::query_as::<_, Lesson>(
-        "INSERT INTO lessons (module_id, title, content_type, content_url, position, transcription, metadata, is_graded, grading_category_id, max_attempts, allow_retry, due_date, important_date_type) 
-         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) RETURNING *"
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(claims.sub),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("USER_EVENT".to_string()),
     )
+    .await
+    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let lesson = sqlx::query_as::<_, Lesson>(
+        "SELECT * FROM fn_create_lesson($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)"
+    )
+    .bind(org_ctx.id)
     .bind(module_id)
     .bind(title)
     .bind(content_type)
@@ -349,22 +449,16 @@ pub async fn create_lesson(
     .bind(allow_retry)
     .bind(due_date)
     .bind(important_date_type)
-    .fetch_one(&pool)
+    .fetch_one(&mut *tx)
     .await
     .map_err(|e| {
         tracing::error!("Create lesson failed: {}", e);
         StatusCode::INTERNAL_SERVER_ERROR
     })?;
 
-    log_action(
-        &pool,
-        claims.sub,
-        "CREATE",
-        "Lesson",
-        lesson.id,
-        json!({ "title": title, "module_id": module_id }),
-    )
-    .await;
+    tx.commit()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
     Ok(Json(lesson))
 }
@@ -724,8 +818,10 @@ pub async fn get_lesson(
 }
 
 pub async fn update_lesson(
+    Org(org_ctx): Org,
     claims: common::auth::Claims,
     State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Path(id): Path<Uuid>,
     Json(payload): Json<serde_json::Value>,
 ) -> Result<Json<Lesson>, StatusCode> {
@@ -742,50 +838,90 @@ pub async fn update_lesson(
         .and_then(|v| v.as_i64())
         .map(|v| v as i32);
     let allow_retry = payload.get("allow_retry").and_then(|v| v.as_bool());
-    let metadata = payload.get("metadata");
+    let metadata = payload.get("metadata").cloned();
     let important_date_type = payload.get("important_date_type").and_then(|v| v.as_str());
+    let summary = payload.get("summary").and_then(|v| v.as_str());
+    let content_blocks = payload.get("content_blocks").cloned();
+    let transcription = payload.get("transcription").cloned();
 
-    let updated_lesson = sqlx::query_as::<_, Lesson>(
-        "UPDATE lessons 
-         SET title = COALESCE($1, title), 
-             content_type = COALESCE($2, content_type), 
-             content_url = COALESCE($3, content_url),
-             position = COALESCE($4, position),
-             is_graded = COALESCE($5, is_graded),
-             grading_category_id = CASE WHEN $6 = 'SET_NULL' THEN NULL WHEN $7::UUID IS NOT NULL THEN $7 ELSE grading_category_id END,
-             metadata = COALESCE($8, metadata),
-             max_attempts = COALESCE($9, max_attempts),
-             allow_retry = COALESCE($10, allow_retry),
-             summary = COALESCE($11, summary),
-             due_date = CASE WHEN $12 = 'SET_NULL' THEN NULL WHEN $13::TIMESTAMPTZ IS NOT NULL THEN $13 ELSE due_date END,
-             important_date_type = COALESCE($14, important_date_type)
-         WHERE id = $15 RETURNING *"
+    let clear_grading_category = payload
+        .get("grading_category_id")
+        .map(|v| v.is_null())
+        .unwrap_or(false);
+    let grading_category_id = payload
+        .get("grading_category_id")
+        .and_then(|v| v.as_str())
+        .and_then(|s| Uuid::parse_str(s).ok());
+
+    let clear_due_date = payload
+        .get("due_date")
+        .map(|v| v.is_null())
+        .unwrap_or(false);
+    let due_date = payload
+        .get("due_date")
+        .and_then(|v| v.as_str())
+        .and_then(|s| s.parse::<DateTime<Utc>>().ok());
+
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(claims.sub),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("USER_EVENT".to_string()),
     )
+    .await
+    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let lesson = sqlx::query_as::<_, Lesson>(
+        "SELECT * FROM fn_update_lesson($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18)"
+    )
+    .bind(id)
+    .bind(org_ctx.id)
     .bind(title)
     .bind(content_type)
     .bind(content_url)
-    .bind(position)
-    .bind(is_graded)
-    .bind(if payload.get("grading_category_id").map(|v| v.is_null()).unwrap_or(false) { "SET_NULL" } else { "" })
-    .bind(payload.get("grading_category_id").and_then(|v| v.as_str()).and_then(|s| Uuid::parse_str(s).ok()))
+    .bind(content_blocks)
+    .bind(transcription)
     .bind(metadata)
+    .bind(is_graded)
+    .bind(grading_category_id)
     .bind(max_attempts)
     .bind(allow_retry)
-    .bind(payload.get("summary").and_then(|v| v.as_str()))
-    .bind(if payload.get("due_date").map(|v| v.is_null()).unwrap_or(false) { "SET_NULL" } else { "" })
-    .bind(payload.get("due_date").and_then(|v| v.as_str()).and_then(|s| s.parse::<DateTime<Utc>>().ok()))
+    .bind(position)
+    .bind(due_date)
     .bind(important_date_type)
-    .bind(id)
-    .fetch_one(&pool)
+    .bind(summary)
+    .bind(clear_due_date)
+    .bind(clear_grading_category)
+    .fetch_one(&mut *tx)
     .await
     .map_err(|e| {
         tracing::error!("Update lesson failed: {}", e);
         StatusCode::INTERNAL_SERVER_ERROR
     })?;
 
-    log_action(&pool, claims.sub, "UPDATE", "Lesson", id, json!(payload)).await;
+    tx.commit()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
-    Ok(Json(updated_lesson))
+    Ok(Json(lesson))
 }
 
 // Grading Policies
@@ -925,53 +1061,115 @@ pub async fn get_lessons(
     Ok(Json(lessons))
 }
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Serialize)]
 pub struct ReorderPayload {
     pub items: Vec<ReorderItem>,
 }
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Serialize)]
 pub struct ReorderItem {
     pub id: Uuid,
     pub position: i32,
 }
 
 pub async fn reorder_modules(
-    _claims: common::auth::Claims,
+    Org(org_ctx): Org,
+    claims: common::auth::Claims,
     State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Json(payload): Json<ReorderPayload>,
 ) -> Result<StatusCode, StatusCode> {
-    for item in payload.items {
-        sqlx::query("UPDATE modules SET position = $1 WHERE id = $2")
-            .bind(item.position)
-            .bind(item.id)
-            .execute(&pool)
-            .await
-            .map_err(|e| {
-                tracing::error!("Reorder modules failed: {}", e);
-                StatusCode::INTERNAL_SERVER_ERROR
-            })?;
-    }
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(claims.sub),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("USER_EVENT".to_string()),
+    )
+    .await
+    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    sqlx::query("CALL pr_reorder_modules($1, $2)")
+        .bind(org_ctx.id)
+        .bind(serde_json::to_value(&payload.items).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?)
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| {
+            tracing::error!("Reorder modules failed: {}", e);
+            StatusCode::INTERNAL_SERVER_ERROR
+        })?;
+
+    tx.commit()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
     Ok(StatusCode::OK)
 }
 
 pub async fn reorder_lessons(
-    _claims: common::auth::Claims,
+    Org(org_ctx): Org,
+    claims: common::auth::Claims,
     State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Json(payload): Json<ReorderPayload>,
 ) -> Result<StatusCode, StatusCode> {
-    for item in payload.items {
-        sqlx::query("UPDATE lessons SET position = $1 WHERE id = $2")
-            .bind(item.position)
-            .bind(item.id)
-            .execute(&pool)
-            .await
-            .map_err(|e| {
-                tracing::error!("Reorder lessons failed: {}", e);
-                StatusCode::INTERNAL_SERVER_ERROR
-            })?;
-    }
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(claims.sub),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("USER_EVENT".to_string()),
+    )
+    .await
+    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    sqlx::query("CALL pr_reorder_lessons($1, $2)")
+        .bind(org_ctx.id)
+        .bind(serde_json::to_value(&payload.items).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?)
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| {
+            tracing::error!("Reorder lessons failed: {}", e);
+            StatusCode::INTERNAL_SERVER_ERROR
+        })?;
+
+    tx.commit()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
     Ok(StatusCode::OK)
 }
@@ -1105,31 +1303,21 @@ pub async fn register(
         .await
         .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
-    let organization = sqlx::query_as::<_, Organization>(
-        "INSERT INTO organizations (name) VALUES ($1) ON CONFLICT (name) DO UPDATE SET name = EXCLUDED.name RETURNING *"
-    )
-    .bind(&org_name)
-    .fetch_one(&mut *tx)
-    .await
-    .map_err(|e| {
-        tracing::error!("Failed to create/find org: {}", e);
-        (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to find or create organization: {}", e))
-    })?;
-
-    let user = sqlx::query_as::<_, User>(
-        "INSERT INTO users (email, password_hash, full_name, role, organization_id) VALUES ($1, $2, $3, $4, $5) RETURNING *"
-    )
-    .bind(&payload.email)
-    .bind(password_hash)
-    .bind(full_name)
-    .bind(&role)
-    .bind(organization.id)
-    .fetch_one(&mut *tx)
-    .await
-    .map_err(|e: sqlx::Error| {
-        tracing::error!("Failed to create user: {}", e);
-        (StatusCode::CONFLICT, format!("User already exists or DB error: {}", e))
-    })?;
+    let user = sqlx::query_as::<_, User>("SELECT * FROM fn_register_user($1, $2, $3, $4, $5)")
+        .bind(&payload.email)
+        .bind(password_hash)
+        .bind(full_name)
+        .bind(&role)
+        .bind(&org_name)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|e: sqlx::Error| {
+            tracing::error!("Failed to create user: {}", e);
+            (
+                StatusCode::CONFLICT,
+                format!("User already exists or DB error: {}", e),
+            )
+        })?;
 
     tx.commit()
         .await
@@ -1160,7 +1348,7 @@ pub async fn login(
     State(pool): State<PgPool>,
     Json(payload): Json<AuthPayload>,
 ) -> Result<Json<AuthResponse>, (StatusCode, String)> {
-    let user = sqlx::query_as::<_, User>("SELECT * FROM users WHERE email = $1")
+    let user = sqlx::query_as::<_, User>("SELECT * FROM fn_get_user_by_email($1)")
         .bind(&payload.email)
         .fetch_one(&pool)
         .await
@@ -1241,6 +1429,55 @@ pub async fn get_course_analytics(
     Ok(Json(analytics))
 }
 
+pub async fn get_advanced_analytics(
+    Org(org_ctx): Org,
+    claims: common::auth::Claims,
+    State(pool): State<PgPool>,
+    Path(id): Path<Uuid>,
+) -> Result<Json<common::models::AdvancedAnalytics>, (StatusCode, String)> {
+    // 1. Fetch Course to check ownership
+    let course =
+        sqlx::query_as::<_, Course>("SELECT * FROM courses WHERE id = $1 AND organization_id = $2")
+            .bind(id)
+            .bind(org_ctx.id)
+            .fetch_one(&pool)
+            .await
+            .map_err(|_| (StatusCode::NOT_FOUND, "Course not found".into()))?;
+
+    // 2. Enforce RBAC
+    if claims.role != "admin" && course.instructor_id != claims.sub {
+        return Err((
+            StatusCode::FORBIDDEN,
+            "You do not have permission to view stats for a course you don't own".into(),
+        ));
+    }
+
+    // 4. Fetch from LMS
+    let client = reqwest::Client::new();
+    let res = client
+        .get(format!(
+            "http://lms-service:3002/courses/{}/analytics/advanced",
+            id
+        ))
+        .send()
+        .await
+        .map_err(|e| (StatusCode::BAD_GATEWAY, e.to_string()))?;
+
+    if !res.status().is_success() {
+        return Err((
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "Failed to fetch advanced analytics from LMS".into(),
+        ));
+    }
+
+    let analytics = res
+        .json::<common::models::AdvancedAnalytics>()
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+
+    Ok(Json(analytics))
+}
+
 #[derive(Deserialize)]
 pub struct AuditQuery {
     pub page: Option<i64>,
@@ -1356,8 +1593,10 @@ pub async fn get_course_outline(
 }
 
 pub async fn update_module(
+    Org(org_ctx): Org,
     claims: common::auth::Claims,
     State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Path(id): Path<Uuid>,
     Json(payload): Json<serde_json::Value>,
 ) -> Result<Json<Module>, StatusCode> {
@@ -1367,61 +1606,168 @@ pub async fn update_module(
         .and_then(|v| v.as_i64())
         .map(|v| v as i32);
 
-    let updated_module = sqlx::query_as::<_, Module>(
-        "UPDATE modules 
-         SET title = COALESCE($1, title), 
-             position = COALESCE($2, position)
-         WHERE id = $3 RETURNING *",
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(claims.sub),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("USER_EVENT".to_string()),
     )
-    .bind(title)
-    .bind(position)
-    .bind(id)
-    .fetch_one(&pool)
     .await
-    .map_err(|e| {
-        tracing::error!("Update module failed: {}", e);
-        StatusCode::INTERNAL_SERVER_ERROR
-    })?;
+    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
-    log_action(&pool, claims.sub, "UPDATE", "Module", id, json!(payload)).await;
+    // Fetch existing to handle COALESCE if not handled in function
+    // For now, let's just use the function logic.
+    // Wait, the DB function I wrote doesn't use COALESCE, it just sets values.
+    // I should fix the DB function or handle COALESCE here.
+    // Let's fix the DB function to use COALESCE for optional updates.
 
-    Ok(Json(updated_module))
+    let module = sqlx::query_as::<_, Module>("SELECT * FROM fn_update_module($1, $2, $3, $4)")
+        .bind(id)
+        .bind(org_ctx.id)
+        .bind(title)
+        .bind(position)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|e| {
+            tracing::error!("Update module failed: {}", e);
+            StatusCode::INTERNAL_SERVER_ERROR
+        })?;
+
+    tx.commit()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    Ok(Json(module))
 }
 
 pub async fn delete_module(
+    Org(org_ctx): Org,
     claims: common::auth::Claims,
     State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Path(id): Path<Uuid>,
 ) -> Result<StatusCode, StatusCode> {
-    sqlx::query("DELETE FROM modules WHERE id = $1")
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(claims.sub),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("USER_EVENT".to_string()),
+    )
+    .await
+    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let success = sqlx::query_scalar::<_, bool>("SELECT fn_delete_module($1, $2)")
         .bind(id)
-        .execute(&pool)
+        .bind(org_ctx.id)
+        .fetch_one(&mut *tx)
         .await
         .map_err(|e| {
             tracing::error!("Delete module failed: {}", e);
             StatusCode::INTERNAL_SERVER_ERROR
         })?;
 
-    log_action(&pool, claims.sub, "DELETE", "Module", id, json!({})).await;
+    if !success {
+        return Err(StatusCode::NOT_FOUND);
+    }
+
+    tx.commit()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
     Ok(StatusCode::OK)
 }
 
 pub async fn delete_lesson(
+    Org(org_ctx): Org,
     claims: common::auth::Claims,
     State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Path(id): Path<Uuid>,
 ) -> Result<StatusCode, StatusCode> {
     if claims.role != "admin" {
         return Err(StatusCode::FORBIDDEN);
     }
-    sqlx::query("DELETE FROM lessons WHERE id = $1")
-        .bind(id)
-        .execute(&pool)
+
+    let mut tx = pool
+        .begin()
         .await
         .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
-    log_action(&pool, claims.sub, "DELETE_LESSON", "Lesson", id, json!({})).await;
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(claims.sub),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("USER_EVENT".to_string()),
+    )
+    .await
+    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let success = sqlx::query_scalar::<_, bool>("SELECT fn_delete_lesson($1, $2)")
+        .bind(id)
+        .bind(org_ctx.id)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|e| {
+            tracing::error!("Delete lesson failed: {}", e);
+            StatusCode::INTERNAL_SERVER_ERROR
+        })?;
+
+    if !success {
+        return Err(StatusCode::NOT_FOUND);
+    }
+
+    tx.commit()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
     Ok(StatusCode::OK)
 }
@@ -1529,3 +1875,102 @@ pub async fn create_organization(
 
     Ok(Json(org))
 }
+
+#[derive(serde::Deserialize)]
+pub struct CreateWebhookPayload {
+    pub url: String,
+    pub events: Vec<String>,
+    pub secret: Option<String>,
+}
+
+pub async fn get_webhooks(
+    Org(org_ctx): Org,
+    claims: common::auth::Claims,
+    State(pool): State<PgPool>,
+) -> Result<Json<Vec<common::models::Webhook>>, (StatusCode, String)> {
+    if claims.role != "admin" {
+        return Err((StatusCode::FORBIDDEN, "Admin access required".into()));
+    }
+
+    let webhooks = sqlx::query_as::<_, common::models::Webhook>(
+        "SELECT * FROM webhooks WHERE organization_id =  ORDER BY created_at DESC",
+    )
+    .bind(org_ctx.id)
+    .fetch_all(&pool)
+    .await
+    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+
+    Ok(Json(webhooks))
+}
+
+pub async fn create_webhook(
+    Org(org_ctx): Org,
+    claims: common::auth::Claims,
+    State(pool): State<PgPool>,
+    Json(payload): Json<CreateWebhookPayload>,
+) -> Result<Json<common::models::Webhook>, (StatusCode, String)> {
+    if claims.role != "admin" {
+        return Err((StatusCode::FORBIDDEN, "Admin access required".into()));
+    }
+
+    let webhook = sqlx::query_as::<_, common::models::Webhook>(
+        r#"
+        INSERT INTO webhooks (organization_id, url, events, secret)
+        VALUES (, , , )
+        RETURNING *
+        "#,
+    )
+    .bind(org_ctx.id)
+    .bind(payload.url)
+    .bind(payload.events)
+    .bind(payload.secret)
+    .fetch_one(&pool)
+    .await
+    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+
+    log_action(
+        &pool,
+        claims.sub,
+        "CREATE_WEBHOOK",
+        "Webhook",
+        webhook.id,
+        serde_json::to_value(&webhook).unwrap(),
+    )
+    .await;
+
+    Ok(Json(webhook))
+}
+
+pub async fn delete_webhook(
+    Org(org_ctx): Org,
+    claims: common::auth::Claims,
+    State(pool): State<PgPool>,
+    Path(id): Path<Uuid>,
+) -> Result<StatusCode, (StatusCode, String)> {
+    if claims.role != "admin" {
+        return Err((StatusCode::FORBIDDEN, "Admin access required".into()));
+    }
+
+    let result = sqlx::query("DELETE FROM webhooks WHERE id =  AND organization_id = ")
+        .bind(id)
+        .bind(org_ctx.id)
+        .execute(&pool)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+
+    if result.rows_affected() == 0 {
+        return Err((StatusCode::NOT_FOUND, "Webhook not found".into()));
+    }
+
+    log_action(
+        &pool,
+        claims.sub,
+        "DELETE_WEBHOOK",
+        "Webhook",
+        id,
+        serde_json::json!({}),
+    )
+    .await;
+
+    Ok(StatusCode::OK)
+}

services/cms-service/src/main.rs

@@ -1,5 +1,7 @@
+mod db_util;
 mod handlers;
 mod handlers_branding;
+mod webhooks;
 
 use axum::{
     Router, middleware,
@@ -50,6 +52,10 @@ async fn main() {
             "/courses/{id}/analytics",
             get(handlers::get_course_analytics),
         )
+        .route(
+            "/courses/{id}/analytics/advanced",
+            get(handlers::get_advanced_analytics),
+        )
         .route(
             "/modules",
             get(handlers::get_modules).post(handlers::create_module),
@@ -86,7 +92,15 @@ async fn main() {
         .route("/users/{id}", axum::routing::put(handlers::update_user))
         .route("/audit-logs", get(handlers::get_audit_logs))
         .route("/assets/upload", post(handlers::upload_asset))
-        .route("/organizations", get(handlers::get_organizations).post(handlers::create_organization))
+        .route(
+            "/organizations",
+            get(handlers::get_organizations).post(handlers::create_organization),
+        )
+        .route(
+            "/webhooks",
+            get(handlers::get_webhooks).post(handlers::create_webhook),
+        )
+        .route("/webhooks/{id}", delete(handlers::delete_webhook))
         .route("/organization", get(handlers::get_organization))
         .route(
             "/organizations/{id}/logo",

services/cms-service/src/webhooks.rs

@@ -0,0 +1,93 @@
+use common::models::Webhook;
+use hex;
+use hmac::{Hmac, Mac};
+use sha2::Sha256;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+pub struct WebhookService {
+    pool: PgPool,
+}
+
+impl WebhookService {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+
+    pub async fn dispatch(&self, org_id: Uuid, event_type: &str, payload: &serde_json::Value) {
+        // 1. Fetch active webhooks for this org that are interested in this event
+        let webhooks = match sqlx::query_as::<_, Webhook>(
+            "SELECT * FROM webhooks WHERE organization_id = $1 AND is_active = TRUE AND $2 = ANY(events)"
+        )
+        .bind(org_id)
+        .bind(event_type)
+        .fetch_all(&self.pool)
+        .await {
+            Ok(w) => w,
+            Err(e) => {
+                tracing::error!("Failed to fetch webhooks for org {}: {}", org_id, e);
+                return;
+            }
+        };
+
+        if webhooks.is_empty() {
+            return;
+        }
+
+        let client = reqwest::Client::new();
+        let payload_str = payload.to_string();
+
+        for webhook in webhooks {
+            let mut request = client
+                .post(&webhook.url)
+                .header("Content-Type", "application/json")
+                .header("X-OpenCCB-Event", event_type)
+                .header("X-OpenCCB-Delivery", Uuid::new_v4().to_string());
+
+            // Add signature if secret exists
+            if let Some(secret) = &webhook.secret {
+                let signature = generate_signature(secret, &payload_str);
+                request = request.header("X-OpenCCB-Signature", signature);
+            }
+
+            let url = webhook.url.clone();
+            let res = request.body(payload_str.clone()).send().await;
+
+            match res {
+                Ok(response) => {
+                    if !response.status().is_success() {
+                        tracing::warn!(
+                            "Webhook delivery to {} (event: {}) failed with status {}",
+                            url,
+                            event_type,
+                            response.status()
+                        );
+                    } else {
+                        tracing::info!(
+                            "Successfully delivered webhook to {} (event: {})",
+                            url,
+                            event_type
+                        );
+                    }
+                }
+                Err(e) => {
+                    tracing::error!(
+                        "Failed to deliver webhook to {} (event: {}): {}",
+                        url,
+                        event_type,
+                        e
+                    );
+                }
+            }
+        }
+    }
+}
+
+fn generate_signature(secret: &str, payload: &str) -> String {
+    type HmacSha256 = Hmac<Sha256>;
+    let mut mac =
+        HmacSha256::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size");
+    mac.update(payload.as_bytes());
+    let result = mac.finalize();
+    hex::encode(result.into_bytes())
+}

services/lms-service/Dockerfile

@@ -1,23 +0,0 @@
-# Build stage
-FROM rustlang/rust:nightly AS builder
-
-WORKDIR /usr/src/app
-COPY . .
-
-# Install necessary build dependencies
-RUN apt-get update && apt-get install -y pkg-config libssl-dev && rm -rf /var/lib/apt/lists/*
-
-# Build the specific service
-RUN cargo build --release -p lms-service
-
-# Final stage
-FROM debian:bookworm-slim
-RUN apt-get update && apt-get install -y libssl3 ca-certificates && rm -rf /var/lib/apt/lists/*
-
-WORKDIR /usr/local/bin
-COPY --from=builder /usr/src/app/target/release/lms-service .
-
-ENV RUST_LOG=info
-EXPOSE 3002
-
-CMD ["./lms-service"]

services/lms-service/migrations/20260111000002_lms_logic.sql

@@ -0,0 +1,138 @@
+-- Migration: LMS Logic Functions (XP and Enrollment)
+
+-- 1. Function to award XP
+CREATE OR REPLACE FUNCTION fn_award_xp(
+    p_user_id UUID,
+    p_org_id UUID,
+    p_amount INTEGER,
+    p_reason TEXT,
+    p_entity_type TEXT DEFAULT NULL,
+    p_entity_id UUID DEFAULT NULL
+) RETURNS VOID AS $$
+BEGIN
+    -- Update XP in users table
+    UPDATE users 
+    SET xp = xp + p_amount,
+        level = FLOOR(SQRT((xp + p_amount)::FLOAT / 100)) + 1,
+        updated_at = NOW()
+    WHERE id = p_user_id;
+
+    -- Log to points_log
+    IF EXISTS (SELECT 1 FROM pg_tables WHERE tablename = 'points_log') THEN
+        INSERT INTO points_log (user_id, organization_id, amount, reason, entity_type, entity_id)
+        VALUES (p_user_id, p_org_id, p_amount, p_reason, p_entity_type, p_entity_id);
+    END IF;
+END;
+$$ LANGUAGE plpgsql;
+
+-- 2. Function for Course Cohort Analytics
+CREATE OR REPLACE FUNCTION fn_get_cohort_analytics(p_course_id UUID)
+RETURNS TABLE (
+    period TEXT,
+    student_count BIGINT,
+    completion_rate FLOAT4
+) AS $$
+BEGIN
+    RETURN QUERY
+    WITH cohort_students AS (
+        SELECT 
+            user_id,
+            TO_CHAR(enrolled_at, 'YYYY-MM') as v_period
+        FROM enrollments
+        WHERE course_id = p_course_id
+    ),
+    course_lesson_count AS (
+        SELECT COUNT(*)::float4 as total_lessons
+        FROM lessons
+        WHERE module_id IN (SELECT id FROM modules WHERE course_id = p_course_id)
+    )
+    SELECT 
+        cs.v_period as period,
+        COUNT(DISTINCT cs.user_id) as student_count,
+        COALESCE(AVG(
+            (SELECT COUNT(DISTINCT lesson_id)::float4 FROM user_grades WHERE user_id = cs.user_id AND course_id = p_course_id) / 
+            NULLIF((SELECT total_lessons FROM course_lesson_count), 0)
+        ), 0)::float4 as completion_rate
+    FROM cohort_students cs
+    GROUP BY cs.v_period
+    ORDER BY cs.v_period DESC;
+END;
+$$ LANGUAGE plpgsql;
+
+-- 3. Retention Data Function
+CREATE OR REPLACE FUNCTION fn_get_retention_data(p_course_id UUID)
+RETURNS TABLE (
+    lesson_id UUID,
+    lesson_title VARCHAR,
+    student_count BIGINT
+) AS $$
+BEGIN
+    RETURN QUERY
+    SELECT 
+        l.id as lesson_id,
+        l.title as lesson_title,
+        COUNT(DISTINCT ug.user_id) as student_count
+    FROM lessons l
+    LEFT JOIN user_grades ug ON l.id = ug.lesson_id
+    WHERE l.module_id IN (SELECT id FROM modules WHERE course_id = p_course_id)
+    GROUP BY l.id, l.title, l.position
+    ORDER BY l.position;
+END;
+$$ LANGUAGE plpgsql;
+
+-- 4. Enrollment Function
+CREATE OR REPLACE FUNCTION fn_enroll_student(
+    p_organization_id UUID,
+    p_user_id UUID,
+    p_course_id UUID
+) RETURNS SETOF enrollments AS $$
+BEGIN
+    RETURN QUERY
+    INSERT INTO enrollments (organization_id, user_id, course_id)
+    VALUES (p_organization_id, p_user_id, p_course_id)
+    ON CONFLICT (user_id, course_id) DO UPDATE SET enrolled_at = NOW()
+    RETURNING *;
+END;
+$$ LANGUAGE plpgsql;
+
+-- 5. Grading Function (Upsert) with Automated Logic
+CREATE OR REPLACE FUNCTION fn_upsert_user_grade(
+    p_organization_id UUID,
+    p_user_id UUID,
+    p_course_id UUID,
+    p_lesson_id UUID,
+    p_score FLOAT4,
+    p_metadata JSONB DEFAULT NULL
+) RETURNS SETOF user_grades AS $$
+DECLARE
+    v_grade user_grades;
+    v_xp_amount INTEGER := 20; -- Default XP for completion
+    v_badge_id UUID;
+BEGIN
+    -- 1. Upsert grade
+    INSERT INTO user_grades (organization_id, user_id, course_id, lesson_id, score, metadata, attempts_count)
+    VALUES (p_organization_id, p_user_id, p_course_id, p_lesson_id, p_score, p_metadata, 1)
+    ON CONFLICT (user_id, lesson_id) DO UPDATE SET
+        score = EXCLUDED.score,
+        metadata = EXCLUDED.metadata,
+        attempts_count = user_grades.attempts_count + 1,
+        updated_at = NOW()
+    RETURNING * INTO v_grade;
+
+    -- 2. Award XP automatically
+    PERFORM fn_award_xp(p_user_id, p_organization_id, v_xp_amount, 'lesson_completion', 'lesson', p_lesson_id);
+
+    -- 3. Check for new badges
+    FOR v_badge_id IN 
+        SELECT id FROM badges 
+        WHERE organization_id = p_organization_id 
+          AND requirement_type = 'points' 
+          AND requirement_value <= (SELECT xp FROM users WHERE id = p_user_id)
+          AND id NOT IN (SELECT badge_id FROM user_badges WHERE user_id = p_user_id)
+    LOOP
+        INSERT INTO user_badges (user_id, badge_id) VALUES (p_user_id, v_badge_id) ON CONFLICT DO NOTHING;
+    END LOOP;
+
+    RETURN NEXT v_grade;
+END;
+$$ LANGUAGE plpgsql;

services/lms-service/migrations/20260111000003_lms_webhooks.sql

@@ -0,0 +1,14 @@
+-- Migration: Create Webhooks Table for LMS
+CREATE TABLE webhooks (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    organization_id UUID NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
+    url VARCHAR(500) NOT NULL,
+    events VARCHAR(50)[] NOT NULL, -- e.g., ['user.enrolled', 'lesson.completed', 'course.completed']
+    secret VARCHAR(255), -- For HMAC-SHA256 signatures
+    is_active BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Index for organization_id
+CREATE INDEX idx_webhooks_organization_id ON webhooks(organization_id);

services/lms-service/src/db_util.rs

@@ -0,0 +1,40 @@
+use sqlx::{Postgres, Transaction};
+
+pub async fn set_session_context(
+    tx: &mut Transaction<'_, Postgres>,
+    user_id: Option<uuid::Uuid>,
+    org_id: Option<uuid::Uuid>,
+    ip: Option<String>,
+    ua: Option<String>,
+    event_type: Option<String>,
+) -> Result<(), sqlx::Error> {
+    if let Some(uid) = user_id {
+        sqlx::query(&format!("SET LOCAL app.current_user_id = '{}'", uid))
+            .execute(&mut **tx)
+            .await?;
+    }
+    if let Some(oid) = org_id {
+        sqlx::query(&format!("SET LOCAL app.current_org_id = '{}'", oid))
+            .execute(&mut **tx)
+            .await?;
+    }
+    if let Some(ip_addr) = ip {
+        sqlx::query(&format!("SET LOCAL app.client_ip = '{}'", ip_addr))
+            .execute(&mut **tx)
+            .await?;
+    }
+    if let Some(user_agent) = ua {
+        // Use set_config for potentially long strings to avoid SQL injection/formatting issues
+        sqlx::query("SELECT set_config('app.user_agent', $1, true)")
+            .bind(user_agent)
+            .execute(&mut **tx)
+            .await?;
+    }
+    if let Some(et) = event_type {
+        sqlx::query("SELECT set_config('app.event_type', $1, true)")
+            .bind(et)
+            .execute(&mut **tx)
+            .await?;
+    }
+    Ok(())
+}

services/lms-service/src/handlers.rs

@@ -15,9 +15,10 @@ use sqlx::{PgPool, Row};
 use uuid::Uuid;
 
 pub async fn enroll_user(
-    State(pool): State<PgPool>,
     Org(org_ctx): Org,
     claims: Claims,
+    State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Json(payload): Json<serde_json::Value>,
 ) -> Result<Json<Enrollment>, StatusCode> {
     let course_id_str = payload
@@ -27,18 +28,61 @@ pub async fn enroll_user(
     let course_id = Uuid::parse_str(course_id_str).map_err(|_| StatusCode::BAD_REQUEST)?;
     let user_id = claims.sub;
 
-    let enrollment = sqlx::query_as::<_, Enrollment>(
-        "INSERT INTO enrollments (user_id, course_id, organization_id) VALUES ($1, $2, $3) RETURNING *"
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(user_id),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("USER_EVENT".to_string()),
     )
-    .bind(user_id)
-    .bind(course_id)
-    .bind(org_ctx.id)
-    .fetch_one(&pool)
     .await
-    .map_err(|e| {
-        tracing::error!("Enrollment failed: {}", e);
-        StatusCode::INTERNAL_SERVER_ERROR
-    })?;
+    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    let enrollment = sqlx::query_as::<_, Enrollment>("SELECT * FROM fn_enroll_student($1, $2, $3)")
+        .bind(org_ctx.id)
+        .bind(user_id)
+        .bind(course_id)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|e| {
+            tracing::error!("Enrollment failed: {}", e);
+            StatusCode::INTERNAL_SERVER_ERROR
+        })?;
+
+    tx.commit()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    // Dispatch Webhook
+    let webhook_service = common::webhooks::WebhookService::new(pool.clone());
+    webhook_service
+        .dispatch(
+            org_ctx.id,
+            "user.enrolled",
+            &serde_json::json!({
+                "user_id": user_id,
+                "course_id": course_id,
+                "enrollment_id": enrollment.id
+            }),
+        )
+        .await;
 
     Ok(Json(enrollment))
 }
@@ -397,16 +441,45 @@ pub async fn get_user_enrollments(
 
 pub async fn submit_lesson_score(
     Org(org_ctx): Org,
+    claims: Claims,
     State(pool): State<PgPool>,
+    headers: axum::http::HeaderMap,
     Json(payload): Json<GradeSubmissionPayload>,
 ) -> Result<Json<common::models::UserGrade>, (StatusCode, String)> {
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+
+    let ip = headers
+        .get("x-forwarded-for")
+        .and_then(|h| h.to_str().ok())
+        .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok()))
+        .map(|s| s.to_string());
+
+    let ua = headers
+        .get("user-agent")
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+
+    crate::db_util::set_session_context(
+        &mut tx,
+        Some(claims.sub),
+        Some(org_ctx.id),
+        ip,
+        ua,
+        Some("SYSTEM_EVENT".to_string()),
+    )
+    .await
+    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+
     // 1. Get lesson attempt rules
     let max_attempts: Option<Option<i32>> = sqlx::query_scalar(
         "SELECT max_attempts FROM lessons WHERE id = $1 AND organization_id = $2",
     )
     .bind(payload.lesson_id)
     .bind(org_ctx.id)
-    .fetch_optional(&pool)
+    .fetch_optional(&mut *tx)
     .await
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
@@ -420,19 +493,13 @@ pub async fn submit_lesson_score(
         .bind(payload.user_id)
         .bind(payload.lesson_id)
         .bind(org_ctx.id)
-        .fetch_optional(&pool)
+        .fetch_optional(&mut *tx)
         .await
         .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
     if let Some(count) = existing_attempts {
         if let Some(max) = max_attempts {
             if count >= max {
-                tracing::warn!(
-                    "User {} attempted to resubmit lesson {} but reached max_attempts ({})",
-                    payload.user_id,
-                    payload.lesson_id,
-                    max
-                );
                 return Err((
                     StatusCode::FORBIDDEN,
                     "Maximum attempts reached for this assessment".into(),
@@ -441,68 +508,66 @@ pub async fn submit_lesson_score(
         }
     }
 
-    // 3. Upsert with increment
+    // 3. Upsert with automated DB logic (XP, Badges)
     let grade = sqlx::query_as::<_, common::models::UserGrade>(
-        "INSERT INTO user_grades (user_id, course_id, lesson_id, score, metadata, attempts_count, organization_id)
-         VALUES ($1, $2, $3, $4, $5, 1, $6)
-         ON CONFLICT (user_id, lesson_id) DO UPDATE SET
-            score = EXCLUDED.score,
-            metadata = EXCLUDED.metadata,
-            attempts_count = user_grades.attempts_count + 1,
-            created_at = CURRENT_TIMESTAMP
-         RETURNING *"
+        "SELECT * FROM fn_upsert_user_grade($1, $2, $3, $4, $5, $6)",
     )
+    .bind(org_ctx.id)
     .bind(payload.user_id)
     .bind(payload.course_id)
     .bind(payload.lesson_id)
     .bind(payload.score)
     .bind(payload.metadata)
-    .bind(org_ctx.id)
-    .fetch_one(&pool)
+    .fetch_one(&mut *tx)
     .await
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
-    // 4. Grant Points
-    let points_to_grant = 20; // Base points for any lesson
-    let _ = sqlx::query(
-        "INSERT INTO points_log (user_id, organization_id, amount, reason, entity_type, entity_id) VALUES ($1, $2, $3, $4, $5, $6)"
+    tx.commit()
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+
+    // 4. Dispatch Webhooks
+    let webhook_service = common::webhooks::WebhookService::new(pool.clone());
+
+    // lesson.completed
+    webhook_service
+        .dispatch(
+            org_ctx.id,
+            "lesson.completed",
+            &serde_json::json!({
+                "user_id": payload.user_id,
+                "course_id": payload.course_id,
+                "lesson_id": payload.lesson_id,
+                "score": payload.score
+            }),
+        )
+        .await;
+
+    // TODO: Detect course completion logic
+    let total_lessons: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM lessons WHERE module_id IN (SELECT id FROM modules WHERE course_id = $1)")
+        .bind(payload.course_id)
+        .fetch_one(&pool).await.unwrap_or(0);
+
+    let completed_lessons: i64 = sqlx::query_scalar(
+        "SELECT COUNT(*) FROM user_grades WHERE user_id = $1 AND course_id = $2",
     )
     .bind(payload.user_id)
-    .bind(org_ctx.id)
-    .bind(points_to_grant)
-    .bind("lesson_completion")
-    .bind("lesson")
-    .bind(payload.lesson_id)
-    .execute(&pool)
-    .await;
+    .bind(payload.course_id)
+    .fetch_one(&pool)
+    .await
+    .unwrap_or(0);
 
-    // 5. Check for new badges (Trigger-like logic in code)
-    // For now, very simple: if they reached a points threshold
-    let total_points: i64 =
-        sqlx::query_scalar("SELECT COALESCE(SUM(amount), 0) FROM points_log WHERE user_id = $1")
-            .bind(payload.user_id)
-            .fetch_one(&pool)
-            .await
-            .unwrap_or(0);
-
-    let eligible_badges = sqlx::query(
-        "SELECT id FROM badges WHERE organization_id = $1 AND requirement_type = 'points' AND requirement_value <= $2 AND id NOT IN (SELECT badge_id FROM user_badges WHERE user_id = $3)"
-    )
-    .bind(org_ctx.id)
-    .bind(total_points as i32)
-    .bind(payload.user_id)
-    .fetch_all(&pool)
-    .await;
-
-    if let Ok(new_badges) = eligible_badges {
-        for b in new_badges {
-            let badge_id: Uuid = b.get("id");
-            let _ = sqlx::query("INSERT INTO user_badges (user_id, badge_id) VALUES ($1, $2) ON CONFLICT DO NOTHING")
-                .bind(payload.user_id)
-                .bind(badge_id)
-                .execute(&pool)
-                .await;
-        }
+    if total_lessons > 0 && completed_lessons >= total_lessons {
+        webhook_service
+            .dispatch(
+                org_ctx.id,
+                "course.completed",
+                &serde_json::json!({
+                    "user_id": payload.user_id,
+                    "course_id": payload.course_id
+                }),
+            )
+            .await;
     }
 
     Ok(Json(grade))
@@ -511,6 +576,7 @@ pub async fn submit_lesson_score(
 #[derive(serde::Serialize)]
 pub struct GamificationStatus {
     pub points: i64,
+    pub level: i32,
     pub badges: Vec<BadgeResponse>,
 }
 
@@ -528,12 +594,11 @@ pub async fn get_user_gamification(
     State(pool): State<PgPool>,
     Path(user_id): Path<Uuid>,
 ) -> Result<Json<GamificationStatus>, StatusCode> {
-    let points: i64 =
-        sqlx::query_scalar("SELECT COALESCE(SUM(amount), 0) FROM points_log WHERE user_id = $1")
-            .bind(user_id)
-            .fetch_one(&pool)
-            .await
-            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    let user_stats: (i32, i32) = sqlx::query_as("SELECT xp, level FROM users WHERE id = $1")
+        .bind(user_id)
+        .fetch_one(&pool)
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
     let badges = sqlx::query_as::<_, BadgeResponse>(
         "SELECT b.id, b.name, b.description, b.icon_url, ub.earned_at 
@@ -546,7 +611,42 @@ pub async fn get_user_gamification(
     .await
     .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
-    Ok(Json(GamificationStatus { points, badges }))
+    Ok(Json(GamificationStatus {
+        points: user_stats.0 as i64,
+        level: user_stats.1,
+        badges,
+    }))
+}
+
+pub async fn get_leaderboard(
+    Org(org_ctx): Org,
+    State(pool): State<PgPool>,
+) -> Result<Json<Vec<UserResponse>>, StatusCode> {
+    let top_users = sqlx::query_as::<_, User>(
+        "SELECT * FROM users WHERE organization_id = $1 ORDER BY xp DESC LIMIT 10",
+    )
+    .bind(org_ctx.id)
+    .fetch_all(&pool)
+    .await
+    .map_err(|e| {
+        tracing::error!("Failed to fetch leaderboard: {}", e);
+        StatusCode::INTERNAL_SERVER_ERROR
+    })?;
+
+    let response = top_users
+        .into_iter()
+        .map(|u| UserResponse {
+            id: u.id,
+            email: u.email,
+            full_name: u.full_name,
+            role: u.role,
+            organization_id: u.organization_id,
+            xp: u.xp,
+            level: u.level,
+        })
+        .collect();
+
+    Ok(Json(response))
 }
 
 pub async fn get_user_course_grades(
@@ -630,3 +730,38 @@ pub async fn get_course_analytics(
         lessons,
     }))
 }
+
+pub async fn get_advanced_analytics(
+    Org(_org_ctx): Org,
+    State(pool): State<PgPool>,
+    Path(course_id): Path<Uuid>,
+) -> Result<Json<common::models::AdvancedAnalytics>, StatusCode> {
+    // 1. Cohort Analysis using DB function
+    let cohort_data = sqlx::query_as::<_, common::models::CohortData>(
+        "SELECT period, student_count as count, completion_rate FROM fn_get_cohort_analytics($1)",
+    )
+    .bind(course_id)
+    .fetch_all(&pool)
+    .await
+    .map_err(|e| {
+        tracing::error!("Cohort query failed: {}", e);
+        StatusCode::INTERNAL_SERVER_ERROR
+    })?;
+
+    // 2. Retention Analysis using DB function
+    let retention_data = sqlx::query_as::<_, common::models::RetentionData>(
+        "SELECT lesson_id, lesson_title, student_count FROM fn_get_retention_data($1)",
+    )
+    .bind(course_id)
+    .fetch_all(&pool)
+    .await
+    .map_err(|e| {
+        tracing::error!("Retention query failed: {}", e);
+        StatusCode::INTERNAL_SERVER_ERROR
+    })?;
+
+    Ok(Json(common::models::AdvancedAnalytics {
+        cohorts: cohort_data,
+        retention: retention_data,
+    }))
+}

services/lms-service/src/main.rs

@@ -1,15 +1,15 @@
+mod db_util;
 mod handlers;
 
 use axum::{
+    Router, middleware,
     routing::{get, post},
-    Router,
-    middleware,
 };
-use tower_http::cors::{Any, CorsLayer};
-use sqlx::postgres::PgPoolOptions;
-use std::net::SocketAddr;
 use dotenvy::dotenv;
+use sqlx::postgres::PgPoolOptions;
 use std::env;
+use std::net::SocketAddr;
+use tower_http::cors::{Any, CorsLayer};
 
 #[tokio::main]
 async fn main() {
@@ -40,10 +40,26 @@ async fn main() {
         .route("/courses/{id}/outline", get(handlers::get_course_outline))
         .route("/lessons/{id}", get(handlers::get_lesson_content))
         .route("/grades", post(handlers::submit_lesson_score))
-        .route("/users/{user_id}/courses/{course_id}/grades", get(handlers::get_user_course_grades))
-        .route("/courses/{id}/analytics", get(handlers::get_course_analytics))
-        .route("/users/{id}/gamification", get(handlers::get_user_gamification))
-        .route_layer(middleware::from_fn(common::middleware::org_extractor_middleware));
+        .route(
+            "/users/{user_id}/courses/{course_id}/grades",
+            get(handlers::get_user_course_grades),
+        )
+        .route(
+            "/courses/{id}/analytics",
+            get(handlers::get_course_analytics),
+        )
+        .route(
+            "/courses/{id}/analytics/advanced",
+            get(handlers::get_advanced_analytics),
+        )
+        .route(
+            "/users/{id}/gamification",
+            get(handlers::get_user_gamification),
+        )
+        .route("/analytics/leaderboard", get(handlers::get_leaderboard))
+        .route_layer(middleware::from_fn(
+            common::middleware::org_extractor_middleware,
+        ));
 
     let public_routes = Router::new()
         .route("/catalog", get(handlers::get_course_catalog))