@check(code="MO007")
def check_model_code_does_not_contain_regexp_pattern(model, *, regexp_pattern: str):
    """The raw code for a model must not match the specified regexp pattern.

    !!! info "Rationale"

        Teams often adopt coding standards that forbid certain SQL patterns — for example, using `ifnull` instead of `coalesce`, or using deprecated functions. This check allows those standards to be enforced automatically at CI time, preventing non-compliant code from reaching production.

    Parameters:
        regexp_pattern (str): The regexp pattern that should not be matched by the model code.

    Receives:
        model (ModelNode): The ModelNode object to check.

    Other Parameters:
        description (str | None): Description of what the check does and why it is implemented.
        exclude (str | list[str] | None): Regex pattern(s) to match the model path. Model paths that match any pattern will not be checked.
        include (str | list[str] | None): Regex pattern(s) to match the model path. Only model paths that match any pattern will be checked.
        materialization (Literal["ephemeral", "incremental", "table", "view"] | None): Limit check to models with the specified materialization.
        severity (Literal["error", "warn"] | None): Severity level of the check. Default: `error`.

    Example(s):
        ```yaml
        manifest_checks:
            # Prefer `coalesce` over `ifnull`: https://docs.sqlfluff.com/en/stable/rules.html#sqlfluff.rules.sphinx.Rule_CV02
            - name: check_model_code_does_not_contain_regexp_pattern
              regexp_pattern: .*[i][f][n][u][l][l].*
        ```

    """
    compiled = compile_pattern(regexp_pattern.strip(), flags=re.DOTALL)
    if compiled.match(str(model.raw_code)) is not None:
        fail(
            f"`{get_clean_model_name(model.unique_id)}` contains a banned string: `{regexp_pattern}`."
        )

@check(code="MO052")
def check_model_does_not_use_cartesian_join(
    model, *, allow_explicit_cross_join: bool = False
):
    """Models must not perform Cartesian or CROSS JOINs.

    !!! info "Rationale"

        Cartesian joins (or CROSS JOINs) join every row of one table to every row
        of another, leading to exponential row explosion, massive warehouse credit
        consumption, and potential query timeouts. Catching unintended cross joins
        at lint time prevents costly query execution errors in production.

    Parameters:
        allow_explicit_cross_join (bool): Whether to allow intentional Cartesian joins. When `True`, both explicit `CROSS JOIN` statements and constant-`ON` conditions (e.g. `ON 1=1`, `ON TRUE`) are permitted; a `JOIN` with no `ON`/`USING` clause still fails, as an omitted clause is more likely accidental. Default: `False`.

    Receives:
        model (ModelNode): The ModelNode object to check.

    Other Parameters:
        description (str | None): Description of what the check does and why it is implemented.
        exclude (str | list[str] | None): Regex pattern(s) to match the model path. Model paths that match any pattern will not be checked.
        include (str | list[str] | None): Regex pattern(s) to match the model path. Only model paths that match any pattern will be checked.
        materialization (Literal["ephemeral", "incremental", "table", "view"] | None): Limit check to models with the specified materialization.
        severity (Literal["error", "warn"] | None): Severity level of the check. Default: `error`.

    !!! info

        Analysis is AST-based (via sqlglot) and flags explicit `CROSS JOIN` keywords,
        missing `ON`/`USING` clauses, and constant `ON` conditions (e.g. `ON 1=1`,
        `ON TRUE`). A condition is constant only when every operand is a literal,
        so a genuine predicate is not flagged whichever side its literal sits on
        (`ON 1 = b.id` and `ON b.id = 1` both pass). `NATURAL JOIN` is not flagged,
        as it joins on the columns the two relations share. Non-SQL (e.g. Python)
        models are skipped.

    !!! warning

        Models that sqlglot cannot parse (e.g. heavy `{% ... %}` control flow)
        fall back to a best-effort regular-expression scan.

    Example(s):
        ```yaml
        manifest_checks:
            - name: check_model_does_not_use_cartesian_join
        ```
        ```yaml
        manifest_checks:
            - name: check_model_does_not_use_cartesian_join
              allow_explicit_cross_join: true
        ```

    """
    if not _is_sql_model(model):
        return

    parsed = parse_sql(neutralize_jinja(model.raw_code or ""))
    if parsed is None:
        # Fallback: best-effort regex on Jinja-stripped raw code.
        cleaned = _strip_sql_comments(model.raw_code or "")
        if not allow_explicit_cross_join and _CROSS_JOIN_PATTERN.search(cleaned):
            fail(
                f"`{get_clean_model_name(model.unique_id)}` uses a Cartesian or `CROSS JOIN`."
            )
        return

    from sqlglot import exp

    model_name = get_clean_model_name(model.unique_id)
    for statement in parsed:
        for join in statement.find_all(exp.Join):
            message = _cartesian_join_failure(
                join, model_name, allow_explicit_cross_join
            )
            if message is not None:
                fail(message)

