"""Field-notes domain: captures, comments, targets, inference.

Port of v1 main.py — load_comments / load_comment_targets /
load_comments_expanded / collect_expansion_targets / post_comment /
delete_target_refs (≈1757-2580), inferred_targets_for / add_inferred_targets
(≈2170-2477), the comment routes' logic (≈6647-6875) and the capture/queue
API logic (≈8352-8485) — restructured onto the v2 dual-table layer.

P1 PARITY DUAL-WRITE (the load-bearing rule of this module)
-----------------------------------------------------------
v2 has BOTH ``field_notes`` (the immutable raw log, carried from v1) and
``notes`` (the visible dual-voice diary layer). Until the P4 assistant
pipeline flips authorship, every comment is written EXACTLY like
scripts/migrate_v1.py derives the diary layer:

- the comment row goes to ``field_notes`` (body/kind/comment_date/targets,
  status='processed'), and
- IF the body is non-empty, a mirror ``notes`` row is written with
  author='user', body_source='user_raw', body VERBATIM,
  source_capture_id=<field_note id>, and targets mirrored to
  ``note_targets`` with link_source='confirmed'.

Media-only comments (empty body) get NO mirror — same as the migration's
"one notes row per bodied field_note" invariant. Edits and deletes keep the
mirror in sync (delete removes it; an edit that empties the body removes it;
an edit that adds a body creates it). ``note_media`` is intentionally NOT
written here: P1 comments carry media in the v1 pipe-delimited
``field_notes.photo_paths`` (deviation D3), exactly like migrated rows.

NARROWED TARGET INFERENCE (the one sanctioned behavior improvement)
-------------------------------------------------------------------
v1's ``inferred_targets_for`` over-linked: a comment on a broad area was
attached to every plant/species/variety in the subtree (the "sawfly comment
linked to apple and alyssum" bug, PLANNING.md backlog + PROCESSING.md Rule
11). The v2 port implements the narrowed scope specified there:

- DEFAULT inference is ONLY: the year (from comment_date) + the serving
  watering station(s) + the parent-area chain of wherever the target lives.
- species / variety / plant-group secondaries are added ONLY when the
  species or variety is explicitly named in the comment body (word-boundary,
  case-insensitive match on name/common_name; names shorter than 3 chars are
  ignored to avoid pathological matches). Plant-group links additionally
  require the plant to be inside the target's own scope (area subtree /
  planting / station coverage) and are capped at 10, v1's PLANT_CAP.
- One definitional exception: a comment on a VARIETY still links its parent
  species — that reference is structural, not over-linking.
- v1's "single planting" inference for species/variety targets is dropped —
  plantings are not part of the narrowed default set.
- Archived entities are never inferred (v1's post-filter, kept verbatim).
"""
from __future__ import annotations

import re
from datetime import date, datetime, timezone

import sqlalchemy as sa
from fastapi import UploadFile
from sqlalchemy import delete, func, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session

from ..errors import AppError
from ..models.catalog import Species, Variety
from ..models.garden import Area, AreaStation, WateringStation, Year
from ..models.history import PipelineRun
from ..models.notes import FieldNote, FieldNoteTarget, Note, NoteRevision, NoteTarget
from ..models.planting import Plant, Planting
from . import media
from .audit import log_add, log_delete, log_edit

# --- vocabularies (v1 main.py:180-188) ---
# "feedback" is v2-only: app feedback captured while testing (not garden
# diary content — the diary feed excludes it; see load_comments_expanded).
COMMENT_KINDS = ("", "observation", "question", "done", "issue", "idea", "note", "feedback")
COMMENT_TARGET_TYPES = frozenset(
    {"area", "plant", "species", "variety", "planting", "year", "station"}
)
NOTE_STATUSES = ("new", "processing", "processed", "archived")

# --- error codes (notes domain) ---
NOTE_NOT_FOUND = "note_not_found"
NOTE_EMPTY = "note_empty"
NOTE_INVALID_STATUS = "note_invalid_status"
NOTE_INVALID_TARGET_TYPE = "note_invalid_target_type"
NOTE_NO_UPDATABLE_FIELDS = "note_no_updatable_fields"
NOTE_LAST_TARGET = "note_last_target"
NOTE_NO_AUDIO = "note_no_audio"
NOTE_AUDIO_FILE_MISSING = "note_audio_file_missing"

# Narrowed-inference caps (PLANT_CAP carried from v1's area rule).
PLANT_CAP = 10
_MIN_MENTION_LEN = 3


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _short(value, limit: int = 80) -> str:
    """v1 main.py:984 — stringify + truncate for compact log display."""
    if value is None:
        return ""
    s = str(value)
    if len(s) > limit:
        s = s[: limit - 1] + "…"
    return s


def _not_archived(col):
    """Nullable-boolean archived filter (v1 stored 0/1; v2 allows NULL)."""
    return col.isnot(True)


# ---------------------------------------------------------------------------
# Area tree helpers (v1 build_tree / get_ancestors / get_descendant_ids /
# area_path_name, trimmed to what this domain needs)
# ---------------------------------------------------------------------------

def load_area_maps(db: Session, garden_id: int, include_archived: bool = True) -> dict:
    """id -> {id, name, short_name, parent_id, children:[ids]} for the garden.

    Defaults to INCLUDING archived areas so labels/ancestor chains still
    resolve for old comments that reference archived areas.
    """
    stmt = select(
        Area.id, Area.name, Area.short_name, Area.parent_id
    ).where(Area.garden_id == garden_id)
    if not include_archived:
        stmt = stmt.where(_not_archived(Area.is_archived))
    by_id: dict = {
        r.id: {
            "id": r.id,
            "name": r.name,
            "short_name": r.short_name,
            "parent_id": r.parent_id,
            "children": [],
        }
        for r in db.execute(stmt)
    }
    for a in by_id.values():
        pid = a["parent_id"]
        if pid and pid in by_id:
            by_id[pid]["children"].append(a["id"])
    return by_id


def get_ancestor_ids(area_id: int, by_id: dict) -> list[int]:
    """[root, ..., parent, self] — empty when the area is unknown."""
    path: list[int] = []
    cur = by_id.get(area_id)
    while cur:
        path.append(cur["id"])
        pid = cur["parent_id"]
        cur = by_id.get(pid) if pid else None
    return list(reversed(path))


def get_descendant_ids(area_id: int, by_id: dict) -> set[int]:
    result: set[int] = set()

    def walk(aid: int) -> None:
        node = by_id.get(aid)
        if not node:
            return
        for child_id in node["children"]:
            result.add(child_id)
            walk(child_id)

    walk(area_id)
    return result


