//! Pipeline lowering: translates an `Expr` AST into a `Pipeline` IR ready for execution.
//!
//! `Pipeline::lower` is the single entry point; it walks the expression, classifies sources,
//! stages, and sinks, and returns `None` for shapes that cannot be a linear pull chain,
//! signalling fallback to the VM opcode path.
//!
//! This module also contains the registry-driven stage factory helpers (`lower_method_from_registry`
//! and friends) that were previously in `stage_factory.rs`.

use std::sync::Arc;

use crate::builtins::registry::{pipeline_accepts_arity, pipeline_lowering, BuiltinId};
use crate::builtins::{
    BuiltinCategory, BuiltinMethod, BuiltinPipelineLowering, BuiltinSelectionPosition,
    BuiltinSinkAccumulator, BuiltinViewStage,
};
use crate::parse::ast::Expr;
use crate::data::value::Val;

use super::{
    expr_label, plan_with_exprs, sink_name, source_name, trace_enabled, BodyKernel, Pipeline,
    PipelineBody, Plan, Sink, SortSpec, Source, Stage,
};

impl Pipeline {
    /// Lowers `expr` into a `Pipeline` IR, returning `None` when the expression shape requires the VM opcode path.
    pub fn lower(expr: &Expr) -> Option<Pipeline> {
        let p = Self::lower_with_reason(expr);
        if trace_enabled() {
            match &p {
                Ok(pipe) => eprintln!(
                    "[pipeline] activated: stages={} sink={} src={}",
                    pipe.stages.len(),
                    sink_name(&pipe.sink),
                    source_name(&pipe.source),
                ),
                Err(reason) => {
                    eprintln!("[pipeline] fallback: ({}) at {}", reason, expr_label(expr),)
                }
            }
        }
        p.ok()
    }

    // Converts `None` from `lower_inner` into `Err(&str)` so the trace path can report the reason.
    fn lower_with_reason(expr: &Expr) -> std::result::Result<Pipeline, &'static str> {
        Self::lower_inner(expr).ok_or("shape not yet supported")
    }

    // Requires `expr` rooted at `$`; extracts the leading field chain and delegates the rest to `lower_from_source`.
    fn lower_inner(expr: &Expr) -> Option<Pipeline> {
        use crate::parse::ast::Step;
        let (base, steps) = match expr {
            Expr::Chain(b, s) => (b.as_ref(), s.as_slice()),
            _ => return None,
        };
        if !matches!(base, Expr::Root) {
            return None;
        }

        let mut field_end = 0;
        for s in steps {
            match s {
                Step::Field(_) => field_end += 1,
                _ => break,
            }
        }
        if field_end == 0 {
            return None;
        }

        let keys: Arc<[Arc<str>]> = steps[..field_end]
            .iter()
            .map(|s| match s {
                Step::Field(k) => Arc::<str>::from(k.as_str()),
                _ => unreachable!(),
            })
            .collect::<Vec<_>>()
            .into();

        let trailing = &steps[field_end..];
        Self::lower_from_source(Source::FieldChain { keys }, trailing)
    }

    /// Lowers `trailing` steps into a `PipelineBody` and attaches `source`, returning `None` when any step is unclassifiable.
    pub(crate) fn lower_from_source(
        source: Source,
        trailing: &[crate::parse::ast::Step],
    ) -> Option<Pipeline> {
        Some(Self::lower_body_from_steps(trailing)?.with_source(source))
    }

    /// Decodes `trailing` steps into stages and a sink, runs rewrite passes, and classifies body kernels.
    pub(crate) fn lower_body_from_steps(
        trailing: &[crate::parse::ast::Step],
    ) -> Option<PipelineBody> {
        let (stages, stage_exprs, sink) = decode_method_chain(trailing)?;
        let mut p = PipelineBody {
            stages,
            stage_exprs,
            sink,
            stage_kernels: Vec::new(),
            sink_kernels: Vec::new(),
        };
        rewrite(&mut p);
        Some(PipelineBody::planned(p.stages, p.stage_exprs, p.sink))
    }

    /// Returns `true` when `step` is a method call that can open a receiver-based pipeline without a field-chain prefix.
    pub(crate) fn is_receiver_pipeline_start(step: &crate::parse::ast::Step) -> bool {
        use crate::parse::ast::Step;

        let Step::Method(name, args) = step else {
            return false;
        };
        is_receiver_pipeline_start_method(name.as_str(), args.len())
    }
}

