cyberyan_linkedin_search/backend/apps/search/indexing.py

"""
Elasticsearch index definition and (re)build helpers.

Index design
------------
* One document per profile, denormalised: the fields a recruiter searches on
  (name, title, employer, skills, education, location) all live in the same
  document, so a query never needs a join.
* Text fields are analysed with an English analyzer (lowercase, ASCII folding,
  stop words, stemming) so "engineering" matches "engineer".
* Every field that is *filtered* or *aggregated* on has a ``keyword``
  counterpart with a lowercase normaliser - analysed text cannot be used for
  exact term filters or facet counts.
* ``full_name.autocomplete`` uses an edge n-gram analyzer for the type-ahead;
  the search analyzer is deliberately different (``standard``) so the query
  term is not itself n-grammed.
* **No personal contact data is indexed.**  E-mail addresses and phone numbers
  stay in PostgreSQL, which removes any chance of confirming someone's e-mail
  through a crafted search query.

Rebuilds use an alias plus a timestamped physical index, so a reindex never
leaves the API pointing at a half-populated index and can be rolled back.
"""

from __future__ import annotations

import logging
from collections.abc import Iterable, Iterator
from datetime import UTC, datetime
from typing import Any

from django.conf import settings
from elasticsearch import BadRequestError, Elasticsearch
from elasticsearch.helpers import bulk

from apps.profiles.models import Profile

logger = logging.getLogger(__name__)

INDEX_SETTINGS: dict[str, Any] = {
    "number_of_shards": 1,
    "number_of_replicas": 0,
    "refresh_interval": "1s",
    "max_result_window": 10_000,
    "analysis": {
        "filter": {
            "english_stop": {"type": "stop", "stopwords": "_english_"},
            "english_stemmer": {"type": "stemmer", "language": "english"},
            "english_possessive": {"type": "stemmer", "language": "possessive_english"},
            "edge_ngrams": {"type": "edge_ngram", "min_gram": 2, "max_gram": 15},
        },
        "normalizer": {
            "lowercase_normalizer": {
                "type": "custom",
                "filter": ["lowercase", "asciifolding"],
            }
        },
        "analyzer": {
            "english_text": {
                "type": "custom",
                "tokenizer": "standard",
                "filter": [
                    "lowercase",
                    "asciifolding",
                    "english_possessive",
                    "english_stop",
                    "english_stemmer",
                ],
            },
            "autocomplete_index": {
                "type": "custom",
                "tokenizer": "standard",
                "filter": ["lowercase", "asciifolding", "edge_ngrams"],
            },
            "autocomplete_search": {
                "type": "custom",
                "tokenizer": "standard",
                "filter": ["lowercase", "asciifolding"],
            },
        },
    },
}

KEYWORD = {"type": "keyword", "normalizer": "lowercase_normalizer"}

INDEX_MAPPINGS: dict[str, Any] = {
    "dynamic": "strict",  # reject unexpected fields instead of guessing a type
    "properties": {
        "id": {"type": "integer"},
        "full_name": {
            "type": "text",
            "analyzer": "english_text",
            "fields": {
                "keyword": KEYWORD,
                "autocomplete": {
                    "type": "text",
                    "analyzer": "autocomplete_index",
                    "search_analyzer": "autocomplete_search",
                },
            },
        },
        "headline": {"type": "text", "analyzer": "english_text"},
        "job_title": {
            "type": "text",
            "analyzer": "english_text",
            "fields": {"keyword": KEYWORD},
        },
        "job_title_role": KEYWORD,
        "job_title_sub_role": KEYWORD,
        "seniority_levels": KEYWORD,
        "company": {
            "type": "text",
            "analyzer": "english_text",
            "fields": {"keyword": KEYWORD},
        },
        "past_companies": {"type": "text", "analyzer": "english_text"},
        "industry": {
            "type": "text",
            "analyzer": "english_text",
            "fields": {"keyword": KEYWORD},
        },
        "skills": {
            "type": "text",
            "analyzer": "english_text",
            "fields": {
                "keyword": KEYWORD,
                "autocomplete": {
                    "type": "text",
                    "analyzer": "autocomplete_index",
                    "search_analyzer": "autocomplete_search",
                },
            },
        },
        "education": {"type": "text", "analyzer": "english_text"},
        "degrees": KEYWORD,
        "majors": {"type": "text", "analyzer": "english_text", "fields": {"keyword": KEYWORD}},
        "summary": {"type": "text", "analyzer": "english_text"},
        "interests": {"type": "text", "analyzer": "english_text"},
        "languages": KEYWORD,
        "location_name": {"type": "text", "analyzer": "english_text"},
        "location_locality": KEYWORD,
        "location_region": KEYWORD,
        "location_country": KEYWORD,
        "location_continent": KEYWORD,
        "years_experience": {"type": "integer"},
        "connections": {"type": "integer"},
        "linkedin_url": {"type": "keyword", "index": False},
        "indexed_at": {"type": "date"},
    },
}


def get_client() -> Elasticsearch:
    """Build an Elasticsearch client from settings (auth optional in dev)."""
    config = settings.SEARCH["ELASTICSEARCH"]
    kwargs: dict[str, Any] = {
        "hosts": config["HOSTS"],
        "request_timeout": config["TIMEOUT"],
        "max_retries": config["MAX_RETRIES"],
        "retry_on_timeout": True,
    }
    if config["USERNAME"] and config["PASSWORD"]:
        kwargs["basic_auth"] = (config["USERNAME"], config["PASSWORD"])
    return Elasticsearch(**kwargs)