def area_path_name(area_id: int, by_id: dict) -> str:
    return " > ".join(
        by_id[a]["name"] for a in get_ancestor_ids(area_id, by_id)
    )


def lca_area_ids(area_ids: list[int], by_id: dict) -> list[int]:
    """v1 main.py:2140 — ancestor chain of the lowest common ancestor."""
    area_ids = [a for a in area_ids if a in by_id]
    if not area_ids:
        return []
    chains = [get_ancestor_ids(a, by_id) for a in area_ids]
    common: list[int] = []
    for items in zip(*chains):
        if len(set(items)) == 1:
            common.append(items[0])
        else:
            break
    return common


# ---------------------------------------------------------------------------
# Target labels / urls (v1 target_label / target_short_label / target_url)
# ---------------------------------------------------------------------------

def target_url(target_type: str, target_id: int) -> str:
    return {
        "area": f"/areas/{target_id}",
        "species": f"/species/{target_id}",
        "variety": f"/varieties/{target_id}",
        "planting": f"/plantings/{target_id}",
        "year": f"/years/{target_id}",
        "plant": f"/plants/{target_id}",
        "station": f"/stations/{target_id}",
    }.get(target_type, "/")


def target_label(
    db: Session, target_type: str, target_id: int, areas_by_id: dict | None = None
) -> str:
    """Human label for any commentable entity (v1 main.py:1682)."""
    if target_type == "area":
        if areas_by_id and target_id in areas_by_id:
            return area_path_name(target_id, areas_by_id)
        name = db.scalar(select(Area.name).where(Area.id == target_id))
        return name if name else f"area #{target_id}"
    if target_type == "species":
        name = db.scalar(select(Species.name).where(Species.id == target_id))
        return name if name else f"species #{target_id}"
    if target_type == "variety":
        row = db.execute(
            select(Variety.name, Species.name.label("species_name"))
            .join(Species, Species.id == Variety.species_id, isouter=True)
            .where(Variety.id == target_id)
        ).first()
        if row:
            return f"{row.name} ({row.species_name})" if row.species_name else row.name
        return f"variety #{target_id}"
    if target_type == "planting":
        row = db.execute(
            select(Planting.name, Year.year.label("year_value"))
            .join(Year, Year.id == Planting.year_id, isouter=True)
            .where(Planting.id == target_id)
        ).first()
        if row:
            return f"{row.year_value} — {row.name}" if row.year_value else row.name
        return f"planting #{target_id}"
    if target_type == "year":
        y = db.scalar(select(Year.year).where(Year.id == target_id))
        return str(y) if y is not None else f"year #{target_id}"
    if target_type == "plant":
        row = db.execute(
            select(
                Plant.quantity,
                Species.name.label("species_name"),
                Variety.name.label("variety_name"),
            )
            .join(Species, Species.id == Plant.species_id, isouter=True)
            .join(Variety, Variety.id == Plant.variety_id, isouter=True)
            .where(Plant.id == target_id)
        ).first()
        if row:
            name = row.variety_name or row.species_name or f"plant #{target_id}"
            qty = (row.quantity or "").strip()
            return f"{qty}× {name}" if qty else name
        return f"plant #{target_id}"
    if target_type == "station":
        name = db.scalar(
            select(WateringStation.name).where(WateringStation.id == target_id)
        )
        return name if name else f"station #{target_id}"
    return f"{target_type} #{target_id}"


def target_short_label(
    db: Session, target_type: str, target_id: int, areas_by_id: dict | None = None
) -> str:
    """Chip label — for areas the leaf name (short_name preferred), not the path."""
    if target_type == "area":
        if areas_by_id and target_id in areas_by_id:
            a = areas_by_id[target_id]
            return a.get("short_name") or a["name"]
        row = db.execute(
            select(Area.name, Area.short_name).where(Area.id == target_id)
        ).first()
        if row:
            return row.short_name or row.name
        return f"area #{target_id}"
    return target_label(db, target_type, target_id, areas_by_id)


def _target_out(
    db: Session, t_type: str, t_id: int, by_id: dict | None, is_primary: bool = False
) -> dict:
    return {
        "type": t_type,
        "id": t_id,
        "is_primary": is_primary,
        "label": target_label(db, t_type, t_id, by_id),
        "short_label": target_short_label(db, t_type, t_id, by_id),
        "url": target_url(t_type, t_id),
    }


# ---------------------------------------------------------------------------
# Row shaping
# ---------------------------------------------------------------------------

def _photos_list(photo_paths: str | None) -> list[str]:
    return [p for p in (photo_paths or "").split("|") if p]


def capture_dict(fn: FieldNote) -> dict:
    """CaptureOut shape — the raw field_notes row + split photos list."""
    return {
        "id": fn.id,
        "created_at": fn.created_at,
        "captured_at": fn.captured_at,
        "audio_path": fn.audio_path,
        "photo_paths": fn.photo_paths or "",
        "photos": _photos_list(fn.photo_paths),
        "text": fn.text or "",
        "transcript": fn.transcript or "",
        "status": fn.status,
        "processed_at": fn.processed_at,
        "processing_notes": fn.processing_notes or "",
        "body": fn.body or "",
        "kind": fn.kind or "",
        "comment_date": fn.comment_date,
        "page_context": fn.page_context or "",
        "parent_id": fn.parent_id,
        "is_archived": bool(fn.is_archived),
    }


def _base_comment(fn: FieldNote) -> dict:
    return {
        "id": fn.id,
        "comment_date": fn.comment_date,
        "created_at": fn.created_at,
        "captured_at": fn.captured_at,
        "body": fn.body or "",
        "kind": fn.kind or "",
        "transcript": fn.transcript or "",
        "audio_path": fn.audio_path,
        "photos": _photos_list(fn.photo_paths),
        "page_context": fn.page_context or "",
        "is_archived": bool(fn.is_archived),
        "primary": None,
        "other_targets": [],
        "is_primary_for_scope": True,
    }


def _targets_by_note(db: Session, note_ids: list[int]) -> dict[int, list]:
    """All field_note_targets for the given notes, primary-first per note."""
    if not note_ids:
        return {}
    rows = db.execute(
        select(
            FieldNoteTarget.field_note_id,
            FieldNoteTarget.target_type,
            FieldNoteTarget.target_id,
            FieldNoteTarget.is_primary,
        )
        .where(FieldNoteTarget.field_note_id.in_(note_ids))
        .order_by(FieldNoteTarget.is_primary.desc().nulls_last(), FieldNoteTarget.id)
    ).all()
    out: dict[int, list] = {}
    for r in rows:
        out.setdefault(r.field_note_id, []).append(r)
    return out


