# (c) cavaliba.com - data - ipam.py

"""
ipam_ip auto-discovery: keep the collected=True corner of the ipam_ip schema
in sync with every plain IPv4 value written anywhere else in the system.

discover_ipam_ip() is the real-time path, called from Instance.save() for
every write. ipam_refresh() is the batch/backfill path (mirrors
app_data.eav.eav_refresh()): a one-off scan of the EAV cache, for anything
written before the real-time hook existed, or via any path that bypasses
Instance.save() (e.g. eav_refresh()'s own direct EavBatch.save() calls).
Both funnel through upsert_ipam_ip(), which never overwrites an existing
entry - manually created or already collected.

app_ipam (the IPAM app) is being deprecated - its views/models for
ipam_subnet/ipam_vlan/calculator moved to app_data (views_ipam.py) or were
removed outright once ipam_subnet/ipam_vlan got full instance_detail/
instance_list pages of their own (see app_data/hook.py). app_ipam/common.py
was ported here too: is_ipv4/is_ipv4_subnet/align_to_subnet/classify_query
are still live (used by views_ipam.py's search box); its other classes
(IpamIP/IpamSubnet/IpamVLAN) and the in-memory subnet-index helpers
(SubnetRef/get_subnet_index/find_containing_subnets/find_child_subnets)
and cached occupancy (compute_subnet_occupancy) were dead code by the time
of the port - test-only, superseded by get_ip_subnet_hierarchy()/
get_subnet_hierarchy() (which climb one ipam_subnet DB query per level
instead of loading the whole subnet list into memory - _climb_ancestors())
and subnet_occupancy()/subnet_ip_list() (single hexip range query) - so
they were deleted rather than ported.

First-level child subnets are a different problem: no bounded candidate
list exists the way _find_containing_subnet() has one for ancestors (a
child could sit at any prefix length), so answering it genuinely requires
looking at every ipam_subnet keyname at least once. build_subnet_forest()
revives the "whole table in memory" idea the old in-memory index had -
but this time cached with a TTL and explicit invalidation (see
Instance.cache_purge()) rather than held forever in an unbounded
process-local dict, and built with a hand-rolled parser instead of
ipaddress.ip_network() (measured ~3x faster at 50k rows - the parse step
dominates build time, not the sort/stack pass).

Method inventory:

Module functions:
    hexip, is_ipv4, is_ipv4_subnet, align_to_subnet, classify_query,
    upsert_ipam_ip, discover_ipam_ip, ipam_refresh, task_ipam_refresh,
    get_related_ipam_ip, subnet_occupancy, _ipam_ip_queryset,
    _enrich_ip_rows, subnet_ip_list, ip_list_filename, subnet_ip_list_csv,
    get_ip_subnet_hierarchy, get_subnet_hierarchy,
    get_vlan_subnet_hierarchies, _find_containing_subnet, _climb_ancestors,
    _build_hierarchy_entries, build_subnet_rows, build_vlan_rows,
    _parse_cidr_fast, build_subnet_forest, get_subnet_forest,
    get_subnet_children
"""

import csv
import ipaddress

from celery import shared_task
from django.conf import settings

import app_home.cache as cache
from app_data.fieldtypes.field_ipv4 import check_valid_ipv4
from app_data.models import DataEAV, DataInstance
from app_data.related import build_related_reply, get_related
from app_data.schema import Schema
from app_home.configuration import get_configuration
from app_home.log import ERROR, INFO, WARNING, log


def is_ipv4(data):
    """Return True if data is a valid single IPv4 address (not a range, not IPv6)."""
    try:
        ipobj = ipaddress.ip_address(data)
    except Exception:
        return False
    return ipobj.version == 4


def is_ipv4_subnet(data):
    """Return True if data is a valid IPv4 CIDR range, aligned or not (host bits allowed)."""
    try:
        netobj = ipaddress.ip_network(data, strict=False)
    except Exception:
        return False
    return netobj.version == 4


def align_to_subnet(ipmask_str):
    """
    align ip/mask to subnet boundaries
    IN:  string   10.1.1.1/24
    OUT: string   10.1.1.0/24

    Example usage
    result = align_to_subnet("10.1.1.1/24")
    print(result)  # Output: 10.1.1.0/24
    """
    iface = ipaddress.ip_interface(ipmask_str)
    network_str = f"{iface.network.network_address}/{iface.network.prefixlen}"
    return network_str


