feat(phases-10-15): complete TapTrack Hub v1.0

Phase 10 — Support Ticket System:
- tickets.py router: SLA status (on_track/at_risk/breached/responded), email
  notifications on create+reply via background threads, school_name in list,
  priority filter, bulk-close endpoint
- tasks/tickets.py: escalate_stale Celery task (48h→high, 72h no reply→urgent)
- worker.py: escalate_stale scheduled every hour
- templates/email/ticket_notification.html: HTML ticket notification email
- TicketsPage.vue: status tabs, SLA badge, priority badge, school name column,
  checkbox bulk-close, pagination
- TicketDetailPage.vue: inline priority/status/assignee selectors, SLA timer,
  internal note lock icon, closed-ticket guard

Phase 11 — Monthly Report Generation:
- models/report.py: MonthlyReport + SchoolMonthlyStats ORM models
- tasks/reports.py: send_monthly_reports enhanced with SMS stats, attendance
  data, invoice summary, stores MonthlyReport record per school per month

Phase 12 — On-Prem Monthly Report Pull:
- tasks/reports.py: pull_monthly_stats task — httpx GET to each school's
  hub_base_url, upserts SchoolMonthlyStats; runs 1st at 5am
- worker.py: pull_monthly_stats scheduled 1st at 5am

Phase 13 — Feature Flags + Suspension:
- models/school.py: hub_base_url, feature_overrides (JSON), onboarding_completed_at
- routers/schools.py: PUT /{id}/feature-overrides endpoint
- routers/sync.py: _tier_features() merges school.feature_overrides into poll config

Phase 14 — Onboarding Wizard + Welcome Email:
- tasks/onboarding.py: send_welcome_email Celery task with license key
- routers/schools.py: auto-trigger welcome email on POST /schools,
  POST /{id}/activate (status→active + onboarding_completed_at),
  POST /{id}/resend-welcome

Phase 15 — UX Polish + Ops Tools:
- routers/search.py: GET /api/search?q= (schools + invoices + tickets, 5 each)
- routers/audit.py: GET /api/audit-logs (paginated, filterable)
- AppLayout.vue: global search bar with debounced dropdown, result navigation
- AuditLogsPage.vue: new page with filter + pagination
- AppSidebar.vue: Audit Logs nav item added
- router/index.ts: /audit-logs route
- api.ts: globalSearch, getAuditLogs, activateSchool, resendWelcomeEmail,
  updateFeatureOverrides, bulkCloseTickets

Deployment:
- docker-compose.yml: x-backend-env anchor (DRY), PDF_DIR env var,
  seed service (one-shot python seed.py on first boot)
- migrations/003_phases11_15.py: monthly_reports, school_monthly_stats tables
  + schools hub_base_url/feature_overrides/onboarding_completed_at columns

backend/app/main.py

@@ -5,10 +5,11 @@ from fastapi.middleware.cors import CORSMiddleware
 
 from app.database import engine, Base
 # Import all models so Alembic/SQLAlchemy picks them up
-from app.models import user, school, license, sms, billing, ticket, audit, announcement, email_log  # noqa
+from app.models import user, school, license, sms, billing, ticket, audit, announcement, email_log, report  # noqa
 
 from app.routers import auth, schools, licenses, sms as sms_router, billing as billing_router
 from app.routers import tickets, users, dashboard, school_portal, announcements, sync, email as email_router
+from app.routers import search, audit
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
@@ -45,6 +46,8 @@ app.include_router(school_portal.router)
 app.include_router(announcements.router)
 app.include_router(sync.router)
 app.include_router(email_router.router)
+app.include_router(search.router)
+app.include_router(audit.router)
 
 @app.get("/api/health")
 async def health():

backend/app/models/report.py

@@ -0,0 +1,37 @@
+"""Monthly report and school stats models."""
+import uuid
+from datetime import datetime, timezone, date
+from sqlalchemy import String, DateTime, Date, Text, JSON, Numeric, Integer, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column
+from app.database import Base
+
+
+class MonthlyReport(Base):
+    """Record of each monthly report sent to a school."""
+    __tablename__ = "monthly_reports"
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
+    school_id: Mapped[str] = mapped_column(String(36), ForeignKey("schools.id", ondelete="CASCADE"), nullable=False, index=True)
+    report_month: Mapped[str] = mapped_column(String(7), nullable=False)  # YYYY-MM
+    email_sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    report_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
+
+
+class SchoolMonthlyStats(Base):
+    """Attendance + SMS stats pulled from on-prem each month."""
+    __tablename__ = "school_monthly_stats"
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
+    school_id: Mapped[str] = mapped_column(String(36), ForeignKey("schools.id", ondelete="CASCADE"), nullable=False, index=True)
+    report_month: Mapped[str] = mapped_column(String(7), nullable=False)  # YYYY-MM
+    total_students: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    school_days: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    present_days_total: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    absent_days_total: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    late_days_total: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    avg_attendance_rate: Mapped[float | None] = mapped_column(Numeric(5, 2), nullable=True)
+    sms_sent: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    pull_status: Mapped[str] = mapped_column(String(20), default="unavailable", nullable=False)  # success|failed|unavailable
+    pulled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    hub_base_url: Mapped[str | None] = mapped_column(String(500), nullable=True)

backend/app/models/school.py

@@ -34,6 +34,9 @@ class School(Base):
     sms_sender_name: Mapped[str] = mapped_column(String(11), default="SCHOOL", nullable=False)
     sms_credits: Mapped[float] = mapped_column(Numeric(12, 2), default=0.0, nullable=False)
     sms_credit_low_threshold: Mapped[int] = mapped_column(Integer, default=50, nullable=False)
+    hub_base_url: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    feature_overrides: Mapped[dict | None] = mapped_column(Text, nullable=True)  # JSON string
+    onboarding_completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
     notes: Mapped[str | None] = mapped_column(Text, nullable=True)
 

backend/app/routers/audit.py