# ---------------------------------------------------------------------------
# Reading comments (v1 load_comments / load_comment_targets /
# collect_expansion_targets / load_comments_expanded)
# ---------------------------------------------------------------------------

def load_comments_for_target(
    db: Session,
    garden_id: int,
    target_type: str,
    target_id: int,
    by_id: dict | None = None,
    include_archived: bool = False,
) -> list[dict]:
    """Comments attached to one target. v1 load_comments shape: ``primary``
    is the note's primary target; ``other_targets`` is every target EXCEPT
    the requested one (v1 kept the primary in that list when it differs)."""
    if by_id is None:
        by_id = load_area_maps(db, garden_id)
    stmt = (
        select(FieldNote)
        .join(FieldNoteTarget, FieldNoteTarget.field_note_id == FieldNote.id)
        .where(
            FieldNote.garden_id == garden_id,
            FieldNoteTarget.target_type == target_type,
            FieldNoteTarget.target_id == target_id,
            # App feedback is QA channel, not diary content (NULL-safe).
            FieldNote.kind.is_distinct_from("feedback"),
        )
        .order_by(
            FieldNote.comment_date.desc().nulls_last(), FieldNote.created_at.desc()
        )
    )
    if not include_archived:
        stmt = stmt.where(_not_archived(FieldNote.is_archived))
    fns = db.scalars(stmt).all()
    targets = _targets_by_note(db, [fn.id for fn in fns])
    result = []
    for fn in fns:
        c = _base_comment(fn)
        trs = targets.get(fn.id, [])
        prim = next((t for t in trs if t.is_primary), None)
        if prim:
            c["primary"] = _target_out(
                db, prim.target_type, prim.target_id, by_id, is_primary=True
            )
        c["other_targets"] = [
            _target_out(db, t.target_type, t.target_id, by_id)
            for t in trs
            if not (t.target_type == target_type and t.target_id == target_id)
        ]
        c["is_primary_for_scope"] = bool(
            prim and prim.target_type == target_type and prim.target_id == target_id
        )
        result.append(c)
    return result


def load_comment_targets(
    db: Session, garden_id: int, comment_id: int, by_id: dict | None = None
) -> list[dict]:
    """All targets for one comment, labeled (v1 load_comment_targets)."""
    if by_id is None:
        by_id = load_area_maps(db, garden_id)
    _get_field_note(db, garden_id, comment_id)  # 404 guard
    rows = db.execute(
        select(
            FieldNoteTarget.target_type,
            FieldNoteTarget.target_id,
            FieldNoteTarget.is_primary,
        )
        .where(FieldNoteTarget.field_note_id == comment_id)
        .order_by(FieldNoteTarget.is_primary.desc().nulls_last(), FieldNoteTarget.id)
    ).all()
    return [
        _target_out(db, r.target_type, r.target_id, by_id, bool(r.is_primary))
        for r in rows
    ]


def collect_expansion_targets(
    db: Session,
    garden_id: int,
    target_type: str,
    target_id: int,
    include_children: bool = False,
    include_related: bool = False,
    by_id: dict | None = None,
) -> list[tuple[str, int]]:
    """(type, id) tuples for the expanded comment-thread lookup.

    Verbatim port of v1 collect_expansion_targets (main.py:1827-1938) —
    the READ-side expansion is unchanged; only WRITE-side inference was
    narrowed. Always includes the direct target first.
    """
    direct: list[tuple[str, int]] = [(target_type, target_id)]
    extras: list[tuple[str, int]] = []

    def _add(t: str, i) -> None:
        if i is not None:
            extras.append((t, int(i)))

    def _plants_where(*conds):
        return db.execute(
            select(
                Plant.id, Plant.planting_id, Plant.species_id,
                Plant.variety_id, Plant.area_id,
            ).where(Plant.garden_id == garden_id, *conds)
        ).all()

    if target_type == "area":
        if include_children and by_id:
            for aid in get_descendant_ids(target_id, by_id):
                extras.append(("area", aid))
        if include_related:
            area_ids = {target_id}
            if by_id:
                area_ids |= get_descendant_ids(target_id, by_id)
            for p in _plants_where(Plant.area_id.in_(area_ids)):
                _add("planting", p.planting_id)
                _add("species", p.species_id)
                _add("variety", p.variety_id)
                _add("plant", p.id)
            for r in db.execute(
                select(AreaStation.station_id.distinct()).where(
                    AreaStation.area_id.in_(area_ids)
                )
            ):
                _add("station", r[0])
    elif target_type == "planting":
        plants = _plants_where(Plant.planting_id == target_id)
        if include_children:
            for p in plants:
                _add("plant", p.id)
        if include_related:
            for p in plants:
                _add("area", p.area_id)
                _add("species", p.species_id)
                _add("variety", p.variety_id)
    elif target_type == "species":
        if include_children:
            for vid in db.scalars(
                select(Variety.id).where(Variety.species_id == target_id)
            ):
                _add("variety", vid)
        if include_related:
            for p in _plants_where(Plant.species_id == target_id):
                _add("planting", p.planting_id)
                _add("area", p.area_id)
                _add("plant", p.id)
    elif target_type == "variety":
        if include_related:
            sp_id = db.scalar(
                select(Variety.species_id).where(Variety.id == target_id)
            )
            if sp_id:
                _add("species", sp_id)
            for p in _plants_where(Plant.variety_id == target_id):
                _add("planting", p.planting_id)
                _add("area", p.area_id)
                _add("plant", p.id)
    elif target_type == "station":
        if include_related:
            for aid in db.scalars(
                select(AreaStation.area_id).where(AreaStation.station_id == target_id)
            ):
                _add("area", aid)
    elif target_type == "year":
        if include_related:
            pids = list(
                db.scalars(
                    select(Planting.id).where(
                        Planting.garden_id == garden_id,
                        Planting.year_id == target_id,
                    )
                )
            )
            for pid in pids:
                _add("planting", pid)
            if pids:
                for p in _plants_where(Plant.planting_id.in_(pids)):
                    _add("plant", p.id)
                    _add("species", p.species_id)
                    _add("variety", p.variety_id)
                    _add("area", p.area_id)
    elif target_type == "plant":
        if include_related:
            row = db.execute(
                select(
                    Plant.planting_id, Plant.area_id, Plant.species_id, Plant.variety_id
                ).where(Plant.id == target_id, Plant.garden_id == garden_id)
            ).first()
            if row:
                _add("planting", row.planting_id)
                _add("area", row.area_id)
                _add("species", row.species_id)
                _add("variety", row.variety_id)

    seen: set = set()
    out: list[tuple[str, int]] = []
    for t in direct + extras:
        if t not in seen:
            seen.add(t)
            out.append(t)
    return out


