use crate::prelude::*;

#[derive(Clone, Debug)]
pub(crate) struct GroupProfileOptions {
    pub(crate) query: Option<String>,
    pub(crate) group_ids: Vec<GroupAuditSelector>,
    pub(crate) all: bool,
    pub(crate) one: bool,
    pub(crate) member_limit: Option<usize>,
    pub(crate) include_fields: Vec<String>,
    pub(crate) page_size: usize,
    pub(crate) concurrency: usize,
}

impl GroupProfileOptions {
    pub(crate) fn from_matches(matches: &ArgMatches) -> Result<Self> {
        let query = matches
            .get_one::<String>("query")
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty());
        let group_ids = group_audit_selectors_from_matches(matches)?;
        let all = matches.get_flag("all");
        if !all && query.is_none() && group_ids.is_empty() {
            return err("groups:profile requires --query, --group-ids, or --all");
        }
        let all_members = matches.get_flag("all-members");
        let member_limit = optional_nonnegative_usize_from_matches(matches, "member-limit")?;
        if all_members && member_limit.is_some() {
            return err("groups:profile accepts --all-members or --member-limit, not both");
        }
        Ok(Self {
            query,
            group_ids,
            all,
            one: matches.get_flag("one"),
            member_limit: if all_members {
                None
            } else {
                Some(member_limit.unwrap_or(GROUP_PROFILE_MEMBER_LIMIT_DEFAULT))
            },
            include_fields: include_fields_from_matches(matches, "include-fields")?,
            page_size: optional_usize_from_matches(matches, "page-size")?
                .unwrap_or(SEARCH_LIMIT_MAX),
            concurrency: contact_fetch_concurrency(matches, "concurrency")?,
        })
    }
}

pub(crate) fn group_audit_selectors_from_matches(
    matches: &ArgMatches,
) -> Result<Vec<GroupAuditSelector>> {
    group_audit_selectors_from_matches_flag(matches, "group-ids")
}

pub(crate) fn group_audit_selectors_from_matches_flag(
    matches: &ArgMatches,
    flag: &str,
) -> Result<Vec<GroupAuditSelector>> {
    let raw = split_list_values(&collect_values(matches, flag));
    let mut seen = BTreeSet::new();
    let mut selectors = Vec::new();
    for value in raw {
        let selector = GroupAuditSelector::parse_with_flag(&value, &format!("--{flag}"))?;
        let key = match selector {
            GroupAuditSelector::Id(id) => format!("id:{id}"),
            GroupAuditSelector::Starred => "starred".to_string(),
        };
        if seen.insert(key) {
            selectors.push(selector);
        }
    }
    Ok(selectors)
}

pub(crate) fn group_resolve_dry_run_plan(options: &GroupResolveOptions) -> Value {
    let mut plan = vec![
        json!({"route": "/tools/v2/get-groups", "payload": {}, "purpose": "read live group catalog without writes"}),
        json!({"local": "groups:resolve", "purpose": "select group candidates by --query, --group-ids, or explicit --all"}),
    ];
    if options.member_counts {
        plan.push(json!({
            "route": "/tools/v2/search",
            "payload": {"group_ids": "one returned candidate group per request", "limit": 0},
            "concurrency": options.concurrency,
            "purpose": "count live members for returned candidates without writes",
        }));
    }
    json!({
        "source": "live",
        "filters": {
            "query": options.query,
            "group_ids": options.group_ids.iter().map(GroupAuditSelector::as_value).collect::<Vec<_>>(),
            "all": options.all,
        },
        "candidate_limit": options.candidate_limit,
        "one": options.one,
        "member_counts": options.member_counts,
        "plan": plan,
    })
}

pub(crate) async fn groups_resolve(
    runtime: &Runtime,
    options: &GroupResolveOptions,
) -> Result<Value> {
    let (source, groups, _) = groups_for_audit_live(runtime).await?;
    let discovered_count = groups.len();
    let mut selected = groups
        .into_iter()
        .filter(|group| {
            group_selection_matches(
                group,
                options.query.as_deref(),
                &options.group_ids,
                options.all,
            )
        })
        .collect::<Vec<_>>();
    selected.sort_by(compare_groups_by_name_then_id);

    let total = selected.len();
    if options.one && total != 1 {
        return err(format!("groups:resolve --one found {total} matches"));
    }

    let truncated = total > options.candidate_limit;
    selected.truncate(options.candidate_limit);
    let (member_counts, member_count_errors) = if options.member_counts {
        group_audit_live_member_counts(runtime, &selected, options.concurrency).await?
    } else {
        (BTreeMap::new(), Vec::new())
    };
    let candidates = selected
        .into_iter()
        .map(|group| group_resolve_candidate(group, &member_counts))
        .collect::<Vec<_>>();
    let ids = candidates
        .iter()
        .filter_map(|candidate| candidate.get("id").cloned())
        .filter(|value| !value.is_null())
        .collect::<Vec<_>>();
    let resolved_id = if total == 1 {
        ids.first().cloned().unwrap_or(Value::Null)
    } else {
        Value::Null
    };
    let status = if total == 0 {
        "none"
    } else if total == 1 {
        "resolved"
    } else {
        "ambiguous"
    };
    Ok(json!({
        "source": source,
        "filters": {
            "query": options.query,
            "group_ids": options.group_ids.iter().map(GroupAuditSelector::as_value).collect::<Vec<_>>(),
            "all": options.all,
        },
        "summary": {
            "status": status,
            "discovered_count": discovered_count,
            "total_matches": total,
            "candidate_count": candidates.len(),
            "candidate_limit": options.candidate_limit,
            "truncated": truncated,
            "resolved_id": resolved_id,
            "ids": ids,
            "member_counts": {
                "enabled": options.member_counts,
                "counted": member_counts.len(),
                "error_count": member_count_errors.len(),
                "errors": member_count_errors,
            },
        },
        "candidates": candidates,
    }))
}

pub(crate) fn group_resolve_candidate(
    group: Value,
    member_counts: &BTreeMap<String, u64>,
) -> Value {
    let id_text = record_id(&group).unwrap_or_default();
    let id = parse_contact_id(&id_text)
        .map(|id| Value::Number(Number::from(id)))
        .unwrap_or_else(|_| {
            if id_text.is_empty() {
                Value::Null
            } else {
                Value::String(id_text.clone())
            }
        });
    let name = group_name(&group).unwrap_or_default();
    let selector = group_audit_member_count_request(&group)
        .map(|(_, _, selector)| selector)
        .unwrap_or(Value::Null);
    json!({
        "id": id,
        "name": name,
        "normalized_name": normalize_group_name_key(&name).unwrap_or_default(),
        "search_selector": selector,
        "member_count": member_counts.get(&id_text).copied().map(Value::from).unwrap_or(Value::Null),
        "raw": group,
    })
}

pub(crate) fn group_resolve_rows(report: &Value) -> Value {
    let summary = report.get("summary").unwrap_or(&Value::Null);
    let candidates = report
        .get("candidates")
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default();
    if candidates.is_empty() {
        return Value::Array(vec![group_resolve_row(summary, None)]);
    }
    Value::Array(
        candidates
            .iter()
            .map(|candidate| group_resolve_row(summary, Some(candidate)))
            .collect(),
    )
}

pub(crate) fn group_resolve_row(summary: &Value, candidate: Option<&Value>) -> Value {
    json!({
        "status": summary.get("status").cloned().unwrap_or(Value::Null),
        "total_matches": summary.get("total_matches").cloned().unwrap_or(Value::Null),
        "candidate_count": summary.get("candidate_count").cloned().unwrap_or(Value::Null),
        "candidate_limit": summary.get("candidate_limit").cloned().unwrap_or(Value::Null),
        "truncated": summary.get("truncated").cloned().unwrap_or(Value::Null),
        "resolved_id": summary.get("resolved_id").cloned().unwrap_or(Value::Null),
        "id": candidate.and_then(|value| value.get("id")).cloned().unwrap_or(Value::Null),
        "name": candidate.and_then(|value| value.get("name")).cloned().unwrap_or(Value::Null),
        "normalized_name": candidate.and_then(|value| value.get("normalized_name")).cloned().unwrap_or(Value::Null),
        "search_selector": candidate.and_then(|value| value.get("search_selector")).cloned().unwrap_or(Value::Null),
        "member_count": candidate.and_then(|value| value.get("member_count")).cloned().unwrap_or(Value::Null),
    })
}