def classify_query(query):
    """
    Classify a free-form IPAM search string (top search box, or any view's own "query" param).

    Returns (kind, value):
    - ("empty", "")                  blank/whitespace-only input
    - ("ipv6_unsupported", query)    valid IPv6 address/range - not supported by this IPv4-only feature
    - ("ip", query)                  valid single IPv4 address
    - ("subnet", cidr)               valid IPv4 CIDR range (aligned), or a partial dotted-octet prefix
                                      ("10", "10.2", "10.2.1") zero-padded and given a /8, /16 or /24 mask
    - ("text", query)                anything else - free-text (VLAN) search
    """
    if not query:
        return ("empty", "")

    query = query.strip()
    if not query:
        return ("empty", "")

    # IPv6 (address or range) ?
    try:
        ifaceobj = ipaddress.ip_interface(query)
        if ifaceobj.version != 4:
            return ("ipv6_unsupported", query)
    except Exception:
        pass

    if is_ipv4(query):
        return ("ip", query)

    if is_ipv4_subnet(query):
        return ("subnet", align_to_subnet(query))

    # partial dotted-octet prefix: "10", "10.2", "10.2.", "10.2.1" -> zero-padded subnet
    candidate = query[:-1] if query.endswith(".") else query
    segments = candidate.split(".")
    if 1 <= len(segments) <= 3:
        try:
            octets = [int(s) for s in segments]
        except ValueError:
            octets = None
        if octets is not None and all(0 <= o <= 255 for o in octets):
            padded = octets + [0] * (4 - len(octets))
            prefixlen = len(octets) * 8
            cidr = "{}/{}".format(".".join(str(o) for o in padded), prefixlen)
            return ("subnet", cidr)

    return ("text", query)


def hexip(ip):
    """Hex encode a single IPv4 address (string or ipaddress.IPv4Address) as
    an 8-char, zero-padded lowercase hex string (e.g. "10.1.2.3" ->
    "0a010203") - sorts lexicographically exactly like the address's own
    numeric value. The single hexip computation in the app - used for
    ipam_ip's keyname_mode="hexip" (Schema.create_keyname_hexip() delegates
    here) and for building range-query boundaries in subnet_occupancy()/
    subnet_ip_list(). Raises ValueError/TypeError for anything that isn't a
    single valid IPv4 address (not a network/CIDR, not IPv6) - callers with
    unvalidated input should catch that or pre-check with check_valid_ipv4()."""
    return format(int(ipaddress.IPv4Address(ip)), "08x")


def upsert_ipam_ip(ip, aaa=None, dryrun=False):
    """Create an ipam_ip instance for ip (collected=True) if one doesn't
    already exist for its hexip keyname. Returns True if created (or, when
    dryrun, would be created); False for an invalid ip, a missing ipam_ip
    schema, or an entry that already exists (manually created or previously
    collected - never overwritten)."""

    if not check_valid_ipv4(ip):
        return False

    if not Schema.exists("ipam_ip"):
        return False

    hexkey = hexip(ip)

    from app_data.data import Instance

    if Instance.from_keyname(classname="ipam_ip", keyname=hexkey, expand=False):
        return False

    if dryrun:
        return True

    ipam_ip = Instance(classname="ipam_ip")
    ipam_ip.displayname = ip
    ipam_ip.set_field_value_single(fieldname="collected", value=True)
    ipam_ip.save(aaa=aaa, action="create", skip_revision=True)
    return True


def discover_ipam_ip(instance, aaa=None):
    """Scan instance's own fields (not the EAV cache) for pure IPv4 values
    (get_eav_format() == "ipv4", i.e. not a "ipv4:subnet" CIDR range) and
    upsert_ipam_ip() each one. Called from Instance.save() after a real DB
    write, reusing the caller's aaa. Never touches ipam_ip itself (no
    infinite loop, and it has no ipv4-format field anyway)."""

    if instance.classname == "ipam_ip":
        return

    ips = set()
    for field in instance.fields.values():
        if field.is_injected:
            continue
        if field.get_eav_format() != "ipv4":
            continue
        ips.update(field.value)

    for ip in ips:
        upsert_ipam_ip(ip, aaa=aaa)


def get_related_ipam_ip(instance, aaa=None):
    """Objects referencing an ipam_ip instance - called from Instance.related()
    instead of app_data.related.get_related() when classname == "ipam_ip".

    ipam_ip's keyname is an opaque hexip string that nothing else ever
    writes into a "schema"-type field, so the generic keyname-match lookup
    finds nothing. What actually "points at" an ipam_ip entry is any
    ipv4-format field elsewhere holding the same IP - instance.displayname,
    not instance.keyname. Same reply shape as get_related(); if aaa
    provided, filtered the same way (app_data.related.is_authorized_schema)."""

    instances_eav = DataEAV.objects.filter(
        format="ipv4", value__iexact=instance.displayname
    ).exclude(classname="ipam_ip")

    return build_related_reply(instances_eav, aaa=aaa)