def load_comments_expanded(
    db: Session,
    garden_id: int,
    target_type: str,
    target_id: int,
    include_children: bool = False,
    include_related: bool = False,
    by_id: dict | None = None,
    primary_only_for_direct: bool = False,
    include_archived: bool = False,
) -> list[dict]:
    """v1 load_comments_expanded — same shape as load_comments_for_target but
    across children/related entities. ``primary_only_for_direct`` restricts
    the direct target to is_primary matches (the /years default)."""
    if by_id is None:
        by_id = load_area_maps(db, garden_id)
    targets = collect_expansion_targets(
        db, garden_id, target_type, target_id, include_children, include_related, by_id
    )
    if not targets:
        return []

    comment_ids: set[int] = set()
    direct = (target_type, target_id)
    indirect = [t for t in targets if t != direct]

    direct_stmt = select(FieldNoteTarget.field_note_id).where(
        FieldNoteTarget.target_type == target_type,
        FieldNoteTarget.target_id == target_id,
    )
    if primary_only_for_direct:
        direct_stmt = direct_stmt.where(FieldNoteTarget.is_primary.is_(True))
    comment_ids.update(db.scalars(direct_stmt))

    if indirect:
        comment_ids.update(
            db.scalars(
                select(FieldNoteTarget.field_note_id.distinct()).where(
                    sa.tuple_(
                        FieldNoteTarget.target_type, FieldNoteTarget.target_id
                    ).in_(indirect)
                )
            )
        )
    if not comment_ids:
        return []

    stmt = (
        select(FieldNote)
        .where(
            FieldNote.garden_id == garden_id,
            FieldNote.id.in_(comment_ids),
            # App feedback is QA channel, not diary content (NULL-safe).
            FieldNote.kind.is_distinct_from("feedback"),
        )
        .order_by(
            FieldNote.comment_date.desc().nulls_last(),
            FieldNote.created_at.desc(),
        )
    )
    if not include_archived:
        stmt = stmt.where(_not_archived(FieldNote.is_archived))
    fns = db.scalars(stmt).all()
    targets_map = _targets_by_note(db, [fn.id for fn in fns])

    result = []
    for fn in fns:
        c = _base_comment(fn)
        trs = targets_map.get(fn.id, [])
        prim = next((t for t in trs if t.is_primary), None)
        if prim:
            c["primary"] = _target_out(
                db, prim.target_type, prim.target_id, by_id, is_primary=True
            )
        c["other_targets"] = [
            _target_out(db, t.target_type, t.target_id, by_id)
            for t in trs
            if not t.is_primary
            and not (
                prim
                and t.target_type == prim.target_type
                and t.target_id == prim.target_id
            )
        ]
        c["is_primary_for_scope"] = bool(
            prim and prim.target_type == target_type and prim.target_id == target_id
        )
        result.append(c)
    return result


# ---------------------------------------------------------------------------
# Narrowed target inference
# ---------------------------------------------------------------------------

def get_or_create_year(db: Session, garden_id: int, year_value: int) -> int:
    yid = db.scalar(
        select(Year.id).where(Year.garden_id == garden_id, Year.year == year_value)
    )
    if yid is not None:
        return yid
    year = Year(garden_id=garden_id, year=year_value, notes="", created_at=_now())
    db.add(year)
    db.flush()
    return year.id


def _stations_for_areas(db: Session, area_ids: set[int]) -> set[int]:
    """Active stations directly attached to any of these areas. Direct
    assignment is the complete serving set in v1's model (the station form
    auto-checks descendants, each saved individually)."""
    if not area_ids:
        return set()
    return set(
        db.scalars(
            select(AreaStation.station_id.distinct())
            .join(WateringStation, WateringStation.id == AreaStation.station_id)
            .where(
                AreaStation.area_id.in_(area_ids),
                _not_archived(WateringStation.is_archived),
            )
        )
    )


def _mentioned_catalog_ids(
    db: Session, garden_id: int, body_text: str
) -> tuple[set[int], set[int]]:
    """Species/variety ids explicitly named in the body (word-boundary,
    case-insensitive; names < 3 chars ignored). Active rows only."""
    if not (body_text or "").strip():
        return set(), set()
    text = body_text.lower()

    def mentioned(name: str | None) -> bool:
        name = (name or "").strip().lower()
        if len(name) < _MIN_MENTION_LEN:
            return False
        return re.search(rf"\b{re.escape(name)}\b", text) is not None

    sp_rows = db.execute(
        select(Species.id, Species.name, Species.common_name).where(
            Species.garden_id == garden_id, _not_archived(Species.is_archived)
        )
    ).all()
    species_ids = {
        r.id for r in sp_rows if mentioned(r.name) or mentioned(r.common_name)
    }
    var_rows = db.execute(
        select(Variety.id, Variety.name)
        .join(Species, Species.id == Variety.species_id, isouter=True)
        .where(
            sa.or_(Species.garden_id == garden_id, Variety.species_id.is_(None)),
            _not_archived(Variety.is_archived),
        )
    ).all()
    variety_ids = {r.id for r in var_rows if mentioned(r.name)}
    return species_ids, variety_ids