def alias_name() -> str:
    return settings.SEARCH["ELASTICSEARCH"]["INDEX"]


def physical_index_name() -> str:
    return f"{alias_name()}-{datetime.now(UTC):%Y%m%d%H%M%S}"


def build_document(profile: Profile) -> dict[str, Any]:
    """Project a ``Profile`` row (plus its relations) into an index document."""
    skills = [link.skill.name for link in profile.profile_skills.all()]
    experiences = list(profile.experiences.all())
    educations = list(profile.educations.all())

    degrees = sorted({degree for edu in educations for degree in edu.degrees})
    majors = sorted({major for edu in educations for major in edu.majors})
    schools = [edu.school_name for edu in educations if edu.school_name]

    return {
        "id": profile.id,
        "full_name": profile.full_name,
        "headline": profile.headline,
        "job_title": profile.job_title,
        "job_title_role": profile.job_title_role,
        "job_title_sub_role": profile.job_title_sub_role,
        "seniority_levels": profile.seniority_levels,
        "company": profile.current_company.name if profile.current_company_id else "",
        "past_companies": sorted({exp.company_name for exp in experiences if exp.company_name}),
        "industry": profile.industry,
        "skills": skills,
        "education": " ".join([*schools, *majors]),
        "degrees": degrees,
        "majors": majors,
        "summary": profile.summary,
        "interests": profile.interests,
        "languages": profile.languages,
        "location_name": profile.location_name,
        "location_locality": profile.location_locality,
        "location_region": profile.location_region,
        "location_country": profile.location_country,
        "location_continent": profile.location_continent,
        "years_experience": profile.years_experience,
        "connections": profile.connections,
        "linkedin_url": profile.linkedin_url,
        "indexed_at": datetime.now(UTC).isoformat(),
    }


def iter_documents(queryset: Iterable[Profile] | None = None) -> Iterator[dict[str, Any]]:
    profiles = queryset if queryset is not None else _default_queryset()
    for profile in profiles:
        yield {"_id": profile.id, "_source": build_document(profile)}


def _default_queryset():  # noqa: ANN202
    return (
        Profile.objects.select_related("current_company")
        .prefetch_related("profile_skills__skill", "experiences", "educations")
        .order_by("id")
        .iterator(chunk_size=200)
    )


# Creating an index waits for the master to ack and the shard to be allocated,
# which can take far longer than the per-request timeout used for searches.
INDEX_CREATE_TIMEOUT = 120


def create_index(client: Elasticsearch, name: str) -> None:
    """
    Create the physical index, tolerating a retried request.

    ``indices.create`` is not idempotent, but the client runs with
    ``retry_on_timeout``: when the cluster acks the creation more slowly than
    the request timeout, the retry comes back as
    ``resource_already_exists_exception`` and the rebuild fails even though the
    index is there.  The name carries a timestamp generated moments earlier, so
    an index already under that name can only be our own first attempt.
    """
    try:
        client.options(request_timeout=INDEX_CREATE_TIMEOUT).indices.create(
            index=name, settings=INDEX_SETTINGS, mappings=INDEX_MAPPINGS
        )
    except BadRequestError as exc:
        if exc.error != "resource_already_exists_exception":
            raise
        logger.warning(
            "index_create_retried",
            extra={"event": "index_create_retried", "index": name},
        )


def rebuild_index(client: Elasticsearch | None = None) -> dict[str, Any]:
    """
    Zero-downtime rebuild.

    Creates a fresh physical index, indexes every profile into it, then flips
    the alias atomically and removes the previous physical index.
    """
    client = client or get_client()
    alias = alias_name()
    new_index = physical_index_name()

    create_index(client, new_index)
    logger.info("index_created", extra={"event": "index_created", "index": new_index})

    # Anything that goes wrong from here on leaves an index nothing points at;
    # drop it rather than accumulating one orphan per failed attempt.
    try:
        succeeded, errors = bulk(
            client,
            ({"_index": new_index, **doc} for doc in iter_documents()),
            raise_on_error=False,
            stats_only=False,
        )
        client.indices.refresh(index=new_index)
    except Exception:
        client.indices.delete(index=new_index, ignore_unavailable=True)
        raise

    previous = _current_indices(client, alias)
    actions: list[dict[str, Any]] = [{"add": {"index": new_index, "alias": alias}}]
    actions += [{"remove": {"index": name, "alias": alias}} for name in previous]
    client.indices.update_aliases(actions=actions)

    for name in previous:
        client.indices.delete(index=name, ignore_unavailable=True)

    result = {
        "alias": alias,
        "index": new_index,
        "indexed": succeeded,
        "errors": len(errors) if isinstance(errors, list) else errors,
        "replaced": previous,
    }
    logger.info("index_rebuilt", extra={"event": "index_rebuilt", **result})
    return result


def _current_indices(client: Elasticsearch, alias: str) -> list[str]:
    if not client.indices.exists_alias(name=alias):
        # A leftover *index* (not alias) with the same name would block the
        # alias creation, so remove it first.
        if client.indices.exists(index=alias):
            client.indices.delete(index=alias, ignore_unavailable=True)
        return []
    return list(client.indices.get_alias(name=alias).keys())


def index_profile(profile: Profile, client: Elasticsearch | None = None) -> None:
    """Index (or re-index) a single profile - used after a targeted update."""
    client = client or get_client()
    client.index(index=alias_name(), id=profile.id, document=build_document(profile))