def _find_containing_subnet(network, include_self):
    """One DB query: the most specific ipam_subnet containing `network` (or
    equal to it, if include_self). Candidates are every supernet CIDR of
    `network` from its own prefixlen (or one less) down to /0 - at most 33
    strings, matched via a single keyname__in query - not a full-table scan.
    None if none of those candidate CIDRs exist as an ipam_subnet."""

    start = network.prefixlen if include_self else network.prefixlen - 1
    if start < 0:
        return None

    candidates = [
        str(ipaddress.ip_network(f"{network.network_address}/{p}", strict=False))
        for p in range(start, -1, -1)
    ]

    rows = DataInstance.objects.filter(classname="ipam_subnet", keyname__in=candidates).values(
        "id", "keyname", "displayname"
    )

    best = None
    for row in rows:
        try:
            prefixlen = int(row["keyname"].rsplit("/", 1)[1])
        except (IndexError, ValueError):
            continue
        if best is None or prefixlen > best["prefixlen"]:
            best = {
                "id": row["id"],
                "keyname": row["keyname"],
                "displayname": row["displayname"],
                "prefixlen": prefixlen,
            }
    return best


def _climb_ancestors(current):
    """Ancestor ipam_subnets of `current` (excluding `current` itself),
    most-specific first: one _find_containing_subnet(..., include_self=
    False) DB query per level, climbing until a step finds nothing (every
    step strictly widens the search by at least one prefix bit, so this is
    bounded by current's own prefixlen, at most 33 steps). Shared climbing
    loop for get_ip_subnet_hierarchy() (whose base subnet is found by
    searching, since it only starts with an IP) and get_subnet_hierarchy()
    (whose base subnet is already known - the instance being viewed)."""

    ancestors = []
    while True:
        found = _find_containing_subnet(current, include_self=False)
        if not found:
            break
        ancestors.append(found)
        current = ipaddress.ip_network(found["keyname"])
    return ancestors


def build_subnet_rows(subnets, aaa=None, with_subnet_detail=False):
    """Shared per-subnet row builder for _build_hierarchy_entries() (the
    ancestor/self chain) and get_subnet_children() (first-level children):
    given `subnets` (list of {id, keyname, displayname} dicts, any order),
    fetch descriptions (one batched EAV query - not a full-table scan) and,
    per subnet, VLAN(s) referencing it via get_related() (an ipam_vlan's
    own "subnet" field lists 0+ ipam_subnet keynames it covers; nothing on
    ipam_subnet points the other way) - every row gets its own VLAN lookup,
    every ipam_subnet here is a real curated subnet someone may want to
    know the VLAN of. If with_subnet_detail, also attach first/last IP,
    prefixlen, dotted netmask, size, and live occupancy (subnet_occupancy())
    per entry.

    Returns entries in the same order as `subnets`: {id, keyname,
    displayname, description, vlans, [first_ip, last_ip, prefixlen,
    netmask, size, occupancy if with_subnet_detail]}. If aaa provided, vlan
    visibility is filtered the same way get_related() filters any other
    related-objects lookup."""

    descriptions = dict(
        DataEAV.objects.filter(
            classname="ipam_subnet",
            fieldname="description",
            keyname__in=[subnet["keyname"] for subnet in subnets],
        ).values_list("keyname", "value")
    )

    def vlans_for(keyname):
        related = get_related(classname="ipam_subnet", keyname=keyname, aaa=aaa)
        return [
            {"id": v.iid, "keyname": v.keyname, "displayname": v.displayname or v.keyname}
            for v in related.get("ipam_vlan", {}).get("instances", [])
        ]

    entries = []
    for subnet in subnets:
        entry = {
            "id": subnet["id"],
            "keyname": subnet["keyname"],
            "displayname": subnet["displayname"] or subnet["keyname"],
            "description": descriptions.get(subnet["keyname"], ""),
            "vlans": vlans_for(subnet["keyname"]),
        }
        if with_subnet_detail:
            try:
                network = ipaddress.ip_network(subnet["keyname"])
            except ValueError:
                log(WARNING, app="data", view="ipam", action="build_subnet_rows", status="KO",
                    data=f"skipping malformed ipam_subnet keyname: {subnet['keyname']!r}")  # fmt: skip
                continue
            entry.update(
                {
                    "first_ip": str(network.network_address),
                    "last_ip": str(network.broadcast_address),
                    "prefixlen": network.prefixlen,
                    "netmask": str(network.netmask),
                    "size": network.num_addresses,
                    "occupancy": subnet_occupancy(subnet["keyname"]),
                }
            )
        entries.append(entry)

    return entries