def inferred_targets_for(
    db: Session,
    garden_id: int,
    target_type: str,
    target_id: int,
    comment_date: date | None = None,
    body_text: str = "",
) -> list[tuple[str, int]]:
    """Inherited secondary targets for a comment on a primary entity —
    NARROWED scope (see module docstring). Cached at comment-creation time;
    represents state at time of posting, exactly as in v1."""
    inferred: list[tuple[str, int]] = []
    by_id = load_area_maps(db, garden_id, include_archived=False)

    # Year: always derived from comment_date when provided (v1 rule kept).
    if comment_date:
        inferred.append(("year", get_or_create_year(db, garden_id, comment_date.year)))

    def add_area_chain(area_ids: set[int]) -> None:
        for aid in area_ids:
            for anc in get_ancestor_ids(aid, by_id):
                inferred.append(("area", anc))

    # --- default set: parent-area chain + serving stations ---
    scope_area_ids: set[int] = set()   # where the target itself lives
    if target_type == "area":
        add_area_chain({target_id})
        scope_area_ids = {target_id}
        inferred.extend(
            ("station", sid) for sid in _stations_for_areas(db, {target_id})
        )
    elif target_type == "plant":
        row = db.execute(
            select(Plant.area_id).where(
                Plant.id == target_id, Plant.garden_id == garden_id
            )
        ).first()
        if row is None:
            return []
        if row.area_id:
            scope_area_ids = {row.area_id}
            add_area_chain(scope_area_ids)
            inferred.extend(
                ("station", sid) for sid in _stations_for_areas(db, scope_area_ids)
            )
    elif target_type == "planting":
        plant_areas = set(
            db.scalars(
                select(Plant.area_id.distinct()).where(
                    Plant.planting_id == target_id,
                    Plant.garden_id == garden_id,
                    Plant.area_id.isnot(None),
                    _not_archived(Plant.is_archived),
                )
            )
        )
        scope_area_ids = plant_areas
        add_area_chain(plant_areas)
        inferred.extend(
            ("station", sid) for sid in _stations_for_areas(db, plant_areas)
        )
    elif target_type in ("species", "variety"):
        if target_type == "variety":
            # Definitional: a variety's parent species is not over-linking.
            sp_id = db.scalar(
                select(Variety.species_id).where(Variety.id == target_id)
            )
            if sp_id:
                inferred.append(("species", sp_id))
            plant_filter = Plant.variety_id == target_id
        else:
            plant_filter = Plant.species_id == target_id
        plant_areas = list(
            db.scalars(
                select(Plant.area_id).where(
                    plant_filter,
                    Plant.garden_id == garden_id,
                    Plant.area_id.isnot(None),
                    _not_archived(Plant.is_archived),
                )
            )
        )
        if plant_areas:
            for aid in lca_area_ids(plant_areas, by_id):
                inferred.append(("area", aid))
            inferred.extend(
                ("station", sid)
                for sid in _stations_for_areas(db, set(plant_areas))
            )
    elif target_type == "station":
        attached = set(
            db.scalars(
                select(AreaStation.area_id).where(
                    AreaStation.station_id == target_id
                )
            )
        )
        add_area_chain(attached)
        served = set(attached)
        for aid in attached:
            served |= get_descendant_ids(aid, by_id)
        scope_area_ids = served
    # target_type == "year": nothing beyond the comment_date year.

    # --- mention-gated species / variety / plant-group secondaries ---
    m_species, m_varieties = _mentioned_catalog_ids(db, garden_id, body_text)
    inferred.extend(("species", sid) for sid in sorted(m_species))
    inferred.extend(("variety", vid) for vid in sorted(m_varieties))
    if (m_species or m_varieties) and target_type in ("area", "planting", "station"):
        if target_type == "area":
            scope_area_ids = scope_area_ids | get_descendant_ids(target_id, by_id)
        conds = [
            Plant.garden_id == garden_id,
            _not_archived(Plant.is_archived),
            sa.or_(
                Plant.species_id.in_(m_species) if m_species else sa.false(),
                Plant.variety_id.in_(m_varieties) if m_varieties else sa.false(),
            ),
        ]
        if target_type == "planting":
            conds.append(Plant.planting_id == target_id)
        else:
            if not scope_area_ids:
                conds.append(sa.false())
            else:
                conds.append(Plant.area_id.in_(scope_area_ids))
        plant_ids = list(
            db.scalars(select(Plant.id).where(*conds).order_by(Plant.id))
        )[:PLANT_CAP]
        inferred.extend(("plant", pid) for pid in plant_ids)

    # --- de-dupe + drop self (v1) ---
    seen: set = set()
    out: list[tuple[str, int]] = []
    for pair in inferred:
        if pair == (target_type, target_id) or pair in seen:
            continue
        seen.add(pair)
        out.append(pair)
    if not out:
        return out

    # --- drop archived entities (v1 post-filter, kept verbatim) ---
    model_for = {
        "area": Area,
        "planting": Planting,
        "plant": Plant,
        "species": Species,
        "variety": Variety,
        "station": WateringStation,
        "year": Year,
    }
    by_type: dict[str, set[int]] = {}
    for t, i in out:
        by_type.setdefault(t, set()).add(i)
    archived: set = set()
    for t, ids in by_type.items():
        model = model_for.get(t)
        if model is None:
            continue
        for rid in db.scalars(
            select(model.id).where(model.id.in_(ids), model.is_archived.is_(True))
        ):
            archived.add((t, rid))
    return [pair for pair in out if pair not in archived]


def add_inferred_targets(
    db: Session,
    garden_id: int,
    comment_id: int,
    target_type: str,
    target_id: int,
    comment_date: date | None = None,
    body_text: str = "",
) -> None:
    """Insert inherited secondary targets for a new comment. Idempotent."""
    for t_type, t_id in inferred_targets_for(
        db, garden_id, target_type, target_id, comment_date, body_text
    ):
        db.execute(
            pg_insert(FieldNoteTarget)
            .values(
                field_note_id=comment_id,
                target_type=t_type,
                target_id=t_id,
                is_primary=False,
            )
            .on_conflict_do_nothing()
        )


# ---------------------------------------------------------------------------
# notes-mirror helpers (P1 dual-write; see module docstring)
# ---------------------------------------------------------------------------

def _mirror_note(db: Session, comment_id: int) -> Note | None:
    return db.scalar(
        select(Note).where(
            Note.source_capture_id == comment_id,
            Note.author == "user",
            Note.body_source == "user_raw",
        )
    )


def _mirror_targets_sync(db: Session, note_id: int, comment_id: int) -> None:
    """Make note_targets equal the comment's field_note_targets (mirrored
    with link_source='confirmed', like migrate_v1.py)."""
    db.execute(delete(NoteTarget).where(NoteTarget.note_id == note_id))
    rows = db.execute(
        select(
            FieldNoteTarget.target_type,
            FieldNoteTarget.target_id,
            FieldNoteTarget.is_primary,
        ).where(FieldNoteTarget.field_note_id == comment_id)
    ).all()
    for r in rows:
        db.add(
            NoteTarget(
                note_id=note_id,
                target_type=r.target_type,
                target_id=r.target_id,
                is_primary=bool(r.is_primary),
                link_source="confirmed",
            )
        )