// Returns `true` when `name`/`arity` is a builtin that can open a receiver-based pipeline.
fn is_receiver_pipeline_start_method(name: &str, arity: usize) -> bool {
    let method = BuiltinMethod::from_name(name);
    if method == BuiltinMethod::Unknown {
        return false;
    }

    pipeline_accepts_arity(BuiltinId::from_method(method), arity, true)
}

/// Decodes a `map(expr)` argument as a nested pipeline `Plan`, enabling the `CompiledMap` stage optimisation.
pub(super) fn try_decode_map_body(arg: &crate::parse::ast::Arg) -> Option<Plan> {
    use crate::parse::ast::{Arg, Step};
    let expr = match arg {
        Arg::Pos(e) => e,
        _ => return None,
    };
    let (base, steps) = match expr {
        Expr::Chain(b, s) => (b.as_ref(), s.as_slice()),
        _ => return None,
    };

    let mut leading_fields: Vec<Arc<str>> = Vec::new();
    match base {
        Expr::Current => {}
        Expr::Ident(name) => leading_fields.push(Arc::from(name.as_str())),
        _ => return None,
    }

    let mut field_end = 0;
    for s in steps {
        match s {
            Step::Field(_) => field_end += 1,
            _ => break,
        }
    }
    let trailing = &steps[field_end..];
    if trailing.is_empty() {
        return None;
    }
    if !trailing_has_collection_operator(trailing) {
        return None;
    }

    let source = if field_end > 0 || !leading_fields.is_empty() {
        leading_fields.extend(steps[..field_end].iter().map(|s| match s {
            Step::Field(k) => Arc::<str>::from(k.as_str()),
            _ => unreachable!(),
        }));
        Source::FieldChain {
            keys: leading_fields.into(),
        }
    } else {
        Source::Receiver(Val::Null)
    };
    let mut stages: Vec<Stage> = Vec::new();
    let (mut more_stages, more_exprs, sink) = decode_method_chain(trailing)?;
    stages.append(&mut more_stages);

    let kernels = Stage::body_kernels(&stages);
    let mut plan = plan_with_exprs(stages, more_exprs, &kernels, sink);
    plan.source = source;
    Some(plan)
}

fn trailing_has_collection_operator(trailing: &[crate::parse::ast::Step]) -> bool {
    use crate::parse::ast::Step;
    trailing.iter().any(|step| {
        let Step::Method(name, _) = step else {
            return false;
        };
        matches!(
            BuiltinMethod::from_name(name.as_str()).spec().category,
            BuiltinCategory::StreamingOneToOne
                | BuiltinCategory::StreamingFilter
                | BuiltinCategory::StreamingExpand
                | BuiltinCategory::Reducer
                | BuiltinCategory::Positional
                | BuiltinCategory::Barrier
                | BuiltinCategory::Deep
                | BuiltinCategory::Relational
        )
    })
}

// Classifies each trailing method step as a stage or sink; `None` on any unrecognised step.
fn decode_method_chain(
    trailing: &[crate::parse::ast::Step],
) -> Option<(Vec<Stage>, Vec<Option<Arc<Expr>>>, Sink)> {
    use crate::parse::ast::Step;
    let mut stages: Vec<Stage> = Vec::new();
    let mut stage_exprs: Vec<Option<Arc<Expr>>> = Vec::new();
    let mut sink: Sink = Sink::Collect;
    for (i, s) in trailing.iter().enumerate() {
        let is_last = i == trailing.len() - 1;
        match s {
            Step::Method(name, args) => {
                if let Some(call) =
                    crate::builtins::BuiltinCall::from_pipeline_literal_args(name.as_str(), args)
                {
                    stages.push(Stage::Builtin(call));
                    stage_exprs.push(None);
                    continue;
                }
                let method = BuiltinMethod::from_name(name.as_str());
                if matches!(
                    method.spec().view_stage,
                    Some(BuiltinViewStage::Compact | BuiltinViewStage::RemoveValue)
                ) {
                    if let Some(call) =
                        crate::builtins::BuiltinCall::from_literal_ast_args(name.as_str(), args)
                    {
                        stages.push(Stage::Builtin(call));
                        stage_exprs.push(None);
                        continue;
                    }
                }
                lower_method_from_registry(
                    method,
                    args,
                    is_last,
                    &mut stages,
                    &mut stage_exprs,
                    &mut sink,
                )?;
            }
            Step::Slice(start, end, step) => {
                // Pipeline lowering only handles step-1 slices today. Step
                // != 1 falls back to interpreted path (returns `None` to
                // signal "lower me elsewhere").
                if step.unwrap_or(1) != 1 {
                    return None;
                }
                push_path_slice_stages(*start, *end, &mut stages, &mut stage_exprs)?;
            }
            Step::Wildcard => {
                // Wildcard `[*]` is a no-op pass-through in pipeline form;
                // upstream already produced the array element stream.
            }
            _ => return None,
        }
    }
    Some((stages, stage_exprs, sink))
}

