ADR LMS 010: External LMS Integration API

ADR-LMS-010: External LMS Integration API

Status: Proposed Date: 2025-12-11 Phase: Phase 3 - Enterprise Integration Deciders: Hal Casteel (Founder/CEO/CTO), CODITECT Core Team Technical Story: Enable CODITECT training content to be consumed by external LMS platforms (Moodle, Canvas, Blackboard, Cornerstone) and allow external systems to integrate with CODITECT via RESTful APIs and webhooks

---

Context and Problem Statement

Enterprise organizations typically have existing LMS investments:

1. LMS Lock-in - Organizations can't easily adopt CODITECT alongside existing LMS
2. Content Portability - CODITECT training content is isolated
3. Data Silos - Learning data doesn't flow to HR/talent systems
4. SSO Requirements - Users expect single sign-on across all learning tools
5. Compliance Integration - Training records must sync with compliance systems
6. Automation Needs - No programmatic access for workflow automation

The Problem: How do we enable bidirectional integration with external LMS platforms and enterprise systems while maintaining data integrity, security, and a seamless user experience?

---

Decision Drivers

Technical Requirements

• R1: RESTful API with OpenAPI specification
• R2: Webhook system for event notifications
• R3: LTI 1.3 (Learning Tools Interoperability) support
• R4: SCORM/xAPI content packaging for export
• R5: OAuth 2.0 for third-party authorization
• R6: Rate limiting and usage quotas
• R7: API versioning strategy

Integration Requirements

• I1: Moodle, Canvas, Blackboard, Cornerstone compatibility
• I2: SSO federation (SAML 2.0, OIDC)
• I3: HR system sync (Workday, SAP SuccessFactors)
• I4: Webhook delivery with retry logic
• I5: Bulk data export capabilities

Security Requirements

• S1: API key and OAuth token management
• S2: Scope-based access control
• S3: Request signing for webhooks
• S4: Audit logging of all API calls
• S5: IP allowlisting option

---

Decision Outcome

Chosen Solution: Implement a comprehensive integration layer with RESTful API, LTI 1.3 provider, webhook system, and pre-built connectors for major LMS platforms.

Architecture Overview