def _mirror_after_write(db: Session, garden_id: int, fn: FieldNote) -> None:
    """Create/update/delete the mirror note so the diary layer keeps the
    migration invariant: one user_raw note per BODIED field_note.

    kind='feedback' (app feedback) never mirrors — it is QA channel, not
    diary content, matching the exclusion in the comment loaders."""
    body = (fn.body or "").strip()
    if (fn.kind or "") == "feedback":
        body = ""  # falls through to the delete-mirror branch below
    mirror = _mirror_note(db, fn.id)
    if body:
        if mirror is None:
            mirror = Note(
                garden_id=garden_id,
                source_capture_id=fn.id,
                author="user",
                body_md=fn.body,
                body_source="user_raw",
                voice_locked=False,
                kind=fn.kind or "",
                note_date=fn.comment_date,
                is_pinned=False,
                created_at=fn.created_at,
                is_archived=bool(fn.is_archived),
            )
            db.add(mirror)
            db.flush()
        else:
            if not mirror.voice_locked and mirror.body_md != fn.body:
                db.add(
                    NoteRevision(
                        note_id=mirror.id,
                        body=mirror.body_md,
                        body_source=mirror.body_source,
                        created_at=_now(),
                    )
                )
                mirror.body_md = fn.body
            mirror.kind = fn.kind or ""
            mirror.note_date = fn.comment_date
            mirror.is_archived = bool(fn.is_archived)
            mirror.updated_at = _now()
        _mirror_targets_sync(db, mirror.id, fn.id)
    elif mirror is not None:
        db.delete(mirror)  # note_targets/media/revisions cascade
        db.flush()


def _delete_mirrors_for(db: Session, comment_id: int) -> None:
    """Remove the user_raw mirror; detach (not delete) any OTHER notes that
    reference this capture (future assistant notes must survive), because
    notes.source_capture_id has no ON DELETE behavior."""
    mirror = _mirror_note(db, comment_id)
    if mirror is not None:
        db.delete(mirror)
        db.flush()
    db.execute(
        update(Note)
        .where(Note.source_capture_id == comment_id)
        .values(source_capture_id=None)
    )


# ---------------------------------------------------------------------------
# Upload helpers
# ---------------------------------------------------------------------------

def _has_file(upload: UploadFile | None) -> bool:
    return upload is not None and bool((upload.filename or "").strip())


def _save_uploads(
    audio: UploadFile | None, photos: list[UploadFile] | None
) -> tuple[str | None, list[str]]:
    """Save all uploads; on any failure delete what was already saved
    (v1 api_note_create cleanup semantics) and re-raise."""
    audio_rel: str | None = None
    photo_rels: list[str] = []
    saved: list[str] = []
    try:
        if _has_file(audio):
            audio_rel = media.save_upload(audio, "audio")
            saved.append(audio_rel)
        for ph in photos or []:
            if _has_file(ph):
                rel = media.save_upload(ph, "photo")
                saved.append(rel)
                photo_rels.append(rel)
    except Exception:
        for rel in saved:
            media.delete_note_file(rel)
        raise
    return audio_rel, photo_rels


# ---------------------------------------------------------------------------
# Captures (v1 /api/notes)
# ---------------------------------------------------------------------------

def _get_field_note(db: Session, garden_id: int, note_id: int) -> FieldNote:
    fn = db.scalar(
        select(FieldNote).where(
            FieldNote.id == note_id, FieldNote.garden_id == garden_id
        )
    )
    if fn is None:
        raise AppError(NOTE_NOT_FOUND, f"note {note_id} not found", 404)
    return fn


def create_capture(
    db: Session,
    garden_id: int,
    *,
    text: str = "",
    transcript: str = "",
    captured_at: datetime | None = None,
    audio: UploadFile | None = None,
    photos: list[UploadFile] | None = None,
    kind: str = "",
    page_context: str = "",
) -> dict:
    """Raw capture intake — a status='new' field_notes row (v1 api_note_create).

    v2 additions: ``kind`` (notably 'feedback' for app feedback captured while
    testing) and ``page_context`` (the SPA route the capture was made from).
    """
    text = (text or "").strip()
    transcript = (transcript or "").strip()
    if kind and kind not in COMMENT_KINDS:
        kind = ""
    page_context = (page_context or "").strip()[:300]
    if not (text or _has_file(audio) or any(_has_file(p) for p in photos or [])):
        raise AppError(NOTE_EMPTY, "note must include audio, a photo, or text", 400)
    audio_rel, photo_rels = _save_uploads(audio, photos)
    fn = FieldNote(
        garden_id=garden_id,
        created_at=_now(),
        captured_at=captured_at,
        audio_path=audio_rel,
        photo_paths="|".join(photo_rels),
        text=text,
        transcript=transcript,
        status="new",
        processing_notes="",
        body="",
        kind=kind,
        page_context=page_context,
    )
    db.add(fn)
    db.flush()
    log_add(db, garden_id, "field_note", fn.id,
            _short(text or transcript or "(media-only capture)", 60))
    db.commit()
    return capture_dict(fn)


def list_captures(
    db: Session, garden_id: int, status: str = "new", limit: int = 50,
    kind: str = "",
) -> list[dict]:
    """v1 api_notes_list: bad status falls back to 'new'; 'all' allowed.

    v2 addition: optional ``kind`` filter (e.g. kind='feedback' lists app
    feedback regardless of processing status).
    """
    if status not in NOTE_STATUSES and status != "all":
        status = "new"
    limit = max(1, min(200, limit))
    stmt = (
        select(FieldNote)
        .where(FieldNote.garden_id == garden_id)
        .order_by(FieldNote.created_at.desc(), FieldNote.id.desc())
        .limit(limit)
    )
    if status != "all":
        stmt = stmt.where(FieldNote.status == status)
    if kind and kind in COMMENT_KINDS:
        stmt = stmt.where(FieldNote.kind == kind)
    return [capture_dict(fn) for fn in db.scalars(stmt)]


def queue_counts(db: Session, garden_id: int) -> dict[str, int]:
    counts = {s: 0 for s in NOTE_STATUSES}
    for status, n in db.execute(
        select(FieldNote.status, func.count())
        .where(FieldNote.garden_id == garden_id)
        .group_by(FieldNote.status)
    ):
        counts[status] = n
    return counts