pub(crate) fn group_profile_dry_run_plan(options: &GroupProfileOptions) -> Value {
    let member_page_limit = options
        .member_limit
        .map(|limit| limit.min(options.page_size))
        .unwrap_or(options.page_size);
    json!({
        "source": "live",
        "filters": group_profile_filters(options),
        "page_size": options.page_size,
        "concurrency": options.concurrency,
        "plan": [
            {"route": "/tools/v2/get-groups", "payload": {}, "purpose": "read live group catalog without writes"},
            {"local": "groups:profile", "purpose": "select groups by --query, --group-ids, or explicit --all and compute local audit signals"},
            {"route": "/tools/v2/search", "payload": {"group_ids": "one selected group per request", "include_fields": options.include_fields, "limit": 0}, "concurrency": options.concurrency, "purpose": "count members for each selected group without writes"},
            {"route": "/tools/v2/search", "enabled": options.member_limit != Some(0), "payload": {"group_ids": "same selected group", "include_fields": options.include_fields, "limit": member_page_limit, "exclude_contact_ids": "accumulated from prior pages"}, "page_size": options.page_size, "member_limit": options.member_limit.map(Value::from).unwrap_or(Value::String("all".to_string())), "purpose": "fetch bounded member samples, or all members with --all-members"}
        ],
    })
}

pub(crate) async fn groups_profile(
    runtime: &Runtime,
    options: &GroupProfileOptions,
) -> Result<Value> {
    let (source, groups, _) = groups_for_audit_live(runtime).await?;
    let discovered_count = groups.len();
    let duplicate_name_counts = group_duplicate_name_counts(&groups);
    let mut selected = groups
        .into_iter()
        .filter(|group| {
            group_selection_matches(
                group,
                options.query.as_deref(),
                &options.group_ids,
                options.all,
            )
        })
        .collect::<Vec<_>>();
    selected.sort_by(compare_groups_by_name_then_id);
    let selected_count = selected.len();
    if options.one && selected_count != 1 {
        return err(format!(
            "groups:profile --one found {selected_count} matches"
        ));
    }

    let member_options = GroupMembersOptions {
        query: None,
        group_ids: Vec::new(),
        all_groups: true,
        include_fields: options.include_fields.clone(),
        limit_per_group: options.member_limit,
        page_size: options.page_size,
        concurrency: options.concurrency,
        flat: false,
    };
    let (member_groups, errors) =
        group_members_fetch_selected(runtime, &selected, &member_options).await?;
    let profiles = member_groups
        .into_iter()
        .map(|group| group_profile_record(group, &duplicate_name_counts))
        .collect::<Vec<_>>();
    let total_members = profiles
        .iter()
        .filter_map(|group| group.get("total_count").and_then(Value::as_u64))
        .sum::<u64>();
    let returned_members = profiles
        .iter()
        .filter_map(|group| group.get("returned_count").and_then(Value::as_u64))
        .sum::<u64>();
    let truncated_group_count = profiles
        .iter()
        .filter(|group| {
            group
                .get("truncated")
                .and_then(Value::as_bool)
                .unwrap_or(false)
        })
        .count();
    let zero_member_count = profiles
        .iter()
        .filter(|group| {
            group
                .get("total_count")
                .and_then(Value::as_u64)
                .is_some_and(|count| count == 0)
        })
        .count();
    let issue_group_count = profiles
        .iter()
        .filter(|group| {
            group
                .get("issue_count")
                .and_then(Value::as_u64)
                .unwrap_or_default()
                > 0
        })
        .count();

    Ok(json!({
        "source": source,
        "filters": group_profile_filters(options),
        "pagination": "exclude_contact_ids",
        "page_size": options.page_size,
        "discovered_group_count": discovered_count,
        "selected_group_count": selected_count,
        "returned_group_count": profiles.len(),
        "summary": {
            "total_members": total_members,
            "returned_members": returned_members,
            "truncated_group_count": truncated_group_count,
            "zero_member_count": zero_member_count,
            "issue_group_count": issue_group_count,
            "error_count": errors.len(),
        },
        "groups": profiles,
        "errors": errors,
    }))
}

pub(crate) fn group_profile_filters(options: &GroupProfileOptions) -> Value {
    json!({
        "query": options.query,
        "group_ids": options.group_ids.iter().map(GroupAuditSelector::as_value).collect::<Vec<_>>(),
        "all": options.all,
        "one": options.one,
        "member_limit": options.member_limit.map(Value::from).unwrap_or(Value::String("all".to_string())),
        "include_fields": options.include_fields,
    })
}

pub(crate) fn group_profile_record(
    member_group: Value,
    duplicate_name_counts: &BTreeMap<String, u64>,
) -> Value {
    let group_data = member_group.get("group").unwrap_or(&Value::Null);
    let id = group_data
        .get("id")
        .cloned()
        .or_else(|| group_data.get("raw").and_then(record_id).map(Value::String))
        .unwrap_or(Value::Null);
    let name = group_data
        .get("name")
        .and_then(value_string)
        .unwrap_or_default();
    let normalized_name = normalize_group_name_key(&name);
    let total_count = member_group
        .get("total_count")
        .cloned()
        .unwrap_or(Value::Null);
    let mut issues = Vec::new();
    if id.as_str().unwrap_or_default().trim().is_empty() && !id.is_number() {
        issues.push("missing_id".to_string());
    }
    if normalized_name.is_none() {
        issues.push("empty_name".to_string());
    }
    if normalized_name
        .as_ref()
        .and_then(|value| duplicate_name_counts.get(value))
        .is_some_and(|count| *count > 1)
    {
        issues.push("duplicate_name".to_string());
    }
    if total_count.as_u64().is_some_and(|count| count == 0) {
        issues.push("zero_members".to_string());
    }
    json!({
        "id": id,
        "name": name,
        "normalized_name": normalized_name.unwrap_or_default(),
        "selector": member_group.get("selector").cloned().unwrap_or(Value::Null),
        "total_count": total_count,
        "returned_count": member_group.get("returned_count").cloned().unwrap_or(Value::Null),
        "truncated": member_group.get("truncated").cloned().unwrap_or(Value::Null),
        "issue_count": issues.len(),
        "issues": issues,
        "raw": group_data.get("raw").cloned().unwrap_or(Value::Null),
        "members": member_group.get("members").cloned().unwrap_or(Value::Array(Vec::new())),
    })
}

pub(crate) fn group_profile_rows(report: &Value) -> Value {
    let groups = report
        .get("groups")
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default();
    Value::Array(
        groups
            .iter()
            .map(|group| {
                let members = group
                    .get("members")
                    .and_then(Value::as_array)
                    .cloned()
                    .unwrap_or_default();
                let member_ids = members
                    .iter()
                    .filter_map(record_id)
                    .collect::<Vec<_>>()
                    .join(",");
                let member_names = members
                    .iter()
                    .filter_map(contact_name)
                    .collect::<Vec<_>>()
                    .join(", ");
                json!({
                    "id": group.get("id").cloned().unwrap_or(Value::Null),
                    "name": group.get("name").cloned().unwrap_or(Value::Null),
                    "selector": group.get("selector").cloned().unwrap_or(Value::Null),
                    "total_count": group.get("total_count").cloned().unwrap_or(Value::Null),
                    "returned_count": group.get("returned_count").cloned().unwrap_or(Value::Null),
                    "truncated": group.get("truncated").cloned().unwrap_or(Value::Null),
                    "issue_count": group.get("issue_count").cloned().unwrap_or(Value::Null),
                    "issues": group.get("issues").map(cell_string).unwrap_or_default(),
                    "sample_member_ids": member_ids,
                    "sample_member_names": member_names,
                })
            })
            .collect(),
    )
}

pub(crate) fn group_overlap_dry_run_plan(options: &GroupOverlapOptions) -> Value {
    json!({
        "source": "live",
        "filters": group_overlap_filters(options),
        "page_size": options.page_size,
        "concurrency": options.concurrency,
        "plan": [
            {"route": "/tools/v2/get-groups", "payload": {}, "purpose": "read live group catalog without writes"},
            {"local": "groups:overlap", "purpose": "select groups by --query, --group-ids, or explicit --all"},
            {"route": "/tools/v2/search", "payload": {"group_ids": "one selected group per request", "limit": 0}, "purpose": "count selected group members without writes"},
            {"route": "/tools/v2/search", "payload": {"group_ids": "one selected group per request", "limit": "page_size", "exclude_contact_ids": "accumulated from prior pages"}, "page_size": options.page_size, "concurrency": options.concurrency, "purpose": "page every selected group member for local pairwise overlap math"}
        ],
    })
}