@check(code="MO008")
def check_model_does_not_use_select_star(model):
    """Models must not use `SELECT *`.

    !!! info "Rationale"

        `SELECT *` makes a model's output schema implicit and brittle to upstream column changes. When a source or upstream model adds, removes, or reorders columns, a `SELECT *` model silently propagates the change, potentially breaking downstream consumers or introducing unexpected columns into the DAG. Explicit column lists are self-documenting, stable, and make schema changes intentional and reviewable.

    Receives:
        model (ModelNode): The ModelNode object to check.

    Other Parameters:
        description (str | None): Description of what the check does and why it is implemented.
        exclude (str | list[str] | None): Regex pattern(s) to match the model path. Model paths that match any pattern will not be checked.
        include (str | list[str] | None): Regex pattern(s) to match the model path. Only model paths that match any pattern will be checked.
        materialization (Literal["ephemeral", "incremental", "table", "view"] | None): Limit check to models with the specified materialization.
        severity (Literal["error", "warn"] | None): Severity level of the check. Default: `error`.

    !!! info

        Analysis is AST-based (via sqlglot) and flags both bare (`SELECT *`) and
        qualified (`SELECT t.*`) stars, while correctly ignoring stars inside
        function calls (`count(*)`), string literals, and comments. It analyses
        the model's `raw_code` with Jinja neutralized. Non-SQL (e.g. Python)
        models are skipped.

    !!! warning

        Models that sqlglot cannot parse (e.g. heavy `{% ... %}` control flow)
        fall back to a best-effort regular-expression scan, which retains the
        original limitations around string literals and Jinja tags.

    Example(s):
        ```yaml
        manifest_checks:
            - name: check_model_does_not_use_select_star
              include: ^models/marts
        ```

    """
    if not _is_sql_model(model):
        return

    parsed = parse_sql(neutralize_jinja(model.raw_code or ""))
    if parsed is None:
        # Fallback: best-effort regex on comment-stripped raw code.
        cleaned = _strip_sql_comments(model.raw_code or "")
        if _SELECT_STAR_PATTERN.search(cleaned):
            fail(
                f"`{get_clean_model_name(model.unique_id)}` uses `SELECT *`; list columns explicitly."
            )
        return

    from sqlglot import exp

    for statement in parsed:
        for select in statement.find_all(exp.Select):
            if _select_uses_star(select):
                fail(
                    f"`{get_clean_model_name(model.unique_id)}` uses `SELECT *`; list columns explicitly."
                )