fn push_path_slice_stages(
    start: Option<i64>,
    end: Option<i64>,
    stages: &mut Vec<Stage>,
    stage_exprs: &mut Vec<Option<Arc<Expr>>>,
) -> Option<()> {
    let start = match start {
        Some(n) if n < 0 => return None,
        Some(n) => n as usize,
        None => 0,
    };
    let end = match end {
        Some(n) if n < 0 => return None,
        Some(n) => Some(n as usize),
        None => None,
    };

    if start > 0 {
        stages.push(Stage::UsizeBuiltin {
            method: BuiltinMethod::Skip,
            value: start,
        });
        stage_exprs.push(None);
    }

    if let Some(end) = end {
        stages.push(Stage::UsizeBuiltin {
            method: BuiltinMethod::Take,
            value: end.saturating_sub(start),
        });
        stage_exprs.push(None);
    }

    Some(())
}

// Repeatedly applies `rewrite_step` until fixpoint (at most 16 iterations).
fn rewrite(p: &mut PipelineBody) {
    let mut fuel = 16usize;
    while fuel > 0 {
        fuel -= 1;
        if rewrite_step(p) {
            continue;
        }
        break;
    }
}

// Applies one rewrite pass: const-false short-circuit, adjacent map/filter fusion, map/take commutation.
fn rewrite_step(p: &mut PipelineBody) -> bool {
    use crate::vm::Opcode;

    let mut const_false_at: Option<usize> = None;
    for (i, s) in p.stages.iter().enumerate() {
        if let Stage::Filter(prog, _) = s {
            if let Some(false) = prog_const_bool(prog) {
                const_false_at = Some(i);
                break;
            }
        }
    }
    if const_false_at.is_some() {
        p.stages.clear();
        p.stage_exprs.clear();
        return true;
    }

    for i in 0..p.stages.len().saturating_sub(1) {
        match (&p.stages[i], &p.stages[i + 1]) {
            (Stage::Map(a_prog, _), Stage::Map(b_prog, _)) => {
                let ka = BodyKernel::classify(a_prog);
                let kb = BodyKernel::classify(b_prog);
                let chain: Option<Vec<Arc<str>>> = match (&ka, &kb) {
                    (BodyKernel::FieldRead(a), BodyKernel::FieldRead(b)) => {
                        Some(vec![a.clone(), b.clone()])
                    }
                    (BodyKernel::FieldRead(a), BodyKernel::FieldChain(bs)) => {
                        let mut v = vec![a.clone()];
                        v.extend(bs.iter().cloned());
                        Some(v)
                    }
                    (BodyKernel::FieldChain(as_), BodyKernel::FieldRead(b)) => {
                        let mut v: Vec<Arc<str>> = as_.iter().cloned().collect();
                        v.push(b.clone());
                        Some(v)
                    }
                    (BodyKernel::FieldChain(as_), BodyKernel::FieldChain(bs)) => {
                        let mut v: Vec<Arc<str>> = as_.iter().cloned().collect();
                        v.extend(bs.iter().cloned());
                        Some(v)
                    }
                    _ => None,
                };
                if let Some(keys) = chain {
                    let fcd = Arc::new(crate::vm::FieldChainData {
                        keys: keys.into(),
                        ics: (0..0)
                            .map(|_| std::sync::atomic::AtomicU64::new(0))
                            .collect::<Vec<_>>()
                            .into_boxed_slice(),
                    });
                    let new_ops = vec![Opcode::PushCurrent, Opcode::FieldChain(fcd)];
                    let merged = Arc::new(crate::vm::Program::new(new_ops, "<map-fused>"));
                    p.stages[i] = Stage::Map(merged, BuiltinViewStage::Map);
                    p.stage_exprs[i] = None;
                    p.stages.remove(i + 1);
                    p.stage_exprs.remove(i + 1);
                    return true;
                }
            }
            (Stage::Filter(p_prog, _), Stage::Filter(q_prog, _)) => {
                let mut ops: Vec<Opcode> = p_prog.ops.as_ref().to_vec();
                ops.push(Opcode::AndOp(Arc::clone(q_prog)));
                let merged = Arc::new(crate::vm::Program {
                    ops: ops.into(),
                    source: p_prog.source.clone(),
                    id: 0,
                    is_structural: false,
                    ics: p_prog.ics.clone(),
                });
                p.stages[i] = Stage::Filter(merged, BuiltinViewStage::Filter);
                p.stage_exprs[i] = None;
                p.stages.remove(i + 1);
                p.stage_exprs.remove(i + 1);
                return true;
            }
            _ => {}
        }
    }

    for i in 0..p.stages.len().saturating_sub(1) {
        if matches!(&p.stages[i], Stage::Map(_, _)) && is_take_stage(&p.stages[i + 1]) {
            p.stages.swap(i, i + 1);
            p.stage_exprs.swap(i, i + 1);
            return true;
        }
    }

    false
}