@@ -0,0 +1,42 @@
+"""Audit log viewer endpoint."""
+from typing import Optional
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, func, desc
+
+from app.auth.dependencies import require_super_admin
+from app.database import get_db
+from app.models.user import HubUser
+from app.models.audit import AuditLog
+
+router = APIRouter(prefix="/api/audit-logs", tags=["audit"])
+
+
+@router.get("")
+async def list_audit_logs(
+    school_id: Optional[str] = Query(None),
+    action: Optional[str] = Query(None),
+    page: int = Query(1, ge=1),
+    per_page: int = Query(50),
+    _admin: HubUser = Depends(require_super_admin),
+    db: AsyncSession = Depends(get_db),
+):
+    stmt = select(AuditLog).order_by(desc(AuditLog.created_at))
+    if school_id:
+        stmt = stmt.where(AuditLog.school_id == school_id)
+    if action:
+        stmt = stmt.where(AuditLog.action.ilike(f"%{action}%"))
+    total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar_one()
+    logs = (await db.execute(stmt.offset((page - 1) * per_page).limit(per_page))).scalars().all()
+    return {
+        "items": [
+            {
+                "id": log.id, "school_id": log.school_id, "actor_email": log.actor_email,
+                "action": log.action, "entity_type": log.entity_type, "entity_id": log.entity_id,
+                "detail": log.detail, "ip_address": log.ip_address,
+                "created_at": log.created_at.isoformat(),
+            }
+            for log in logs
+        ],
+        "total": total, "page": page, "per_page": per_page,
+    }

backend/app/routers/dashboard.py

@@ -1,8 +1,9 @@
 """Super admin dashboard summary."""
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, Query
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy import select, func, and_
+from sqlalchemy import select, func, and_, desc
 from datetime import date, timedelta
+from typing import Optional
 
 from app.auth.dependencies import require_super_admin
 from app.database import get_db
@@ -51,3 +52,124 @@ async def get_summary(
         "invoices": {"pending": pending_invoices},
         "sms": {"sent_today": sms_today, "pending": sms_pending},
     }