pub(crate) async fn groups_overlap(
    runtime: &Runtime,
    options: &GroupOverlapOptions,
) -> Result<Value> {
    let (source, groups, _) = groups_for_audit_live(runtime).await?;
    let discovered_count = groups.len();
    let mut selected = groups
        .into_iter()
        .filter(|group| {
            group_selection_matches(
                group,
                options.query.as_deref(),
                &options.group_ids,
                options.all,
            )
        })
        .collect::<Vec<_>>();
    selected.sort_by(compare_groups_by_name_then_id);
    let selected_count = selected.len();
    let member_options = GroupMembersOptions {
        query: None,
        group_ids: Vec::new(),
        all_groups: true,
        include_fields: Vec::new(),
        limit_per_group: None,
        page_size: options.page_size,
        concurrency: options.concurrency,
        flat: false,
    };
    let (member_groups, errors) =
        group_members_fetch_selected(runtime, &selected, &member_options).await?;
    let profiles = member_groups
        .iter()
        .map(group_overlap_profile)
        .collect::<Result<Vec<_>>>()?;
    let pair_count = profiles
        .len()
        .saturating_mul(profiles.len().saturating_sub(1))
        / 2;
    let mut pairs = Vec::new();
    let mut identical_pair_count = 0_usize;
    let mut subset_pair_count = 0_usize;
    let mut disjoint_pair_count = 0_usize;

    for left_index in 0..profiles.len() {
        for right_index in (left_index + 1)..profiles.len() {
            let pair = group_overlap_pair(&profiles[left_index], &profiles[right_index]);
            let overlap = pair
                .get("overlap_count")
                .and_then(Value::as_u64)
                .unwrap_or_default() as usize;
            let jaccard = pair
                .get("jaccard")
                .and_then(Value::as_f64)
                .unwrap_or_default();
            let relationship = pair
                .get("relationship")
                .and_then(Value::as_str)
                .unwrap_or_default();
            if relationship == "same_members" {
                identical_pair_count += 1;
            }
            if relationship.contains("subset") || relationship == "same_members" {
                subset_pair_count += 1;
            }
            if relationship == "disjoint" {
                disjoint_pair_count += 1;
            }
            if overlap >= options.min_overlap && jaccard >= options.min_jaccard {
                pairs.push(pair);
            }
        }
    }

    pairs.sort_by(|left, right| {
        let left_overlap = left
            .get("overlap_count")
            .and_then(Value::as_u64)
            .unwrap_or_default();
        let right_overlap = right
            .get("overlap_count")
            .and_then(Value::as_u64)
            .unwrap_or_default();
        let left_jaccard = left
            .get("jaccard")
            .and_then(Value::as_f64)
            .unwrap_or_default();
        let right_jaccard = right
            .get("jaccard")
            .and_then(Value::as_f64)
            .unwrap_or_default();
        right_overlap
            .cmp(&left_overlap)
            .then_with(|| right_jaccard.total_cmp(&left_jaccard))
            .then_with(|| {
                left.get("left_name")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .cmp(
                        right
                            .get("left_name")
                            .and_then(Value::as_str)
                            .unwrap_or_default(),
                    )
            })
            .then_with(|| {
                left.get("right_name")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .cmp(
                        right
                            .get("right_name")
                            .and_then(Value::as_str)
                            .unwrap_or_default(),
                    )
            })
    });
    let matching_pair_count = pairs.len();
    if let Some(top) = options.top {
        pairs.truncate(top);
    }
    let zero_member_group_count = profiles
        .iter()
        .filter(|profile| profile.member_ids.is_empty())
        .count();

    Ok(json!({
        "source": source,
        "filters": group_overlap_filters(options),
        "pagination": "exclude_contact_ids",
        "page_size": options.page_size,
        "discovered_group_count": discovered_count,
        "selected_group_count": selected_count,
        "scanned_group_count": profiles.len(),
        "summary": {
            "group_count": profiles.len(),
            "pair_count": pair_count,
            "matching_pair_count": matching_pair_count,
            "returned_pair_count": pairs.len(),
            "zero_member_group_count": zero_member_group_count,
            "identical_pair_count": identical_pair_count,
            "subset_pair_count": subset_pair_count,
            "disjoint_pair_count": disjoint_pair_count,
            "error_count": errors.len(),
        },
        "groups": profiles.iter().map(GroupOverlapProfile::summary_value).collect::<Vec<_>>(),
        "pairs": pairs,
        "errors": errors,
    }))
}

#[derive(Clone, Debug)]
pub(crate) struct GroupOverlapProfile {
    pub(crate) id: Value,
    pub(crate) id_key: String,
    pub(crate) name: String,
    pub(crate) selector: Value,
    pub(crate) total_count: Value,
    pub(crate) member_ids: BTreeSet<u64>,
}

impl GroupOverlapProfile {
    pub(crate) fn summary_value(&self) -> Value {
        json!({
            "id": self.id,
            "name": self.name,
            "selector": self.selector,
            "total_count": self.total_count,
            "member_count": self.member_ids.len(),
        })
    }
}

pub(crate) fn group_overlap_profile(group: &Value) -> Result<GroupOverlapProfile> {
    let group_data = group.get("group").unwrap_or(&Value::Null);
    let id = group_data.get("id").cloned().unwrap_or(Value::Null);
    let id_key = id
        .as_str()
        .map(ToOwned::to_owned)
        .or_else(|| id.as_u64().map(|id| id.to_string()))
        .unwrap_or_default();
    let name = group_data
        .get("name")
        .and_then(value_string)
        .unwrap_or_default();
    let mut member_ids = BTreeSet::new();
    for member in group
        .get("members")
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
    {
        let id = contact_id_from_value(member)
            .ok_or_else(|| miette!("groups:overlap member row did not include numeric id"))?;
        member_ids.insert(id);
    }
    Ok(GroupOverlapProfile {
        id,
        id_key,
        name,
        selector: group.get("selector").cloned().unwrap_or(Value::Null),
        total_count: group.get("total_count").cloned().unwrap_or(Value::Null),
        member_ids,
    })
}

pub(crate) fn group_overlap_pair(left: &GroupOverlapProfile, right: &GroupOverlapProfile) -> Value {
    let left_count = left.member_ids.len();
    let right_count = right.member_ids.len();
    let overlap_count = left.member_ids.intersection(&right.member_ids).count();
    let union_count = left.member_ids.union(&right.member_ids).count();
    let left_only_count = left_count.saturating_sub(overlap_count);
    let right_only_count = right_count.saturating_sub(overlap_count);
    let jaccard = if union_count == 0 {
        1.0
    } else {
        overlap_count as f64 / union_count as f64
    };
    let left_subset = overlap_count == left_count;
    let right_subset = overlap_count == right_count;
    let relationship = if left_subset && right_subset {
        "same_members"
    } else if overlap_count == 0 {
        "disjoint"
    } else if left_subset {
        "left_subset_of_right"
    } else if right_subset {
        "right_subset_of_left"
    } else {
        "overlap"
    };
    json!({
        "left_id": left.id,
        "left_id_key": left.id_key,
        "left_name": left.name,
        "left_count": left_count,
        "right_id": right.id,
        "right_id_key": right.id_key,
        "right_name": right.name,
        "right_count": right_count,
        "overlap_count": overlap_count,
        "left_only_count": left_only_count,
        "right_only_count": right_only_count,
        "union_count": union_count,
        "jaccard": number_value(jaccard).unwrap_or(Value::Null),
        "relationship": relationship,
    })
}

pub(crate) fn group_overlap_filters(options: &GroupOverlapOptions) -> Value {
    json!({
        "query": options.query,
        "group_ids": options.group_ids.iter().map(GroupAuditSelector::as_value).collect::<Vec<_>>(),
        "all": options.all,
        "min_overlap": options.min_overlap,
        "min_jaccard": number_value(options.min_jaccard).unwrap_or(Value::Null),
        "top": options.top,
    })
}

pub(crate) fn group_compare_dry_run_plan(options: &GroupCompareOptions) -> Value {
    json!({
        "source": "live",
        "filters": group_compare_filters(options),
        "page_size": options.page_size,
        "concurrency": options.concurrency,
        "plan": [
            {"route": "/tools/v2/get-groups", "payload": {}, "purpose": "read live group catalog without writes"},
            {"local": "groups:compare", "purpose": "resolve left and right targets to exactly one group each"},
            {"route": "/tools/v2/search", "payload": {"group_ids": "left and right group selectors", "include_fields": options.include_fields, "limit": 0}, "purpose": "count both group member sets without writes"},
            {"route": "/tools/v2/search", "payload": {"group_ids": "left and right group selectors", "include_fields": options.include_fields, "limit": "page_size", "exclude_contact_ids": "accumulated from prior pages"}, "page_size": options.page_size, "concurrency": options.concurrency, "purpose": "page every member in both groups for local overlap and difference math"}
        ],
    })
}