def build_vlan_rows(vlans, aaa=None):
    """Per-VLAN row builder for instance_list's ipam_vlan hook (mirrors
    build_subnet_rows()): given `vlans` (list of {id, keyname, displayname}
    dicts, any order), batch-fetch description/vlan_id (one EAV query each)
    and the "subnet" field's values (one EAV query, multi-value - 0+ rows
    per VLAN), then resolve each referenced subnet keyname to its own
    id/displayname/size/live occupancy (subnet_occupancy(), memoized per
    unique keyname so a subnet referenced by several VLANs is only queried
    once).

    A listed subnet keyname with no matching/bound ipam_subnet instance is
    skipped, same as get_vlan_subnet_hierarchies(). `aaa` is accepted for
    signature symmetry with build_subnet_rows() but unused - a VLAN's own
    "subnet" field values are just data on an instance whose own read
    permission was already checked upstream, unlike the reverse
    get_related() lookup build_subnet_rows() does for VLANs-of-a-subnet.

    Returns entries in the same order as `vlans`: {id, keyname, displayname,
    description, vlan_id, subnets: [{id, keyname, displayname, size,
    occupancy}, ...]}."""

    keynames = [vlan["keyname"] for vlan in vlans]

    descriptions = dict(
        DataEAV.objects.filter(
            classname="ipam_vlan", fieldname="description", keyname__in=keynames
        ).values_list("keyname", "value")
    )

    vlan_ids = dict(
        DataEAV.objects.filter(
            classname="ipam_vlan", fieldname="vlan_id", keyname__in=keynames
        ).values_list("keyname", "value")
    )

    subnet_keynames_by_vlan = {}
    for vlan_keyname, subnet_keyname in DataEAV.objects.filter(
        classname="ipam_vlan", fieldname="subnet", keyname__in=keynames
    ).values_list("keyname", "value"):
        subnet_keynames_by_vlan.setdefault(vlan_keyname, []).append(subnet_keyname)

    all_subnet_keynames = {
        subnet_keyname
        for subnet_list in subnet_keynames_by_vlan.values()
        for subnet_keyname in subnet_list
    }

    subnet_rows = DataInstance.objects.filter(
        classname="ipam_subnet", keyname__in=all_subnet_keynames
    ).values("id", "keyname", "displayname")
    subnet_by_keyname = {row["keyname"]: row for row in subnet_rows}

    occupancy_cache = {}

    def subnet_entry(subnet_keyname):
        subnet_row = subnet_by_keyname.get(subnet_keyname)
        if not subnet_row:
            return None
        try:
            size = ipaddress.ip_network(subnet_keyname).num_addresses
        except ValueError:
            log(WARNING, app="data", view="ipam", action="build_vlan_rows", status="KO",
                data=f"skipping malformed ipam_subnet keyname: {subnet_keyname!r}")  # fmt: skip
            return None
        if subnet_keyname not in occupancy_cache:
            occupancy_cache[subnet_keyname] = subnet_occupancy(subnet_keyname)
        return {
            "id": subnet_row["id"],
            "keyname": subnet_row["keyname"],
            "displayname": subnet_row["displayname"] or subnet_row["keyname"],
            "size": size,
            "occupancy": occupancy_cache[subnet_keyname],
        }

    entries = []
    for vlan in vlans:
        subnets = [
            entry
            for entry in (
                subnet_entry(subnet_keyname)
                for subnet_keyname in subnet_keynames_by_vlan.get(vlan["keyname"], [])
            )
            if entry
        ]
        entries.append(
            {
                "id": vlan["id"],
                "keyname": vlan["keyname"],
                "displayname": vlan["displayname"] or vlan["keyname"],
                "description": descriptions.get(vlan["keyname"], ""),
                "vlan_id": vlan_ids.get(vlan["keyname"], ""),
                "subnets": subnets,
            }
        )

    return entries


def _build_hierarchy_entries(chain, aaa=None, vlan_for_all=False, with_subnet_detail=False):
    """Ancestor/self-chain wrapper around build_subnet_rows(): given `chain`
    (most-specific-first list of {id, keyname, displayname} dicts), builds
    the shared per-subnet rows, then layers on chain-specific semantics -
    "is_containing" (depth 0 = the instance itself/found subnet), VLANs
    blanked out on every row except the innermost unless vlan_for_all, and
    reversed order with a 2-space-per-position "indent" field, meant for a
    monospace/white-space:pre element (see .cv-ipam-hierarchy-mono in
    cavaliba.css) rather than CSS padding, so depth isn't capped by a fixed
    number of indent classes - outermost supernet at the top, narrowing
    down to the innermost/is_containing entry at the bottom, the natural
    reading order for a nested-subnet display."""

    entries = build_subnet_rows(chain, aaa=aaa, with_subnet_detail=with_subnet_detail)

    for depth, entry in enumerate(entries):
        is_containing = depth == 0
        entry["is_containing"] = is_containing
        if not (vlan_for_all or is_containing):
            entry["vlans"] = []

    entries.reverse()
    for position, entry in enumerate(entries):
        entry["indent"] = "  " * position

    return entries