fn is_take_stage(stage: &Stage) -> bool {
    matches!(
        stage,
        Stage::UsizeBuiltin {
            method: BuiltinMethod::Take,
            ..
        }
    )
}

// Returns `Some(b)` when `prog` is a single `PushBool(b)` opcode; detects constant filter stages.
fn prog_const_bool(prog: &crate::vm::Program) -> Option<bool> {
    use crate::vm::Opcode;
    let ops = prog.ops.as_ref();
    if ops.len() != 1 {
        return None;
    }
    match &ops[0] {
        Opcode::PushBool(b) => Some(*b),
        _ => None,
    }
}

/// Compiles a positional argument into a VM `Program`, rewriting bare `Ident` nodes into `@.<ident>` field accesses.
pub(super) fn compile_subexpr(arg: &crate::parse::ast::Arg) -> Option<Arc<crate::vm::Program>> {
    use crate::parse::ast::{Arg, Step};
    let inner = match arg {
        Arg::Pos(e) => e,
        _ => return None,
    };
    // Single-param lambda: route through `compile_lambda_arg` so the body
    // is substituted and, when nested-lambda refs to the param remain,
    // wrapped in `BindLamCurrent` for correct outer-row binding.
    if matches!(inner, Expr::Lambda { params, .. } if params.len() == 1) {
        return Some(crate::compile::lambda_lower::compile_lambda_arg(inner, ""));
    }
    let rooted: Expr = match inner {
        Expr::Ident(name) => Expr::Chain(Box::new(Expr::Current), vec![Step::Field(name.clone())]),
        Expr::Chain(base, _) if matches!(base.as_ref(), Expr::Current) => inner.clone(),
        other => other.clone(),
    };
    Some(Arc::new(crate::compile::compiler::Compiler::compile(
        &rooted, "",
    )))
}

/// Compiles a pipeline stage body expression using the same current-row
/// binding rules as method-chain lowering.
pub(crate) fn compile_pipeline_expr_body(expr: &Expr) -> Arc<crate::vm::Program> {
    compile_subexpr(&crate::parse::ast::Arg::Pos(expr.clone()))
        .expect("positional pipeline expression must compile")
}

fn compile_raw_arg_expr(arg: &crate::parse::ast::Arg) -> Option<Arc<crate::vm::Program>> {
    use crate::parse::ast::Arg;
    let expr = match arg {
        Arg::Pos(e) => e,
        Arg::Named(_, _) => return None,
    };
    Some(crate::compile::lambda_lower::compile_lambda_arg(expr, ""))
}

