Add role-aware global grouped search

containers/backupchecks/src/backend/app/main/routes.py

@@ -26,5 +26,6 @@ from . import routes_feedback  # noqa: F401
 from . import routes_api  # noqa: F401
 from . import routes_reporting_api  # noqa: F401
 from . import routes_user_settings  # noqa: F401
+from . import routes_search  # noqa: F401
 
 __all__ = ["main_bp", "roles_required"]

containers/backupchecks/src/backend/app/main/routes_search.py

@@ -0,0 +1,574 @@
+from .routes_shared import *  # noqa: F401,F403
+from .routes_shared import _format_datetime
+
+from sqlalchemy import and_, cast, func, or_, String
+
+
+SEARCH_LIMIT_PER_SECTION = 10
+
+
+def _is_section_allowed(section: str) -> bool:
+    role = get_active_role()
+    allowed = {
+        "inbox": {"admin", "operator", "viewer"},
+        "customers": {"admin", "operator", "viewer"},
+        "jobs": {"admin", "operator", "viewer"},
+        "daily_jobs": {"admin", "operator", "viewer"},
+        "run_checks": {"admin", "operator"},
+        "tickets": {"admin", "operator", "viewer"},
+        "overrides": {"admin", "operator", "viewer"},
+        "reports": {"admin", "operator", "viewer", "reporter"},
+    }
+    return role in allowed.get(section, set())
+
+
+def _build_patterns(raw_query: str) -> list[str]:
+    tokens = [t.strip() for t in (raw_query or "").split() if t.strip()]
+    patterns: list[str] = []
+    for token in tokens:
+        p = token.replace("\\", "\\\\")
+        p = p.replace("%", "\\%").replace("_", "\\_")
+        p = p.replace("*", "%")
+        if not p.startswith("%"):
+            p = f"%{p}"
+        if not p.endswith("%"):
+            p = f"{p}%"
+        patterns.append(p)
+    return patterns
+
+
+def _contains_all_terms(columns: list, patterns: list[str]):
+    if not patterns or not columns:
+        return None
+    term_filters = []
+    for pattern in patterns:
+        per_term = [col.ilike(pattern, escape="\\") for col in columns]
+        term_filters.append(or_(*per_term))
+    return and_(*term_filters)
+
+
+def _build_inbox_results(patterns: list[str]) -> dict:
+    section = {
+        "key": "inbox",
+        "title": "Inbox",
+        "view_all_url": url_for("main.inbox"),
+        "total": 0,
+        "items": [],
+    }
+    if not _is_section_allowed("inbox"):
+        return section
+
+    query = MailMessage.query
+    if hasattr(MailMessage, "location"):
+        query = query.filter(MailMessage.location == "inbox")
+
+    match_expr = _contains_all_terms(
+        [
+            func.coalesce(MailMessage.from_address, ""),
+            func.coalesce(MailMessage.subject, ""),
+            cast(MailMessage.received_at, String),
+            func.coalesce(MailMessage.backup_software, ""),
+            func.coalesce(MailMessage.backup_type, ""),
+            func.coalesce(MailMessage.job_name, ""),
+            func.coalesce(MailMessage.parse_result, ""),
+            cast(MailMessage.parsed_at, String),
+        ],
+        patterns,
+    )
+    if match_expr is not None:
+        query = query.filter(match_expr)
+
+    section["total"] = query.count()
+    rows = (
+        query.order_by(MailMessage.received_at.desc().nullslast(), MailMessage.id.desc())
+        .limit(SEARCH_LIMIT_PER_SECTION)
+        .all()
+    )
+
+    for msg in rows:
+        parsed_flag = bool(getattr(msg, "parsed_at", None) or (msg.parse_result or ""))
+        section["items"].append(
+            {
+                "title": msg.subject or f"Message #{msg.id}",
+                "subtitle": f"{msg.from_address or '-'} | {_format_datetime(msg.received_at)}",
+                "meta": f"{msg.backup_software or '-'} / {msg.backup_type or '-'} / {msg.job_name or '-'} | Parsed: {'Yes' if parsed_flag else 'No'}",
+                "link": url_for("main.inbox"),
+            }
+        )
+
+    return section
+
+
+def _build_customers_results(patterns: list[str]) -> dict:
+    section = {
+        "key": "customers",
+        "title": "Customers",
+        "view_all_url": url_for("main.customers"),
+        "total": 0,
+        "items": [],
+    }
+    if not _is_section_allowed("customers"):
+        return section
+
+    query = Customer.query
+    match_expr = _contains_all_terms([func.coalesce(Customer.name, "")], patterns)
+    if match_expr is not None:
+        query = query.filter(match_expr)
+
+    section["total"] = query.count()
+    rows = query.order_by(Customer.name.asc()).limit(SEARCH_LIMIT_PER_SECTION).all()
+    for c in rows:
+        try:
+            job_count = c.jobs.count()
+        except Exception:
+            job_count = 0
+        section["items"].append(
+            {
+                "title": c.name or f"Customer #{c.id}",
+                "subtitle": f"Jobs: {job_count}",
+                "meta": "Active" if c.active else "Inactive",
+                "link": url_for("main.jobs", customer_id=c.id),
+            }
+        )
+
+    return section
+
+
+def _build_jobs_results(patterns: list[str]) -> dict:
+    section = {
+        "key": "jobs",
+        "title": "Jobs",
+        "view_all_url": url_for("main.jobs"),
+        "total": 0,
+        "items": [],
+    }
+    if not _is_section_allowed("jobs"):
+        return section
+
+    query = (
+        db.session.query(
+            Job.id.label("job_id"),
+            Job.backup_software.label("backup_software"),
+            Job.backup_type.label("backup_type"),
+            Job.job_name.label("job_name"),
+            Customer.name.label("customer_name"),
+        )
+        .select_from(Job)
+        .outerjoin(Customer, Customer.id == Job.customer_id)
+        .filter(Job.archived.is_(False))
+        .filter(db.or_(Customer.id.is_(None), Customer.active.is_(True)))
+    )
+
+    match_expr = _contains_all_terms(
+        [
+            func.coalesce(Customer.name, ""),
+            func.coalesce(Job.backup_software, ""),
+            func.coalesce(Job.backup_type, ""),
+            func.coalesce(Job.job_name, ""),
+        ],
+        patterns,
+    )
+    if match_expr is not None:
+        query = query.filter(match_expr)
+
+    section["total"] = query.count()
+    rows = (
+        query.order_by(
+            Customer.name.asc().nullslast(),
+            Job.backup_software.asc(),
+            Job.backup_type.asc(),
+            Job.job_name.asc(),
+        )
+        .limit(SEARCH_LIMIT_PER_SECTION)
+        .all()
+    )
+    for row in rows:
+        section["items"].append(
+            {
+                "title": row.job_name or f"Job #{row.job_id}",
+                "subtitle": f"{row.customer_name or '-'} | {row.backup_software or '-'} / {row.backup_type or '-'}",
+                "meta": "",
+                "link": url_for("main.job_detail", job_id=row.job_id),
+            }
+        )
+
+    return section
+
+
+def _build_daily_jobs_results(patterns: list[str]) -> dict:
+    section = {
+        "key": "daily_jobs",
+        "title": "Daily Jobs",
+        "view_all_url": url_for("main.daily_jobs"),
+        "total": 0,
+        "items": [],
+    }
+    if not _is_section_allowed("daily_jobs"):
+        return section
+
+    query = (
+        db.session.query(
+            Job.id.label("job_id"),
+            Job.job_name.label("job_name"),
+            Job.backup_software.label("backup_software"),
+            Job.backup_type.label("backup_type"),
+            Customer.name.label("customer_name"),
+        )
+        .select_from(Job)
+        .outerjoin(Customer, Customer.id == Job.customer_id)
+        .filter(Job.archived.is_(False))
+        .filter(db.or_(Customer.id.is_(None), Customer.active.is_(True)))
+    )
+
+    match_expr = _contains_all_terms(
+        [
+            func.coalesce(Customer.name, ""),
+            func.coalesce(Job.backup_software, ""),
+            func.coalesce(Job.backup_type, ""),
+            func.coalesce(Job.job_name, ""),
+        ],
+        patterns,
+    )
+    if match_expr is not None:
+        query = query.filter(match_expr)
+
+    section["total"] = query.count()
+    rows = (
+        query.order_by(
+            Customer.name.asc().nullslast(),
+            Job.backup_software.asc(),
+            Job.backup_type.asc(),
+            Job.job_name.asc(),
+        )
+        .limit(SEARCH_LIMIT_PER_SECTION)
+        .all()
+    )
+    for row in rows:
+        section["items"].append(
+            {
+                "title": row.job_name or f"Job #{row.job_id}",
+                "subtitle": f"{row.customer_name or '-'} | {row.backup_software or '-'} / {row.backup_type or '-'}",
+                "meta": "",
+                "link": url_for("main.daily_jobs"),
+            }
+        )
+
+    return section
+
+
+def _build_run_checks_results(patterns: list[str]) -> dict:
+    section = {
+        "key": "run_checks",
+        "title": "Run Checks",
+        "view_all_url": url_for("main.run_checks_page"),
+        "total": 0,
+        "items": [],
+    }
+    if not _is_section_allowed("run_checks"):
+        return section
+
+    agg = (
+        db.session.query(
+            JobRun.job_id.label("job_id"),
+            func.count(JobRun.id).label("run_count"),
+        )
+        .filter(JobRun.reviewed_at.is_(None))
+        .group_by(JobRun.job_id)
+        .subquery()
+    )
+
+    query = (
+        db.session.query(
+            Job.id.label("job_id"),
+            Job.job_name.label("job_name"),
+            Job.backup_software.label("backup_software"),
+            Job.backup_type.label("backup_type"),
+            Customer.name.label("customer_name"),
+            agg.c.run_count.label("run_count"),
+        )
+        .select_from(Job)
+        .join(agg, agg.c.job_id == Job.id)
+        .outerjoin(Customer, Customer.id == Job.customer_id)
+        .filter(Job.archived.is_(False))
+    )
+
+    match_expr = _contains_all_terms(
+        [
+            func.coalesce(Customer.name, ""),
+            func.coalesce(Job.backup_software, ""),
+            func.coalesce(Job.backup_type, ""),
+            func.coalesce(Job.job_name, ""),
+        ],
+        patterns,
+    )
+    if match_expr is not None:
+        query = query.filter(match_expr)
+
+    section["total"] = query.count()
+    rows = (
+        query.order_by(
+            Customer.name.asc().nullslast(),
+            Job.backup_software.asc().nullslast(),
+            Job.backup_type.asc().nullslast(),
+            Job.job_name.asc().nullslast(),
+        )
+        .limit(SEARCH_LIMIT_PER_SECTION)
+        .all()
+    )
+    for row in rows:
+        section["items"].append(
+            {
+                "title": row.job_name or f"Job #{row.job_id}",
+                "subtitle": f"{row.customer_name or '-'} | {row.backup_software or '-'} / {row.backup_type or '-'}",
+                "meta": f"Unreviewed runs: {int(row.run_count or 0)}",
+                "link": url_for("main.run_checks_page"),
+            }
+        )
+
+    return section
+
+
+def _build_tickets_results(patterns: list[str]) -> dict:
+    section = {
+        "key": "tickets",
+        "title": "Tickets",
+        "view_all_url": url_for("main.tickets_page"),
+        "total": 0,
+        "items": [],
+    }
+    if not _is_section_allowed("tickets"):
+        return section
+
+    query = (
+        db.session.query(Ticket)
+        .select_from(Ticket)
+        .outerjoin(TicketScope, TicketScope.ticket_id == Ticket.id)
+        .outerjoin(Customer, Customer.id == TicketScope.customer_id)
+        .outerjoin(Job, Job.id == TicketScope.job_id)
+    )
+
+    match_expr = _contains_all_terms(
+        [
+            func.coalesce(Ticket.ticket_code, ""),
+            func.coalesce(Customer.name, ""),
+            func.coalesce(TicketScope.scope_type, ""),
+            func.coalesce(TicketScope.backup_software, ""),
+            func.coalesce(TicketScope.backup_type, ""),
+            func.coalesce(TicketScope.job_name_match, ""),
+            func.coalesce(Job.job_name, ""),
+        ],
+        patterns,
+    )
+    if match_expr is not None:
+        query = query.filter(match_expr)
+
+    query = query.distinct()
+    section["total"] = query.count()
+    rows = query.order_by(Ticket.start_date.desc().nullslast()).limit(SEARCH_LIMIT_PER_SECTION).all()
+
+    for t in rows:
+        customer_display = "-"
+        scope_summary = "-"
+        try:
+            scope_rows = (
+                db.session.query(
+                    TicketScope.scope_type.label("scope_type"),
+                    TicketScope.backup_software.label("backup_software"),
+                    TicketScope.backup_type.label("backup_type"),
+                    Customer.name.label("customer_name"),
+                )
+                .select_from(TicketScope)
+                .outerjoin(Customer, Customer.id == TicketScope.customer_id)
+                .filter(TicketScope.ticket_id == t.id)
+                .all()
+            )
+            customer_names = []
+            for s in scope_rows:
+                cname = getattr(s, "customer_name", None)
+                if cname and cname not in customer_names:
+                    customer_names.append(cname)
+            if customer_names:
+                customer_display = customer_names[0]
+                if len(customer_names) > 1:
+                    customer_display = f"{customer_display} +{len(customer_names)-1}"
+
+            if scope_rows:
+                s = scope_rows[0]
+                bits = []
+                if getattr(s, "scope_type", None):
+                    bits.append(str(getattr(s, "scope_type")))
+                if getattr(s, "backup_software", None):
+                    bits.append(str(getattr(s, "backup_software")))
+                if getattr(s, "backup_type", None):
+                    bits.append(str(getattr(s, "backup_type")))
+                scope_summary = " / ".join(bits) if bits else "-"
+        except Exception:
+            customer_display = "-"
+            scope_summary = "-"
+
+        section["items"].append(
+            {
+                "title": t.ticket_code or f"Ticket #{t.id}",
+                "subtitle": f"{customer_display} | {scope_summary}",
+                "meta": _format_datetime(t.start_date),
+                "link": url_for("main.ticket_detail", ticket_id=t.id),
+            }
+        )
+
+    return section
+
+
+def _build_overrides_results(patterns: list[str]) -> dict:
+    section = {
+        "key": "overrides",
+        "title": "Existing overrides",
+        "view_all_url": url_for("main.overrides"),
+        "total": 0,
+        "items": [],
+    }
+    if not _is_section_allowed("overrides"):
+        return section
+
+    query = (
+        db.session.query(
+            Override.id.label("id"),
+            Override.level.label("level"),
+            Override.backup_software.label("backup_software"),
+            Override.backup_type.label("backup_type"),
+            Override.object_name.label("object_name"),
+            Override.start_at.label("start_at"),
+            Override.end_at.label("end_at"),
+            Override.comment.label("comment"),
+            Customer.name.label("customer_name"),
+            Job.job_name.label("job_name"),
+        )
+        .select_from(Override)
+        .outerjoin(Job, Job.id == Override.job_id)
+        .outerjoin(Customer, Customer.id == Job.customer_id)
+    )
+
+    match_expr = _contains_all_terms(
+        [
+            func.coalesce(Override.level, ""),
+            func.coalesce(Customer.name, ""),
+            func.coalesce(Override.backup_software, ""),
+            func.coalesce(Override.backup_type, ""),
+            func.coalesce(Job.job_name, ""),
+            func.coalesce(Override.object_name, ""),
+            cast(Override.start_at, String),
+            cast(Override.end_at, String),
+            func.coalesce(Override.comment, ""),
+        ],
+        patterns,
+    )
+    if match_expr is not None:
+        query = query.filter(match_expr)
+
+    section["total"] = query.count()
+    rows = query.order_by(Override.level.asc(), Override.start_at.desc()).limit(SEARCH_LIMIT_PER_SECTION).all()
+    for row in rows:
+        scope_bits = []
+        if row.customer_name:
+            scope_bits.append(row.customer_name)
+        if row.backup_software:
+            scope_bits.append(row.backup_software)
+        if row.backup_type:
+            scope_bits.append(row.backup_type)
+        if row.job_name:
+            scope_bits.append(row.job_name)
+        if row.object_name:
+            scope_bits.append(f"object: {row.object_name}")
+        scope_text = " / ".join(scope_bits) if scope_bits else "All jobs"
+
+        section["items"].append(
+            {
+                "title": (row.level or "override").capitalize(),
+                "subtitle": scope_text,
+                "meta": f"From {_format_datetime(row.start_at)} to {_format_datetime(row.end_at) if row.end_at else '-'} | {row.comment or ''}",
+                "link": url_for("main.overrides"),
+            }
+        )
+
+    return section
+
+
+def _build_reports_results(patterns: list[str]) -> dict:
+    section = {
+        "key": "reports",
+        "title": "Reports",
+        "view_all_url": url_for("main.reports"),
+        "total": 0,
+        "items": [],
+    }
+    if not _is_section_allowed("reports"):
+        return section
+
+    query = ReportDefinition.query
+    match_expr = _contains_all_terms(
+        [
+            func.coalesce(ReportDefinition.name, ""),
+            func.coalesce(ReportDefinition.report_type, ""),
+            cast(ReportDefinition.period_start, String),
+            cast(ReportDefinition.period_end, String),
+            func.coalesce(ReportDefinition.output_format, ""),
+        ],
+        patterns,
+    )
+    if match_expr is not None:
+        query = query.filter(match_expr)
+
+    section["total"] = query.count()
+    rows = query.order_by(ReportDefinition.created_at.desc()).limit(SEARCH_LIMIT_PER_SECTION).all()
+
+    can_edit = get_active_role() in ("admin", "operator", "reporter")
+    for r in rows:
+        section["items"].append(
+            {
+                "title": r.name or f"Report #{r.id}",
+                "subtitle": f"{r.report_type or '-'} | {r.output_format or '-'}",
+                "meta": f"{_format_datetime(r.period_start)} -> {_format_datetime(r.period_end)}",
+                "link": (url_for("main.reports_edit", report_id=r.id) if can_edit else url_for("main.reports")),
+            }
+        )
+
+    return section
+
+
+@main_bp.route("/search")
+@login_required
+def search_page():
+    query = (request.args.get("q") or "").strip()
+    patterns = _build_patterns(query)
+
+    sections = []
+    if patterns:
+        sections.append(_build_inbox_results(patterns))
+        sections.append(_build_customers_results(patterns))
+        sections.append(_build_jobs_results(patterns))
+        sections.append(_build_daily_jobs_results(patterns))
+        sections.append(_build_run_checks_results(patterns))
+        sections.append(_build_tickets_results(patterns))
+        sections.append(_build_overrides_results(patterns))
+        sections.append(_build_reports_results(patterns))
+    else:
+        sections = [
+            {"key": "inbox", "title": "Inbox", "view_all_url": url_for("main.inbox"), "total": 0, "items": []},
+            {"key": "customers", "title": "Customers", "view_all_url": url_for("main.customers"), "total": 0, "items": []},
+            {"key": "jobs", "title": "Jobs", "view_all_url": url_for("main.jobs"), "total": 0, "items": []},
+            {"key": "daily_jobs", "title": "Daily Jobs", "view_all_url": url_for("main.daily_jobs"), "total": 0, "items": []},
+            {"key": "run_checks", "title": "Run Checks", "view_all_url": url_for("main.run_checks_page"), "total": 0, "items": []},
+            {"key": "tickets", "title": "Tickets", "view_all_url": url_for("main.tickets_page"), "total": 0, "items": []},
+            {"key": "overrides", "title": "Existing overrides", "view_all_url": url_for("main.overrides"), "total": 0, "items": []},
+            {"key": "reports", "title": "Reports", "view_all_url": url_for("main.reports"), "total": 0, "items": []},
+        ]
+
+    visible_sections = [s for s in sections if _is_section_allowed(s["key"])]
+    total_hits = sum(int(s.get("total", 0) or 0) for s in visible_sections)
+
+    return render_template(
+        "main/search.html",
+        query=query,
+        sections=visible_sections,
+        total_hits=total_hits,
+        limit_per_section=SEARCH_LIMIT_PER_SECTION,
+    )