pub(crate) async fn groups_compare(
    runtime: &Runtime,
    options: &GroupCompareOptions,
) -> Result<Value> {
    let (source, groups, _) = groups_for_audit_live(runtime).await?;
    let discovered_count = groups.len();
    let left_group = group_compare_resolve_target(&groups, &options.left, "left")?;
    let right_group = group_compare_resolve_target(&groups, &options.right, "right")?;

    let (left_report, right_report) = if group_compare_same_group(&left_group, &right_group) {
        let report = group_members_fetch_one(
            runtime.clone(),
            left_group.clone(),
            options.include_fields.clone(),
            None,
            options.page_size,
        )
        .await?;
        (report.clone(), report)
    } else if options.concurrency > 1 {
        let left_group = left_group.clone();
        let right_group = right_group.clone();
        let include_fields = options.include_fields.clone();
        let page_size = options.page_size;
        let left_runtime = runtime.clone();
        let right_runtime = runtime.clone();
        tokio::try_join!(
            group_members_fetch_one(
                left_runtime,
                left_group,
                include_fields.clone(),
                None,
                page_size
            ),
            group_members_fetch_one(right_runtime, right_group, include_fields, None, page_size),
        )?
    } else {
        let left = group_members_fetch_one(
            runtime.clone(),
            left_group.clone(),
            options.include_fields.clone(),
            None,
            options.page_size,
        )
        .await?;
        let right = group_members_fetch_one(
            runtime.clone(),
            right_group.clone(),
            options.include_fields.clone(),
            None,
            options.page_size,
        )
        .await?;
        (left, right)
    };

    let left_index = group_compare_member_index(&left_report)?;
    let right_index = group_compare_member_index(&right_report)?;
    let left_ids = left_index.keys().copied().collect::<BTreeSet<_>>();
    let right_ids = right_index.keys().copied().collect::<BTreeSet<_>>();
    let overlap = left_ids
        .intersection(&right_ids)
        .copied()
        .collect::<Vec<_>>();
    let left_only = left_ids.difference(&right_ids).copied().collect::<Vec<_>>();
    let right_only = right_ids.difference(&left_ids).copied().collect::<Vec<_>>();
    let union_count = left_ids.union(&right_ids).count();
    let overlap_count = overlap.len();
    let jaccard = if union_count == 0 {
        1.0
    } else {
        overlap_count as f64 / union_count as f64
    };
    let overlap_refs =
        group_compare_contact_refs(&overlap, &left_index, &right_index, options.id_limit);
    let left_only_refs =
        group_compare_contact_refs(&left_only, &left_index, &right_index, options.id_limit);
    let right_only_refs =
        group_compare_contact_refs(&right_only, &left_index, &right_index, options.id_limit);

    Ok(json!({
        "source": source,
        "filters": group_compare_filters(options),
        "pagination": "exclude_contact_ids",
        "page_size": options.page_size,
        "discovered_group_count": discovered_count,
        "groups": {
            "left": group_compare_group_meta(&left_report, &options.left),
            "right": group_compare_group_meta(&right_report, &options.right),
        },
        "summary": {
            "left_count": left_ids.len(),
            "right_count": right_ids.len(),
            "overlap_count": overlap_count,
            "left_only_count": left_only.len(),
            "right_only_count": right_only.len(),
            "union_count": union_count,
            "same_members": left_ids == right_ids,
            "jaccard": number_value(jaccard).unwrap_or(Value::Null),
            "id_limit": options.id_limit.map(Value::from).unwrap_or(Value::String("all".to_string())),
            "returned_overlap_count": overlap_refs.len(),
            "returned_left_only_count": left_only_refs.len(),
            "returned_right_only_count": right_only_refs.len(),
            "error_count": 0,
        },
        "sets": {
            "overlap": overlap_refs,
            "left_only": left_only_refs,
            "right_only": right_only_refs,
        },
        "errors": [],
    }))
}

pub(crate) fn group_compare_resolve_target(
    groups: &[Value],
    target: &GroupCompareTarget,
    side: &str,
) -> Result<Value> {
    let mut selected = groups
        .iter()
        .filter(|group| target.matches_group(group))
        .cloned()
        .collect::<Vec<_>>();
    selected.sort_by(compare_groups_by_name_then_id);
    if selected.len() != 1 {
        return err(format!(
            "groups:compare {side} target found {} matches",
            selected.len()
        ));
    }
    Ok(selected.remove(0))
}

pub(crate) fn group_compare_same_group(left: &Value, right: &Value) -> bool {
    record_id(left)
        .zip(record_id(right))
        .is_some_and(|(left, right)| left == right)
}

pub(crate) fn group_compare_filters(options: &GroupCompareOptions) -> Value {
    json!({
        "left": options.left.as_value(),
        "right": options.right.as_value(),
        "include_fields": options.include_fields,
        "id_limit": options.id_limit.map(Value::from).unwrap_or(Value::String("all".to_string())),
    })
}

pub(crate) fn group_compare_group_meta(report: &Value, target: &GroupCompareTarget) -> Value {
    let group = report.get("group").unwrap_or(&Value::Null);
    json!({
        "target": target.as_value(),
        "id": group.get("id").cloned().unwrap_or(Value::Null),
        "name": group.get("name").cloned().unwrap_or(Value::Null),
        "selector": report.get("selector").cloned().unwrap_or(Value::Null),
        "total_count": report.get("total_count").cloned().unwrap_or(Value::Null),
        "returned_count": report.get("returned_count").cloned().unwrap_or(Value::Null),
        "truncated": report.get("truncated").cloned().unwrap_or(Value::Null),
        "raw": group.get("raw").cloned().unwrap_or(Value::Null),
    })
}

pub(crate) fn group_compare_member_index(report: &Value) -> Result<BTreeMap<u64, Value>> {
    let mut index = BTreeMap::new();
    for member in report
        .get("members")
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
    {
        let id = contact_id_from_value(member)
            .ok_or_else(|| miette!("groups:compare member row did not include numeric id"))?;
        index.entry(id).or_insert_with(|| member.clone());
    }
    Ok(index)
}

pub(crate) fn group_compare_contact_refs(
    ids: &[u64],
    left_index: &BTreeMap<u64, Value>,
    right_index: &BTreeMap<u64, Value>,
    id_limit: Option<usize>,
) -> Vec<Value> {
    let limit = id_limit.unwrap_or(usize::MAX);
    ids.iter()
        .take(limit)
        .map(|id| {
            let contact = left_index
                .get(id)
                .or_else(|| right_index.get(id))
                .unwrap_or(&Value::Null);
            json!({
                "id": id,
                "name": contact_name(contact).unwrap_or_default(),
                "display_name": contact.get("displayName").cloned().unwrap_or(Value::Null),
                "url": contact.get("url").cloned().unwrap_or(Value::Null),
                "score": contact.get("score").cloned().unwrap_or(Value::Null),
                "raw": contact,
            })
        })
        .collect()
}

pub(crate) fn group_compare_rows(report: &Value) -> Value {
    let groups = report.get("groups").unwrap_or(&Value::Null);
    let left = groups.get("left").unwrap_or(&Value::Null);
    let right = groups.get("right").unwrap_or(&Value::Null);
    let summary = report.get("summary").unwrap_or(&Value::Null);
    let sets = report.get("sets").unwrap_or(&Value::Null);
    let mut rows = Vec::new();
    for (set_name, count_key, returned_key) in [
        ("overlap", "overlap_count", "returned_overlap_count"),
        ("left_only", "left_only_count", "returned_left_only_count"),
        (
            "right_only",
            "right_only_count",
            "returned_right_only_count",
        ),
    ] {
        let refs = sets
            .get(set_name)
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default();
        if refs.is_empty() {
            rows.push(group_compare_row(
                set_name,
                summary,
                left,
                right,
                count_key,
                returned_key,
                None,
            ));
        } else {
            for contact in refs {
                rows.push(group_compare_row(
                    set_name,
                    summary,
                    left,
                    right,
                    count_key,
                    returned_key,
                    Some(&contact),
                ));
            }
        }
    }
    Value::Array(rows)
}

pub(crate) fn group_compare_row(
    set_name: &str,
    summary: &Value,
    left: &Value,
    right: &Value,
    count_key: &str,
    returned_key: &str,
    contact: Option<&Value>,
) -> Value {
    json!({
        "set": set_name,
        "set_count": summary.get(count_key).cloned().unwrap_or(Value::Null),
        "set_returned": summary.get(returned_key).cloned().unwrap_or(Value::Null),
        "contact_id": contact.and_then(|value| value.get("id")).cloned().unwrap_or(Value::Null),
        "contact_name": contact.and_then(|value| value.get("name")).cloned().unwrap_or(Value::Null),
        "contact_display_name": contact.and_then(|value| value.get("display_name")).cloned().unwrap_or(Value::Null),
        "contact_url": contact.and_then(|value| value.get("url")).cloned().unwrap_or(Value::Null),
        "left_group_id": left.get("id").cloned().unwrap_or(Value::Null),
        "left_group_name": left.get("name").cloned().unwrap_or(Value::Null),
        "left_count": summary.get("left_count").cloned().unwrap_or(Value::Null),
        "right_group_id": right.get("id").cloned().unwrap_or(Value::Null),
        "right_group_name": right.get("name").cloned().unwrap_or(Value::Null),
        "right_count": summary.get("right_count").cloned().unwrap_or(Value::Null),
        "overlap_count": summary.get("overlap_count").cloned().unwrap_or(Value::Null),
        "left_only_count": summary.get("left_only_count").cloned().unwrap_or(Value::Null),
        "right_only_count": summary.get("right_only_count").cloned().unwrap_or(Value::Null),
        "union_count": summary.get("union_count").cloned().unwrap_or(Value::Null),
        "same_members": summary.get("same_members").cloned().unwrap_or(Value::Null),
        "jaccard": summary.get("jaccard").cloned().unwrap_or(Value::Null),
    })
}