/// Compiles a sort-key argument into a `SortSpec`, interpreting `UnaryNeg`-wrapping as descending order.
///
/// Returns `None` for multi-param `Expr::Lambda` arguments — those are
/// 2-arg comparator lambdas (`.sort((a, b) => …)`) which the pipeline IR
/// cannot represent as a single-key `SortSpec`. Bailing out forces the
/// router to fall back to the VM path, where `exec_lambda_method` handles
/// the comparator via `sort_comparator_apply`.
pub(crate) fn compile_sort_spec(
    arg: &crate::parse::ast::Arg,
) -> Option<(SortSpec, Option<Arc<Expr>>)> {
    use crate::parse::ast::{Arg, Expr};
    let expr = match arg {
        Arg::Pos(e) => e,
        _ => return None,
    };
    if matches!(expr, Expr::Lambda { params, .. } if params.len() >= 2) {
        return None;
    }
    let (key_expr, descending) = match expr {
        Expr::UnaryNeg(inner) => (inner.as_ref().clone(), true),
        other => (other.clone(), false),
    };
    let key_arg = Arg::Pos(key_expr.clone());
    Some((
        SortSpec::keyed(compile_subexpr(&key_arg)?, descending),
        Some(Arc::new(expr.clone())),
    ))
}

/// Wraps the inner `Expr` of a positional argument as `Arc<Expr>`, returning `None` for named arguments.
pub(super) fn arg_expr(arg: &crate::parse::ast::Arg) -> Option<Arc<Expr>> {
    match arg {
        crate::parse::ast::Arg::Pos(e) => Some(Arc::new(e.clone())),
        _ => None,
    }
}

// ---------------------------------------------------------------------------
// Registry-driven stage factory (formerly stage_factory.rs)
// ---------------------------------------------------------------------------

use super::{
    ArgExtremeSinkSpec, MembershipSinkSpec, MembershipSinkTarget, PredicateSinkSpec, ReducerSpec,
};

/// Lowers a `BuiltinMethod` call to a concrete `Stage` or `Sink`, returning `None` when the method cannot be lowered at this position.
pub(super) fn lower_method_from_registry(
    method: BuiltinMethod,
    args: &[crate::parse::ast::Arg],
    is_last: bool,
    stages: &mut Vec<Stage>,
    stage_exprs: &mut Vec<Option<Arc<Expr>>>,
    sink: &mut Sink,
) -> Option<()> {
    let Some(lowering) = pipeline_lowering(BuiltinId::from_method(method)) else {
        if is_last {
            *sink = terminal_sink_for_method(method, args)?;
            return Some(());
        }
        return None;
    };
    match lowering {
        BuiltinPipelineLowering::ExprArg => {
            if args.len() != 1 {
                return None;
            }
            push_expr_stage(method, &args[0], stages, stage_exprs)
        }
        BuiltinPipelineLowering::TerminalExprArg { terminal } => {
            if args.len() != 1 {
                return None;
            }
            push_expr_stage(method, &args[0], stages, stage_exprs)?;
            if is_last {
                set_terminal_sink(terminal, sink)?;
            }
            Some(())
        }
        BuiltinPipelineLowering::Nullary => {
            if !args.is_empty() {
                return None;
            }
            match method {
                BuiltinMethod::Reverse => {
                    let cancel = method
                        .spec()
                        .cancellation
                        .expect("reverse builtin must define cancellation metadata");
                    stages.push(Stage::Reverse(cancel));
                }
                BuiltinMethod::Unique => stages.push(Stage::UniqueBy(None)),
                _ if method.is_pipeline_element_method() => {
                    stages.push(Stage::Builtin(crate::builtins::BuiltinCall::new(
                        method,
                        crate::builtins::BuiltinArgs::None,
                    )));
                }
                _ => return None,
            }
            stage_exprs.push(None);
            Some(())
        }
        BuiltinPipelineLowering::UsizeArg { min } => {
            if args.len() != 1 {
                return None;
            }
            let n = usize_arg_at_least(&args[0], min)?;
            stages.push(Stage::UsizeBuiltin { method, value: n });
            stage_exprs.push(None);
            Some(())
        }
        BuiltinPipelineLowering::StringArg => {
            if args.len() != 1 {
                return None;
            }
            stages.push(Stage::StringBuiltin {
                method,
                value: string_arg(&args[0])?,
            });
            stage_exprs.push(None);
            Some(())
        }
        BuiltinPipelineLowering::StringPairArg => {
            if args.len() != 2 {
                return None;
            }
            stages.push(Stage::StringPairBuiltin {
                method,
                first: string_arg(&args[0])?,
                second: string_arg(&args[1])?,
            });
            stage_exprs.push(None);
            Some(())
        }
        BuiltinPipelineLowering::Sort => match args {
            [] => {
                stages.push(Stage::Sort(super::SortSpec::identity()));
                stage_exprs.push(None);
                Some(())
            }
            [arg] => {
                let (spec, expr) = compile_sort_spec(arg)?;
                stages.push(Stage::Sort(spec));
                stage_exprs.push(expr);
                Some(())
            }
            _ => None,
        },
        BuiltinPipelineLowering::IntRangeArg => match args {
            [arg] => {
                stages.push(Stage::IntRangeBuiltin {
                    method,
                    start: int_arg(arg)?,
                    end: None,
                });
                stage_exprs.push(None);
                Some(())
            }
            [start, end] => {
                stages.push(Stage::IntRangeBuiltin {
                    method,
                    start: int_arg(start)?,
                    end: Some(int_arg(end)?),
                });
                stage_exprs.push(None);
                Some(())
            }
            _ => None,
        },
        BuiltinPipelineLowering::TerminalSink if is_last => {
            *sink = terminal_sink_for_method(method, args)?;
            Some(())
        }
        BuiltinPipelineLowering::TerminalSink => None,
        BuiltinPipelineLowering::TerminalUsizeSink { min } if is_last => {
            if args.len() != 1 {
                return None;
            }
            *sink = Sink::Nth(usize_arg_at_least(&args[0], min)?);
            Some(())
        }
        BuiltinPipelineLowering::TerminalUsizeSink { .. } => None,
    }
}