containers/backupchecks/src/templates/layout/base.html

@@ -157,6 +157,18 @@
             </li>
             {% endif %}
           </ul>
+          <form method="get" action="{{ url_for('main.search_page') }}" class="d-flex me-3 mb-2 mb-lg-0" role="search" autocomplete="off">
+            <input
+              class="form-control form-control-sm me-2"
+              type="search"
+              name="q"
+              placeholder="Search"
+              aria-label="Search"
+              value="{{ request.args.get('q','') if request.path == url_for('main.search_page') else '' }}"
+              style="min-width: 220px;"
+            />
+            <button class="btn btn-outline-secondary btn-sm" type="submit">Search</button>
+          </form>
           <span class="navbar-text me-3">
             <a class="text-decoration-none" href="{{ url_for('main.user_settings') }}">
               {{ current_user.username }} ({{ active_role }})
@@ -361,4 +373,4 @@
       })();
     </script>
   </body>
-</html>
+</html>

containers/backupchecks/src/templates/main/search.html

@@ -0,0 +1,60 @@
+{% extends "layout/base.html" %}
+{% block content %}
+<h2 class="mb-3">Search</h2>
+
+{% if query %}
+<p class="text-muted mb-3">
+  Query: <strong>{{ query }}</strong> | Total hits: <strong>{{ total_hits }}</strong>
+</p>
+{% else %}
+<div class="alert alert-secondary py-2">
+  Enter a search term in the top navigation bar.
+</div>
+{% endif %}
+
+{% for section in sections %}
+<div class="card mb-3">
+  <div class="card-header d-flex justify-content-between align-items-center">
+    <span>{{ section.title }} ({{ section.total }})</span>
+    <a href="{{ section.view_all_url }}" class="btn btn-sm btn-outline-secondary">Open {{ section.title }}</a>
+  </div>
+  <div class="card-body p-0">
+    {% if section.items %}
+    <div class="table-responsive">
+      <table class="table table-sm mb-0 align-middle">
+        <thead class="table-light">
+          <tr>
+            <th>Result</th>
+            <th>Details</th>
+            <th>Meta</th>
+          </tr>
+        </thead>
+        <tbody>
+          {% for item in section.items %}
+          <tr>
+            <td>
+              {% if item.link %}
+              <a href="{{ item.link }}">{{ item.title }}</a>
+              {% else %}
+              {{ item.title }}
+              {% endif %}
+            </td>
+            <td>{{ item.subtitle }}</td>
+            <td>{{ item.meta }}</td>
+          </tr>
+          {% endfor %}
+        </tbody>
+      </table>
+    </div>
+    {% else %}
+    <div class="p-3 text-muted">No results in this section.</div>
+    {% endif %}
+  </div>
+  {% if section.total > limit_per_section %}
+  <div class="card-footer small text-muted">
+    Showing first {{ limit_per_section }} of {{ section.total }} results.
+  </div>
+  {% endif %}
+</div>
+{% endfor %}
+{% endblock %}

docs/changelog-claude.md

@@ -7,11 +7,15 @@ This file documents all changes made to this project via Claude Code.
 ### Added
 - Added customer-to-jobs navigation by making customer names clickable on the Customers page, linking to `/jobs?customer_id=<id>`
 - Added Jobs page customer filter context UI with an active filter banner and a "Clear filter" action
+- Added global search page (`/search`) with grouped results for Inbox, Customers, Jobs, Daily Jobs, Run Checks, Tickets, Existing overrides, and Reports
+- Added navbar search form to trigger global search from all authenticated pages
 
 ### Changed
 - Changed `/jobs` route to accept optional `customer_id` query parameter and return only jobs for that customer when provided
 - Changed default Jobs listing behavior to keep existing inactive-customer filtering only when no `customer_id` filter is applied
 - Changed `docs/technical-notes-codex.md` with updated "Last updated" date, Customers→Jobs navigation notes, and test build/push validation snapshot
+- Changed search matching to be case-insensitive with wildcard support (`*`) and automatic contains behavior (`*term*`) per search term
+- Changed global search visibility to only include sections accessible to the currently active role
 
 ## [2026-02-13]