pub(crate) fn group_audit_dry_run_plan(options: &GroupAuditOptions) -> Value {
    let filters = json!({
        "query": options.query,
        "group_ids": options.group_ids.iter().map(GroupAuditSelector::as_value).collect::<Vec<_>>(),
        "issues_only": options.issues_only,
    });
    if let Some(dir) = &options.snapshot_dir {
        json!({
            "source": "snapshot",
            "filters": filters,
            "member_counts": false,
            "plan": [
                {"local_file": dir.join("manifest.json").display().to_string(), "purpose": "verify snapshot hashes"},
                {"local_file": dir.join("groups.json").display().to_string(), "purpose": "read snapshot group catalog"},
                {"local": "groups:audit", "purpose": "audit group names, IDs, duplicate names, and local catalog issues"}
            ],
        })
    } else {
        let mut plan = vec![
            json!({"route": "/tools/v2/get-groups", "payload": {}, "purpose": "read live group catalog without writes"}),
            json!({"local": "groups:audit", "purpose": "audit group names, IDs, duplicate names, and local catalog issues"}),
        ];
        if options.member_counts {
            plan.push(json!({
                "route": "/tools/v2/search",
                "payload": {"group_ids": "one selected group per request", "limit": 0},
                "concurrency": options.concurrency,
                "purpose": "count live members for each selected group without writes",
            }));
        }
        json!({
            "source": "live",
            "filters": filters,
            "member_counts": options.member_counts,
            "top": options.top,
            "plan": plan,
        })
    }
}

pub(crate) async fn groups_audit(runtime: &Runtime, options: GroupAuditOptions) -> Result<Value> {
    let (source, mut groups, source_is_live) = if let Some(dir) = &options.snapshot_dir {
        groups_for_audit_snapshot(dir)?
    } else {
        groups_for_audit_live(runtime).await?
    };
    let discovered_count = groups.len();
    groups.retain(|group| group_audit_matches(group, &options));
    groups.sort_by(compare_groups_by_name_then_id);

    let (member_counts, member_count_errors) = if options.member_counts && source_is_live {
        group_audit_live_member_counts(runtime, &groups, options.concurrency).await?
    } else {
        (BTreeMap::new(), Vec::new())
    };
    Ok(group_audit_report(
        source,
        discovered_count,
        groups,
        member_counts,
        member_count_errors,
        &options,
    ))
}

pub(crate) async fn groups_for_audit_live(runtime: &Runtime) -> Result<(Value, Vec<Value>, bool)> {
    let data = runtime.call_tool(route::GET_GROUPS, json!({})).await?;
    let groups = snapshot_group_rows_from_response(&data)?;
    Ok((
        json!({
            "type": "live",
            "group_count": groups.len(),
        }),
        groups,
        true,
    ))
}

pub(crate) fn groups_for_audit_snapshot(dir: &Path) -> Result<(Value, Vec<Value>, bool)> {
    let verify = verify_snapshot(dir)?;
    if !verify.get("ok").and_then(Value::as_bool).unwrap_or(false) {
        return err("snapshot failed manifest verification");
    }
    let entry = snapshot_manifest_file_entry(dir, "groups")?
        .ok_or_else(|| miette!("snapshot does not contain file groups"))?;
    let path = entry
        .get("path")
        .and_then(Value::as_str)
        .ok_or_else(|| miette!("snapshot file entry groups is missing path"))?;
    let values = read_snapshot_array_record_values_at_path(&safe_snapshot_file_path(dir, path)?)?;
    let groups = values.into_values().collect::<Vec<_>>();
    Ok((
        json!({
            "type": "snapshot",
            "dir": dir.display().to_string(),
            "file": path,
            "group_count": groups.len(),
        }),
        groups,
        false,
    ))
}

pub(crate) fn group_audit_matches(group: &Value, options: &GroupAuditOptions) -> bool {
    if let Some(query) = &options.query
        && !group_name_matches_query(group, query)
    {
        return false;
    }
    options.group_ids.is_empty()
        || options
            .group_ids
            .iter()
            .any(|selector| selector.matches_group(group))
}

pub(crate) async fn group_audit_live_member_counts(
    runtime: &Runtime,
    groups: &[Value],
    concurrency: usize,
) -> Result<(BTreeMap<String, u64>, Vec<Value>)> {
    if concurrency == 0 {
        return err("group audit concurrency must be greater than zero");
    }
    let mut counts = BTreeMap::new();
    let mut errors = Vec::new();
    for chunk in groups.chunks(concurrency) {
        let mut handles = Vec::new();
        for group in chunk {
            let Some((key, name, selector)) = group_audit_member_count_request(group) else {
                errors.push(json!({
                    "id": record_id(group).unwrap_or_default(),
                    "name": group_name(group).unwrap_or_default(),
                    "error": "group has no usable ID for member count",
                }));
                continue;
            };
            let runtime = runtime.clone();
            handles.push(tokio::spawn(async move {
                let mut payload = Map::new();
                payload.set("group_ids", Value::Array(vec![selector]));
                let result = runtime.search_total(payload).await;
                (key, name, result)
            }));
        }
        for handle in handles {
            let (key, name, result) = handle
                .await
                .into_diagnostic()
                .wrap_err("joining groups:audit member-count task")?;
            match result {
                Ok(total) => {
                    counts.insert(key, total as u64);
                }
                Err(error) => {
                    errors.push(json!({
                        "id": key,
                        "name": name,
                        "error": error.to_string(),
                    }));
                }
            }
        }
    }
    Ok((counts, errors))
}

pub(crate) fn group_audit_member_count_request(group: &Value) -> Option<(String, String, Value)> {
    let key = record_id(group)?;
    let name = group_name(group).unwrap_or_default();
    let selector = if normalize_group_name_key(&name).is_some_and(|value| value == "starred") {
        Value::String("starred".to_string())
    } else {
        Value::Number(Number::from(parse_contact_id(&key).ok()?))
    };
    Some((key, name, selector))
}

pub(crate) fn group_audit_report(
    source: Value,
    discovered_count: usize,
    groups: Vec<Value>,
    member_counts: BTreeMap<String, u64>,
    member_count_errors: Vec<Value>,
    options: &GroupAuditOptions,
) -> Value {
    let mut name_counts: BTreeMap<String, u64> = BTreeMap::new();
    for group in &groups {
        if let Some(name) = group_name(group).and_then(|value| normalize_group_name_key(&value)) {
            *name_counts.entry(name).or_default() += 1;
        }
    }
    let duplicate_name_counts = name_counts
        .iter()
        .filter(|(_, count)| **count > 1)
        .map(|(name, count)| (name.clone(), *count))
        .collect::<BTreeMap<_, _>>();

    let mut issue_counts: BTreeMap<String, u64> = BTreeMap::new();
    let mut records = Vec::new();
    let mut zero_member_count = 0_u64;
    let mut total_members_counted = 0_u64;
    let selected_count = groups.len();
    for group in groups {
        let id = record_id(&group).unwrap_or_default();
        let name = group_name(&group).unwrap_or_default();
        let normalized_name = normalize_group_name_key(&name);
        let member_count = member_counts.get(&id).copied();
        let mut issues = Vec::new();
        if id.trim().is_empty() {
            issues.push("missing_id".to_string());
        }
        if normalized_name.is_none() {
            issues.push("empty_name".to_string());
        }
        if normalized_name
            .as_ref()
            .and_then(|value| duplicate_name_counts.get(value))
            .is_some_and(|count| *count > 1)
        {
            issues.push("duplicate_name".to_string());
        }
        if let Some(count) = member_count {
            total_members_counted += count;
            if count == 0 {
                zero_member_count += 1;
                issues.push("zero_members".to_string());
            }
        }
        if options.issues_only && issues.is_empty() {
            continue;
        }
        for issue in &issues {
            *issue_counts.entry(issue.clone()).or_default() += 1;
        }
        records.push(json!({
            "id": id,
            "name": name,
            "normalized_name": normalized_name.unwrap_or_default(),
            "member_count": member_count.map(Value::from).unwrap_or(Value::Null),
            "issue_count": issues.len(),
            "issues": issues,
            "raw": group,
        }));
    }
    records.sort_by(|left, right| {
        let left_issues = left
            .get("issue_count")
            .and_then(Value::as_u64)
            .unwrap_or_default();
        let right_issues = right
            .get("issue_count")
            .and_then(Value::as_u64)
            .unwrap_or_default();
        right_issues
            .cmp(&left_issues)
            .then_with(|| {
                right
                    .get("member_count")
                    .and_then(Value::as_u64)
                    .unwrap_or_default()
                    .cmp(
                        &left
                            .get("member_count")
                            .and_then(Value::as_u64)
                            .unwrap_or_default(),
                    )
            })
            .then_with(|| {
                left.get("name")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .cmp(
                        right
                            .get("name")
                            .and_then(Value::as_str)
                            .unwrap_or_default(),
                    )
            })
    });

    json!({
        "source": source,
        "filters": {
            "query": options.query,
            "group_ids": options.group_ids.iter().map(GroupAuditSelector::as_value).collect::<Vec<_>>(),
            "issues_only": options.issues_only,
        },
        "discovered_count": discovered_count,
        "analyzed_count": selected_count,
        "returned_count": records.len(),
        "summary": {
            "selected_count": selected_count,
            "issue_group_count": records.iter().filter(|record| record.get("issue_count").and_then(Value::as_u64).unwrap_or_default() > 0).count(),
            "total_issues": issue_counts.values().sum::<u64>(),
            "top_issues": top_count_entries(&issue_counts, options.top),
            "duplicate_name_count": duplicate_name_counts.len(),
            "top_duplicate_names": top_count_entries(&duplicate_name_counts, options.top),
            "member_counts": {
                "enabled": options.member_counts,
                "counted": member_counts.len(),
                "error_count": member_count_errors.len(),
                "zero_member_count": zero_member_count,
                "total_members_counted": total_members_counted,
                "top_groups": group_audit_top_member_rows(&records, options.top),
                "errors": member_count_errors,
            },
        },
        "groups": records,
    })
}