@check(code="MO009")
def check_model_hard_coded_references(model):
    """A model must not contain hard-coded table references; use ref() or source() instead.

    Flags table references qualified with a schema or catalog (e.g.
    ``FROM schema.table`` or ``JOIN catalog.schema.table``). Hard-coded
    references bypass the dbt DAG, break lineage, and are environment-specific.

    !!! info "Rationale"

        Hard-coded table references bypass dbt's dependency graph, break lineage tracking, and are environment-specific — a reference that works in production will silently read the wrong data in development. Using `ref()` or `source()` ensures models run in the correct order, compile to the right environment, and appear correctly in lineage tools.

    !!! info

        Analysis is AST-based (via sqlglot) on the model's `raw_code` with Jinja
        neutralized, so that `{{ ref(...) }}` / `{{ source(...) }}` are not
        mistaken for hard-coded references. Compiled SQL is intentionally *not*
        used: dbt renders every `ref()`/`source()` into a physical
        schema-qualified relation, which is indistinguishable from a hand-written
        hard-coded reference. Non-SQL (e.g. Python) models are skipped.

    !!! warning

        Models that sqlglot cannot parse (e.g. heavy `{% ... %}` control flow)
        fall back to a best-effort regular-expression scan, which may miss
        references inside complex Jinja logic.

    Receives:
        model (ModelNode): The ModelNode object to check.

    Other Parameters:
        description (str | None): Description of what the check does and why it is implemented.
        exclude (str | list[str] | None): Regex pattern(s) to match the model path. Model paths that match any pattern will not be checked.
        include (str | list[str] | None): Regex pattern(s) to match the model path. Only model paths that match any pattern will be checked.
        materialization (Literal["ephemeral", "incremental", "table", "view"] | None): Limit check to models with the specified materialization.
        severity (Literal["error", "warn"] | None): Severity level of the check. Default: `error`.

    Example(s):
        ```yaml
        manifest_checks:
            - name: check_model_hard_coded_references
        ```

    """
    if not _is_sql_model(model):
        return

    parsed = parse_sql(neutralize_jinja(model.raw_code or ""))
    if parsed is None:
        # Fallback: best-effort regex on Jinja-stripped raw code.
        cleaned = _JINJA_PATTERN.sub("", model.raw_code or "")
        matches = _HARD_CODED_REF_PATTERN.findall(cleaned)
        if matches:
            fail(
                f"`{get_clean_model_name(model.unique_id)}` contains hard-coded table "
                f"references: {matches}. Use `{{{{ ref(...) }}}}` or `{{{{ source(..., ...) }}}}` instead."
            )
        return

    tables = _hard_coded_tables(parsed)
    if tables:
        fail(
            f"`{get_clean_model_name(model.unique_id)}` contains hard-coded table "
            f"references: {tables}. Use `{{{{ ref(...) }}}}` or `{{{{ source(..., ...) }}}}` instead."
        )

@check(code="MO010")
def check_model_has_semi_colon(model):
    """Model may not end with a semi-colon (`;`).

    !!! info "Rationale"

        dbt automatically wraps model SQL before executing it, so a trailing semi-colon can cause syntax errors in certain warehouse adapters. This check catches the mistake at lint time, preventing obscure build failures that can be hard to diagnose in CI.

    Receives:
        model (ModelNode): The ModelNode object to check.

    Other Parameters:
        description (str | None): Description of what the check does and why it is implemented.
        exclude (str | list[str] | None): Regex pattern(s) to match the model path. Model paths that match any pattern will not be checked.
        include (str | list[str] | None): Regex pattern(s) to match the model path. Only model paths that match any pattern will be checked.
        materialization (Literal["ephemeral", "incremental", "table", "view"] | None): Limit check to models with the specified materialization.
        severity (Literal["error", "warn"] | None): Severity level of the check. Default: `error`.

    Example(s):
        ```yaml
        manifest_checks:
            - name: check_model_has_semi_colon
              include: ^models/marts
        ```

    """
    raw_code = (model.raw_code or "").strip()
    if raw_code and raw_code[-1] == ";":
        fail(
            f"`{get_clean_model_name(model.unique_id)}` ends with a semi-colon, this is not permitted."
        )

@check(code="MO011")
def check_model_incremental_has_unique_key(model):
    """Incremental models must declare a `unique_key`.

    !!! info "Rationale"

        Incremental models without a `unique_key` perform insert-only loads.
        Late-arriving or updated rows are silently duplicated on every run,
        leading to over-counting in downstream aggregations and broken
        idempotency. Declaring a `unique_key` enables dbt to merge or
        delete-insert matching rows, keeping the table correct across reruns.

    Receives:
        model (ModelNode): The ModelNode object to check.

    Other Parameters:
        description (str | None): Description of what the check does and why it is implemented.
        exclude (str | list[str] | None): Regex pattern(s) to match the model path. Model paths that match any pattern will not be checked.
        include (str | list[str] | None): Regex pattern(s) to match the model path. Only model paths that match any pattern will be checked.
        materialization (Literal["ephemeral", "incremental", "table", "view"] | None): Limit check to models with the specified materialization.
        severity (Literal["error", "warn"] | None): Severity level of the check. Default: `error`.

    Example(s):
        ```yaml
        manifest_checks:
            - name: check_model_incremental_has_unique_key
        ```

    """
    config = model.config
    if (
        config
        and config.materialized == "incremental"
        and not getattr(config, "unique_key", None)
    ):
        fail(
            f"`{get_clean_model_name(model.unique_id)}` is incremental but has no `unique_key`."
        )