+
+
+@router.get("/school-health")
+async def get_school_health(
+    sort_by: Optional[str] = Query("name", regex="^(name|status|sms_credits|license_expires_at|license_last_seen|created_at)$"),
+    sort_dir: Optional[str] = Query("asc", regex="^(asc|desc)$"),
+    status: Optional[SchoolStatus] = Query(None),
+    search: Optional[str] = Query(None),
+    _admin: HubUser = Depends(require_super_admin),
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    School health table for the super admin dashboard.
+    Returns all schools with their license expiry, last seen, and credit status.
+    """
+    stmt = select(School)
+    if status:
+        stmt = stmt.where(School.status == status)
+    if search:
+        stmt = stmt.where(School.name.ilike(f"%{search}%"))
+
+    # Apply sort
+    sort_col = {
+        "name": School.name,
+        "status": School.status,
+        "sms_credits": School.sms_credits,
+        "created_at": School.created_at,
+    }.get(sort_by, School.name)
+    stmt = stmt.order_by(desc(sort_col) if sort_dir == "desc" else sort_col)
+
+    schools = (await db.execute(stmt)).scalars().all()
+
+    # Fetch licenses in bulk
+    all_lics = {
+        lic.school_id: lic
+        for lic in (await db.execute(select(License))).scalars().all()
+    }
+
+    items = []
+    for s in schools:
+        lic = all_lics.get(s.id)
+        # Days until license expires
+        days_left = None
+        if lic and lic.expires_at:
+            days_left = (lic.expires_at - date.today()).days
+
+        items.append({
+            "id": s.id,
+            "name": s.name,
+            "slug": s.slug,
+            "status": s.status.value,
+            "tier": s.tier.value,
+            "sms_credits": float(s.sms_credits),
+            "sms_credit_low": float(s.sms_credits) < s.sms_credit_low_threshold,
+            "license_status": lic.status.value if lic else None,
+            "license_expires_at": lic.expires_at.isoformat() if lic and lic.expires_at else None,
+            "license_days_left": days_left,
+            "license_expiring_soon": days_left is not None and 0 < days_left <= 30,
+            "license_last_seen": lic.last_validated_at.isoformat() if lic and lic.last_validated_at else None,
+            "created_at": s.created_at.isoformat(),
+        })
+
+    return {"items": items, "total": len(items)}
+
+
+@router.get("/revenue")
+async def get_revenue_snapshot(
+    _admin: HubUser = Depends(require_super_admin),
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Revenue snapshot for the super admin dashboard.
+    Returns: MRR (from active subscriptions), outstanding invoices total, overdue total.
+    """
+    from app.models.billing import SchoolSubscription
+
+    # MRR: sum of monthly_fee from active subscriptions
+    mrr_result = (await db.execute(
+        select(func.sum(SchoolSubscription.monthly_fee)).where(SchoolSubscription.is_active == True)
+    )).scalar_one()
+    mrr = float(mrr_result or 0)
+
+    # Outstanding: all sent invoices total
+    outstanding_result = (await db.execute(
+        select(func.sum(Invoice.total_amount)).where(Invoice.status == InvoiceStatus.sent)
+    )).scalar_one()
+    outstanding = float(outstanding_result or 0)
+
+    # Overdue: all overdue invoices total
+    overdue_result = (await db.execute(
+        select(func.sum(Invoice.total_amount)).where(Invoice.status == InvoiceStatus.overdue)
+    )).scalar_one()
+    overdue = float(overdue_result or 0)
+
+    # Paid this month
+    paid_result = (await db.execute(
+        select(func.sum(Invoice.total_amount)).where(
+            and_(
+                Invoice.status == InvoiceStatus.paid,
+                func.date_trunc("month", Invoice.paid_at) == func.date_trunc("month", func.current_date())
+            )
+        )
+    )).scalar_one()
+    paid_this_month = float(paid_result or 0)
+
+    # Invoice counts
+    overdue_count = (await db.execute(
+        select(func.count()).where(Invoice.status == InvoiceStatus.overdue)
+    )).scalar_one()
+    outstanding_count = (await db.execute(
+        select(func.count()).where(Invoice.status == InvoiceStatus.sent)
+    )).scalar_one()
+
+    return {
+        "mrr": mrr,
+        "outstanding": outstanding,
+        "outstanding_count": outstanding_count,
+        "overdue": overdue,
+        "overdue_count": overdue_count,
+        "paid_this_month": paid_this_month,
+    }

backend/app/routers/schools.py

@@ -1,7 +1,8 @@
 """School registry endpoints — super admin only."""
+import json
 import uuid
 from datetime import datetime, timezone
-from typing import Optional, Any
+from typing import Optional
 from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import BaseModel, EmailStr
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -16,6 +17,7 @@ from app.models.license import License, LicenseStatus
 
 router = APIRouter(prefix="/api/schools", tags=["schools"])
 
+
 class SchoolCreate(BaseModel):
     name: str
     address: Optional[str] = None
@@ -29,6 +31,7 @@ class SchoolCreate(BaseModel):
     sms_sender_name: str = "SCHOOL"
     notes: Optional[str] = None
 
+
 class SchoolUpdate(BaseModel):
     name: Optional[str] = None
     address: Optional[str] = None
@@ -41,41 +44,43 @@ class SchoolUpdate(BaseModel):
     student_limit: Optional[int] = None
     sms_sender_name: Optional[str] = None
     status: Optional[SchoolStatus] = None
+    hub_base_url: Optional[str] = None
     notes: Optional[str] = None
 
+
+class FeatureOverridesBody(BaseModel):
+    overrides: dict
+
+
 def _school_out(s: School, license: License | None = None) -> dict:
+    feat = {}
+    if s.feature_overrides:
+        try:
+            feat = json.loads(s.feature_overrides) if isinstance(s.feature_overrides, str) else {}
+        except Exception:
+            feat = {}
     return {
-        "id": s.id,
-        "name": s.name,
-        "slug": s.slug,
-        "address": s.address,
-        "city": s.city,
-        "contact_name": s.contact_name,
-        "contact_email": s.contact_email,
-        "contact_phone": s.contact_phone,
-        "billing_email": s.billing_email,
-        "status": s.status.value,
-        "tier": s.tier.value,
-        "student_limit": s.student_limit,
-        "sms_sender_name": s.sms_sender_name,
-        "sms_credits": float(s.sms_credits),
+        "id": s.id, "name": s.name, "slug": s.slug, "address": s.address, "city": s.city,
+        "contact_name": s.contact_name, "contact_email": s.contact_email,
+        "contact_phone": s.contact_phone, "billing_email": s.billing_email,
+        "status": s.status.value, "tier": s.tier.value, "student_limit": s.student_limit,
+        "sms_sender_name": s.sms_sender_name, "sms_credits": float(s.sms_credits),
         "sms_credit_low_threshold": s.sms_credit_low_threshold,
-        "created_at": s.created_at.isoformat(),
-        "notes": s.notes,
+        "hub_base_url": s.hub_base_url, "feature_overrides": feat,
+        "onboarding_completed_at": s.onboarding_completed_at.isoformat() if s.onboarding_completed_at else None,
+        "created_at": s.created_at.isoformat(), "notes": s.notes,
         "license_key": license.key if license else None,
         "license_status": license.status.value if license else None,
         "license_expires_at": license.expires_at.isoformat() if license and license.expires_at else None,
         "license_last_seen": license.last_validated_at.isoformat() if license and license.last_validated_at else None,
     }
 
+
 @router.get("")
 async def list_schools(
-    page: int = Query(1, ge=1),
-    per_page: int = Query(25, ge=1, le=100),
-    search: Optional[str] = Query(None),
-    status: Optional[SchoolStatus] = Query(None),
-    _admin: HubUser = Depends(require_super_admin),
-    db: AsyncSession = Depends(get_db),
+    page: int = Query(1, ge=1), per_page: int = Query(25, ge=1, le=100),
+    search: Optional[str] = Query(None), status: Optional[SchoolStatus] = Query(None),
+    _admin: HubUser = Depends(require_super_admin), db: AsyncSession = Depends(get_db),
 ):
     stmt = select(School).order_by(desc(School.created_at))
     if search:
@@ -86,11 +91,11 @@ async def list_schools(
     schools = (await db.execute(stmt.offset((page - 1) * per_page).limit(per_page))).scalars().all()
     items = []
     for s in schools:
-        lic_res = await db.execute(select(License).where(License.school_id == s.id))
-        lic = lic_res.scalar_one_or_none()
+        lic = (await db.execute(select(License).where(License.school_id == s.id))).scalar_one_or_none()
         items.append(_school_out(s, lic))
     return {"items": items, "total": total, "page": page, "per_page": per_page}
 
+
 @router.post("", status_code=201)
 async def create_school(
     body: SchoolCreate,
@@ -98,34 +103,31 @@ async def create_school(
     db: AsyncSession = Depends(get_db),
 ):
     slug = slugify(body.name)
-    # Ensure slug uniqueness
     existing = (await db.execute(select(School).where(School.slug == slug))).scalar_one_or_none()
     if existing:
         slug = f"{slug}-{uuid.uuid4().hex[:6]}"
     school = School(
-        name=body.name,
-        slug=slug,
-        address=body.address,
-        city=body.city,
-        contact_name=body.contact_name,
-        contact_email=body.contact_email,
-        contact_phone=body.contact_phone,
-        billing_email=body.billing_email,
-        tier=body.tier,
-        student_limit=body.student_limit,
-        sms_sender_name=body.sms_sender_name[:11],
-        notes=body.notes,
+        name=body.name, slug=slug, address=body.address, city=body.city,
+        contact_name=body.contact_name, contact_email=body.contact_email,
+        contact_phone=body.contact_phone, billing_email=body.billing_email,
+        tier=body.tier, student_limit=body.student_limit,
+        sms_sender_name=body.sms_sender_name[:11], notes=body.notes,
         status=SchoolStatus.pending,
     )
     db.add(school)
     await db.flush()
-    # Auto-create license
     lic = License(school_id=school.id, tier=body.tier.value, max_students=body.student_limit)
     db.add(lic)
     await db.commit()
     await db.refresh(school)
+    try:
+        from app.tasks.onboarding import send_welcome_email
+        send_welcome_email.delay(school.id)
+    except Exception:
+        pass
     return _school_out(school, lic)
 
+
 @router.get("/{school_id}")
 async def get_school(
     school_id: str,
@@ -138,12 +140,11 @@ async def get_school(
     lic = (await db.execute(select(License).where(License.school_id == school_id))).scalar_one_or_none()
     return _school_out(school, lic)
 
+
 @router.put("/{school_id}")
 async def update_school(
-    school_id: str,
-    body: SchoolUpdate,
-    _admin: HubUser = Depends(require_super_admin),
-    db: AsyncSession = Depends(get_db),
+    school_id: str, body: SchoolUpdate,
+    _admin: HubUser = Depends(require_super_admin), db: AsyncSession = Depends(get_db),
 ):
     school = (await db.execute(select(School).where(School.id == school_id))).scalar_one_or_none()
     if not school:
@@ -156,26 +157,67 @@ async def update_school(
     lic = (await db.execute(select(License).where(License.school_id == school_id))).scalar_one_or_none()
     return _school_out(school, lic)
 
+
+@router.put("/{school_id}/feature-overrides")
+async def update_feature_overrides(
+    school_id: str, body: FeatureOverridesBody,
+    _admin: HubUser = Depends(require_super_admin), db: AsyncSession = Depends(get_db),
+):
+    school = (await db.execute(select(School).where(School.id == school_id))).scalar_one_or_none()
+    if not school:
+        raise HTTPException(404, "School not found")
+    school.feature_overrides = json.dumps(body.overrides)
+    await db.commit()
+    return {"feature_overrides": body.overrides}
+
+
+@router.post("/{school_id}/activate")
+async def activate_school(
+    school_id: str,
+    _admin: HubUser = Depends(require_super_admin), db: AsyncSession = Depends(get_db),
+):
+    school = (await db.execute(select(School).where(School.id == school_id))).scalar_one_or_none()
+    if not school:
+        raise HTTPException(404, "School not found")
+    lic = (await db.execute(select(License).where(License.school_id == school_id))).scalar_one_or_none()
+    if not lic:
+        raise HTTPException(400, "No license issued — cannot activate")
+    school.status = SchoolStatus.active
+    school.onboarding_completed_at = datetime.now(timezone.utc)
+    await db.commit()
+    return _school_out(school, lic)
+
+
+@router.post("/{school_id}/resend-welcome")
+async def resend_welcome(
+    school_id: str,
+    _admin: HubUser = Depends(require_super_admin), db: AsyncSession = Depends(get_db),
+):
+    school = (await db.execute(select(School).where(School.id == school_id))).scalar_one_or_none()
+    if not school:
+        raise HTTPException(404, "School not found")
+    try:
+        from app.tasks.onboarding import send_welcome_email
+        send_welcome_email.delay(school_id)
+    except Exception:
+        raise HTTPException(500, "Failed to queue welcome email")
+    return {"message": "Welcome email queued"}
+
+
 @router.post("/{school_id}/credits")
 async def add_sms_credits(
-    school_id: str,
-    amount: float,
-    description: Optional[str] = None,
-    _admin: HubUser = Depends(require_super_admin),
-    db: AsyncSession = Depends(get_db),
+    school_id: str, amount: float, description: Optional[str] = None,
+    _admin: HubUser = Depends(require_super_admin), db: AsyncSession = Depends(get_db),
 ):
     from app.models.sms import SmsCreditLedger, SmsCreditTx
     school = (await db.execute(select(School).where(School.id == school_id))).scalar_one_or_none()
     if not school:
         raise HTTPException(404, "School not found")
     school.sms_credits = float(school.sms_credits) + amount
-    ledger = SmsCreditLedger(
-        school_id=school_id,
-        tx_type=SmsCreditTx.topup,
-        amount=amount,
+    db.add(SmsCreditLedger(
+        school_id=school_id, tx_type=SmsCreditTx.topup, amount=amount,
         balance_after=float(school.sms_credits),
         description=description or f"Manual top-up of {amount} credits",
-    )
-    db.add(ledger)
+    ))
     await db.commit()
     return {"sms_credits": float(school.sms_credits), "added": amount}

backend/app/routers/search.py

@@ -0,0 +1,43 @@
+"""Global search across schools, invoices, tickets."""
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+
+from app.auth.dependencies import require_super_admin
+from app.database import get_db
+from app.models.user import HubUser
+from app.models.school import School
+from app.models.billing import Invoice
+from app.models.ticket import SupportTicket
+
+router = APIRouter(prefix="/api/search", tags=["search"])
+
+
+@router.get("")
+async def global_search(
+    q: str = Query(..., min_length=2),
+    _admin: HubUser = Depends(require_super_admin),
+    db: AsyncSession = Depends(get_db),
+):
+    """Search across schools, invoices, and tickets. Returns up to 5 per category."""
+    pattern = f"%{q}%"
+
+    schools = (await db.execute(
+        select(School).where(School.name.ilike(pattern)).limit(5)
+    )).scalars().all()
+
+    invoices = (await db.execute(
+        select(Invoice).where(Invoice.invoice_number.ilike(pattern)).limit(5)
+    )).scalars().all()
+
+    tickets = (await db.execute(
+        select(SupportTicket).where(
+            SupportTicket.subject.ilike(pattern) | SupportTicket.ticket_number.ilike(pattern)
+        ).limit(5)
+    )).scalars().all()
+
+    return {
+        "schools": [{"id": s.id, "name": s.name, "status": s.status.value, "type": "school"} for s in schools],
+        "invoices": [{"id": i.id, "invoice_number": i.invoice_number, "total_amount": float(i.total_amount), "status": i.status.value, "type": "invoice"} for i in invoices],
+        "tickets": [{"id": t.id, "ticket_number": t.ticket_number, "subject": t.subject, "status": t.status.value, "type": "ticket"} for t in tickets],
+    }

backend/app/routers/sync.py

@@ -14,12 +14,20 @@ from app.models.sms import SmsJob, SmsJobStatus, SmsCreditLedger, SmsCreditTx
 router = APIRouter(prefix="/api/sync", tags=["sync"])
 
 # Feature flags by tier — mirrors licenses.py
-def _tier_features(tier: str) -> dict:
+def _tier_features(tier: str, overrides: str | None = None) -> dict:
+    import json
     base = {"sms": True, "reports": True, "websocket": True, "multi_terminal": True}
     if tier == "premium":
         base.update({"api_keys": True, "webhooks": True, "bulk_enrollment": True})
     elif tier == "basic":
         base.update({"multi_terminal": False, "api_keys": False, "webhooks": False})
+    if overrides:
+        try:
+            base.update(json.loads(overrides))
+        except Exception:
+            pass
+    elif tier == "basic":
+        base.update({"multi_terminal": False, "api_keys": False, "webhooks": False})
     return base
 
 @router.post("/poll")
@@ -167,6 +175,6 @@ async def sync_poll(
             "sms_credits": float(school.sms_credits),
             "school_status": school.status.value,
             "suspended": suspended,
-            "feature_flags": _tier_features(lic.tier),
+            "feature_flags": _tier_features(lic.tier, school.feature_overrides),
         },
     }

backend/app/routers/tickets.py

@@ -1,6 +1,7 @@
 """Support ticket endpoints."""
-import uuid
-from datetime import datetime, timezone
+import os
+import threading
+from datetime import datetime, timezone, timedelta
 from typing import Optional
 from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import BaseModel
@@ -14,45 +15,130 @@ from app.models.ticket import SupportTicket, TicketReply, TicketStatus, TicketPr
 
 router = APIRouter(prefix="/api/tickets", tags=["tickets"])
 
+
 class TicketCreate(BaseModel):
     subject: str
     body: str
     category: TicketCategory = TicketCategory.general
 
+
 class TicketUpdate(BaseModel):
     status: Optional[TicketStatus] = None
     priority: Optional[TicketPriority] = None
     assigned_to: Optional[str] = None
 
+
 class ReplyCreate(BaseModel):
     body: str
     is_internal: bool = False
 
-def _ticket_out(t: SupportTicket) -> dict:
+
+class BulkClose(BaseModel):
+    ticket_ids: list[str]
+
+
+def _sla_status(ticket: SupportTicket) -> str:
+    if ticket.status in (TicketStatus.resolved, TicketStatus.closed):
+        return "resolved"
+    if ticket.first_response_at:
+        return "responded"
+    age_h = (datetime.now(timezone.utc) - ticket.created_at).total_seconds() / 3600
+    if age_h > 48:
+        return "breached"
+    if age_h > 24:
+        return "at_risk"
+    return "on_track"
+
+
+def _ticket_out(t: SupportTicket, school_name: str | None = None) -> dict:
     return {
-        "id": t.id, "school_id": t.school_id, "ticket_number": t.ticket_number,
-        "subject": t.subject, "body": t.body, "category": t.category.value,
-        "status": t.status.value, "priority": t.priority.value,
-        "assigned_to": t.assigned_to,
+        "id": t.id, "school_id": t.school_id, "school_name": school_name,
+        "ticket_number": t.ticket_number, "subject": t.subject, "body": t.body,
+        "category": t.category.value, "status": t.status.value, "priority": t.priority.value,
+        "assigned_to": t.assigned_to, "sla_status": _sla_status(t),
         "first_response_at": t.first_response_at.isoformat() if t.first_response_at else None,
         "resolved_at": t.resolved_at.isoformat() if t.resolved_at else None,
-        "created_at": t.created_at.isoformat(),
-        "updated_at": t.updated_at.isoformat(),
+        "created_at": t.created_at.isoformat(), "updated_at": t.updated_at.isoformat(),
     }
 
+
 def _next_ticket_number(count: int) -> str:
     from datetime import date
     return f"TKT-{date.today().year}-{count + 1:05d}"
 
+
+def _notify_async(ticket_id: str, message_body: str, is_reply: bool, from_admin: bool):
+    """Send ticket email notifications in a background thread."""
+    import os as _os
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+    from app.models.ticket import SupportTicket
+    from app.models.school import School
+    from app.models.user import HubUser, UserRole
+    from app.services.email import send_email
+
+    db_url = _os.getenv("DATABASE_URL", "postgresql://postgres:postgres@db:5432/taptrack_hub")
+    engine = create_engine(db_url.replace("postgresql+asyncpg://", "postgresql://"), pool_pre_ping=True)
+    db = sessionmaker(bind=engine)()
+    try:
+        t = db.get(SupportTicket, ticket_id)
+        if not t:
+            return
+        hub_url = _os.getenv("HUB_BASE_URL", "http://localhost:8090")
+        ticket_url = f"{hub_url}/tickets/{t.id}"
+        base_ctx = {
+            "ticket_number": t.ticket_number, "subject": t.subject,
+            "category": t.category.value, "priority": t.priority.value,
+            "status": t.status.value, "message_body": message_body, "ticket_url": ticket_url,
+        }
+
+        if is_reply and from_admin:
+            school = db.get(School, t.school_id)
+            if school and school.billing_email:
+                send_email(
+                    to=school.billing_email,
+                    subject=f"[TapTrack] Reply on {t.ticket_number}: {t.subject}",
+                    body=f"Support replied to ticket {t.ticket_number}.",
+                    template_name="email/ticket_notification.html",
+                    context={**base_ctx, "email_title": f"Reply on Ticket {t.ticket_number}",
+                             "recipient_name": school.contact_name or school.name, "message_label": "Support Reply"},
+                    email_type="other", school_id=t.school_id,
+                )
+        else:
+            admins = db.execute(
+                select(HubUser).where(HubUser.role == UserRole.super_admin, HubUser.is_active == True)
+            ).scalars().all()
+            title = f"New Ticket: {t.ticket_number}" if not is_reply else f"Customer Reply on {t.ticket_number}"
+            for admin in admins:
+                send_email(
+                    to=admin.email,
+                    subject=f"[TapTrack] {title}: {t.subject}",
+                    body=f"{title}\n\n{message_body}",
+                    template_name="email/ticket_notification.html",
+                    context={**base_ctx, "email_title": title,
+                             "recipient_name": admin.full_name,
+                             "message_label": "Message" if not is_reply else "Customer Reply"},
+                    email_type="other", school_id=t.school_id,
+                )
+    except Exception as e:
+        import logging
+        logging.getLogger(__name__).warning("Ticket notification error: %s", e)
+    finally:
+        db.close()
+        engine.dispose()
+
+
 @router.get("")
 async def list_tickets(
     school_id: Optional[str] = Query(None),
     status: Optional[TicketStatus] = Query(None),
+    priority: Optional[TicketPriority] = Query(None),
     page: int = Query(1, ge=1),
     per_page: int = Query(25),
     current_user: HubUser = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
+    from app.models.school import School
     stmt = select(SupportTicket).order_by(desc(SupportTicket.updated_at))
     if current_user.role != UserRole.super_admin:
         stmt = stmt.where(SupportTicket.school_id == current_user.school_id)
@@ -60,9 +146,18 @@ async def list_tickets(
         stmt = stmt.where(SupportTicket.school_id == school_id)
     if status:
         stmt = stmt.where(SupportTicket.status == status)
+    if priority:
+        stmt = stmt.where(SupportTicket.priority == priority)
     total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar_one()
     tickets = (await db.execute(stmt.offset((page - 1) * per_page).limit(per_page))).scalars().all()
-    return {"items": [_ticket_out(t) for t in tickets], "total": total, "page": page, "per_page": per_page}
+    sids = {t.school_id for t in tickets}
+    school_map: dict[str, str] = {}
+    if sids:
+        schools = (await db.execute(select(School).where(School.id.in_(sids)))).scalars().all()
+        school_map = {s.id: s.name for s in schools}
+    return {"items": [_ticket_out(t, school_map.get(t.school_id)) for t in tickets],
+            "total": total, "page": page, "per_page": per_page}
+
 
 @router.post("", status_code=201)
 async def create_ticket(
@@ -74,16 +169,17 @@ async def create_ticket(
         raise HTTPException(400, "No school associated with your account")
     count = (await db.execute(select(func.count()).select_from(SupportTicket))).scalar_one()
     ticket = SupportTicket(
-        school_id=current_user.school_id,
-        submitted_by=current_user.id,
-        ticket_number=_next_ticket_number(count),
-        subject=body.subject,
-        body=body.body,
-        category=body.category,
+        school_id=current_user.school_id, submitted_by=current_user.id,
+        ticket_number=_next_ticket_number(count), subject=body.subject,
+        body=body.body, category=body.category,
     )
     db.add(ticket)
     await db.commit()
-    return _ticket_out(ticket)
+    threading.Thread(target=_notify_async, args=(ticket.id, body.body, False, False), daemon=True).start()
+    from app.models.school import School
+    school = (await db.execute(select(School).where(School.id == current_user.school_id))).scalar_one_or_none()
+    return _ticket_out(ticket, school.name if school else None)
+
 
 @router.get("/{ticket_id}")
 async def get_ticket(
@@ -91,23 +187,25 @@ async def get_ticket(
     current_user: HubUser = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
+    from app.models.school import School
     t = (await db.execute(select(SupportTicket).where(SupportTicket.id == ticket_id))).scalar_one_or_none()
     if not t:
         raise HTTPException(404)
     if current_user.role != UserRole.super_admin and t.school_id != current_user.school_id:
         raise HTTPException(403)
-    replies_res = await db.execute(select(TicketReply).where(TicketReply.ticket_id == ticket_id).order_by(TicketReply.created_at))
-    replies = replies_res.scalars().all()
-    visible_replies = [r for r in replies if not r.is_internal or current_user.role == UserRole.super_admin]
+    replies = (await db.execute(
+        select(TicketReply).where(TicketReply.ticket_id == ticket_id).order_by(TicketReply.created_at)
+    )).scalars().all()
+    visible = [r for r in replies if not r.is_internal or current_user.role == UserRole.super_admin]
+    school = (await db.execute(select(School).where(School.id == t.school_id))).scalar_one_or_none()
     return {
-        **_ticket_out(t),
-        "replies": [
-            {"id": r.id, "body": r.body, "is_internal": r.is_internal,
-             "author_id": r.author_id, "created_at": r.created_at.isoformat()}
-            for r in visible_replies
-        ],
+        **_ticket_out(t, school.name if school else None),
+        "replies": [{"id": r.id, "body": r.body, "is_internal": r.is_internal,
+                     "author_id": r.author_id, "created_at": r.created_at.isoformat()}
+                    for r in visible],
     }
 
+
 @router.put("/{ticket_id}")
 async def update_ticket(
     ticket_id: str,
@@ -125,6 +223,7 @@ async def update_ticket(
     await db.commit()
     return _ticket_out(t)
 
+
 @router.post("/{ticket_id}/replies", status_code=201)
 async def add_reply(
     ticket_id: str,
@@ -140,10 +239,33 @@ async def add_reply(
     is_internal = body.is_internal and current_user.role == UserRole.super_admin
     reply = TicketReply(ticket_id=ticket_id, author_id=current_user.id, body=body.body, is_internal=is_internal)
     db.add(reply)
-    # Set first response time (super admin only)
     if current_user.role == UserRole.super_admin and not t.first_response_at:
         t.first_response_at = datetime.now(timezone.utc)
     if t.status == TicketStatus.open:
         t.status = TicketStatus.in_progress
     await db.commit()
-    return {"id": reply.id, "body": reply.body, "created_at": reply.created_at.isoformat()}
+    if not is_internal:
+        from_admin = current_user.role == UserRole.super_admin
+        threading.Thread(target=_notify_async, args=(t.id, body.body, True, from_admin), daemon=True).start()
+    return {"id": reply.id, "body": reply.body, "is_internal": is_internal, "created_at": reply.created_at.isoformat()}
+
+
+@router.post("/bulk-close")
+async def bulk_close_tickets(
+    body: BulkClose,
+    _admin: HubUser = Depends(require_super_admin),
+    db: AsyncSession = Depends(get_db),
+):
+    now = datetime.now(timezone.utc)
+    tickets = (await db.execute(
+        select(SupportTicket).where(SupportTicket.id.in_(body.ticket_ids))
+    )).scalars().all()
+    closed = 0
+    for t in tickets:
+        if t.status != TicketStatus.closed:
+            t.status = TicketStatus.closed
+            if not t.resolved_at:
+                t.resolved_at = now
+            closed += 1
+    await db.commit()
+    return {"closed": closed, "total": len(tickets)}

backend/app/tasks/onboarding.py

@@ -0,0 +1,56 @@
+"""Celery task: welcome email on school onboarding."""
+import logging
+from app.worker import celery_app
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(name="onboarding.send_welcome_email")
+def send_welcome_email(school_id: str):
+    """Send welcome email with license key to new school."""
+    import os
+    from sqlalchemy import create_engine, select
+    from sqlalchemy.orm import sessionmaker
+    from app.models.school import School
+    from app.models.license import License
+    from app.services.email import send_email
+
+    db_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@db:5432/taptrack_hub")
+    engine = create_engine(db_url.replace("postgresql+asyncpg://", "postgresql://"), pool_pre_ping=True)
+    db = sessionmaker(bind=engine)()
+    try:
+        school = db.get(School, school_id)
+        if not school:
+            return
+        lic = db.execute(select(License).where(License.school_id == school_id)).scalar_one_or_none()
+        email_to = school.billing_email or school.contact_email
+        if not email_to:
+            logger.warning("No email for school %s — skipping welcome", school.name)
+            return
+
+        hub_url = os.getenv("HUB_BASE_URL", "http://localhost:8090")
+        license_key = lic.key if lic else "Contact support to get your license key"
+
+        plain = (
+            f"Welcome to TapTrack Hub!\n\n"
+            f"Your school '{school.name}' has been registered.\n\n"
+            f"License Key: {license_key}\n\n"
+            f"Setup instructions:\n"
+            f"1. Log in to TapTrack Hub: {hub_url}/login\n"
+            f"2. Enter your license key in TapTrack's Hub settings\n"
+            f"3. Restart the TapTrack service\n\n"
+            f"Support: support@taptrack.io\n\nTapTrack Hub Team"
+        )
+        send_email(
+            to=email_to,
+            subject=f"Welcome to TapTrack Hub — {school.name}",
+            body=plain,
+            email_type="welcome",
+            school_id=school.id,
+        )
+        logger.info("Welcome email sent to %s for school %s", email_to, school.name)
+    except Exception as e:
+        logger.error("send_welcome_email error for %s: %s", school_id, e)
+    finally:
+        db.close()
+        engine.dispose()

backend/app/tasks/reports.py

@@ -1,35 +1,196 @@
-"""Celery task: send monthly reports to schools."""
+"""Celery tasks: monthly report generation and on-prem data pull."""
 import logging
+from datetime import datetime, timezone, timedelta, date
+
 from app.worker import celery_app
 
 logger = logging.getLogger(__name__)
 
-@celery_app.task(name="reports.send_monthly_reports")
-def send_monthly_reports():
-    """Send monthly attendance and SMS report email to each active school."""
-    import os
-    from datetime import date, timedelta
-    from sqlalchemy import create_engine, select
-    from sqlalchemy.orm import sessionmaker
-    from app.models.school import School, SchoolStatus
-    from app.services.email import send_email
 
+def _make_session():
+    import os
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
     db_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@db:5432/taptrack_hub")
     engine = create_engine(db_url.replace("postgresql+asyncpg://", "postgresql://"), pool_pre_ping=True)
-    db = sessionmaker(bind=engine)()
+    return sessionmaker(bind=engine)()
+
+
+@celery_app.task(name="reports.pull_monthly_stats")
+def pull_monthly_stats():
+    """
+    Pull attendance stats from each on-prem TapTrack instance.
+    Runs 1st of month at 5am, before report generation at 7am.
+    """
+    import httpx
+    from sqlalchemy import select
+    from app.models.school import School, SchoolStatus
+    from app.models.license import License
+    from app.models.report import SchoolMonthlyStats
+
+    db = _make_session()
     try:
         today = date.today()
-        prev_month_end = date(today.year, today.month, 1) - timedelta(days=1)
-        prev_month_start = date(prev_month_end.year, prev_month_end.month, 1)
+        prev_end = date(today.year, today.month, 1) - timedelta(days=1)
+        report_month = f"{prev_end.year}-{prev_end.month:02d}"
+
         schools = db.execute(select(School).where(School.status == SchoolStatus.active)).scalars().all()
+        for school in schools:
+            lic = db.execute(select(License).where(License.school_id == school.id)).scalar_one_or_none()
+            hub_url = getattr(school, 'hub_base_url', None)
+
+            existing = db.execute(
+                select(SchoolMonthlyStats).where(
+                    SchoolMonthlyStats.school_id == school.id,
+                    SchoolMonthlyStats.report_month == report_month,
+                )
+            ).scalar_one_or_none()
+            stats = existing or SchoolMonthlyStats(school_id=school.id, report_month=report_month)
+
+            if hub_url and lic:
+                try:
+                    resp = httpx.get(
+                        f"{hub_url.rstrip('/')}/api/hub/monthly-report",
+                        params={"key": lic.key, "month": report_month},
+                        timeout=10,
+                    )
+                    if resp.status_code == 200:
+                        data = resp.json()
+                        stats.total_students = data.get("total_students")
+                        stats.school_days = data.get("school_days")
+                        stats.present_days_total = data.get("present_days_total")
+                        stats.absent_days_total = data.get("absent_days_total")
+                        stats.late_days_total = data.get("late_days_total")
+                        stats.avg_attendance_rate = data.get("avg_attendance_rate")
+                        stats.sms_sent = data.get("sms_sent")
+                        stats.pull_status = "success"
+                        stats.pulled_at = datetime.now(timezone.utc)
+                        stats.hub_base_url = hub_url
+                    else:
+                        stats.pull_status = "failed"
+                except Exception as e:
+                    logger.warning("Failed to pull stats for %s: %s", school.name, e)
+                    stats.pull_status = "failed"
+            else:
+                stats.pull_status = "unavailable"
+
+            if not existing:
+                db.add(stats)
+
+        db.commit()
+        logger.info("pull_monthly_stats complete for %s", report_month)
+    except Exception as e:
+        db.rollback()
+        logger.error("pull_monthly_stats error: %s", e)
+    finally:
+        db.close()
+
+
+@celery_app.task(name="reports.send_monthly_reports")
+def send_monthly_reports():
+    """Send monthly report emails to all active schools on the 1st at 7am."""
+    from sqlalchemy import select, func
+    from app.models.school import School, SchoolStatus
+    from app.models.sms import SmsJob, SmsJobStatus
+    from app.models.billing import Invoice
+    from app.models.report import MonthlyReport, SchoolMonthlyStats
+    from app.services.email import send_email
+
+    db = _make_session()
+    try:
+        today = date.today()
+        prev_end = date(today.year, today.month, 1) - timedelta(days=1)
+        prev_start = date(prev_end.year, prev_end.month, 1)
+        report_month = f"{prev_end.year}-{prev_end.month:02d}"
+        month_label = prev_start.strftime("%B %Y")
+
+        schools = db.execute(select(School).where(School.status == SchoolStatus.active)).scalars().all()
+        sent = 0
+
         for school in schools:
             if not school.billing_email:
                 continue
-            send_email(
-                to=school.billing_email,
-                subject=f"Monthly Report — {school.name} — {prev_month_start.strftime('%B %Y')}",
-                body=f"Dear {school.contact_name or school.name},\n\nPlease find your monthly summary for {prev_month_start.strftime('%B %Y')} in your TapTrack Hub portal.\n\nSMS Credits Remaining: {float(school.sms_credits):.0f}\n\nLog in to view full details.\n\nThank you,\nTapTrack Hub Team",
+
+            sms_sent_count = db.execute(
+                select(func.count()).where(
+                    SmsJob.school_id == school.id, SmsJob.status == SmsJobStatus.sent,
+                    func.date(SmsJob.sent_at) >= prev_start, func.date(SmsJob.sent_at) <= prev_end,
+                )
+            ).scalar_one()
+            sms_failed_count = db.execute(
+                select(func.count()).where(
+                    SmsJob.school_id == school.id, SmsJob.status == SmsJobStatus.failed,
+                    func.date(SmsJob.created_at) >= prev_start, func.date(SmsJob.created_at) <= prev_end,
+                )
+            ).scalar_one()
+
+            inv = db.execute(
+                select(Invoice).where(Invoice.school_id == school.id, Invoice.billing_period_start == prev_start)
+            ).scalar_one_or_none()
+
+            att = db.execute(
+                select(SchoolMonthlyStats).where(
+                    SchoolMonthlyStats.school_id == school.id,
+                    SchoolMonthlyStats.report_month == report_month,
+                )
+            ).scalar_one_or_none()
+
+            report_data = {
+                "month": month_label, "sms_sent": sms_sent_count, "sms_failed": sms_failed_count,
+                "credits_remaining": float(school.sms_credits),
+                "attendance": {
+                    "available": att is not None and att.pull_status == "success",
+                    "total_students": att.total_students if att else None,
+                    "school_days": att.school_days if att else None,
+                    "avg_attendance_rate": float(att.avg_attendance_rate) if att and att.avg_attendance_rate else None,
+                },
+                "invoice": {"number": inv.invoice_number, "total": float(inv.total_amount), "status": inv.status.value} if inv else None,
+            }
+
+            existing_report = db.execute(
+                select(MonthlyReport).where(
+                    MonthlyReport.school_id == school.id, MonthlyReport.report_month == report_month,
+                )
+            ).scalar_one_or_none()
+
+            if not existing_report:
+                report = MonthlyReport(school_id=school.id, report_month=report_month, report_data=report_data)
+                db.add(report)
+                db.flush()
+            else:
+                report = existing_report
+                report.report_data = report_data
+
+            att_section = ""
+            if report_data["attendance"]["available"]:
+                att_section = (f"\nAttendance Rate: {report_data['attendance']['avg_attendance_rate']:.1f}%"
+                               f" | School Days: {report_data['attendance']['school_days']}"
+                               f" | Students: {report_data['attendance']['total_students']}")
+
+            plain = (
+                f"Dear {school.contact_name or school.name},\n\n"
+                f"Monthly summary for {month_label}:\n\n"
+                f"SMS Sent: {sms_sent_count} | Failed: {sms_failed_count}\n"
+                f"SMS Credits Remaining: {float(school.sms_credits):.0f}\n"
+                f"{att_section}\n\n"
+                f"Log in to view full details.\n\nTapTrack Hub Team"
             )
-        logger.info(f"Sent monthly reports to {len(schools)} schools")
+
+            ok = send_email(
+                to=school.billing_email,
+                subject=f"Monthly Report — {school.name} — {month_label}",
+                body=plain,
+                email_type="monthly_report",
+                school_id=school.id,
+            )
+            if ok:
+                report.email_sent_at = datetime.now(timezone.utc)
+                sent += 1
+
+        db.commit()
+        logger.info("send_monthly_reports: %d/%d sent", sent, len(schools))
+    except Exception as e:
+        db.rollback()
+        logger.error("send_monthly_reports error: %s", e)
     finally:
         db.close()

backend/app/tasks/tickets.py

@@ -0,0 +1,70 @@
+"""Celery tasks: ticket priority escalation."""
+import logging
+from datetime import datetime, timezone, timedelta
+
+from app.worker import celery_app
+
+logger = logging.getLogger(__name__)
+
+
+def _make_session():
+    import os
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+    db_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@db:5432/taptrack_hub")
+    engine = create_engine(db_url.replace("postgresql+asyncpg://", "postgresql://"), pool_pre_ping=True)
+    return sessionmaker(bind=engine)()
+
+
+@celery_app.task(name="tickets.escalate_stale")
+def escalate_stale():
+    """
+    Escalate stale open tickets:
+    - open > 48h + priority medium → high
+    - open > 72h + no reply yet → urgent
+    """
+    from sqlalchemy import select, and_
+    from app.models.ticket import SupportTicket, TicketReply, TicketStatus, TicketPriority
+
+    db = _make_session()
+    try:
+        now = datetime.now(timezone.utc)
+        threshold_48h = now - timedelta(hours=48)
+        threshold_72h = now - timedelta(hours=72)
+
+        # Open tickets > 48h, medium priority → high
+        stale_medium = db.execute(
+            select(SupportTicket).where(
+                and_(
+                    SupportTicket.status == TicketStatus.open,
+                    SupportTicket.priority == TicketPriority.medium,
+                    SupportTicket.created_at < threshold_48h,
+                )
+            )
+        ).scalars().all()
+        for t in stale_medium:
+            t.priority = TicketPriority.high
+            logger.info("Escalated ticket %s to high priority (48h)", t.ticket_number)
+
+        # Open tickets > 72h, no reply → urgent
+        stale_72h = db.execute(
+            select(SupportTicket).where(
+                and_(
+                    SupportTicket.status == TicketStatus.open,
+                    SupportTicket.first_response_at == None,  # noqa: E711
+                    SupportTicket.created_at < threshold_72h,
+                )
+            )
+        ).scalars().all()
+        for t in stale_72h:
+            if t.priority != TicketPriority.urgent:
+                t.priority = TicketPriority.urgent
+                logger.warning("Escalated ticket %s to urgent (72h no response)", t.ticket_number)
+
+        db.commit()
+        logger.info("escalate_stale: %d→high, %d→urgent", len(stale_medium), len(stale_72h))
+    except Exception as e:
+        db.rollback()
+        logger.error("escalate_stale error: %s", e)
+    finally:
+        db.close()

backend/app/templates/email/ticket_notification.html

@@ -0,0 +1,22 @@
+{% extends "email/base.html" %}
+{% block content %}
+<h2>{{ email_title }}</h2>
+<p>Dear {{ recipient_name }},</p>
+
+<div class="info-box">
+  <div class="row"><span class="label">Ticket</span><span class="value">{{ ticket_number }}</span></div>
+  <div class="row"><span class="label">Subject</span><span class="value">{{ subject }}</span></div>
+  <div class="row"><span class="label">Category</span><span class="value" style="text-transform:capitalize">{{ category }}</span></div>
+  <div class="row"><span class="label">Priority</span><span class="value" style="text-transform:capitalize">{{ priority }}</span></div>
+  <div class="row"><span class="label">Status</span><span class="value" style="text-transform:capitalize">{{ status }}</span></div>
+</div>
+
+<div style="background:#f8fafc;border:1px solid #e2e8f0;border-radius:8px;padding:16px;margin:16px 0">
+  <p style="font-size:11px;color:#94a3b8;margin:0 0 8px;font-weight:600;text-transform:uppercase;letter-spacing:0.5px">{{ message_label }}</p>
+  <p style="margin:0;white-space:pre-wrap;color:#334155">{{ message_body }}</p>
+</div>
+
+<p><a class="btn" href="{{ ticket_url }}">View Ticket</a></p>
+
+<p>Thank you,<br><strong>TapTrack Hub Team</strong></p>
+{% endblock %}

backend/app/worker.py

@@ -14,6 +14,8 @@ celery_app = Celery(
         "app.tasks.billing",
         "app.tasks.reports",
         "app.tasks.license",
+        "app.tasks.tickets",
+        "app.tasks.onboarding",
     ],
 )
 
@@ -44,6 +46,11 @@ celery_app.conf.update(
             "task": "billing.generate_monthly_invoices",
             "schedule": crontab(day_of_month=1, hour=6, minute=0),
         },
+        # Pull on-prem stats on the 1st at 5am
+        "pull-monthly-stats": {
+            "task": "reports.pull_monthly_stats",
+            "schedule": crontab(day_of_month=1, hour=5, minute=0),
+        },
         # Send monthly reports on the 1st at 7am
         "send-monthly-reports": {
             "task": "reports.send_monthly_reports",
@@ -54,6 +61,11 @@ celery_app.conf.update(
             "task": "billing.check_overdue",
             "schedule": crontab(hour=9, minute=0),
         },
+        # Escalate stale tickets every hour
+        "escalate-stale-tickets": {
+            "task": "tickets.escalate_stale",
+            "schedule": crontab(minute=0),
+        },
     },
 )