pub(crate) fn group_audit_top_member_rows(records: &[Value], top: usize) -> Value {
    let mut rows = records
        .iter()
        .filter(|record| record.get("member_count").and_then(Value::as_u64).is_some())
        .map(|record| {
            json!({
                "id": record.get("id").cloned().unwrap_or(Value::Null),
                "name": record.get("name").cloned().unwrap_or(Value::Null),
                "member_count": record.get("member_count").cloned().unwrap_or(Value::Null),
            })
        })
        .collect::<Vec<_>>();
    rows.sort_by(|left, right| {
        right
            .get("member_count")
            .and_then(Value::as_u64)
            .unwrap_or_default()
            .cmp(
                &left
                    .get("member_count")
                    .and_then(Value::as_u64)
                    .unwrap_or_default(),
            )
            .then_with(|| {
                left.get("name")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .cmp(
                        right
                            .get("name")
                            .and_then(Value::as_str)
                            .unwrap_or_default(),
                    )
            })
    });
    rows.truncate(top);
    Value::Array(rows)
}

pub(crate) fn group_audit_table_rows(report: &Value) -> Value {
    let Some(rows) = report.get("groups").and_then(Value::as_array) else {
        return Value::Array(Vec::new());
    };
    Value::Array(
        rows.iter()
            .map(|row| {
                json!({
                    "id": row.get("id").cloned().unwrap_or(Value::Null),
                    "name": row.get("name").cloned().unwrap_or(Value::Null),
                    "member_count": row.get("member_count").cloned().unwrap_or(Value::Null),
                    "issues": row.get("issues").map(cell_string).unwrap_or_default(),
                })
            })
            .collect(),
    )
}

pub(crate) fn group_members_dry_run_plan(options: &GroupMembersOptions) -> Value {
    json!({
        "source": "live",
        "filters": {
            "query": options.query,
            "group_ids": options.group_ids.iter().map(GroupAuditSelector::as_value).collect::<Vec<_>>(),
            "all_groups": options.all_groups,
            "include_fields": options.include_fields,
            "limit_per_group": options.limit_per_group,
        },
        "page_size": options.page_size,
        "concurrency": options.concurrency,
        "flat": options.flat,
        "plan": [
            {"route": "/tools/v2/get-groups", "payload": {}, "purpose": "read live group catalog without writes"},
            {"local": "group selection", "purpose": "select groups by --group-ids, --query, or --all-groups"},
            {"route": "/tools/v2/search", "payload": {"group_ids": "one selected group per request", "include_fields": options.include_fields, "limit": 0}, "concurrency": options.concurrency, "purpose": "count members for each selected group"},
            {"route": "/tools/v2/search", "payload": {"group_ids": "same selected group", "include_fields": options.include_fields, "limit": options.page_size, "exclude_contact_ids": "accumulated from prior pages"}, "page_size": options.page_size, "purpose": "fetch group members until count or --limit-per-group is reached"}
        ],
    })
}

pub(crate) async fn groups_members(
    runtime: &Runtime,
    options: GroupMembersOptions,
) -> Result<Value> {
    let (source, groups, _) = groups_for_audit_live(runtime).await?;
    let discovered_count = groups.len();
    let selected_groups = groups
        .into_iter()
        .filter(|group| {
            group_selection_matches(
                group,
                options.query.as_deref(),
                &options.group_ids,
                options.all_groups,
            )
        })
        .collect::<Vec<_>>();
    let selected_count = selected_groups.len();
    let (groups, errors) =
        group_members_fetch_selected(runtime, &selected_groups, &options).await?;
    let total_members = groups
        .iter()
        .filter_map(|group| group.get("total_count").and_then(Value::as_u64))
        .sum::<u64>();
    let returned_members = groups
        .iter()
        .filter_map(|group| group.get("returned_count").and_then(Value::as_u64))
        .sum::<u64>();
    let truncated_group_count = groups
        .iter()
        .filter(|group| {
            group
                .get("truncated")
                .and_then(Value::as_bool)
                .unwrap_or(false)
        })
        .count();
    Ok(json!({
        "source": source,
        "filters": {
            "query": options.query,
            "group_ids": options.group_ids.iter().map(GroupAuditSelector::as_value).collect::<Vec<_>>(),
            "all_groups": options.all_groups,
            "include_fields": options.include_fields,
            "limit_per_group": options.limit_per_group,
        },
        "pagination": "exclude_contact_ids",
        "page_size": options.page_size,
        "discovered_group_count": discovered_count,
        "selected_group_count": selected_count,
        "returned_group_count": groups.len(),
        "summary": {
            "total_members": total_members,
            "returned_members": returned_members,
            "truncated_group_count": truncated_group_count,
            "error_count": errors.len(),
        },
        "groups": groups,
        "errors": errors,
    }))
}

pub(crate) async fn group_members_fetch_selected(
    runtime: &Runtime,
    groups: &[Value],
    options: &GroupMembersOptions,
) -> Result<(Vec<Value>, Vec<Value>)> {
    let mut results = Vec::new();
    let mut errors = Vec::new();
    for chunk in groups.chunks(options.concurrency) {
        let mut handles = Vec::new();
        for group in chunk {
            let group = group.clone();
            let id = record_id(&group).unwrap_or_default();
            let name = group_name(&group).unwrap_or_default();
            let runtime = runtime.clone();
            let include_fields = options.include_fields.clone();
            let limit_per_group = options.limit_per_group;
            let page_size = options.page_size;
            handles.push(tokio::spawn(async move {
                let result = group_members_fetch_one(
                    runtime,
                    group,
                    include_fields,
                    limit_per_group,
                    page_size,
                )
                .await;
                (id, name, result)
            }));
        }
        for handle in handles {
            let (id, name, result) = handle
                .await
                .into_diagnostic()
                .wrap_err("joining groups:members fetch task")?;
            match result {
                Ok(value) => results.push(value),
                Err(error) => errors.push(json!({
                    "id": id,
                    "name": name,
                    "error": error.to_string(),
                })),
            }
        }
    }
    results.sort_by(|left, right| {
        left.get("group")
            .and_then(|group| group.get("name"))
            .and_then(Value::as_str)
            .unwrap_or_default()
            .cmp(
                right
                    .get("group")
                    .and_then(|group| group.get("name"))
                    .and_then(Value::as_str)
                    .unwrap_or_default(),
            )
    });
    Ok((results, errors))
}

pub(crate) async fn group_members_fetch_one(
    runtime: Runtime,
    group: Value,
    include_fields: Vec<String>,
    limit_per_group: Option<usize>,
    page_size: usize,
) -> Result<Value> {
    let (group_id, group_name, selector) = group_audit_member_count_request(&group)
        .ok_or_else(|| miette!("group has no usable ID for member search"))?;
    let mut payload = Map::new();
    payload.insert(
        "group_ids".to_string(),
        Value::Array(vec![selector.clone()]),
    );
    if !include_fields.is_empty() {
        payload.insert(
            "include_fields".to_string(),
            Value::Array(include_fields.into_iter().map(Value::String).collect()),
        );
    }
    let mut members = Vec::new();
    let (returned_count, total_count) =
        export_contacts_each_limited(&runtime, payload, page_size, limit_per_group, |row| {
            members.push(row);
            Ok(())
        })
        .await?;
    Ok(json!({
        "group": {
            "id": group_id,
            "name": group_name,
            "raw": group,
        },
        "selector": selector,
        "total_count": total_count,
        "returned_count": returned_count,
        "truncated": returned_count < total_count,
        "members": members,
    }))
}