def update_capture(db: Session, garden_id: int, note_id: int, data: dict) -> dict:
    """PATCH semantics (v1 api_note_update): status / transcript / text /
    processing_notes; status='processed' also stamps processed_at."""
    fn = _get_field_note(db, garden_id, note_id)
    changed = False
    if "status" in data:
        s = (data.get("status") or "").strip()
        if s not in NOTE_STATUSES:
            raise AppError(
                NOTE_INVALID_STATUS,
                f"invalid status; must be one of {list(NOTE_STATUSES)}",
                400,
            )
        if s != fn.status:
            log_edit(db, garden_id, "field_note", fn.id, "status", fn.status, s)
        fn.status = s
        if s == "processed":
            fn.processed_at = _now()
        changed = True
    for field in ("transcript", "text", "processing_notes"):
        if field in data:
            new_val = data.get(field) or ""
            old_val = getattr(fn, field) or ""
            if new_val != old_val:
                log_edit(
                    db, garden_id, "field_note", fn.id, field,
                    _short(old_val, 500), _short(new_val, 500),
                )
            setattr(fn, field, new_val)
            changed = True
    if not changed:
        raise AppError(NOTE_NO_UPDATABLE_FIELDS, "no updatable fields provided", 400)
    db.commit()
    return capture_dict(fn)


def delete_field_note(db: Session, garden_id: int, note_id: int) -> None:
    """Delete a capture OR comment row: media files cleaned from disk, the
    user_raw notes mirror removed, other referencing notes detached
    (v1 api_note_delete + comment_delete, with the sanctioned addition that
    comment deletion also cleans its media files)."""
    fn = _get_field_note(db, garden_id, note_id)
    paths = ([fn.audio_path] if fn.audio_path else []) + _photos_list(fn.photo_paths)
    entity = "comment" if (fn.body or "").strip() else "field_note"
    label = _short(fn.body or fn.text or fn.transcript or "(media-only)", 60)
    _delete_mirrors_for(db, note_id)
    # pipeline_runs.capture_id has no ON DELETE — remove the capture's
    # pipeline audit rows so the FK doesn't block (P4 pipeline).
    db.execute(delete(PipelineRun).where(PipelineRun.capture_id == note_id))
    db.delete(fn)  # field_note_targets cascade; child parent_id SET NULL
    log_delete(db, garden_id, entity, note_id, label)
    db.commit()
    for rel in paths:
        media.delete_note_file(rel)


def get_capture_audio_abs_path(db: Session, garden_id: int, note_id: int) -> str:
    """For the transcribe route: the note's audio file, resolved safely."""
    fn = _get_field_note(db, garden_id, note_id)
    if not fn.audio_path:
        raise AppError(NOTE_NO_AUDIO, "note has no audio attached", 400)
    abs_path = media.safe_note_path(fn.audio_path)
    if not abs_path:
        raise AppError(NOTE_AUDIO_FILE_MISSING, "audio file missing on disk", 404)
    return abs_path


def set_capture_transcript(
    db: Session, garden_id: int, note_id: int, transcript: str
) -> None:
    fn = _get_field_note(db, garden_id, note_id)
    log_edit(
        db, garden_id, "field_note", fn.id, "transcript",
        _short(fn.transcript or "", 500), _short(transcript, 500),
    )
    fn.transcript = transcript
    db.commit()


# ---------------------------------------------------------------------------
# Comments (v1 post_comment + /comments routes)
# ---------------------------------------------------------------------------

def get_comment(
    db: Session, garden_id: int, comment_id: int, by_id: dict | None = None
) -> dict:
    """One comment in the CommentOut shape (expanded-style targets)."""
    if by_id is None:
        by_id = load_area_maps(db, garden_id)
    fn = _get_field_note(db, garden_id, comment_id)
    c = _base_comment(fn)
    trs = _targets_by_note(db, [fn.id]).get(fn.id, [])
    prim = next((t for t in trs if t.is_primary), None)
    if prim:
        c["primary"] = _target_out(
            db, prim.target_type, prim.target_id, by_id, is_primary=True
        )
    c["other_targets"] = [
        _target_out(db, t.target_type, t.target_id, by_id)
        for t in trs
        if not t.is_primary
    ]
    return c


def create_comment(
    db: Session,
    garden_id: int,
    *,
    target_type: str,
    target_id: int,
    comment_date: date,
    body: str = "",
    kind: str = "",
    transcript: str = "",
    captured_at: datetime | None = None,
    audio: UploadFile | None = None,
    photos: list[UploadFile] | None = None,
    explicit_secondaries: list[tuple[str, int]] | None = None,
    page_context: str = "",
) -> dict:
    """v1 post_comment + comment_create, plus the P1 notes-mirror dual-write.

    ``explicit_secondaries``: when not None, used as-is (auto-inference is
    skipped even when empty — the user removed everything pre-submit).
    """
    if target_type not in COMMENT_TARGET_TYPES:
        raise AppError(NOTE_INVALID_TARGET_TYPE, "invalid target type", 400)
    if kind and kind not in COMMENT_KINDS:
        kind = ""
    body = (body or "").strip()
    transcript = (transcript or "").strip()
    has_media = _has_file(audio) or any(_has_file(p) for p in photos or [])
    if not body and not has_media:
        # v1's HTML flow silently redirected; the API is explicit.
        raise AppError(NOTE_EMPTY, "comment must include a body or media", 400)
    audio_rel, photo_rels = _save_uploads(audio, photos)

    ts = _now()
    fn = FieldNote(
        garden_id=garden_id,
        created_at=ts,
        captured_at=captured_at,
        audio_path=audio_rel,
        photo_paths="|".join(photo_rels),
        text="",
        transcript=transcript,
        status="processed",
        processed_at=ts,
        processing_notes="",
        body=body,
        kind=kind,
        comment_date=comment_date,
        page_context=(page_context or "").strip()[:300],
    )
    db.add(fn)
    db.flush()
    db.add(
        FieldNoteTarget(
            field_note_id=fn.id,
            target_type=target_type,
            target_id=target_id,
            is_primary=True,
        )
    )
    db.flush()
    if explicit_secondaries is not None:
        for t_type, t_id in explicit_secondaries:
            if (t_type, t_id) == (target_type, target_id):
                continue
            if t_type not in COMMENT_TARGET_TYPES:
                continue
            db.execute(
                pg_insert(FieldNoteTarget)
                .values(
                    field_note_id=fn.id,
                    target_type=t_type,
                    target_id=int(t_id),
                    is_primary=False,
                )
                .on_conflict_do_nothing()
            )
    else:
        add_inferred_targets(
            db, garden_id, fn.id, target_type, target_id, comment_date, body
        )

    _mirror_after_write(db, garden_id, fn)

    primary_label = target_label(db, target_type, target_id)
    body_short = _short(body or transcript or "(media-only)", 60)
    log_add(
        db, garden_id, "comment", fn.id,
        f"on {target_type}: {primary_label} — {body_short}",
    )
    db.commit()
    return get_comment(db, garden_id, fn.id)