// Compiles `arg` into a sub-expression program and appends the corresponding `Stage` variant; `None` on compile failure.
fn push_expr_stage(
    method: BuiltinMethod,
    arg: &crate::parse::ast::Arg,
    stages: &mut Vec<Stage>,
    stage_exprs: &mut Vec<Option<Arc<Expr>>>,
) -> Option<()> {
    match method {
        BuiltinMethod::Filter
        | BuiltinMethod::Find
        | BuiltinMethod::FindAll
        | BuiltinMethod::FindFirst => {
            stages.push(Stage::Filter(
                compile_subexpr(arg)?,
                BuiltinViewStage::Filter,
            ));
            stage_exprs.push(arg_expr(arg));
        }
        BuiltinMethod::Map => match try_decode_map_body(arg) {
            Some(plan) => {
                stages.push(Stage::CompiledMap(Arc::new(plan)));
                stage_exprs.push(arg_expr(arg));
            }
            None => {
                stages.push(Stage::Map(compile_subexpr(arg)?, BuiltinViewStage::Map));
                stage_exprs.push(arg_expr(arg));
            }
        },
        BuiltinMethod::FlatMap => {
            stages.push(Stage::FlatMap(
                compile_subexpr(arg)?,
                BuiltinViewStage::FlatMap,
            ));
            stage_exprs.push(arg_expr(arg));
        }
        BuiltinMethod::UniqueBy => {
            stages.push(Stage::UniqueBy(Some(compile_subexpr(arg)?)));
            stage_exprs.push(arg_expr(arg));
        }
        // Remaining expression-argument lowerings route through the generic
        // ExprBuiltin stage. The caller has already checked registry lowering
        // metadata, so new supported builtins do not need another list here.
        _ => {
            push_expr_builtin(method, arg, stages, stage_exprs)?;
        }
    }
    Some(())
}

fn push_expr_builtin(
    method: BuiltinMethod,
    arg: &crate::parse::ast::Arg,
    stages: &mut Vec<Stage>,
    stage_exprs: &mut Vec<Option<Arc<Expr>>>,
) -> Option<()> {
    stages.push(Stage::ExprBuiltin {
        method,
        body: compile_subexpr(arg)?,
    });
    stage_exprs.push(arg_expr(arg));
    Some(())
}

// Writes the no-arg terminal `Sink` for `method` into `*sink`.
fn set_terminal_sink(method: BuiltinMethod, sink: &mut Sink) -> Option<()> {
    *sink = terminal_sink_for_method(method, &[])?;
    Some(())
}

// Extracts a `usize` integer literal from `arg` and enforces `value >= min`.
fn usize_arg_at_least(arg: &crate::parse::ast::Arg, min: usize) -> Option<usize> {
    match arg {
        crate::parse::ast::Arg::Pos(Expr::Int(n)) if *n >= min as i64 => Some(*n as usize),
        _ => None,
    }
}