@check(code="MO012")
def check_model_materialization_permitted(
    model, *, permitted_materializations: list[Materialization]
):
    """Models must use a permitted materialization for their location.

    !!! info "Rationale"

        A project may declare a directory-wide materialization in `dbt_project.yml`
        (e.g. all `staging` models are views), but that default can be silently
        overridden by an in-model `config()` block or a nested properties file.
        Asserting the resolved materialization directly - rather than comparing the
        sources of configuration - catches any model whose final materialization was
        overridden away from what the directory is supposed to guarantee.

    Parameters:
        permitted_materializations (list[Materialization]): List of materializations that models are permitted to use, e.g. `ephemeral`, `incremental`, `table`, `view`.

    Receives:
        model (ModelNode): The ModelNode object to check.

    Other Parameters:
        description (str | None): Description of what the check does and why it is implemented.
        exclude (str | list[str] | None): Regex pattern(s) to match the model path. Model paths that match any pattern will not be checked.
        include (str | list[str] | None): Regex pattern(s) to match the model path. Only model paths that match any pattern will be checked.
        materialization (Literal["ephemeral", "incremental", "table", "view"] | None): Limit check to models with the specified materialization.
        severity (Literal["error", "warn"] | None): Severity level of the check. Default: `error`.

    Example(s):
        ```yaml
        manifest_checks:
            - name: check_model_materialization_permitted
              include: ^models/staging
              permitted_materializations:
                - view
        ```

    """
    materialized = model.config.materialized if model.config else None
    if materialized not in permitted_materializations:
        fail(
            f"`{get_clean_model_name(model.unique_id)}` is materialized as "
            f"`{materialized}`, expected one of "
            f"{[m.value for m in permitted_materializations]}."
        )

@check(code="MO013")
def check_model_max_number_of_lines(model, *, max_number_of_lines: int = 100):
    """Models may not have more than the specified number of lines.

    !!! info "Rationale"

        Very long SQL files are a code smell that often indicates a model is doing too much — mixing staging, joining, and aggregating in a single file. Capping line counts encourages splitting large transformations into smaller, focused models that are easier to test, understand, and reuse.

    Parameters:
        max_number_of_lines (int): The maximum number of permitted lines.

    Receives:
        model (ModelNode): The ModelNode object to check.

    Other Parameters:
        description (str | None): Description of what the check does and why it is implemented.
        exclude (str | list[str] | None): Regex pattern(s) to match the model path. Model paths that match any pattern will not be checked.
        include (str | list[str] | None): Regex pattern(s) to match the model path. Only model paths that match any pattern will be checked.
        materialization (Literal["ephemeral", "incremental", "table", "view"] | None): Limit check to models with the specified materialization.
        severity (Literal["error", "warn"] | None): Severity level of the check. Default: `error`.

    Example(s):
        ```yaml
        manifest_checks:
            - name: check_model_max_number_of_lines
        ```
        ```yaml
        manifest_checks:
            - name: check_model_max_number_of_lines
              max_number_of_lines: 150
        ```

    """
    if max_number_of_lines <= 0:
        raise ValueError(
            f"`max_number_of_lines` must be greater than 0, got {max_number_of_lines}."
        )

    actual_number_of_lines = (model.raw_code or "").count("\n") + 1

    if actual_number_of_lines > max_number_of_lines:
        fail(
            f"`{get_clean_model_name(model.unique_id)}` has {actual_number_of_lines} lines, this is more than the maximum permitted number of lines ({max_number_of_lines})."
        )