pub(crate) fn group_members_flat_rows(report: &Value) -> Value {
    let Some(groups) = report.get("groups").and_then(Value::as_array) else {
        return Value::Array(Vec::new());
    };
    let mut rows = Vec::new();
    for group in groups {
        let group_data = group.get("group").unwrap_or(&Value::Null);
        let group_id = group_data.get("id").cloned().unwrap_or(Value::Null);
        let group_name = group_data.get("name").cloned().unwrap_or(Value::Null);
        let selector = group.get("selector").cloned().unwrap_or(Value::Null);
        let total_count = group.get("total_count").cloned().unwrap_or(Value::Null);
        let returned_count = group.get("returned_count").cloned().unwrap_or(Value::Null);
        let truncated = group
            .get("truncated")
            .cloned()
            .unwrap_or(Value::Bool(false));
        let members = group
            .get("members")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default();
        if members.is_empty() {
            rows.push(json!({
                "group_id": group_id,
                "group_name": group_name,
                "group_selector": selector,
                "group_total": total_count,
                "group_returned": returned_count,
                "group_truncated": truncated,
                "contact_id": Value::Null,
                "contact_name": Value::Null,
                "contact_display_name": Value::Null,
                "contact_url": Value::Null,
                "contact_score": Value::Null,
            }));
            continue;
        }
        for member in members {
            rows.push(json!({
                "group_id": group_id,
                "group_name": group_name,
                "group_selector": selector,
                "group_total": total_count,
                "group_returned": returned_count,
                "group_truncated": truncated,
                "contact_id": record_id(&member).map(Value::String).unwrap_or(Value::Null),
                "contact_name": contact_name(&member).map(Value::String).unwrap_or(Value::Null),
                "contact_display_name": member.get("displayName").cloned().unwrap_or(Value::Null),
                "contact_url": member.get("url").cloned().unwrap_or(Value::Null),
                "contact_score": member.get("score").cloned().unwrap_or(Value::Null),
            }));
        }
    }
    Value::Array(rows)
}

pub(crate) async fn group_bulk_membership_plan(
    runtime: &Runtime,
    options: &GroupBulkMembershipOptions,
) -> Result<GroupBulkMembershipPlan> {
    let (_, groups, _) = groups_for_audit_live(runtime).await?;
    let mut selected_groups = groups
        .into_iter()
        .filter(|group| {
            group_selection_matches(
                group,
                options.query.as_deref(),
                &options.target_group_ids,
                options.all_groups,
            )
        })
        .collect::<Vec<_>>();
    selected_groups.sort_by(compare_groups_by_name_then_id);

    let mut writable_groups = Vec::new();
    let mut skipped_special_groups = Vec::new();
    for group in selected_groups {
        if group_is_special_starred(&group) {
            skipped_special_groups.push(group);
        } else {
            writable_groups.push(group);
        }
    }

    if options.one && writable_groups.len() != 1 {
        return err(format!(
            "{} --one found {} writable groups",
            options.command,
            writable_groups.len()
        ));
    }
    if let Some(limit) = options.group_limit {
        writable_groups.truncate(limit);
    }
    if writable_groups.is_empty() {
        return err(format!("{} resolved zero writable groups", options.command));
    }

    let explicit_ids = options.target_ids.clone();
    let mut target_ids = explicit_ids.clone();
    if let Some(limit) = options.target_limit {
        target_ids.truncate(limit);
    }

    let mut search_exported_count = None;
    let mut search_match_count = None;
    if let Some(search_payload) = &options.search_payload {
        let remaining = options
            .target_limit
            .map(|limit| limit.saturating_sub(target_ids.len()));
        if remaining != Some(0) {
            let mut payload = search_payload.clone();
            append_exclude_contact_ids(&mut payload, &target_ids)?;
            let mut search_ids = Vec::new();
            let (exported, total) = export_contacts_each_limited(
                runtime,
                payload,
                options.page_size,
                remaining,
                |row| {
                    let id = contact_id_from_value(&row)
                        .ok_or_else(|| miette!("me.sh search row did not include numeric id"))?;
                    search_ids.push(id);
                    Ok(())
                },
            )
            .await?;
            search_exported_count = Some(exported);
            search_match_count = Some(total);
            target_ids.extend(search_ids);
            target_ids = dedupe_ids(target_ids);
            if let Some(limit) = options.target_limit {
                target_ids.truncate(limit);
            }
        } else {
            search_exported_count = Some(0);
            search_match_count = Some(0);
        }
    }

    if target_ids.is_empty() {
        return err(format!("{} resolved zero target contacts", options.command));
    }

    let mut actions = Vec::new();
    let mut row = 1_usize;
    for group in &writable_groups {
        let (group_id, group_name) = group_bulk_membership_group_ref(group)?;
        for chunk in target_ids.chunks(options.chunk_size) {
            actions.push(group_bulk_membership_action(
                row,
                options.kind,
                group_id,
                group_name.clone(),
                chunk,
            ));
            row = row.saturating_add(1);
        }
    }

    let group_source = match (
        options.target_group_ids.is_empty(),
        options.query.is_some(),
        options.all_groups,
    ) {
        (_, _, true) => "all-groups",
        (false, true, false) => "explicit+query",
        (false, false, false) => "explicit",
        (true, true, false) => "query",
        (true, false, false) => "none",
    }
    .to_string();
    let target_source = match (explicit_ids.is_empty(), options.search_payload.is_some()) {
        (false, true) => "explicit+search",
        (false, false) => "explicit",
        (true, true) => "search",
        (true, false) => "none",
    }
    .to_string();

    Ok(GroupBulkMembershipPlan {
        kind: options.kind,
        command: options.command,
        group_source,
        target_source,
        query: options.query.clone(),
        target_group_ids: options.target_group_ids.clone(),
        selected_groups: writable_groups,
        skipped_special_groups,
        explicit_ids,
        search_payload: options.search_payload.clone(),
        search_exported_count,
        search_match_count,
        target_ids,
        actions,
        page_size: options.page_size,
        target_limit: options.target_limit,
        group_limit: options.group_limit,
        chunk_size: options.chunk_size,
        concurrency: options.concurrency,
    })
}

pub(crate) fn group_bulk_membership_group_ref(group: &Value) -> Result<(u64, String)> {
    let id = record_id(group)
        .ok_or_else(|| miette!("group did not include an ID"))?
        .parse::<u64>()
        .into_diagnostic()
        .wrap_err("group ID was not numeric")?;
    Ok((id, group_name(group).unwrap_or_default()))
}

pub(crate) fn group_bulk_membership_action(
    row: usize,
    kind: GroupApplyKind,
    group_id: u64,
    group_name: String,
    contact_ids: &[u64],
) -> GroupBulkMembershipAction {
    let mut payload = Map::new();
    payload.insert(
        "group_id".to_string(),
        Value::Number(Number::from(group_id)),
    );
    payload.insert(
        if kind == GroupApplyKind::Add {
            "add_contact_ids".to_string()
        } else {
            "remove_contact_ids".to_string()
        },
        json!(contact_ids),
    );
    GroupBulkMembershipAction {
        row,
        kind,
        group_id,
        group_name,
        route: route::UPDATE_GROUP,
        payload,
    }
}

pub(crate) fn group_bulk_membership_plan_value(plan: &GroupBulkMembershipPlan) -> Value {
    json!({
        "source": "live",
        "action": plan.kind.as_str(),
        "command": plan.command,
        "group_source": plan.group_source,
        "target_source": plan.target_source,
        "filters": {
            "query": plan.query,
            "target_group_ids": plan.target_group_ids.iter().map(GroupAuditSelector::as_value).collect::<Vec<_>>(),
            "contact_ids": plan.explicit_ids,
            "from_search": plan.search_payload.is_some(),
            "search_payload": plan.search_payload,
            "group_limit": plan.group_limit,
            "target_limit": plan.target_limit,
        },
        "summary": {
            "group_count": plan.selected_groups.len(),
            "skipped_special_group_count": plan.skipped_special_groups.len(),
            "target_count": plan.target_ids.len(),
            "pair_count": plan.selected_groups.len().saturating_mul(plan.target_ids.len()),
            "explicit_count": plan.explicit_ids.len(),
            "search_exported_count": plan.search_exported_count,
            "search_match_count": plan.search_match_count,
            "write_chunk_count": plan.actions.len(),
            "write_required": !plan.actions.is_empty(),
        },
        "page_size": plan.page_size,
        "chunk_size": plan.chunk_size,
        "concurrency": plan.concurrency,
        "groups": plan.selected_groups.iter().map(group_bulk_membership_group_value).collect::<Vec<_>>(),
        "skipped_special_groups": plan.skipped_special_groups.iter().map(group_bulk_membership_group_value).collect::<Vec<_>>(),
        "target_contact_ids": plan.target_ids,
        "plan": [
            {"route": "/tools/v2/get-groups", "payload": {}, "purpose": "read live group catalog and select writable target groups"},
            {"route": "/tools/v2/search", "enabled": plan.search_payload.is_some(), "payload": "same contact search filters with limit set to page_size and exclude_contact_ids accumulated from explicit/prior IDs", "page_size": plan.page_size, "purpose": "resolve target contact IDs without writes"},
            {"route": "/tools/v2/update-group", "payload": {"group_id": "one selected group", "contact_ids": "up to chunk_size target IDs per request"}, "chunk_size": plan.chunk_size, "concurrency": plan.concurrency, "purpose": format!("{} selected contacts for each selected writable group; requires --yes outside dry-run", plan.kind.as_str())}
        ],
        "actions": plan.actions.iter().map(group_bulk_membership_action_value).collect::<Vec<_>>(),
    })
}

