use std::collections::HashSet;
use serde_json::{json, Map, Value};
use crate::command::{forbid_t2q, ERR_PERMISSION};
use crate::permission::{is_field_readable, Context};
use crate::schema::{Registry, Schema};
use crate::types::{validate_condition, AGG_OPS};
use super::util::{append_order, collect_having_agg_refs, non_nullish};
#[derive(Debug, Clone)]
pub struct AggDef {
pub op: String,
pub field: Option<String>,
}
#[derive(Debug, Clone)]
pub struct GroupSpec {
pub by: Vec<String>,
pub agg: Vec<(String, AggDef)>,
}
impl GroupSpec {
fn is_by(&self, name: &str) -> bool {
self.by.iter().any(|k| k == name)
}
fn agg_def(&self, name: &str) -> Option<&AggDef> {
self.agg.iter().find(|(k, _)| k == name).map(|(_, d)| d)
}
}
fn validate_by_key(schema: &Schema, key: &str, registry: &Registry) -> Result<(), String> {
if schema.relations.contains_key(key) {
return Err(format!(
"$group 的 by 键 \"{key}\" 是关系名(分组键仅支持标量域,不支持关系字段)"
));
}
if let Some((head, rest)) = key.split_once('.') {
if let Some(rel) = schema.relations.get(head) {
if rel.rel_type != "one" {
return Err(format!(
"$group 的 by 键 \"{key}\" 走 many 关系 \"{}\"(会扇出行数破坏聚合语义;仅支持 one 关系路径)",
head
));
}
let root = rest.split('.').next().unwrap_or(rest);
if !schema_fields_of(registry, &rel.model).contains(&root.to_string()) {
return Err(format!(
"$group 的 by 键 \"{key}\" 引用了关系 \"{}\" 的 schema 外字段 \"{root}\"",
head
));
}
return Ok(());
}
}
let root = key.split('.').next().unwrap_or(key);
let Some(fd) = schema.fields.get(root) else {
return Err(format!(
"$group 的 by 键 \"{key}\" 引用了 schema 外字段 \"{root}\""
));
};
match fd.field_type.as_str() {
"array" => Err(format!(
"$group 的 by 键 \"{key}\" 是数组字段(分组键仅支持标量域)"
)),
"object" if key.contains('.') => forbid_t2q(registry, "$group.by object 点号路径"),
"object" => Err(format!(
"$group 的 by 键 \"{key}\" 是对象字段(须用点号路径指明子字段,如 \"{key}.<子字段>\")"
)),
_ => Ok(()),
}
}
fn schema_fields_of(registry: &Registry, model: &str) -> Vec<String> {
registry
.get(model)
.map(|s| s.fields.keys().cloned().collect())
.unwrap_or_default()
}
fn validate_agg_field(schema: &Schema, op: &str, alias: &str, field: &str) -> Result<(), String> {
if schema.relations.contains_key(field) {
return Err(format!(
"$group 的 agg \"{alias}\"({op})引用了关系 \"{field}\"(仅支持标量字段)"
));
}
if field.contains('.') {
return Err(format!(
"$group 的 agg \"{alias}\"({op})字段 \"{field}\" 不可为点号路径(仅支持标量字段)"
));
}
match schema.fields.get(field) {
None => Err(format!(
"$group 的 agg \"{alias}\"({op})引用了 schema 外字段 \"{field}\""
)),
Some(fd) if fd.field_type == "object" || fd.field_type == "array" => Err(format!(
"$group 的 agg \"{alias}\"({op})字段 \"{field}\" 必须是标量字段"
)),
Some(_) => Ok(()),
}
}
pub fn parse(schema: &Schema, group_v: &Value, registry: &Registry) -> Result<GroupSpec, String> {
let obj = group_v
.as_object()
.ok_or("$group 参数必须是对象 { by, agg }")?;
let mut by: Vec<String> = Vec::new();
match obj.get("by") {
None | Some(Value::Null) => {}
Some(Value::Array(arr)) => {
for v in arr {
let key = v.as_str().ok_or("$group.by 的元素必须是字段名字符串")?;
validate_by_key(schema, key, registry)?;
if by.iter().any(|k| k == key) {
return Err(format!("$group.by 存在重复键 \"{key}\""));
}
by.push(key.to_string());
}
}
Some(_) => return Err("$group.by 必须是字段名数组(省略 / [] = 全表单组)".to_string()),
}
let mut agg: Vec<(String, AggDef)> = Vec::new();
if let Some(v) = obj.get("agg") {
if !v.is_null() {
let m = v
.as_object()
.ok_or("$group.agg 必须是对象 { 别名: { 算子: 字段 } }")?;
for (alias, def) in m {
if alias.is_empty() {
return Err("$group.agg 的别名不能为空".to_string());
}
if by.iter().any(|k| k == alias) {
return Err(format!("$group.agg 别名 \"{alias}\" 与 by 键冲突"));
}
let d = def
.as_object()
.ok_or_else(|| format!("$group.agg \"{alias}\" 的定义必须是对象"))?;
let mut it = d.iter();
let (op, arg) = match (it.next(), it.next()) {
(Some(kv), None) => kv,
_ => return Err(format!("$group.agg \"{alias}\" 必须恰有一个算子键")),
};
if !AGG_OPS.contains(&op.as_str()) {
return Err(format!(
"$group.agg \"{alias}\" 的算子 {op} 不在白名单($count/$sum/$avg/$min/$max)"
));
}
let field = if op == "$count" && arg.as_str() == Some("*") {
None
} else {
let f = arg.as_str().ok_or_else(|| {
format!(
"$group.agg \"{alias}\"({op})必须带标量字段名($count 可用 \"*\")"
)
})?;
validate_agg_field(schema, op, alias, f)?;
Some(f.to_string())
};
agg.push((
alias.clone(),
AggDef {
op: op.clone(),
field,
},
));
}
}
}
if by.is_empty() && agg.is_empty() {
return Err("$group 至少需要 by 键或 agg 别名之一".to_string());
}
Ok(GroupSpec { by, agg })
}
pub fn validate_read_permission(
schema: &Schema,
ctx: Option<&Context>,
spec: &GroupSpec,
) -> Result<(), String> {
if ctx.is_none() {
return Ok(());
}
for key in &spec.by {
if !is_field_readable(schema, ctx, key) {
return Err(ERR_PERMISSION.to_string());
}
}
for (_, def) in &spec.agg {
if let Some(field) = &def.field {
if !is_field_readable(schema, ctx, field) {
return Err(ERR_PERMISSION.to_string());
}
}
}
Ok(())
}
pub fn validate_having(spec: &GroupSpec, having: &Value) -> Result<(), String> {
fn walk(spec: &GroupSpec, v: &Value) -> Result<(), String> {
let Some(m) = v.as_object() else {
return Ok(());
};
for (k, val) in m {
match k.as_str() {
"$and" | "$or" | "$nor" => {
let arr = val
.as_array()
.ok_or_else(|| format!("$having 的 {k} 需要数组"))?;
for it in arr {
walk(spec, it)?;
}
}
_ if k.starts_with('$') => {
return Err(format!(
"$having 不支持操作符 {k}(仅可引用 by 键 / agg 别名,支持 $and/$or/$nor)"
));
}
_ => {
if !spec.is_by(k) && spec.agg_def(k).is_none() {
return Err(format!(
"$having 引用了未声明的字段 \"{k}\"(仅可引用 by 键 / agg 别名)"
));
}
}
}
}
Ok(())
}
if !having.is_object() {
return Err("$having 必须是条件对象".to_string());
}
walk(spec, having)
}
#[allow(clippy::too_many_arguments)]
pub fn build_stages(
spec: &GroupSpec,
fields: &[String],
relations_empty: bool,
condition: Option<&Value>,
having: Option<&Value>,
sort: Option<&Value>,
skip: Option<&Value>,
limit: Option<&Value>,
schema: &Schema,
registry: &Registry,
) -> Result<Vec<Value>, String> {
if !relations_empty {
return Err(
"$group 查询不支持关系字段(选择集仅可为 by 键 / agg 别名,无关系下钻)".to_string(),
);
}
if fields.is_empty() {
return Err("$group 查询必须显式选择输出字段(by 键 / agg 别名)".to_string());
}
for f in fields {
if !spec.is_by(f) && spec.agg_def(f).is_none() {
return Err(format!(
"选择集字段 \"{f}\" 未在 $group 的 by 键 / agg 别名中声明"
));
}
}
if let Some(h) = non_nullish(having) {
validate_condition(h)?;
validate_having(spec, h)?;
}
if let Some(s) = non_nullish(sort) {
let m = s.as_object().ok_or("$sort 必须是对象")?;
for k in m.keys() {
if !spec.is_by(k) && spec.agg_def(k).is_none() {
return Err(format!(
"$sort 键 \"{k}\" 不在 $group 的 by 键 / agg 别名域内"
));
}
}
}
let mut refs: HashSet<String> = HashSet::new();
if let Some(h) = non_nullish(having) {
collect_having_agg_refs(h, &|k| spec.agg_def(k).is_some(), &mut refs);
}
let needed: Vec<&str> = spec
.agg
.iter()
.filter(|(alias, _)| fields.iter().any(|f| f == alias) || refs.contains(alias))
.map(|(alias, _)| alias.as_str())
.collect();
let mut stages: Vec<Value> = Vec::new();
let mut rel_by_keys: Vec<(String, &str, String, String)> = Vec::new(); for key in &spec.by {
if let Some((head, rest)) = key.split_once('.') {
if let Some(rel) = schema.relations.get(head) {
rel_by_keys.push((
key.clone(),
head,
rel.local_field.clone(),
rel.foreign_field.clone(),
));
let _ = rest;
}
}
}
if !rel_by_keys.is_empty() {
let mut seen: Vec<&str> = Vec::new();
for (key, head, local, foreign) in &rel_by_keys {
if seen.contains(head) {
continue;
}
seen.push(head);
let rel = &schema.relations[*head];
let rel_schema = registry.get(&rel.model)?;
let mut let_map = Map::new();
let_map.insert(
format!("g_{}", head),
json!(format!("${}", local)),
);
let mut inner = Map::new();
inner.insert("from".to_string(), json!(rel_schema.collection));
inner.insert("let".to_string(), Value::Object(let_map));
inner.insert(
"pipeline".to_string(),
json!([{ "$match": { "$expr": { "$eq": [
format!("${}", foreign),
format!("$$g_{}", head)
] } } }]),
);
inner.insert("as".to_string(), json!(head));
stages.push(json!({ "$lookup": Value::Object(inner) }));
stages.push(json!({
"$unwind": {
"path": format!("${}", head),
"preserveNullAndEmptyArrays": true
}
}));
let _ = key;
}
}
if let Some(cond) = non_nullish(condition) {
validate_condition(cond)?;
stages.push(json!({ "$match": cond }));
}
let mut group_map = Map::new();
group_map.insert("_id".to_string(), id_expr(&spec.by));
for (alias, def) in &spec.agg {
if !needed.contains(&alias.as_str()) {
continue;
}
group_map.insert(alias.clone(), acc_expr(def));
}
stages.push(json!({ "$group": Value::Object(group_map) }));
if spec.by.is_empty() {
let mut def_row = Map::new();
def_row.insert("_id".to_string(), Value::Null);
for (alias, def) in &spec.agg {
if !needed.contains(&alias.as_str()) {
continue;
}
def_row.insert(
alias.clone(),
if def.op == "$count" {
json!(0)
} else {
Value::Null
},
);
}
stages.push(json!({ "$facet": { "__rows": [] } }));
stages.push(json!({
"$replaceRoot": { "newRoot": { "$ifNull": [
{ "$arrayElemAt": ["$__rows", 0] },
Value::Object(def_row)
] } }
}));
}
if let Some(h) = non_nullish(having) {
stages.push(json!({ "$match": rewrite_having(spec, h) }));
}
let order_src = non_nullish(sort).map(|s| rewrite_order(spec, s));
append_order(&mut stages, order_src.as_ref(), skip, limit);
stages.push(json!({ "$project": projection(spec, fields) }));
Ok(stages)
}
fn id_expr(by: &[String]) -> Value {
match by.len() {
0 => Value::Null,
1 => json!(format!("${}", by[0])),
_ => {
let mut root = Map::new();
for key in by {
let parts: Vec<&str> = key.split('.').collect();
insert_nested(&mut root, &parts, json!(format!("${}", key)));
}
Value::Object(root)
}
}
}
fn insert_nested(obj: &mut Map<String, Value>, parts: &[&str], val: Value) {
if parts.len() == 1 {
obj.insert(parts[0].to_string(), val);
return;
}
let entry = obj.entry(parts[0].to_string()).or_insert_with(|| json!({}));
if let Value::Object(m) = entry {
insert_nested(m, &parts[1..], val);
}
}
pub(crate) fn acc_expr(def: &AggDef) -> Value {
let field_ref = format!("${}", def.field.clone().unwrap_or_default());
match def.op.as_str() {
"$count" => match &def.field {
None => json!({ "$sum": 1 }),
Some(_) => json!({
"$sum": { "$cond": [
{ "$eq": [ { "$ifNull": [field_ref, null] }, null ] },
0,
1
] }
}),
},
"$sum" => json!({ "$sum": field_ref }),
"$avg" => json!({ "$avg": field_ref }),
"$min" => json!({ "$min": field_ref }),
_ => json!({ "$max": field_ref }),
}
}
fn projection(spec: &GroupSpec, fields: &[String]) -> Value {
let mut proj = Map::new();
let requested_by: Vec<&String> = spec
.by
.iter()
.filter(|k| fields.iter().any(|f| f == *k))
.collect();
let id_in_by = requested_by.iter().any(|k| k.as_str() == "_id");
if !id_in_by {
proj.insert("_id".to_string(), json!(0));
}
for key in &requested_by {
let expr = if key.as_str() == "_id" {
if spec.by.len() == 1 {
json!(1)
} else {
json!("$_id._id")
}
} else if spec.by.len() == 1 {
json!("$_id")
} else {
json!(format!("$_id.{}", key))
};
proj.insert((*key).clone(), expr);
}
for (alias, _) in &spec.agg {
if fields.iter().any(|f| f == alias) {
proj.insert(alias.clone(), json!(1));
}
}
Value::Object(proj)
}
fn rewrite_having(spec: &GroupSpec, h: &Value) -> Value {
match h {
Value::Object(m) => {
let mut out = Map::new();
for (k, v) in m {
if matches!(k.as_str(), "$and" | "$or" | "$nor") {
let arr = v
.as_array()
.map(|a| a.iter().map(|it| rewrite_having(spec, it)).collect())
.unwrap_or_default();
out.insert(k.clone(), Value::Array(arr));
} else if spec.is_by(k) {
let nk = if spec.by.len() == 1 {
"_id".to_string()
} else {
format!("_id.{}", k)
};
out.insert(nk, v.clone());
} else {
out.insert(k.clone(), v.clone());
}
}
Value::Object(out)
}
other => other.clone(),
}
}
fn rewrite_order(spec: &GroupSpec, s: &Value) -> Value {
let Some(m) = s.as_object() else {
return s.clone();
};
let mut out = Map::new();
for (k, v) in m {
if spec.is_by(k) {
let nk = if spec.by.len() == 1 {
"_id".to_string()
} else {
format!("_id.{}", k)
};
out.insert(nk, v.clone());
} else {
out.insert(k.clone(), v.clone());
}
}
Value::Object(out)
}