def get_ip_subnet_hierarchy(ip, aaa=None):
    """For ip, return the full containing-subnet ancestry (climbed via
    _climb_ancestors(), see _build_hierarchy_entries() for the reply shape,
    including per-subnet detail - same columns as get_subnet_hierarchy()).
    [] if ip is invalid or not contained in any known subnet. Only the
    innermost (containing) subnet's VLAN is shown - ancestors here are just
    context for locating one IP, not necessarily subnets of interest in
    their own right."""

    try:
        network = ipaddress.ip_network(f"{ip}/32")
    except (ValueError, TypeError):
        return []

    containing = _find_containing_subnet(network, include_self=True)
    if not containing:
        return []

    chain = [containing] + _climb_ancestors(ipaddress.ip_network(containing["keyname"]))

    return _build_hierarchy_entries(chain, aaa=aaa, with_subnet_detail=True)


def get_subnet_hierarchy(instance, aaa=None):
    """For instance (an ipam_subnet Instance), return itself plus its full
    ancestry (climbed via _climb_ancestors() from instance's own network -
    no search needed, instance is already the known base; see
    _build_hierarchy_entries() for the reply shape, here including
    first/last IP, mask, size and live occupancy per entry). [] if
    instance.keyname isn't a valid CIDR. Unlike get_ip_subnet_hierarchy(),
    every entry gets its own VLAN lookup - every row here is itself a real
    curated ipam_subnet someone may want to know the VLAN of, not merely an
    ancestor found by best-effort containment."""

    try:
        network = ipaddress.ip_network(instance.keyname)
    except (ValueError, TypeError):
        return []

    base = {"id": instance.id, "keyname": instance.keyname, "displayname": instance.displayname}
    chain = [base] + _climb_ancestors(network)

    return _build_hierarchy_entries(chain, aaa=aaa, vlan_for_all=True, with_subnet_detail=True)


def get_vlan_subnet_hierarchies(instance, aaa=None):
    """For instance (an ipam_vlan Instance), return one subnet hierarchy -
    via get_subnet_hierarchy(), same reply shape - per subnet its own
    "subnet" field lists (0+ ipam_subnet keynames; a VLAN can cover several
    subnets, or none). Each entry: {"subnet_keyname": CIDR, "subnet_id":
    int, "hierarchy": [...], "ip_list": {...}}. ip_list is
    subnet_ip_list(keyname) (its own first-N dict), and subnet_id is what
    the template needs to link to that subnet's full IP-list CSV. A listed
    keyname with no matching/bound ipam_subnet instance is skipped. [] if
    the field is empty or absent."""

    from app_data.data import Instance

    try:
        subnet_keynames = instance.fields["subnet"].get_value()
    except KeyError:
        return []

    reply = []
    for keyname in subnet_keynames:
        subnet_instance = Instance.from_keyname(classname="ipam_subnet", keyname=keyname)
        if not subnet_instance or not subnet_instance.is_bound:
            continue
        reply.append(
            {
                "subnet_keyname": keyname,
                "subnet_id": subnet_instance.id,
                "hierarchy": get_subnet_hierarchy(subnet_instance, aaa=aaa),
                "ip_list": subnet_ip_list(keyname),
            }
        )

    return reply


def _parse_cidr_fast(keyname):
    """Parse an ipam_subnet keyname ("A.B.C.D/E") into (start, end, prefixlen)
    integers without ipaddress.ip_network()'s object-construction overhead -
    only used by build_subnet_forest(), where it's called once per row in
    the whole table (measured ~3x faster than ipaddress at 50k rows, and
    that parse step dominates build time). None if keyname isn't a
    well-formed, in-range CIDR string."""

    try:
        ip_part, mask_part = keyname.split("/")
        prefixlen = int(mask_part)
        a, b, c, d = (int(x) for x in ip_part.split("."))
    except (ValueError, AttributeError):
        return None

    if not (0 <= prefixlen <= 32):
        return None
    if not all(0 <= octet <= 255 for octet in (a, b, c, d)):
        return None

    start = (a << 24) | (b << 16) | (c << 8) | d
    end = start + (1 << (32 - prefixlen)) - 1
    return start, end, prefixlen