pub(crate) fn group_bulk_membership_group_value(group: &Value) -> Value {
    let (id, name) = group_bulk_membership_group_ref(group).unwrap_or((0, String::new()));
    json!({
        "id": if id == 0 { Value::Null } else { Value::Number(Number::from(id)) },
        "name": name,
        "raw": group,
    })
}

pub(crate) async fn apply_group_bulk_membership(
    runtime: &Runtime,
    plan: &GroupBulkMembershipPlan,
) -> Result<Value> {
    let mut results = Vec::with_capacity(plan.actions.len());
    let mut failures = 0_u64;
    let mut changed_pair_count = 0_usize;
    let mut failed_pair_count = 0_usize;
    let outcomes = run_bulk_tool_calls(
        runtime,
        plan.actions.clone(),
        plan.concurrency,
        &format!("joining {} write task", plan.command),
        |action| (action.route, Value::Object(action.payload.clone())),
    )
    .await?;
    for (action, result) in outcomes {
        let contact_ids = group_bulk_membership_action_contact_ids(&action);
        let target_count = contact_ids.len();
        match result {
            Ok(data) => {
                changed_pair_count = changed_pair_count.saturating_add(target_count);
                results.push(json!({
                    "row": action.row,
                    "action": action.kind.as_str(),
                    "group_id": action.group_id,
                    "group_name": action.group_name,
                    "contact_ids": contact_ids,
                    "target_count": target_count,
                    "route": format!("/tools/v2{}", action.route),
                    "ok": true,
                    "result_id": record_id(&data),
                    "result": data,
                }));
            }
            Err(error) => {
                failures = failures.saturating_add(1);
                failed_pair_count = failed_pair_count.saturating_add(target_count);
                results.push(json!({
                    "row": action.row,
                    "action": action.kind.as_str(),
                    "group_id": action.group_id,
                    "group_name": action.group_name,
                    "contact_ids": contact_ids,
                    "target_count": target_count,
                    "route": format!("/tools/v2{}", action.route),
                    "ok": false,
                    "error": error.to_string(),
                }));
            }
        }
    }
    Ok(json!({
        "source": "live",
        "action": plan.kind.as_str(),
        "command": plan.command,
        "group_source": plan.group_source,
        "target_source": plan.target_source,
        "summary": {
            "group_count": plan.selected_groups.len(),
            "target_count": plan.target_ids.len(),
            "pair_count": plan.selected_groups.len().saturating_mul(plan.target_ids.len()),
            "write_chunk_count": plan.actions.len(),
            "changed_pair_count": changed_pair_count,
            "failed_pair_count": failed_pair_count,
            "failure_count": failures,
            "ok": failures == 0,
        },
        "results": results,
    }))
}

pub(crate) fn group_bulk_membership_action_value(action: &GroupBulkMembershipAction) -> Value {
    let contact_ids = group_bulk_membership_action_contact_ids(action);
    json!({
        "row": action.row,
        "action": action.kind.as_str(),
        "group_id": action.group_id,
        "group_name": action.group_name,
        "contact_ids": contact_ids,
        "target_count": contact_ids.len(),
        "route": format!("/tools/v2{}", action.route),
        "payload": action.payload,
    })
}

pub(crate) fn group_bulk_membership_action_contact_ids(
    action: &GroupBulkMembershipAction,
) -> Vec<u64> {
    let key = if action.kind == GroupApplyKind::Add {
        "add_contact_ids"
    } else {
        "remove_contact_ids"
    };
    action
        .payload
        .get(key)
        .and_then(Value::as_array)
        .map(|ids| {
            ids.iter()
                .filter_map(|value| value.as_u64())
                .collect::<Vec<_>>()
        })
        .unwrap_or_default()
}

pub(crate) fn group_bulk_membership_rows(report: &Value) -> Value {
    let rows = report
        .get("results")
        .or_else(|| report.get("actions"))
        .and_then(Value::as_array)
        .map(|items| {
            items
                .iter()
                .filter_map(group_bulk_membership_row)
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();
    Value::Array(rows)
}

pub(crate) fn group_bulk_membership_row(value: &Value) -> Option<Value> {
    let object = value.as_object()?;
    let payload = object.get("payload").unwrap_or(&Value::Null);
    let contact_ids = object
        .get("contact_ids")
        .cloned()
        .or_else(|| payload.get("add_contact_ids").cloned())
        .or_else(|| payload.get("remove_contact_ids").cloned())
        .unwrap_or_else(|| Value::Array(Vec::new()));
    let target_count = object.get("target_count").cloned().unwrap_or_else(|| {
        Value::Number(Number::from(
            contact_ids.as_array().map(Vec::len).unwrap_or_default(),
        ))
    });
    Some(json!({
        "row": object.get("row").cloned().unwrap_or(Value::Null),
        "action": object.get("action").cloned().unwrap_or(Value::Null),
        "group_id": object
            .get("group_id")
            .cloned()
            .or_else(|| payload.get("group_id").cloned())
            .unwrap_or(Value::Null),
        "group_name": object.get("group_name").cloned().unwrap_or(Value::Null),
        "route": object.get("route").cloned().unwrap_or(Value::Null),
        "ok": object.get("ok").cloned().unwrap_or(Value::Null),
        "target_count": target_count,
        "contact_ids": contact_ids,
        "result_id": object.get("result_id").cloned().unwrap_or(Value::Null),
        "error": object.get("error").cloned().unwrap_or(Value::Null),
    }))
}

#[cfg(test)]
mod tests {
    use super::*;

    fn temp_snapshot_dir(label: &str) -> PathBuf {
        std::env::temp_dir().join(format!(
            "meshx-group-audit-{label}-{}-{}",
            std::process::id(),
            now_millis()
        ))
    }

    fn write_groups_snapshot(dir: &Path, path: &str, content: &str) -> Result<()> {
        let file_path = safe_snapshot_file_path(dir, path)?;
        if let Some(parent) = file_path.parent() {
            fs::create_dir_all(parent).into_diagnostic()?;
        }
        fs::write(file_path, content).into_diagnostic()?;
        fs::write(
            dir.join("manifest.json"),
            serde_json::to_string(&json!({
                "files": {
                    "groups": {
                        "path": path,
                        "bytes": content.len() as u64,
                        "sha256": sha256_hex(content.as_bytes()),
                    }
                }
            }))
            .into_diagnostic()?,
        )
        .into_diagnostic()
    }

    #[test]
    fn groups_for_audit_snapshot_reads_groups_from_manifest_path() -> Result<()> {
        let dir = temp_snapshot_dir("manifest-path");
        write_groups_snapshot(&dir, "data/groups.json", r#"[{"id":7,"name":"Nested"}]"#)?;

        let (source, groups, live_member_counts) = groups_for_audit_snapshot(&dir)?;

        fs::remove_dir_all(&dir).ok();
        assert_eq!(source.get("file"), Some(&json!("data/groups.json")));
        assert_eq!(groups, vec![json!({"id":7,"name":"Nested"})]);
        assert!(!live_member_counts);
        Ok(())
    }

    #[test]
    fn group_bulk_membership_action_builds_add_payload() {
        let action =
            group_bulk_membership_action(3, GroupApplyKind::Add, 5, "Friends".to_string(), &[1, 2]);

        assert_eq!(action.row, 3);
        assert_eq!(action.kind, GroupApplyKind::Add);
        assert_eq!(action.group_id, 5);
        assert_eq!(action.group_name, "Friends");
        assert_eq!(action.route, "/update-group");
        assert_eq!(
            group_bulk_membership_action_contact_ids(&action),
            vec![1, 2]
        );
        assert_eq!(
            Value::Object(action.payload.clone()),
            json!({"group_id": 5, "add_contact_ids": [1, 2]})
        );
        assert_eq!(
            group_bulk_membership_action_value(&action).get("contact_ids"),
            Some(&json!([1, 2]))
        );
    }

    #[test]
    fn group_bulk_membership_action_builds_remove_payload() {
        let action = group_bulk_membership_action(
            4,
            GroupApplyKind::Remove,
            5,
            "Friends".to_string(),
            &[8, 9],
        );

        assert_eq!(action.kind, GroupApplyKind::Remove);
        assert_eq!(
            group_bulk_membership_action_contact_ids(&action),
            vec![8, 9]
        );
        assert_eq!(
            Value::Object(action.payload),
            json!({"group_id": 5, "remove_contact_ids": [8, 9]})
        );
    }
}