┌─────────────────────────────────────────────────────────────────────┐
│              External LMS Integration Architecture                   │
├─────────────────────────────────────────────────────────────────────┤
│                                                                       │
│  ┌─────────────────────────────────────────────────────────────┐    │
│  │                    API Gateway Layer                         │    │
│  │                                                              │    │
│  │  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │    │
│  │  │ Rate        │  │ Auth        │  │ Request             │ │    │
│  │  │ Limiting    │──│ Middleware  │──│ Routing             │ │    │
│  │  │             │  │ (API Key/   │  │                     │ │    │
│  │  │ 1000/min    │  │  OAuth)     │  │ /api/v1/*           │ │    │
│  │  └─────────────┘  └─────────────┘  └─────────────────────┘ │    │
│  └─────────────────────────────────────────────────────────────┘    │
│                               │                                      │
│           ┌───────────────────┼───────────────────┐                 │
│           │                   │                   │                 │
│           ▼                   ▼                   ▼                 │
│  ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐       │
│  │   REST API      │ │   LTI 1.3       │ │   Webhooks      │       │
│  │                 │ │   Provider      │ │   System        │       │
│  │  /courses       │ │                 │ │                 │       │
│  │  /enrollments   │ │  Launch URL     │ │  course.created │       │
│  │  /users         │ │  Deep Linking   │ │  user.enrolled  │       │
│  │  /analytics     │ │  Grade Passback │ │  progress.update│       │
│  │  /certificates  │ │  NRPS           │ │  cert.issued    │       │
│  └─────────────────┘ └─────────────────┘ └─────────────────┘       │
│           │                   │                   │                 │
│           └───────────────────┼───────────────────┘                 │
│                               │                                      │
│                               ▼                                      │
│  ┌─────────────────────────────────────────────────────────────┐    │
│  │                   Integration Services                       │    │
│  │                                                              │    │
│  │  ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐   │    │
│  │  │  Moodle   │ │  Canvas   │ │ Blackboard│ │ Custom    │   │    │
│  │  │ Connector │ │ Connector │ │ Connector │ │ Connector │   │    │
│  │  └───────────┘ └───────────┘ └───────────┘ └───────────┘   │    │
│  │                                                              │    │
│  │  ┌───────────────────────────────────────────────────────┐  │    │
│  │  │            Webhook Delivery Engine                     │  │    │
│  │  │  • Guaranteed delivery with retry                      │  │    │
│  │  │  • HMAC signature verification                         │  │    │
│  │  │  • Dead letter queue                                   │  │    │
│  │  └───────────────────────────────────────────────────────┘  │    │
│  └─────────────────────────────────────────────────────────────┘    │
│                               │                                      │
│                               ▼                                      │
│  ┌─────────────────────────────────────────────────────────────┐    │
│  │                    Data Layer                                │    │
│  │                                                              │    │
│  │  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │    │
│  │  │ API Keys    │  │ Webhooks    │  │ Integration         │ │    │
│  │  │ OAuth Apps  │  │ Endpoints   │  │ Sync Status         │ │    │
│  │  │ Scopes      │  │ Events      │  │ Audit Log           │ │    │
│  │  └─────────────┘  └─────────────┘  └─────────────────────┘ │    │
│  └─────────────────────────────────────────────────────────────┘    │
│                                                                       │
└─────────────────────────────────────────────────────────────────────┘

---

Database Schema

New Tables

-- API Applications (OAuth clients and API keys)
CREATE TABLE IF NOT EXISTS api_applications (
    id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
    organization_id TEXT NOT NULL,
    name TEXT NOT NULL,
    description TEXT,
    client_id TEXT UNIQUE NOT NULL,
    client_secret_hash TEXT,
    api_key_hash TEXT,
    application_type TEXT DEFAULT 'api' CHECK(application_type IN ('api', 'oauth', 'lti')),
    redirect_uris JSON DEFAULT '[]',
    scopes JSON DEFAULT '[]',
    rate_limit_per_minute INTEGER DEFAULT 1000,
    ip_allowlist JSON DEFAULT '[]',
    is_active INTEGER DEFAULT 1,
    created_by TEXT NOT NULL,
    created_at TEXT DEFAULT (datetime('now')),
    updated_at TEXT DEFAULT (datetime('now')),
    last_used_at TEXT,
    FOREIGN KEY (organization_id) REFERENCES organizations(id),
    FOREIGN KEY (created_by) REFERENCES auth_users(id)
);

-- OAuth Access Tokens
CREATE TABLE IF NOT EXISTS oauth_tokens (
    id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
    application_id TEXT NOT NULL,
    user_id TEXT,
    access_token_hash TEXT NOT NULL,
    refresh_token_hash TEXT,
    scopes JSON NOT NULL,
    expires_at TEXT NOT NULL,
    created_at TEXT DEFAULT (datetime('now')),
    revoked_at TEXT,
    FOREIGN KEY (application_id) REFERENCES api_applications(id),
    FOREIGN KEY (user_id) REFERENCES auth_users(id)
);

-- Webhook Endpoints
CREATE TABLE IF NOT EXISTS webhook_endpoints (
    id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
    organization_id TEXT NOT NULL,
    url TEXT NOT NULL,
    secret_hash TEXT NOT NULL,
    events JSON NOT NULL,
    is_active INTEGER DEFAULT 1,
    created_by TEXT NOT NULL,
    created_at TEXT DEFAULT (datetime('now')),
    updated_at TEXT DEFAULT (datetime('now')),
    last_triggered_at TEXT,
    failure_count INTEGER DEFAULT 0,
    FOREIGN KEY (organization_id) REFERENCES organizations(id),
    FOREIGN KEY (created_by) REFERENCES auth_users(id)
);

-- Webhook Deliveries (for retry and audit)
CREATE TABLE IF NOT EXISTS webhook_deliveries (
    id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
    endpoint_id TEXT NOT NULL,
    event_type TEXT NOT NULL,
    payload JSON NOT NULL,
    status TEXT DEFAULT 'pending' CHECK(status IN ('pending', 'delivered', 'failed', 'dead_letter')),
    attempt_count INTEGER DEFAULT 0,
    last_attempt_at TEXT,
    next_retry_at TEXT,
    response_status INTEGER,
    response_body TEXT,
    error_message TEXT,
    created_at TEXT DEFAULT (datetime('now')),
    delivered_at TEXT,
    FOREIGN KEY (endpoint_id) REFERENCES webhook_endpoints(id)
);

-- LTI Platforms (external LMS registrations)
CREATE TABLE IF NOT EXISTS lti_platforms (
    id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
    organization_id TEXT NOT NULL,
    name TEXT NOT NULL,
    issuer TEXT UNIQUE NOT NULL,
    client_id TEXT NOT NULL,
    deployment_id TEXT,
    auth_login_url TEXT NOT NULL,
    auth_token_url TEXT NOT NULL,
    keyset_url TEXT NOT NULL,
    public_keyset JSON,
    private_key TEXT NOT NULL,
    is_active INTEGER DEFAULT 1,
    created_at TEXT DEFAULT (datetime('now')),
    updated_at TEXT DEFAULT (datetime('now')),
    FOREIGN KEY (organization_id) REFERENCES organizations(id)
);

-- LTI Resource Links (content links in external LMS)
CREATE TABLE IF NOT EXISTS lti_resource_links (
    id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
    platform_id TEXT NOT NULL,
    resource_link_id TEXT NOT NULL,
    course_id TEXT,
    lesson_id TEXT,
    title TEXT,
    custom_parameters JSON DEFAULT '{}',
    created_at TEXT DEFAULT (datetime('now')),
    FOREIGN KEY (platform_id) REFERENCES lti_platforms(id),
    FOREIGN KEY (course_id) REFERENCES courses(id),
    FOREIGN KEY (lesson_id) REFERENCES course_lessons(id),
    UNIQUE(platform_id, resource_link_id)
);

-- LTI Grade Submissions
CREATE TABLE IF NOT EXISTS lti_grade_submissions (
    id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
    resource_link_id TEXT NOT NULL,
    user_id TEXT NOT NULL,
    score REAL NOT NULL,
    score_max REAL DEFAULT 100,
    comment TEXT,
    submitted_at TEXT DEFAULT (datetime('now')),
    synced_at TEXT,
    sync_status TEXT DEFAULT 'pending' CHECK(sync_status IN ('pending', 'synced', 'failed')),
    sync_error TEXT,
    FOREIGN KEY (resource_link_id) REFERENCES lti_resource_links(id),
    FOREIGN KEY (user_id) REFERENCES auth_users(id)
);

-- API Audit Log
CREATE TABLE IF NOT EXISTS api_audit_log (
    id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
    application_id TEXT,
    user_id TEXT,
    method TEXT NOT NULL,
    path TEXT NOT NULL,
    query_params TEXT,
    request_body_hash TEXT,
    response_status INTEGER,
    response_time_ms INTEGER,
    ip_address TEXT,
    user_agent TEXT,
    timestamp TEXT DEFAULT (datetime('now')),
    FOREIGN KEY (application_id) REFERENCES api_applications(id),
    FOREIGN KEY (user_id) REFERENCES auth_users(id)
);

-- Integration Sync Status
CREATE TABLE IF NOT EXISTS integration_sync_status (
    id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
    organization_id TEXT NOT NULL,
    integration_type TEXT NOT NULL,
    external_system TEXT NOT NULL,
    last_sync_at TEXT,
    last_sync_status TEXT,
    records_synced INTEGER DEFAULT 0,
    errors JSON DEFAULT '[]',
    next_sync_at TEXT,
    FOREIGN KEY (organization_id) REFERENCES organizations(id)
);

-- Indexes
CREATE INDEX IF NOT EXISTS idx_api_applications_org ON api_applications(organization_id);
CREATE INDEX IF NOT EXISTS idx_api_applications_client ON api_applications(client_id);
CREATE INDEX IF NOT EXISTS idx_oauth_tokens_app ON oauth_tokens(application_id);
CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_status ON webhook_deliveries(status, next_retry_at);
CREATE INDEX IF NOT EXISTS idx_api_audit_log_app ON api_audit_log(application_id, timestamp);
CREATE INDEX IF NOT EXISTS idx_lti_platforms_issuer ON lti_platforms(issuer);

---

Implementation

REST API Service

# coditect_lms/api/rest_api.py
"""
REST API Service for CODITECT LMS.
Provides external API access with rate limiting and authentication.
"""

import sqlite3
import json
import hashlib
import hmac
import secrets
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from functools import wraps

@dataclass
class APIContext:
    application_id: str
    organization_id: str
    user_id: Optional[str]
    scopes: List[str]

class APIService:
    def __init__(self, db_path: str = "~/.coditect/lms.db"):
        import os
        self.db_path = os.path.expanduser(db_path)

    def _get_connection(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn

    # =====================
    # Application Management
    # =====================

    def create_application(
        self,
        organization_id: str,
        name: str,
        created_by: str,
        application_type: str = "api",
        scopes: List[str] = None,
        description: str = None
    ) -> Dict[str, str]:
        """Create a new API application with credentials."""
        conn = self._get_connection()
        try:
            cursor = conn.cursor()

            app_id = secrets.token_hex(16)
            client_id = f"coditect_{secrets.token_hex(8)}"
            client_secret = secrets.token_urlsafe(32)
            api_key = f"sk_live_{secrets.token_urlsafe(32)}"

            cursor.execute("""
                INSERT INTO api_applications (
                    id, organization_id, name, description,
                    client_id, client_secret_hash, api_key_hash,
                    application_type, scopes, created_by
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                app_id,
                organization_id,
                name,
                description,
                client_id,
                self._hash_secret(client_secret),
                self._hash_secret(api_key),
                application_type,
                json.dumps(scopes or ["read"]),
                created_by
            ))

            conn.commit()

            # Return credentials (only shown once!)
            return {
                "application_id": app_id,
                "client_id": client_id,
                "client_secret": client_secret,  # Only returned at creation
                "api_key": api_key,  # Only returned at creation
                "message": "Store these credentials securely - they cannot be retrieved again"
            }
        finally:
            conn.close()

    def authenticate_api_key(self, api_key: str) -> Optional[APIContext]:
        """Authenticate request using API key."""
        conn = self._get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute("""
                SELECT id, organization_id, scopes, is_active
                FROM api_applications
                WHERE api_key_hash = ? AND is_active = 1
            """, (self._hash_secret(api_key),))

            row = cursor.fetchone()
            if row:
                # Update last used
                cursor.execute(
                    "UPDATE api_applications SET last_used_at = datetime('now') WHERE id = ?",
                    (row['id'],)
                )
                conn.commit()

                return APIContext(
                    application_id=row['id'],
                    organization_id=row['organization_id'],
                    user_id=None,
                    scopes=json.loads(row['scopes'])
                )
            return None
        finally:
            conn.close()

    def check_rate_limit(self, application_id: str) -> bool:
        """Check if application is within rate limit."""
        conn = self._get_connection()
        try:
            cursor = conn.cursor()

            # Get rate limit
            cursor.execute(
                "SELECT rate_limit_per_minute FROM api_applications WHERE id = ?",
                (application_id,)
            )
            app = cursor.fetchone()
            limit = app['rate_limit_per_minute'] if app else 1000

            # Count recent requests
            cursor.execute("""
                SELECT COUNT(*) as count FROM api_audit_log
                WHERE application_id = ?
                AND timestamp > datetime('now', '-1 minute')
            """, (application_id,))

            count = cursor.fetchone()['count']
            return count < limit
        finally:
            conn.close()

    def log_api_call(
        self,
        application_id: str,
        method: str,
        path: str,
        response_status: int,
        response_time_ms: int,
        user_id: str = None,
        query_params: str = None,
        ip_address: str = None,
        user_agent: str = None
    ):
        """Log API call for audit and analytics."""
        conn = self._get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT INTO api_audit_log (
                    id, application_id, user_id, method, path,
                    query_params, response_status, response_time_ms,
                    ip_address, user_agent
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                secrets.token_hex(16),
                application_id,
                user_id,
                method,
                path,
                query_params,
                response_status,
                response_time_ms,
                ip_address,
                user_agent
            ))
            conn.commit()
        finally:
            conn.close()

    def _hash_secret(self, secret: str) -> str:
        """Hash a secret for storage."""
        return hashlib.sha256(secret.encode()).hexdigest()

    # =====================
    # API Endpoints
    # =====================

    def get_courses(
        self,
        context: APIContext,
        status: str = None,
        limit: int = 50,
        offset: int = 0
    ) -> Dict[str, Any]:
        """GET /api/v1/courses - List courses."""
        if "read" not in context.scopes and "courses:read" not in context.scopes:
            raise PermissionError("Missing required scope: courses:read")

        conn = self._get_connection()
        try:
            cursor = conn.cursor()

            query = """
                SELECT id, title, slug, description, status,
                       visibility, created_at, updated_at, published_at
                FROM courses
                WHERE organization_id = ?
            """
            params = [context.organization_id]

            if status:
                query += " AND status = ?"
                params.append(status)

            query += " ORDER BY updated_at DESC LIMIT ? OFFSET ?"
            params.extend([limit, offset])

            cursor.execute(query, params)
            courses = [dict(row) for row in cursor.fetchall()]

            # Get total count
            count_query = "SELECT COUNT(*) FROM courses WHERE organization_id = ?"
            cursor.execute(count_query, [context.organization_id])
            total = cursor.fetchone()[0]

            return {
                "data": courses,
                "meta": {
                    "total": total,
                    "limit": limit,
                    "offset": offset,
                    "has_more": offset + limit < total
                }
            }
        finally:
            conn.close()

    def get_enrollments(
        self,
        context: APIContext,
        course_id: str = None,
        user_id: str = None,
        status: str = None,
        limit: int = 100,
        offset: int = 0
    ) -> Dict[str, Any]:
        """GET /api/v1/enrollments - List enrollments."""
        if "read" not in context.scopes and "enrollments:read" not in context.scopes:
            raise PermissionError("Missing required scope: enrollments:read")

        conn = self._get_connection()
        try:
            cursor = conn.cursor()

            query = """
                SELECT e.id, e.course_id, e.user_id, e.status,
                       e.enrollment_type, e.enrolled_at, e.completed_at,
                       e.progress_percent, u.email, u.display_name,
                       c.title as course_title
                FROM course_enrollments e
                JOIN auth_users u ON e.user_id = u.id
                JOIN courses c ON e.course_id = c.id
                WHERE c.organization_id = ?
            """
            params = [context.organization_id]

            if course_id:
                query += " AND e.course_id = ?"
                params.append(course_id)
            if user_id:
                query += " AND e.user_id = ?"
                params.append(user_id)
            if status:
                query += " AND e.status = ?"
                params.append(status)

            query += " ORDER BY e.enrolled_at DESC LIMIT ? OFFSET ?"
            params.extend([limit, offset])

            cursor.execute(query, params)
            enrollments = [dict(row) for row in cursor.fetchall()]

            return {
                "data": enrollments,
                "meta": {
                    "limit": limit,
                    "offset": offset
                }
            }
        finally:
            conn.close()

    def create_enrollment(
        self,
        context: APIContext,
        course_id: str,
        user_email: str
    ) -> Dict[str, Any]:
        """POST /api/v1/enrollments - Create enrollment."""
        if "write" not in context.scopes and "enrollments:write" not in context.scopes:
            raise PermissionError("Missing required scope: enrollments:write")

        conn = self._get_connection()
        try:
            cursor = conn.cursor()

            # Verify course belongs to organization
            cursor.execute(
                "SELECT id FROM courses WHERE id = ? AND organization_id = ?",
                (course_id, context.organization_id)
            )
            if not cursor.fetchone():
                raise ValueError("Course not found")

            # Find or create user
            cursor.execute("SELECT id FROM auth_users WHERE email = ?", (user_email,))
            user_row = cursor.fetchone()

            if user_row:
                user_id = user_row['id']
            else:
                # Create user account (pending activation)
                user_id = secrets.token_hex(16)
                cursor.execute("""
                    INSERT INTO auth_users (id, email, status)
                    VALUES (?, ?, 'pending')
                """, (user_id, user_email))

            # Create enrollment
            enrollment_id = secrets.token_hex(16)
            cursor.execute("""
                INSERT INTO course_enrollments (
                    id, course_id, user_id, enrollment_type
                ) VALUES (?, ?, ?, 'integration')
            """, (enrollment_id, course_id, user_id))

            conn.commit()

            # Trigger webhook
            self._trigger_webhook(
                context.organization_id,
                "enrollment.created",
                {
                    "enrollment_id": enrollment_id,
                    "course_id": course_id,
                    "user_id": user_id,
                    "user_email": user_email
                }
            )

            return {
                "data": {
                    "id": enrollment_id,
                    "course_id": course_id,
                    "user_id": user_id,
                    "status": "active"
                }
            }
        finally:
            conn.close()

    def get_analytics_summary(
        self,
        context: APIContext,
        start_date: str = None,
        end_date: str = None
    ) -> Dict[str, Any]:
        """GET /api/v1/analytics/summary - Get analytics summary."""
        if "read" not in context.scopes and "analytics:read" not in context.scopes:
            raise PermissionError("Missing required scope: analytics:read")

        conn = self._get_connection()
        try:
            cursor = conn.cursor()

            # Default to last 30 days
            if not start_date:
                start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
            if not end_date:
                end_date = datetime.now().strftime('%Y-%m-%d')

            # Course stats
            cursor.execute("""
                SELECT COUNT(*) as total_courses,
                       SUM(CASE WHEN status = 'published' THEN 1 ELSE 0 END) as published
                FROM courses WHERE organization_id = ?
            """, (context.organization_id,))
            course_stats = dict(cursor.fetchone())

            # Enrollment stats
            cursor.execute("""
                SELECT COUNT(*) as total_enrollments,
                       SUM(CASE WHEN e.status = 'completed' THEN 1 ELSE 0 END) as completed,
                       AVG(e.progress_percent) as avg_progress
                FROM course_enrollments e
                JOIN courses c ON e.course_id = c.id
                WHERE c.organization_id = ?
                AND e.enrolled_at BETWEEN ? AND ?
            """, (context.organization_id, start_date, end_date))
            enrollment_stats = dict(cursor.fetchone())

            # Certificate stats
            cursor.execute("""
                SELECT COUNT(*) as certificates_issued
                FROM cert_issued ci
                JOIN cert_definitions cd ON ci.definition_id = cd.id
                WHERE cd.organization_id = ?
                AND ci.issued_at BETWEEN ? AND ?
            """, (context.organization_id, start_date, end_date))
            cert_stats = dict(cursor.fetchone())

            return {
                "data": {
                    "period": {"start": start_date, "end": end_date},
                    "courses": course_stats,
                    "enrollments": enrollment_stats,
                    "certificates": cert_stats
                }
            }
        finally:
            conn.close()

    def _trigger_webhook(
        self,
        organization_id: str,
        event_type: str,
        payload: Dict[str, Any]
    ):
        """Queue webhook delivery for an event."""
        conn = self._get_connection()
        try:
            cursor = conn.cursor()

            # Find matching webhooks
            cursor.execute("""
                SELECT id, url, events FROM webhook_endpoints
                WHERE organization_id = ? AND is_active = 1
            """, (organization_id,))

            for row in cursor.fetchall():
                events = json.loads(row['events'])
                if event_type in events or "*" in events:
                    # Queue delivery
                    cursor.execute("""
                        INSERT INTO webhook_deliveries (
                            id, endpoint_id, event_type, payload, next_retry_at
                        ) VALUES (?, ?, ?, ?, datetime('now'))
                    """, (
                        secrets.token_hex(16),
                        row['id'],
                        event_type,
                        json.dumps(payload)
                    ))

            conn.commit()
        finally:
            conn.close()

Webhook Delivery Service

# coditect_lms/services/webhook_service.py
"""
Webhook Delivery Service for CODITECT LMS.
Handles reliable webhook delivery with retries and signing.
"""

import sqlite3
import json
import hmac
import hashlib
import time
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
import urllib.request
import urllib.error

class WebhookService:
    RETRY_DELAYS = [60, 300, 900, 3600, 14400]  # 1m, 5m, 15m, 1h, 4h

    def __init__(self, db_path: str = "~/.coditect/lms.db"):
        import os
        self.db_path = os.path.expanduser(db_path)

    def _get_connection(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn

    def create_webhook(
        self,
        organization_id: str,
        url: str,
        events: List[str],
        created_by: str
    ) -> Dict[str, str]:
        """Create a new webhook endpoint."""
        import secrets as sec

        webhook_id = sec.token_hex(16)
        secret = f"whsec_{sec.token_urlsafe(32)}"

        conn = self._get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT INTO webhook_endpoints (
                    id, organization_id, url, secret_hash, events, created_by
                ) VALUES (?, ?, ?, ?, ?, ?)
            """, (
                webhook_id,
                organization_id,
                url,
                hashlib.sha256(secret.encode()).hexdigest(),
                json.dumps(events),
                created_by
            ))
            conn.commit()

            return {
                "webhook_id": webhook_id,
                "secret": secret,  # Only returned at creation
                "message": "Store the secret securely - it cannot be retrieved again"
            }
        finally:
            conn.close()

    def process_pending_deliveries(self, batch_size: int = 100):
        """Process pending webhook deliveries."""
        conn = self._get_connection()
        try:
            cursor = conn.cursor()

            # Get pending deliveries ready for retry
            cursor.execute("""
                SELECT d.id, d.endpoint_id, d.event_type, d.payload,
                       d.attempt_count, e.url, e.secret_hash
                FROM webhook_deliveries d
                JOIN webhook_endpoints e ON d.endpoint_id = e.id
                WHERE d.status = 'pending'
                AND d.next_retry_at <= datetime('now')
                AND e.is_active = 1
                ORDER BY d.next_retry_at
                LIMIT ?
            """, (batch_size,))

            deliveries = cursor.fetchall()

            for delivery in deliveries:
                self._deliver_webhook(dict(delivery))

        finally:
            conn.close()

    def _deliver_webhook(self, delivery: Dict[str, Any]):
        """Attempt to deliver a single webhook."""
        conn = self._get_connection()
        try:
            cursor = conn.cursor()

            # Build payload
            payload_data = {
                "id": delivery['id'],
                "event": delivery['event_type'],
                "timestamp": datetime.utcnow().isoformat() + "Z",
                "data": json.loads(delivery['payload'])
            }
            payload_json = json.dumps(payload_data)

            # Sign payload
            signature = self._sign_payload(payload_json, delivery['secret_hash'])

            # Attempt delivery
            try:
                start_time = time.time()

                req = urllib.request.Request(
                    delivery['url'],
                    data=payload_json.encode('utf-8'),
                    headers={
                        'Content-Type': 'application/json',
                        'X-Webhook-Signature': signature,
                        'X-Webhook-ID': delivery['id'],
                        'X-Webhook-Timestamp': str(int(time.time())),
                        'User-Agent': 'CODITECT-Webhook/1.0'
                    },
                    method='POST'
                )

                with urllib.request.urlopen(req, timeout=30) as response:
                    response_time = int((time.time() - start_time) * 1000)
                    response_body = response.read().decode('utf-8')[:1000]

                    # Success!
                    cursor.execute("""
                        UPDATE webhook_deliveries
                        SET status = 'delivered',
                            attempt_count = attempt_count + 1,
                            last_attempt_at = datetime('now'),
                            delivered_at = datetime('now'),
                            response_status = ?,
                            response_body = ?
                        WHERE id = ?
                    """, (response.status, response_body, delivery['id']))

                    # Reset failure count on endpoint
                    cursor.execute("""
                        UPDATE webhook_endpoints
                        SET failure_count = 0,
                            last_triggered_at = datetime('now')
                        WHERE id = ?
                    """, (delivery['endpoint_id'],))

            except urllib.error.HTTPError as e:
                self._handle_delivery_failure(
                    cursor, delivery,
                    e.code, str(e.reason)
                )

            except Exception as e:
                self._handle_delivery_failure(
                    cursor, delivery,
                    0, str(e)
                )

            conn.commit()

        finally:
            conn.close()

    def _handle_delivery_failure(
        self,
        cursor,
        delivery: Dict[str, Any],
        status_code: int,
        error: str
    ):
        """Handle failed webhook delivery."""
        attempt_count = delivery['attempt_count'] + 1

        if attempt_count >= len(self.RETRY_DELAYS):
            # Move to dead letter queue
            cursor.execute("""
                UPDATE webhook_deliveries
                SET status = 'dead_letter',
                    attempt_count = ?,
                    last_attempt_at = datetime('now'),
                    response_status = ?,
                    error_message = ?
                WHERE id = ?
            """, (attempt_count, status_code, error, delivery['id']))

            # Increment endpoint failure count
            cursor.execute("""
                UPDATE webhook_endpoints
                SET failure_count = failure_count + 1
                WHERE id = ?
            """, (delivery['endpoint_id'],))

        else:
            # Schedule retry
            retry_delay = self.RETRY_DELAYS[attempt_count - 1]
            cursor.execute("""
                UPDATE webhook_deliveries
                SET attempt_count = ?,
                    last_attempt_at = datetime('now'),
                    next_retry_at = datetime('now', '+' || ? || ' seconds'),
                    response_status = ?,
                    error_message = ?
                WHERE id = ?
            """, (attempt_count, retry_delay, status_code, error, delivery['id']))

    def _sign_payload(self, payload: str, secret_hash: str) -> str:
        """Create HMAC signature for webhook payload."""
        # In production, we'd use the actual secret, not hash
        # This is a simplified version
        timestamp = str(int(time.time()))
        signed_payload = f"{timestamp}.{payload}"
        signature = hmac.new(
            secret_hash.encode(),
            signed_payload.encode(),
            hashlib.sha256
        ).hexdigest()
        return f"v1={signature}"

    def get_delivery_history(
        self,
        endpoint_id: str,
        limit: int = 50
    ) -> List[Dict[str, Any]]:
        """Get delivery history for an endpoint."""
        conn = self._get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute("""
                SELECT id, event_type, status, attempt_count,
                       created_at, delivered_at, response_status, error_message
                FROM webhook_deliveries
                WHERE endpoint_id = ?
                ORDER BY created_at DESC
                LIMIT ?
            """, (endpoint_id, limit))

            return [dict(row) for row in cursor.fetchall()]
        finally:
            conn.close()

LTI 1.3 Provider Service

# coditect_lms/services/lti_service.py
"""
LTI 1.3 Provider Service for CODITECT LMS.
Implements Learning Tools Interoperability for external LMS integration.
"""

import sqlite3
import json
import jwt
import time
from datetime import datetime
from typing import Optional, Dict, Any
from dataclasses import dataclass
from urllib.parse import urlencode

@dataclass
class LTILaunchContext:
    platform_id: str
    user_id: str
    user_email: str
    user_name: str
    resource_link_id: str
    course_id: Optional[str]
    roles: list

class LTIService:
    def __init__(self, db_path: str = "~/.coditect/lms.db"):
        import os
        self.db_path = os.path.expanduser(db_path)

    def _get_connection(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn

    def register_platform(
        self,
        organization_id: str,
        name: str,
        issuer: str,
        client_id: str,
        auth_login_url: str,
        auth_token_url: str,
        keyset_url: str
    ) -> Dict[str, Any]:
        """Register an external LMS platform for LTI integration."""
        from cryptography.hazmat.primitives import serialization
        from cryptography.hazmat.primitives.asymmetric import rsa
        import secrets

        # Generate RSA key pair
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048
        )

        private_pem = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        ).decode()

        public_key = private_key.public_key()
        public_pem = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        ).decode()

        platform_id = secrets.token_hex(16)

        conn = self._get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT INTO lti_platforms (
                    id, organization_id, name, issuer, client_id,
                    auth_login_url, auth_token_url, keyset_url, private_key
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                platform_id, organization_id, name, issuer, client_id,
                auth_login_url, auth_token_url, keyset_url, private_pem
            ))
            conn.commit()

            return {
                "platform_id": platform_id,
                "public_key": public_pem,
                "tool_url": f"https://lms.coditect.ai/lti/launch",
                "login_url": f"https://lms.coditect.ai/lti/login",
                "jwks_url": f"https://lms.coditect.ai/lti/jwks/{platform_id}",
                "deep_link_url": f"https://lms.coditect.ai/lti/deep-link"
            }
        finally:
            conn.close()

    def initiate_login(
        self,
        iss: str,
        login_hint: str,
        target_link_uri: str,
        lti_message_hint: str = None
    ) -> str:
        """Handle OIDC login initiation (step 1 of LTI launch)."""
        import secrets

        conn = self._get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT * FROM lti_platforms WHERE issuer = ? AND is_active = 1",
                (iss,)
            )
            platform = cursor.fetchone()

            if not platform:
                raise ValueError(f"Unknown platform issuer: {iss}")

            # Generate state and nonce
            state = secrets.token_urlsafe(32)
            nonce = secrets.token_urlsafe(32)

            # Store state for validation (would use Redis in production)
            # For now, encode in state itself

            # Build auth request URL
            params = {
                "scope": "openid",
                "response_type": "id_token",
                "client_id": platform['client_id'],
                "redirect_uri": target_link_uri,
                "login_hint": login_hint,
                "state": state,
                "nonce": nonce,
                "response_mode": "form_post",
                "prompt": "none"
            }

            if lti_message_hint:
                params["lti_message_hint"] = lti_message_hint

            return f"{platform['auth_login_url']}?{urlencode(params)}"

        finally:
            conn.close()

    def validate_launch(self, id_token: str, state: str) -> LTILaunchContext:
        """Validate LTI launch request and extract context."""
        # Decode without verification first to get issuer
        unverified = jwt.decode(id_token, options={"verify_signature": False})
        issuer = unverified.get("iss")

        conn = self._get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT * FROM lti_platforms WHERE issuer = ?",
                (issuer,)
            )
            platform = cursor.fetchone()

            if not platform:
                raise ValueError("Unknown platform")

            # Fetch platform's public keys
            # In production, fetch from keyset_url and cache
            # For now, use stored keyset

            # Verify token signature
            # (simplified - production would fetch JWKS and verify properly)
            claims = unverified  # Would be verified in production

            # Validate required LTI claims
            message_type = claims.get(
                "https://purl.imsglobal.org/spec/lti/claim/message_type"
            )
            if message_type != "LtiResourceLinkRequest":
                raise ValueError(f"Unsupported message type: {message_type}")

            # Extract user info
            user_id = claims.get("sub")
            user_email = claims.get("email", "")
            user_name = claims.get("name", "")

            # Extract resource link
            resource_link = claims.get(
                "https://purl.imsglobal.org/spec/lti/claim/resource_link", {}
            )
            resource_link_id = resource_link.get("id")

            # Extract roles
            roles = claims.get(
                "https://purl.imsglobal.org/spec/lti/claim/roles", []
            )

            # Look up or create resource link mapping
            cursor.execute("""
                SELECT course_id, lesson_id FROM lti_resource_links
                WHERE platform_id = ? AND resource_link_id = ?
            """, (platform['id'], resource_link_id))
            link = cursor.fetchone()

            return LTILaunchContext(
                platform_id=platform['id'],
                user_id=user_id,
                user_email=user_email,
                user_name=user_name,
                resource_link_id=resource_link_id,
                course_id=link['course_id'] if link else None,
                roles=roles
            )

        finally:
            conn.close()

    def submit_grade(
        self,
        platform_id: str,
        resource_link_id: str,
        user_id: str,
        score: float,
        score_max: float = 100,
        comment: str = None
    ):
        """Submit grade back to external LMS via Assignment and Grade Services."""
        import secrets

        conn = self._get_connection()
        try:
            cursor = conn.cursor()

            # Store grade submission
            cursor.execute("""
                INSERT INTO lti_grade_submissions (
                    id, resource_link_id, user_id, score, score_max, comment
                ) VALUES (?, ?, ?, ?, ?, ?)
            """, (
                secrets.token_hex(16),
                resource_link_id,
                user_id,
                score,
                score_max,
                comment
            ))

            # Get platform info for grade passback
            cursor.execute("""
                SELECT p.* FROM lti_platforms p
                JOIN lti_resource_links r ON r.platform_id = p.id
                WHERE r.resource_link_id = ?
            """, (resource_link_id,))
            platform = cursor.fetchone()

            if platform:
                # Queue async grade submission
                # In production, this would call the platform's AGS endpoint
                pass

            conn.commit()

        finally:
            conn.close()

---

CLI Commands

API Management Commands

# Create API application
/api create "My Integration" --org my-org --scopes read,write

# List applications
/api list --org my-org

# View application details
/api view <app-id>

# Rotate credentials
/api rotate-key <app-id>
/api rotate-secret <app-id>

# Deactivate application
/api deactivate <app-id>

# View API usage
/api usage <app-id> --last 30d
/api audit <app-id> --limit 100

Webhook Management Commands

# Create webhook
/webhook create https://example.com/webhook --events enrollment.created,course.completed

# List webhooks
/webhook list --org my-org

# View webhook details and history
/webhook view <webhook-id>
/webhook history <webhook-id>

# Test webhook
/webhook test <webhook-id>

# Update webhook
/webhook update <webhook-id> --events "*"
/webhook update <webhook-id> --url https://new-url.com/webhook

# Disable/enable webhook
/webhook disable <webhook-id>
/webhook enable <webhook-id>

# Replay failed deliveries
/webhook replay <webhook-id> --failed-only

LTI Management Commands

# Register LTI platform
/lti register "Canvas LMS" \
  --issuer https://canvas.instructure.com \
  --client-id 12345 \
  --auth-url https://canvas.instructure.com/api/lti/authorize_redirect \
  --token-url https://canvas.instructure.com/login/oauth2/token \
  --keyset-url https://canvas.instructure.com/api/lti/security/jwks

# List registered platforms
/lti list

# View platform details
/lti view <platform-id>

# Link content to resource
/lti link <resource-link-id> --course <course-id>
/lti link <resource-link-id> --lesson <lesson-id>

# View grade submissions
/lti grades <platform-id> --pending

---

API Documentation (OpenAPI)

openapi: 3.0.3
info:
  title: CODITECT LMS API
  version: 1.0.0
  description: External API for CODITECT Learning Management System

servers:
  - url: https://api.coditect.ai/v1
    description: Production

security:
  - ApiKeyAuth: []
  - OAuth2: [read]

paths:
  /courses:
    get:
      summary: List courses
      parameters:
        - name: status
          in: query
          schema:
            type: string
            enum: [draft, published, archived]
        - name: limit
          in: query
          schema:
            type: integer
            default: 50
        - name: offset
          in: query
          schema:
            type: integer
            default: 0
      responses:
        '200':
          description: Success
          content:
            application/json:
              schema:
                type: object
                properties:
                  data:
                    type: array
                    items:
                      $ref: '#/components/schemas/Course'
                  meta:
                    $ref: '#/components/schemas/PaginationMeta'

  /courses/{course_id}:
    get:
      summary: Get course details
      parameters:
        - name: course_id
          in: path
          required: true
          schema:
            type: string
      responses:
        '200':
          description: Success
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Course'

  /enrollments:
    get:
      summary: List enrollments
      parameters:
        - name: course_id
          in: query
          schema:
            type: string
        - name: user_id
          in: query
          schema:
            type: string
        - name: status
          in: query
          schema:
            type: string
            enum: [active, completed, dropped]
      responses:
        '200':
          description: Success

    post:
      summary: Create enrollment
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [course_id, user_email]
              properties:
                course_id:
                  type: string
                user_email:
                  type: string
                  format: email
      responses:
        '201':
          description: Created

  /analytics/summary:
    get:
      summary: Get analytics summary
      parameters:
        - name: start_date
          in: query
          schema:
            type: string
            format: date
        - name: end_date
          in: query
          schema:
            type: string
            format: date
      responses:
        '200':
          description: Success

  /webhooks:
    get:
      summary: List webhook endpoints
      responses:
        '200':
          description: Success

    post:
      summary: Create webhook endpoint
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [url, events]
              properties:
                url:
                  type: string
                  format: uri
                events:
                  type: array
                  items:
                    type: string

components:
  securitySchemes:
    ApiKeyAuth:
      type: apiKey
      in: header
      name: X-API-Key

    OAuth2:
      type: oauth2
      flows:
        clientCredentials:
          tokenUrl: /oauth/token
          scopes:
            read: Read access
            write: Write access
            admin: Admin access

  schemas:
    Course:
      type: object
      properties:
        id:
          type: string
        title:
          type: string
        slug:
          type: string
        status:
          type: string
        created_at:
          type: string
          format: date-time

    PaginationMeta:
      type: object
      properties:
        total:
          type: integer
        limit:
          type: integer
        offset:
          type: integer
        has_more:
          type: boolean

---

Consequences

Positive

• Ecosystem Integration: CODITECT can work alongside existing LMS investments
• Enterprise Ready: API and webhook support enables enterprise automation
• Standards Compliance: LTI 1.3 ensures broad LMS compatibility
• Developer Friendly: RESTful API with OpenAPI documentation

Negative

• Complexity: Multiple integration protocols increase maintenance burden
• Security Surface: External API exposure requires robust security measures
• Support Load: Integration troubleshooting can be time-consuming

Risks and Mitigations

Risk	Impact	Mitigation
API abuse	High	Rate limiting, usage monitoring, IP allowlisting
Webhook failures	Medium	Retry logic, dead letter queue, alerting
LTI compatibility	Medium	Comprehensive testing with major LMS platforms
Data sync issues	Medium	Idempotent operations, conflict resolution

---

Implementation Phases

Phase 3A: REST API (Weeks 1-2)

• Core API endpoints (courses, enrollments, users)
• API key authentication
• Rate limiting
• API audit logging

Phase 3B: Webhooks (Week 3)

• Webhook registration
• Event triggering
• Delivery with retries
• Webhook testing tools

Phase 3C: OAuth 2.0 (Week 4)

• Client credentials flow
• Token management
• Scope-based access
• Token refresh

Phase 3D: LTI 1.3 (Weeks 5-6)

• OIDC login flow
• Resource link launch
• Deep linking
• Grade passback (AGS)

Phase 3E: Connectors (Weeks 7-8)

• Canvas connector
• Moodle connector
• Generic LTI setup guide
• Integration testing

---

Related ADRs

• ADR-LMS-004: User Authentication (OAuth prerequisite)
• ADR-LMS-006: SCORM/xAPI (content export)
• ADR-LMS-008: Analytics (API data source)
• ADR-LMS-009: Dashboard (integration management UI)

---

References

• LTI 1.3 Specification
• OAuth 2.0 RFC 6749
• OpenAPI Specification
• Webhook Best Practices
• Canvas LTI Developer Guide

---

Document History:

Version	Date	Author	Changes
1.0	2025-12-11	Claude Code	Initial ADR