def build_subnet_forest():
    """
    Build the complete ipam_subnet parent/child forest in one pass over
    every keyname in the table (no EAV, no data_json - ipam_subnet's
    keyname IS the CIDR, nothing else to read). CIDR blocks are always
    either disjoint or fully nested, never partially overlapping, so
    sorting by (start, prefixlen) and walking with a stack yields every
    parent -> first-level-children edge in one O(N log N) pass: for each
    subnet in start order, pop any stack entry it has moved past (nothing
    left can be inside a range once we're processing later starts), and
    whatever remains on top of the stack (if anything) is its immediate
    parent.

    Returns {parent_keyname: [child_keyname, ...]}. A subnet with no
    children has no entry as a key; a subnet with no containing subnet in
    this table (a root) never appears as anyone's value-list entry via a
    parent it doesn't have - it simply isn't listed as a child anywhere.

    Not meant to be called directly per-request - see get_subnet_forest()
    for the cached wrapper actually used by callers.
    """

    keynames = DataInstance.objects.filter(classname="ipam_subnet").values_list(
        "keyname", flat=True
    )

    parsed = []
    for keyname in keynames:
        result = _parse_cidr_fast(keyname)
        if result is None:
            log(WARNING, app="data", view="ipam", action="build_subnet_forest", status="KO",
                data=f"skipping malformed ipam_subnet keyname: {keyname!r}")  # fmt: skip
            continue
        start, end, prefixlen = result
        parsed.append((start, end, prefixlen, keyname))

    parsed.sort(key=lambda row: (row[0], row[2]))

    children = {}
    stack = []
    for start, end, prefixlen, keyname in parsed:
        while stack and stack[-1][1] < start:
            stack.pop()
        if stack:
            children.setdefault(stack[-1][3], []).append(keyname)
        stack.append((start, end, prefixlen, keyname))

    return children


def get_subnet_forest():
    """Cached wrapper around build_subnet_forest(): app_home.cache holds the
    whole {parent_keyname: [children]} dict under one fixed key (it's one
    global structure for the whole ipam_subnet table, not one entry per
    instance). Invalidated explicitly by Instance.cache_purge() on any
    ipam_subnet create/update/delete, with CAVALIBA_CACHE_IPAM_SUBNET_TIMEOUT
    as a TTL backstop.

    Deep-copies on every DataCache.get() call, same as every other cache in
    this app - cheap for the small objects those other caches hold, but
    measurably not free here (tens of ms at 50k subnets). Callers should
    fetch this once per request/task and reuse the local reference rather
    than calling get_subnet_forest() repeatedly for different subnets."""

    forest = cache.cache2_ipam_subnet_forest.get("forest")
    if forest is None:
        forest = build_subnet_forest()
        cache.cache2_ipam_subnet_forest.set("forest", forest)
    return forest


def get_subnet_children(instance, aaa=None):
    """For instance (an ipam_subnet Instance), return its direct (first-level)
    child subnets from the cached forest - descendants with no other known
    subnet in between (a deeper descendant nested inside one of these is
    excluded, it's that child's child, not this instance's). Same reply
    shape/columns as get_subnet_hierarchy() (via the shared
    build_subnet_rows()): {id, keyname, displayname, description, vlans,
    first_ip, last_ip, prefixlen, netmask, size, occupancy}. [] if instance
    has no children.

    Only the cheap topology lookup (get_subnet_forest()) is cached - id and
    every display column are resolved fresh here, scoped to just this
    instance's (typically few) children, same as _build_hierarchy_entries()
    already does for the ancestor chain. Nothing about individual subnets'
    display data is ever cached.

    Order is whatever get_subnet_forest() already produced - ascending by
    CIDR start address (see build_subnet_forest()'s stack pass), not
    keyname string (CIDR text doesn't sort numerically, e.g.
    "10.19.0.0/16" < "10.2.0.0/16" as strings)."""

    forest = get_subnet_forest()
    child_keynames = forest.get(instance.keyname, [])
    if not child_keynames:
        return []

    rows = DataInstance.objects.filter(classname="ipam_subnet", keyname__in=child_keynames).values(
        "id", "keyname", "displayname"
    )
    by_keyname = {row["keyname"]: row for row in rows}

    subnets = [by_keyname[keyname] for keyname in child_keynames if keyname in by_keyname]

    return build_subnet_rows(subnets, aaa=aaa, with_subnet_detail=True)


def subnet_occupancy(cidr):
    """Live occupancy for an ipam_subnet CIDR: how many ipam_ip entries fall
    within it, computed with a single query - no caching, no size gating.

    ipam_ip's keyname is hexip: a fixed-width, zero-padded hex encoding of
    the IP (Schema.create_keyname_hexip), which sorts lexicographically
    exactly like the IP's own numeric value. So a subnet's address range
    maps directly to a keyname BETWEEN clause - counting occupancy is one
    range query on ipam_ip, not a scan of the EAV cache across every schema
    (contrast app_ipam/common.py's compute_subnet_occupancy(), cached and
    gated by CAVALIBA_IPAM_OCCUPANCY_MINPREFIX specifically because that
    full EAV scan is expensive). Correct as long as ipam_ip stays in sync
    with the rest of the app, which upsert_ipam_ip()'s two callers
    (discover_ipam_ip() in real time, ipam_refresh() as a batch backfill)
    are responsible for.

    Returns {"count": int, "percent": float}. Invalid cidr -> count=0."""

    try:
        network = ipaddress.ip_network(cidr)
    except (ValueError, TypeError):
        return {"count": 0, "percent": 0.0}

    start_hex = hexip(network.network_address)
    end_hex = hexip(network.broadcast_address)

    count = DataInstance.objects.filter(
        classname="ipam_ip", keyname__gte=start_hex, keyname__lte=end_hex
    ).count()

    size = network.num_addresses
    percent = min(round(count / size * 100, 2), 100.0) if size else 0.0

    return {"count": count, "percent": percent}