// Extracts a signed integer literal from `arg`.
fn int_arg(arg: &crate::parse::ast::Arg) -> Option<i64> {
    match arg {
        crate::parse::ast::Arg::Pos(Expr::Int(n)) => Some(*n),
        _ => None,
    }
}

// Extracts a string literal from `arg` and interns it as `Arc<str>`.
fn string_arg(arg: &crate::parse::ast::Arg) -> Option<Arc<str>> {
    match arg {
        crate::parse::ast::Arg::Pos(Expr::Str(s)) => Some(Arc::<str>::from(s.as_str())),
        _ => None,
    }
}

// Extracts a literal value from `arg` for value-membership terminal sinks.
fn literal_arg_value(arg: &crate::parse::ast::Arg) -> Option<Val> {
    match arg {
        crate::parse::ast::Arg::Pos(Expr::Null) => Some(Val::Null),
        crate::parse::ast::Arg::Pos(Expr::Bool(b)) => Some(Val::Bool(*b)),
        crate::parse::ast::Arg::Pos(Expr::Int(n)) => Some(Val::Int(*n)),
        crate::parse::ast::Arg::Pos(Expr::Float(n)) => Some(Val::Float(*n)),
        crate::parse::ast::Arg::Pos(Expr::Str(s)) => Some(Val::Str(Arc::from(s.as_str()))),
        _ => None,
    }
}

// Constructs the terminal `Sink` for `method`, handling count predicates, numeric reducers, and positional selects.
fn terminal_sink_for_method(
    method: BuiltinMethod,
    args: &[crate::parse::ast::Arg],
) -> Option<Sink> {
    if let Some(sink) = predicate_sink_for_method(method, args) {
        return Some(sink);
    }
    if let Some(sink) = membership_sink_for_method(method, args) {
        return Some(sink);
    }
    if let Some(sink) = arg_extreme_sink_for_method(method, args) {
        return Some(sink);
    }
    let spec = method.spec();
    match spec.sink?.accumulator {
        BuiltinSinkAccumulator::ApproxDistinct if args.is_empty() => {
            Some(Sink::ApproxCountDistinct)
        }
        BuiltinSinkAccumulator::Count => match args {
            [] => Some(Sink::Reducer(ReducerSpec::count())),
            [arg] if method == BuiltinMethod::Count => Some(Sink::Reducer(
                ReducerSpec::count_with_predicate(compile_subexpr(arg)?, arg_expr(arg)),
            )),
            _ => None,
        },
        BuiltinSinkAccumulator::Numeric => {
            let projection = match args {
                [] => None,
                [arg] => Some(compile_subexpr(arg)?),
                _ => return None,
            };
            let projection_expr = match args {
                [] => None,
                [arg] => arg_expr(arg),
                _ => return None,
            };
            Some(Sink::Reducer(ReducerSpec::numeric(
                method,
                projection,
                projection_expr,
            )?))
        }
        BuiltinSinkAccumulator::SelectOne(position) => match args {
            [] => Some(Sink::Terminal(method)),
            [arg] => Some(Sink::SelectMany {
                n: usize_arg_at_least(arg, 1)?,
                from_end: matches!(position, BuiltinSelectionPosition::Last),
            }),
            _ => None,
        },
        _ => None,
    }
}

fn predicate_sink_for_method(
    method: BuiltinMethod,
    args: &[crate::parse::ast::Arg],
) -> Option<Sink> {
    let [arg] = args else {
        return None;
    };
    Some(Sink::Predicate(PredicateSinkSpec::from_method(
        method,
        compile_subexpr(arg)?,
    )?))
}

fn membership_sink_for_method(
    method: BuiltinMethod,
    args: &[crate::parse::ast::Arg],
) -> Option<Sink> {
    let [arg] = args else {
        return None;
    };
    Some(Sink::Membership(MembershipSinkSpec::from_method(
        method,
        literal_arg_value(arg)
            .map(MembershipSinkTarget::Literal)
            .or_else(|| Some(MembershipSinkTarget::Program(compile_raw_arg_expr(arg)?)))?,
    )?))
}

fn arg_extreme_sink_for_method(
    method: BuiltinMethod,
    args: &[crate::parse::ast::Arg],
) -> Option<Sink> {
    let [arg] = args else {
        return None;
    };
    Some(Sink::ArgExtreme(ArgExtremeSinkSpec::from_method(
        method,
        compile_subexpr(arg)?,
    )?))
}