def update_comment(
    db: Session,
    garden_id: int,
    comment_id: int,
    *,
    comment_date: date | None = None,
    body: str | None = None,
    kind: str | None = None,
    audio: UploadFile | None = None,
    remove_audio: bool = False,
    photos: list[UploadFile] | None = None,
    remove_photos: list[str] | None = None,
) -> dict:
    """PATCH semantics over v1 comment_update: only provided fields change.
    Media: remove/replace audio; drop listed photos (files deleted); append
    new uploads. Mirror note kept in sync."""
    fn = _get_field_note(db, garden_id, comment_id)
    to_delete: list[str] = []

    if kind is not None and kind not in COMMENT_KINDS:
        kind = ""

    cur_audio = fn.audio_path
    new_audio_uploaded = _has_file(audio)
    if (remove_audio or new_audio_uploaded) and cur_audio:
        to_delete.append(cur_audio)
        cur_audio = None
    kept_photos = []
    removal_set = set(remove_photos or [])
    for p in _photos_list(fn.photo_paths):
        if p in removal_set:
            to_delete.append(p)
        else:
            kept_photos.append(p)
    new_audio_rel, new_photo_rels = _save_uploads(
        audio if new_audio_uploaded else None, photos
    )
    if new_audio_uploaded:
        cur_audio = new_audio_rel
    kept_photos.extend(new_photo_rels)

    if comment_date is not None and comment_date != fn.comment_date:
        log_edit(
            db, garden_id, "comment", fn.id, "comment_date",
            fn.comment_date, comment_date,
        )
        fn.comment_date = comment_date
    if body is not None:
        body = body.strip()
        if body != (fn.body or ""):
            log_edit(
                db, garden_id, "comment", fn.id, "body",
                _short(fn.body or "", 500), _short(body, 500),
            )
        fn.body = body
    if kind is not None and kind != (fn.kind or ""):
        log_edit(db, garden_id, "comment", fn.id, "kind", fn.kind, kind)
        fn.kind = kind
    old_media = (fn.audio_path or "", fn.photo_paths or "")
    fn.audio_path = cur_audio
    fn.photo_paths = "|".join(kept_photos)
    if (fn.audio_path or "", fn.photo_paths or "") != old_media:
        log_edit(
            db, garden_id, "comment", fn.id, "media",
            "|".join(filter(None, old_media)),
            "|".join(filter(None, (fn.audio_path or "", fn.photo_paths or ""))),
        )

    _mirror_after_write(db, garden_id, fn)
    db.commit()
    for rel in to_delete:
        media.delete_note_file(rel)
    return get_comment(db, garden_id, comment_id)


def add_comment_target(
    db: Session, garden_id: int, comment_id: int, target_type: str, target_id: int
) -> list[dict]:
    """v1 comment_target_add: secondary target, idempotent."""
    if target_type not in COMMENT_TARGET_TYPES:
        raise AppError(NOTE_INVALID_TARGET_TYPE, "invalid target type", 400)
    fn = _get_field_note(db, garden_id, comment_id)
    db.execute(
        pg_insert(FieldNoteTarget)
        .values(
            field_note_id=comment_id,
            target_type=target_type,
            target_id=target_id,
            is_primary=False,
        )
        .on_conflict_do_nothing()
    )
    mirror = _mirror_note(db, comment_id)
    if mirror is not None:
        _mirror_targets_sync(db, mirror.id, comment_id)
    log_edit(
        db, garden_id, "comment", fn.id, "targets",
        None, f"added {target_type} #{target_id}",
    )
    db.commit()
    return load_comment_targets(db, garden_id, comment_id)


def remove_comment_target(
    db: Session, garden_id: int, comment_id: int, target_type: str, target_id: int
) -> list[dict]:
    """v1 comment_target_delete: refuse to remove the last target; when the
    primary is removed, promote the first remaining target (by id)."""
    fn = _get_field_note(db, garden_id, comment_id)
    count = db.scalar(
        select(func.count()).select_from(FieldNoteTarget).where(
            FieldNoteTarget.field_note_id == comment_id
        )
    )
    if count <= 1:
        raise AppError(
            NOTE_LAST_TARGET,
            "cannot remove last target; delete the note instead",
            400,
        )
    row = db.execute(
        select(FieldNoteTarget).where(
            FieldNoteTarget.field_note_id == comment_id,
            FieldNoteTarget.target_type == target_type,
            FieldNoteTarget.target_id == target_id,
        )
    ).scalar_one_or_none()
    if row is None:
        return load_comment_targets(db, garden_id, comment_id)  # v1: silent no-op
    was_primary = bool(row.is_primary)
    db.delete(row)
    db.flush()
    if was_primary:
        remaining = db.scalar(
            select(FieldNoteTarget)
            .where(FieldNoteTarget.field_note_id == comment_id)
            .order_by(FieldNoteTarget.id)
            .limit(1)
        )
        if remaining is not None:
            remaining.is_primary = True
    mirror = _mirror_note(db, comment_id)
    if mirror is not None:
        _mirror_targets_sync(db, mirror.id, comment_id)
    log_edit(
        db, garden_id, "comment", fn.id, "targets",
        f"{target_type} #{target_id}", "removed",
    )
    db.commit()
    return load_comment_targets(db, garden_id, comment_id)


def delete_target_refs(db: Session, garden_id: int, target_type: str, target_id: int) -> None:
    """v1 delete_target_refs — for entity-deletion flows in other domains:
    remove all target rows pointing at an entity, delete orphaned comments
    (mirror included), re-point primary when it was lost."""
    affected = list(
        db.scalars(
            select(FieldNoteTarget.field_note_id).where(
                FieldNoteTarget.target_type == target_type,
                FieldNoteTarget.target_id == target_id,
            )
        )
    )
    db.execute(
        delete(FieldNoteTarget).where(
            FieldNoteTarget.target_type == target_type,
            FieldNoteTarget.target_id == target_id,
        )
    )
    for cid in affected:
        remaining = db.execute(
            select(FieldNoteTarget)
            .where(FieldNoteTarget.field_note_id == cid)
            .order_by(FieldNoteTarget.is_primary.desc().nulls_last(), FieldNoteTarget.id)
        ).scalars().all()
        if not remaining:
            _delete_mirrors_for(db, cid)
            db.execute(delete(PipelineRun).where(PipelineRun.capture_id == cid))
            db.execute(delete(FieldNote).where(FieldNote.id == cid))
            continue
        if not any(r.is_primary for r in remaining):
            remaining[0].is_primary = True
        mirror = _mirror_note(db, cid)
        if mirror is not None:
            _mirror_targets_sync(db, mirror.id, cid)
    db.flush()