def _ipam_ip_queryset(cidr):
    """The ipam_ip rows inside cidr, ordered by keyname - the shared base
    queryset of subnet_ip_list()/subnet_ip_list_csv(), built with the same
    fast hexip range-query technique as subnet_occupancy() (keyname BETWEEN
    the subnet's two boundary hexips, one indexed range scan - not an EAV
    scan). None if cidr isn't a valid CIDR, so callers can tell "bad input"
    from "no IP in range" without parsing cidr themselves."""

    try:
        network = ipaddress.ip_network(cidr)
    except (ValueError, TypeError):
        return None

    start_hex = hexip(network.network_address)
    end_hex = hexip(network.broadcast_address)

    return (
        DataInstance.objects.filter(
            classname="ipam_ip", keyname__gte=start_hex, keyname__lte=end_hex
        )
        .order_by("keyname")
        .values("id", "keyname", "displayname")
    )


def _enrich_ip_rows(ipam_ip_rows):
    """Pair each ipam_ip row with its own collected/description fields and
    with whatever else references that IP via an ipv4-format field elsewhere
    - three batched EAV queries for the whole chunk handed in, not one query
    per IP. Callers pass a bounded chunk (subnet_ip_list()'s first N, or one
    of subnet_ip_list_csv()'s batches), never a whole wide subnet at once.

    `ipam_ip_rows` is a list of {"id", "keyname", "displayname"} dicts
    (DataInstance .values()). Returns one entry per input row, same order:
    {"id", "keyname", "displayname", "collected", "description",
    "referrers": [{"id", "classname", "keyname", "displayname"}, ...]}.
    "description" is ipam_ip's own field (only meaningful for
    manually-curated, non-collected entries - a collected one was
    auto-discovered from another schema's ipv4 field and never gets one
    written)."""

    if not ipam_ip_rows:
        return []

    keynames = [row["keyname"] for row in ipam_ip_rows]

    collected_by_keyname = dict(
        DataEAV.objects.filter(
            classname="ipam_ip",
            fieldname="collected",
            keyname__in=keynames,
        ).values_list("keyname", "value")
    )

    description_by_keyname = dict(
        DataEAV.objects.filter(
            classname="ipam_ip",
            fieldname="description",
            keyname__in=keynames,
        ).values_list("keyname", "value")
    )

    ips = [row["displayname"] for row in ipam_ip_rows]
    referrers_by_ip = {}
    for eav_row in DataEAV.objects.filter(format="ipv4", value__in=ips).exclude(
        classname="ipam_ip"
    ):
        referrers_by_ip.setdefault(eav_row.value, []).append(
            {
                "id": eav_row.iid,
                "classname": eav_row.classname,
                "keyname": eav_row.keyname,
                "displayname": eav_row.displayname,
            }
        )

    return [
        {
            "id": row["id"],
            "keyname": row["keyname"],
            "displayname": row["displayname"],
            "collected": collected_by_keyname.get(row["keyname"]) in settings.TRUE_LIST,
            "description": description_by_keyname.get(row["keyname"], ""),
            "referrers": referrers_by_ip.get(row["displayname"], []),
        }
        for row in ipam_ip_rows
    ]


def subnet_ip_list(cidr, limit=None):
    """The first `limit` ipam_ip entries within cidr (default
    settings.CAVALIBA_IPAM_IP_LIST_MAX), enriched by _enrich_ip_rows(),
    plus how many there are in total.

    Gated on the number of ipam_ip entries actually found, not on the
    subnet's own size: a wide subnet holding a handful of IPs now lists
    them (refusing outright past a /22, as this did before, hid real data),
    and the cost is one indexed COUNT plus `limit` enriched rows whatever
    the prefix length. Past `limit`, the UI shows these first rows and
    links to the full CSV - see subnet_ip_list_csv().

    Returns {"rows": [...], "total": int, "truncated": bool} - rows ordered
    by keyname (hexip, which sorts the same as the IP's own numeric value).
    An invalid or empty cidr is total=0, not an error."""

    if limit is None:
        limit = settings.CAVALIBA_IPAM_IP_LIST_MAX

    queryset = _ipam_ip_queryset(cidr)
    if queryset is None:
        return {"rows": [], "total": 0, "truncated": False}

    total = queryset.count()
    if not total:
        return {"rows": [], "total": 0, "truncated": False}

    return {
        "rows": _enrich_ip_rows(list(queryset[:limit])),
        "total": total,
        "truncated": total > limit,
    }


def ip_list_filename(cidr):
    """Download filename for a subnet's IP-list CSV: the CIDR with its "/"
    and "." flattened, so it stays one safe path-free filename
    ("10.1.0.0/16" -> "ipam_ip_list_10-1-0-0_16.csv")."""

    slug = str(cidr).replace("/", "_").replace(".", "-")
    return f"ipam_ip_list_{slug}.csv"


def subnet_ip_list_csv(cidr, out, max_rows=None, batch_size=1000):
    """Write every ipam_ip entry within cidr to `out` as CSV - the full list
    the subnet/VLAN detail page only shows the first rows of.

    Streamed in `batch_size` chunks, one _enrich_ip_rows() call per chunk,
    so a subnet holding 100k IPs never builds 100k enriched rows (or their
    referrer map) in memory at once. Each chunk resumes from the last
    keyname written rather than at an OFFSET, so the 100th batch costs the
    same indexed seek as the first. Capped at max_rows (default
    settings.CAVALIBA_IPAM_IP_LIST_FILE_MAX); when the cap bites, a final
    "# truncated ..." comment line says so rather than letting the file end
    looking complete.

    One row per IP - unlike the HTML table, which repeats an IP once per
    referrer - with the referrers joined as space-separated
    "classname:keyname". Delimiter comes from the CSV_DELIMITER
    configuration, same as the exporter's.

    Returns the number of data rows written (header/comment excluded)."""

    if max_rows is None:
        max_rows = settings.CAVALIBA_IPAM_IP_LIST_FILE_MAX

    delimiter = get_configuration(keyname="CSV_DELIMITER") or ","
    writer = csv.writer(out, delimiter=delimiter)
    writer.writerow(["ip", "collected", "description", "referrers"])

    queryset = _ipam_ip_queryset(cidr)
    if queryset is None:
        return 0

    count = 0
    after = ""
    while count < max_rows:
        chunk = list(queryset.filter(keyname__gt=after)[: min(batch_size, max_rows - count)])
        if not chunk:
            break
        after = chunk[-1]["keyname"]
        for entry in _enrich_ip_rows(chunk):
            writer.writerow(
                [
                    entry["displayname"],
                    "yes" if entry["collected"] else "no",
                    entry["description"],
                    " ".join(f"{ref['classname']}:{ref['keyname']}" for ref in entry["referrers"]),
                ]
            )
            count += 1

    if count >= max_rows and queryset.filter(keyname__gt=after).exists():
        out.write(f"# truncated at {count} rows\n")

    return count


@shared_task(ignore_result=True)
def task_ipam_refresh(verbose=False, dryrun=False, progress=False):
    """async to celery ipam_refresh"""
    ipam_refresh(verbose=verbose, dryrun=dryrun, progress=progress)


def ipam_refresh(verbose=False, dryrun=False, progress=False):
    """Loop over the EAV cache for every pure IPv4 value (format == "ipv4")
    and upsert_ipam_ip() each one.

    dryrun defaults to False (real write), same rule as eav_purge()/eav_refresh():
    a maintenance job called with no argument must do its job."""

    from app_data.eav import batch_qs
    from app_data.permissions import permission_all_keynames

    if not Schema.exists("ipam_ip"):
        log(ERROR, app="data", action="ipam_refresh", status="KO",
             data="ipam_ip schema not found, skipping")  # fmt: skip
        return 0, 0

    aaa = {"perms": permission_all_keynames()}

    ips = (
        DataEAV.objects.filter(format="ipv4")
        .exclude(classname="ipam_ip")
        .values_list("value", flat=True)
        .distinct()
    )

    count = 0
    count_create = 0
    for start, end, total, qs in batch_qs(ips, batch_size=500):
        if progress:
            print(f"ipam_refresh - BATCH - processing {start + 1} - {end} of {total}")
        for ip in qs:
            count += 1
            if verbose:
                print(f"        {ip}")
            if upsert_ipam_ip(ip, aaa=aaa, dryrun=dryrun):
                count_create += 1

    log(INFO, app="data", action="ipam_refresh", status="OK",
        data=f"ipam_refresh done. Created {count_create}/{count}")  # fmt: skip

    print("COUNT FOUND:         ", count)
    print("COUNT CREATED:       ", count_create)
    return count_create, count