toko_feed_cli/

lib.rs

//! Installed operator CLI for a Toko Feed canister.
//!
//! The process boundary delegates identity and network handling to `icp`, but
//! encodes and decodes the canister protocol with shared Rust types. Users do
//! not need to write Candid arguments or interpret Candid values.

use std::{
    collections::{HashSet, VecDeque},
    io::{self, Write},
    path::PathBuf,
    process::Command,
};

use candid::{CandidType, decode_one, encode_args};
use serde::{Serialize, de::DeserializeOwned};
use serde_json::{Value, json};
use thiserror::Error;
use toko_feed::{
    CollectionDetails, CollectionIngestReceipt, CollectionPage, CuratePokemonSetArgs, FeedError,
    FeedStatus, IngestReceipt, IngestionRunPage, LockPokemonSetArgs, OperationalLogPage,
    PokemonCardDetails, PokemonCardPage, PokemonSetDetails, PokemonSetPage, PokemonSetView,
    PriceObservationPage, SchedulerStatus, SetIngestReceipt, SetLifecycleStatus, TimeCursor,
};

const DEFAULT_LIMIT: u16 = 20;
const DEFAULT_MAX_PAGES: usize = 1_000;
const MAX_RAW_REPLY_BYTES: usize = 16 * 1024 * 1024;

const HELP: &str = r#"toko-feed

Operate and query a Toko Feed canister without writing Candid arguments.

Usage:
  toko-feed [GLOBAL OPTIONS] status
  toko-feed [GLOBAL OPTIONS] scheduler
  toko-feed [GLOBAL OPTIONS] runs list [HISTORY OPTIONS]
  toko-feed [GLOBAL OPTIONS] logs list [HISTORY OPTIONS]
  toko-feed [GLOBAL OPTIONS] collections ingest
  toko-feed [GLOBAL OPTIONS] collections list [LIST OPTIONS]
  toko-feed [GLOBAL OPTIONS] collections get <ULID>
  toko-feed [GLOBAL OPTIONS] sets ingest
  toko-feed [GLOBAL OPTIONS] sets list [LIST OPTIONS]
  toko-feed [GLOBAL OPTIONS] sets get <ULID>
  toko-feed [GLOBAL OPTIONS] sets curate <ULID> --revision <N> --name <NAME> (--release-date <DATE> | --no-release-date) [--active | --retired]
  toko-feed [GLOBAL OPTIONS] sets lock <ULID> --revision <N>
  toko-feed [GLOBAL OPTIONS] cards ingest
  toko-feed [GLOBAL OPTIONS] cards list [LIST OPTIONS]
  toko-feed [GLOBAL OPTIONS] cards get <ULID>
  toko-feed [GLOBAL OPTIONS] cards prices <ULID> [HISTORY OPTIONS]

Global options:
  --environment <NAME>          ICP environment (default: local)
  --canister <NAME|ID>          Canister name or principal (default: toko-feed)
  --identity <NAME>             ICP identity used for the call (default: anonymous)
  --identity-password-file <PATH>
                                Read an encrypted identity password from a file
  --project-root <PATH>         Override ICP project discovery
  --icp <PATH>                  ICP CLI executable (default: icp)
  --compact                     Print compact JSON instead of pretty JSON
  -h, --help                    Print this help
  -V, --version                 Print the CLI version

List options:
  --limit <1..100>              Records requested per canister call (default: 20)
  --after <ULID>                Start strictly after this local ID
  --all                         Follow next_after until the listing is complete
  --max-pages <COUNT>           Safety bound for --all (default: 1000)

History options:
  --limit <1..100>              Records requested per canister call (default: 20)
  --before-time <INTEGER>       Timestamp from a returned next_before cursor
  --before-id <ULID>            ULID from the same next_before cursor
  --all                         Follow next_before until the listing is complete
  --max-pages <COUNT>           Safety bound for --all (default: 1000)

Examples:
  toko-feed status
  toko-feed scheduler
  toko-feed runs list --limit 100
  toko-feed logs list --all
  toko-feed collections ingest
  toko-feed collections list --all
  toko-feed sets list --limit 10
  toko-feed sets list --limit 100 --all
  toko-feed sets list --after 01KZ9GFKW3SY1G000000000001
  toko-feed sets get 01KZ9GFKW3SY1G000000000001
  toko-feed --identity operator sets curate 01KZ9GFKW3SY1G000000000001 --revision 0 --name "Base Set" --release-date 1999-01-09
  toko-feed --identity operator sets lock 01KZ9GFKW3SY1G000000000001 --revision 1
  toko-feed sets ingest
  toko-feed cards ingest
  toko-feed cards list --limit 100 --all
  toko-feed cards prices 01KZ9GFKW3SY1G000000000001 --limit 100
"#;

/// Failure reported by CLI parsing, the ICP process boundary, or the typed
/// canister protocol.
#[derive(Debug, Error)]
pub enum CliError {
    /// Command-line arguments did not describe a valid operation.
    #[error("{0}\n\nRun with --help for usage.")]
    Usage(String),
    /// The ICP CLI process could not be started.
    #[error("could not start icp: {0}")]
    StartIcp(#[source] io::Error),
    /// The ICP CLI process reported a failed call.
    #[error("icp call failed{status}: {message}")]
    Icp {
        /// Numeric process exit status, when the platform supplied one.
        status: String,
        /// Bounded diagnostic written by the ICP CLI.
        message: String,
    },
    /// Typed Candid argument encoding failed before the process was started.
    #[error("could not encode arguments for `{method}`: {source}")]
    Encode {
        /// Canister method being prepared.
        method: &'static str,
        /// Candid codec failure.
        #[source]
        source: candid::Error,
    },
    /// The canister reply was not valid for the shared interface type.
    #[error("could not decode the typed reply from `{method}`: {source}")]
    Decode {
        /// Canister method whose reply was being decoded.
        method: &'static str,
        /// Candid codec failure.
        #[source]
        source: candid::Error,
    },
    /// The ICP CLI returned something other than one bounded hexadecimal reply.
    #[error("icp returned an invalid raw reply: {0}")]
    RawReply(&'static str),
    /// A typed application error was returned by the canister.
    #[error("canister method `{method}` returned {error}")]
    Canister {
        /// Called canister method.
        method: &'static str,
        /// Credential-free bounded application error.
        error: String,
    },
    /// JSON rendering failed.
    #[error("could not render JSON output: {0}")]
    Json(#[from] serde_json::Error),
    /// Writing the final report failed.
    #[error("could not write output: {0}")]
    Io(#[source] io::Error),
    /// A canister returned a continuation that had already been observed.
    #[error("pagination cursor did not advance: {0}")]
    PaginationStalled(String),
    /// Automatic pagination reached its operator-provided safety bound.
    #[error("listing still had another page after the --max-pages limit of {0}")]
    PaginationLimit(usize),
}

impl CliError {
    /// Shell exit status for this failure category.
    #[must_use]
    pub const fn exit_code(&self) -> i32 {
        if matches!(self, Self::Usage(_)) { 2 } else { 1 }
    }

    /// Whether a downstream pipeline closed standard output intentionally.
    #[must_use]
    pub fn is_broken_pipe(&self) -> bool {
        matches!(self, Self::Io(error) if error.kind() == io::ErrorKind::BrokenPipe)
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct Target {
    environment: String,
    canister: String,
    identity: Option<String>,
    identity_password_file: Option<PathBuf>,
    project_root: Option<PathBuf>,
    icp: PathBuf,
    compact: bool,
}

impl Default for Target {
    fn default() -> Self {
        Self {
            environment: "local".to_owned(),
            canister: "toko-feed".to_owned(),
            identity: Some("anonymous".to_owned()),
            identity_password_file: None,
            project_root: None,
            icp: PathBuf::from("icp"),
            compact: false,
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum Resource {
    Collections,
    Sets,
    Cards,
}

impl Resource {
    const fn collection_field(self) -> &'static str {
        match self {
            Self::Collections => "collections",
            Self::Sets => "sets",
            Self::Cards => "cards",
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct ListOptions {
    limit: u16,
    after: Option<String>,
    all: bool,
    max_pages: usize,
}

impl Default for ListOptions {
    fn default() -> Self {
        Self {
            limit: DEFAULT_LIMIT,
            after: None,
            all: false,
            max_pages: DEFAULT_MAX_PAGES,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct HistoryOptions {
    limit: u16,
    before: Option<TimeCursor>,
    all: bool,
    max_pages: usize,
}

impl Default for HistoryOptions {
    fn default() -> Self {
        Self {
            limit: DEFAULT_LIMIT,
            before: None,
            all: false,
            max_pages: DEFAULT_MAX_PAGES,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
enum Action {
    Status,
    Scheduler,
    Runs(HistoryOptions),
    Logs(HistoryOptions),
    Ingest(Resource),
    List(Resource, ListOptions),
    Get(Resource, String),
    CurateSet(CuratePokemonSetArgs),
    LockSet(LockPokemonSetArgs),
    Prices(String, HistoryOptions),
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct Invocation {
    target: Target,
    action: Action,
}

enum Parsed {
    Help,
    Version,
    Run(Box<Invocation>),
}

/// Run the CLI from the current process arguments.
///
/// # Errors
///
/// Returns a bounded [`CliError`] when arguments are invalid, the `icp`
/// process fails, the canister rejects the request, or output cannot be
/// rendered.
pub fn run_from_env() -> Result<(), CliError> {
    let arguments = std::env::args().skip(1).collect::<Vec<_>>();
    run(&arguments)
}

fn run(arguments: &[String]) -> Result<(), CliError> {
    let invocation = match parse_arguments(arguments)? {
        Parsed::Help => return write_text(HELP),
        Parsed::Version => {
            return write_text(&format!("toko-feed {}\n", env!("CARGO_PKG_VERSION")));
        }
        Parsed::Run(invocation) => invocation,
    };

    let output = execute(&invocation)?;
    let rendered = if invocation.target.compact {
        serde_json::to_string(&output)?
    } else {
        serde_json::to_string_pretty(&output)?
    };
    write_text(&format!("{rendered}\n"))
}

fn write_text(text: &str) -> Result<(), CliError> {
    io::stdout()
        .lock()
        .write_all(text.as_bytes())
        .map_err(CliError::Io)
}

fn parse_arguments(arguments: &[String]) -> Result<Parsed, CliError> {
    let mut arguments = arguments
        .iter()
        .map(String::as_str)
        .collect::<VecDeque<_>>();
    if arguments.is_empty() {
        return Ok(Parsed::Help);
    }
    if arguments
        .iter()
        .any(|argument| matches!(*argument, "-h" | "--help"))
    {
        return Ok(Parsed::Help);
    }

    let mut target = Target::default();
    loop {
        match arguments.front().copied() {
            Some("-V" | "--version") => return Ok(Parsed::Version),
            Some("--environment") => {
                arguments.pop_front();
                take_value(&mut arguments, "--environment")?.clone_into(&mut target.environment);
            }
            Some("--canister") => {
                arguments.pop_front();
                take_value(&mut arguments, "--canister")?.clone_into(&mut target.canister);
            }
            Some("--identity") => {
                arguments.pop_front();
                target.identity = Some(take_value(&mut arguments, "--identity")?.to_owned());
            }
            Some("--identity-password-file") => {
                arguments.pop_front();
                target.identity_password_file = Some(PathBuf::from(take_value(
                    &mut arguments,
                    "--identity-password-file",
                )?));
            }
            Some("--project-root") => {
                arguments.pop_front();
                target.project_root =
                    Some(PathBuf::from(take_value(&mut arguments, "--project-root")?));
            }
            Some("--icp") => {
                arguments.pop_front();
                target.icp = PathBuf::from(take_value(&mut arguments, "--icp")?);
            }
            Some("--compact") => {
                arguments.pop_front();
                target.compact = true;
            }
            Some(option) if option.starts_with('-') => {
                return Err(CliError::Usage(format!(
                    "unknown global option `{option}`; global options must precede the command"
                )));
            }
            _ => break,
        }
    }

    let command = arguments
        .pop_front()
        .ok_or_else(|| CliError::Usage("missing command".to_owned()))?;
    let action = match command {
        "status" => {
            reject_extra(&arguments, "status")?;
            Action::Status
        }
        "scheduler" => {
            reject_extra(&arguments, "scheduler")?;
            Action::Scheduler
        }
        "runs" => parse_history_action("runs", &mut arguments, Action::Runs)?,
        "logs" => parse_history_action("logs", &mut arguments, Action::Logs)?,
        "collections" => parse_resource_action(Resource::Collections, &mut arguments)?,
        "sets" => parse_resource_action(Resource::Sets, &mut arguments)?,
        "cards" => parse_resource_action(Resource::Cards, &mut arguments)?,
        unknown => return Err(CliError::Usage(format!("unknown command `{unknown}`"))),
    };

    Ok(Parsed::Run(Box::new(Invocation { target, action })))
}

fn parse_resource_action(
    resource: Resource,
    arguments: &mut VecDeque<&str>,
) -> Result<Action, CliError> {
    let resource_name = resource.collection_field();
    let subcommand = arguments
        .pop_front()
        .ok_or_else(|| CliError::Usage(format!("missing {resource_name} subcommand")))?;

    match subcommand {
        "ingest" => {
            reject_extra(arguments, &format!("{resource_name} ingest"))?;
            Ok(Action::Ingest(resource))
        }
        "get" => {
            let id = arguments
                .pop_front()
                .ok_or_else(|| CliError::Usage(format!("{resource_name} get requires a ULID")))?;
            validate_local_id(id)?;
            reject_extra(arguments, &format!("{resource_name} get"))?;
            Ok(Action::Get(resource, id.to_owned()))
        }
        "prices" if resource == Resource::Cards => {
            let id = arguments
                .pop_front()
                .ok_or_else(|| CliError::Usage("cards prices requires a ULID".to_owned()))?;
            validate_local_id(id)?;
            Ok(Action::Prices(
                id.to_owned(),
                parse_history_options(arguments)?,
            ))
        }
        "curate" if resource == Resource::Sets => {
            let id = arguments
                .pop_front()
                .ok_or_else(|| CliError::Usage("sets curate requires a ULID".to_owned()))?;
            validate_local_id(id)?;
            parse_set_curate(id, arguments).map(Action::CurateSet)
        }
        "lock" if resource == Resource::Sets => {
            let id = arguments
                .pop_front()
                .ok_or_else(|| CliError::Usage("sets lock requires a ULID".to_owned()))?;
            validate_local_id(id)?;
            parse_set_lock(id, arguments).map(Action::LockSet)
        }
        "list" => parse_list_options(resource, arguments),
        unknown => Err(CliError::Usage(format!(
            "unknown {resource_name} subcommand `{unknown}`"
        ))),
    }
}

fn parse_set_curate(
    id: &str,
    arguments: &mut VecDeque<&str>,
) -> Result<CuratePokemonSetArgs, CliError> {
    let mut expected_revision = None;
    let mut name = None;
    let mut release_date = None;
    let mut release_date_selected = false;
    let mut lifecycle_status = SetLifecycleStatus::Active;
    let mut lifecycle_selected = false;
    while let Some(option) = arguments.pop_front() {
        match option {
            "--revision" => {
                if expected_revision.is_some() {
                    return Err(CliError::Usage(
                        "--revision may be supplied once".to_owned(),
                    ));
                }
                expected_revision = Some(parse_revision(take_value(arguments, "--revision")?)?);
            }
            "--name" => {
                if name.is_some() {
                    return Err(CliError::Usage("--name may be supplied once".to_owned()));
                }
                name = Some(take_value(arguments, "--name")?.to_owned());
            }
            "--release-date" => {
                if release_date_selected {
                    return Err(CliError::Usage(
                        "choose exactly one of --release-date or --no-release-date".to_owned(),
                    ));
                }
                release_date = Some(take_value(arguments, "--release-date")?.to_owned());
                release_date_selected = true;
            }
            "--no-release-date" => {
                if release_date_selected {
                    return Err(CliError::Usage(
                        "choose exactly one of --release-date or --no-release-date".to_owned(),
                    ));
                }
                release_date_selected = true;
            }
            "--active" | "--retired" => {
                if lifecycle_selected {
                    return Err(CliError::Usage(
                        "choose at most one of --active or --retired".to_owned(),
                    ));
                }
                lifecycle_status = if option == "--retired" {
                    SetLifecycleStatus::Retired
                } else {
                    SetLifecycleStatus::Active
                };
                lifecycle_selected = true;
            }
            unknown => {
                return Err(CliError::Usage(format!(
                    "unknown sets curate option `{unknown}`"
                )));
            }
        }
    }
    if !release_date_selected {
        return Err(CliError::Usage(
            "sets curate requires --release-date or --no-release-date".to_owned(),
        ));
    }
    Ok(CuratePokemonSetArgs {
        id: id.to_owned(),
        expected_revision: expected_revision
            .ok_or_else(|| CliError::Usage("sets curate requires --revision".to_owned()))?,
        name: name.ok_or_else(|| CliError::Usage("sets curate requires --name".to_owned()))?,
        release_date,
        lifecycle_status,
    })
}

fn parse_set_lock(
    id: &str,
    arguments: &mut VecDeque<&str>,
) -> Result<LockPokemonSetArgs, CliError> {
    if arguments.pop_front() != Some("--revision") {
        return Err(CliError::Usage(
            "sets lock requires --revision <N>".to_owned(),
        ));
    }
    let expected_revision = parse_revision(take_value(arguments, "--revision")?)?;
    reject_extra(arguments, "sets lock")?;
    Ok(LockPokemonSetArgs {
        id: id.to_owned(),
        expected_revision,
    })
}

fn parse_revision(value: &str) -> Result<u64, CliError> {
    value
        .parse()
        .map_err(|_| CliError::Usage("--revision must be an unsigned integer".to_owned()))
}

fn parse_history_action(
    resource: &str,
    arguments: &mut VecDeque<&str>,
    build: impl FnOnce(HistoryOptions) -> Action,
) -> Result<Action, CliError> {
    let subcommand = arguments
        .pop_front()
        .ok_or_else(|| CliError::Usage(format!("missing {resource} subcommand")))?;
    if subcommand != "list" {
        return Err(CliError::Usage(format!(
            "unknown {resource} subcommand `{subcommand}`"
        )));
    }
    parse_history_options(arguments).map(build)
}

fn parse_history_options(arguments: &mut VecDeque<&str>) -> Result<HistoryOptions, CliError> {
    let mut options = HistoryOptions::default();
    let mut before_time = None;
    let mut before_id = None;
    while let Some(option) = arguments.pop_front() {
        match option {
            "--limit" => {
                let value = take_value(arguments, "--limit")?;
                options.limit = parse_page_limit(value)?;
            }
            "--before-time" => {
                let value = take_value(arguments, "--before-time")?;
                before_time = Some(value.parse::<u64>().map_err(|_| {
                    CliError::Usage("--before-time must be an unsigned integer".into())
                })?);
            }
            "--before-id" => {
                let value = take_value(arguments, "--before-id")?;
                validate_local_id(value)?;
                before_id = Some(value.to_owned());
            }
            "--all" => options.all = true,
            "--max-pages" => {
                let value = take_value(arguments, "--max-pages")?;
                options.max_pages = parse_max_pages(value)?;
            }
            unknown => {
                return Err(CliError::Usage(format!(
                    "unknown history option `{unknown}`"
                )));
            }
        }
    }
    options.before = match (before_time, before_id) {
        (Some(timestamp), Some(id)) => Some(TimeCursor { timestamp, id }),
        (None, None) => None,
        _ => {
            return Err(CliError::Usage(
                "--before-time and --before-id must be supplied together".into(),
            ));
        }
    };
    if !options.all && options.max_pages != DEFAULT_MAX_PAGES {
        return Err(CliError::Usage(
            "--max-pages is only meaningful together with --all".into(),
        ));
    }
    Ok(options)
}

fn parse_list_options(
    resource: Resource,
    arguments: &mut VecDeque<&str>,
) -> Result<Action, CliError> {
    let mut options = ListOptions::default();

    while let Some(option) = arguments.pop_front() {
        match option {
            "--limit" => {
                let value = take_value(arguments, "--limit")?;
                options.limit = parse_page_limit(value)?;
            }
            "--after" => {
                let value = take_value(arguments, "--after")?;
                validate_local_id(value)?;
                options.after = Some(value.to_owned());
            }
            "--all" => options.all = true,
            "--max-pages" => {
                let value = take_value(arguments, "--max-pages")?;
                options.max_pages = parse_max_pages(value)?;
            }
            unknown => return Err(CliError::Usage(format!("unknown list option `{unknown}`"))),
        }
    }

    if !options.all && options.max_pages != DEFAULT_MAX_PAGES {
        return Err(CliError::Usage(
            "--max-pages is only meaningful together with --all".into(),
        ));
    }

    Ok(Action::List(resource, options))
}

fn parse_page_limit(value: &str) -> Result<u16, CliError> {
    value
        .parse::<u16>()
        .ok()
        .filter(|limit| (1..=100).contains(limit))
        .ok_or_else(|| CliError::Usage("--limit must be an integer from 1 to 100".into()))
}

fn parse_max_pages(value: &str) -> Result<usize, CliError> {
    value
        .parse::<usize>()
        .ok()
        .filter(|maximum| *maximum > 0)
        .ok_or_else(|| CliError::Usage("--max-pages must be a positive integer".into()))
}

fn take_value<'a>(arguments: &mut VecDeque<&'a str>, option: &str) -> Result<&'a str, CliError> {
    arguments
        .pop_front()
        .filter(|value| !value.is_empty() && !value.starts_with('-'))
        .ok_or_else(|| CliError::Usage(format!("{option} requires a value")))
}

fn reject_extra(arguments: &VecDeque<&str>, command: &str) -> Result<(), CliError> {
    if let Some(extra) = arguments.front() {
        return Err(CliError::Usage(format!(
            "unexpected argument `{extra}` after `{command}`"
        )));
    }
    Ok(())
}

fn validate_local_id(id: &str) -> Result<(), CliError> {
    if is_valid_local_id(id) {
        Ok(())
    } else {
        Err(CliError::Usage(format!(
            "`{id}` is not a 26-character uppercase ULID"
        )))
    }
}

fn is_valid_local_id(id: &str) -> bool {
    id.len() == 26
        && id.bytes().all(|byte| {
            matches!(
                byte,
                b'0'..=b'9' | b'A'..=b'H' | b'J'..=b'K' | b'M'..=b'N' | b'P'..=b'T' | b'V'..=b'Z'
            )
        })
}

fn execute(invocation: &Invocation) -> Result<Value, CliError> {
    match &invocation.action {
        Action::Status => output(call_empty::<FeedStatus>(
            &invocation.target,
            "toko_feed_status",
            true,
        )?),
        Action::Scheduler => output(call_empty::<SchedulerStatus>(
            &invocation.target,
            "toko_feed_scheduler",
            true,
        )?),
        Action::Runs(options) if options.all => list_all_runs(&invocation.target, options),
        Action::Logs(options) if options.all => list_all_logs(&invocation.target, options),
        Action::Prices(card_id, options) if options.all => {
            list_all_prices(&invocation.target, card_id, options)
        }
        Action::Runs(options) => output(fetch_run_page(
            &invocation.target,
            options.before.clone(),
            options.limit,
        )?),
        Action::Logs(options) => output(fetch_log_page(
            &invocation.target,
            options.before.clone(),
            options.limit,
        )?),
        Action::Prices(card_id, options) => output(fetch_price_page(
            &invocation.target,
            card_id.clone(),
            options.before.clone(),
            options.limit,
        )?),
        Action::Ingest(Resource::Collections) => output(call_empty::<CollectionIngestReceipt>(
            &invocation.target,
            "toko_feed_ingest_collections",
            false,
        )?),
        Action::Ingest(Resource::Sets) => output(call_empty::<SetIngestReceipt>(
            &invocation.target,
            "toko_feed_ingest_sets",
            false,
        )?),
        Action::Get(Resource::Collections, id) => output(call_one::<_, Option<CollectionDetails>>(
            &invocation.target,
            "toko_feed_collection",
            id.clone(),
            true,
        )?),
        Action::Ingest(Resource::Cards) => output(call_empty::<IngestReceipt>(
            &invocation.target,
            "toko_feed_ingest",
            false,
        )?),
        Action::Get(Resource::Sets, id) => output(call_one::<_, Option<PokemonSetDetails>>(
            &invocation.target,
            "toko_feed_set",
            id.clone(),
            true,
        )?),
        Action::CurateSet(args) => set_mutation(&invocation.target, "toko_feed_curate_set", args),
        Action::LockSet(args) => set_mutation(&invocation.target, "toko_feed_lock_set", args),
        Action::Get(Resource::Cards, id) => output(call_one::<_, Option<PokemonCardDetails>>(
            &invocation.target,
            "toko_feed_card",
            id.clone(),
            true,
        )?),
        Action::List(Resource::Sets, options) if options.all => {
            list_all_sets(&invocation.target, options)
        }
        Action::List(Resource::Collections, options) if options.all => {
            list_all_collections(&invocation.target, options)
        }
        Action::List(Resource::Cards, options) if options.all => {
            list_all_cards(&invocation.target, options)
        }
        Action::List(Resource::Sets, options) => output(fetch_set_page(
            &invocation.target,
            options.after.clone(),
            options.limit,
        )?),
        Action::List(Resource::Collections, options) => output(fetch_collection_page(
            &invocation.target,
            options.after.clone(),
            options.limit,
        )?),
        Action::List(Resource::Cards, options) => output(fetch_card_page(
            &invocation.target,
            options.after.clone(),
            options.limit,
        )?),
    }
}

fn output(value: impl Serialize) -> Result<Value, CliError> {
    serde_json::to_value(value).map_err(CliError::Json)
}

fn set_mutation<A>(target: &Target, method: &'static str, args: &A) -> Result<Value, CliError>
where
    A: CandidType + Clone,
{
    output(call_one::<_, PokemonSetView>(
        target,
        method,
        args.clone(),
        false,
    )?)
}

fn call_empty<T>(target: &Target, method: &'static str, query: bool) -> Result<T, CliError>
where
    T: CandidType + DeserializeOwned,
{
    let arguments = encode_args(()).map_err(|source| CliError::Encode { method, source })?;
    call_and_decode(target, method, &arguments, query)
}

fn call_one<A, T>(
    target: &Target,
    method: &'static str,
    argument: A,
    query: bool,
) -> Result<T, CliError>
where
    A: CandidType,
    T: CandidType + DeserializeOwned,
{
    let arguments =
        encode_args((argument,)).map_err(|source| CliError::Encode { method, source })?;
    call_and_decode(target, method, &arguments, query)
}

fn fetch_set_page(
    target: &Target,
    after: Option<String>,
    limit: u16,
) -> Result<PokemonSetPage, CliError> {
    call_page(target, "toko_feed_sets", after, limit)
}

fn fetch_collection_page(
    target: &Target,
    after: Option<String>,
    limit: u16,
) -> Result<CollectionPage, CliError> {
    call_page(target, "toko_feed_collections", after, limit)
}

fn fetch_card_page(
    target: &Target,
    after: Option<String>,
    limit: u16,
) -> Result<PokemonCardPage, CliError> {
    call_page(target, "toko_feed_cards", after, limit)
}

fn fetch_run_page(
    target: &Target,
    before: Option<TimeCursor>,
    limit: u16,
) -> Result<IngestionRunPage, CliError> {
    call_history_page(target, "toko_feed_runs", before, limit)
}

fn fetch_log_page(
    target: &Target,
    before: Option<TimeCursor>,
    limit: u16,
) -> Result<OperationalLogPage, CliError> {
    call_history_page(target, "toko_feed_logs", before, limit)
}

fn fetch_price_page(
    target: &Target,
    card_id: String,
    before: Option<TimeCursor>,
    limit: u16,
) -> Result<PriceObservationPage, CliError> {
    let method = "toko_feed_card_prices";
    let arguments = encode_args((card_id, before, limit))
        .map_err(|source| CliError::Encode { method, source })?;
    call_and_decode(target, method, &arguments, true)
}

fn call_history_page<T>(
    target: &Target,
    method: &'static str,
    before: Option<TimeCursor>,
    limit: u16,
) -> Result<T, CliError>
where
    T: CandidType + DeserializeOwned,
{
    let arguments =
        encode_args((before, limit)).map_err(|source| CliError::Encode { method, source })?;
    call_and_decode(target, method, &arguments, true)
}

fn call_page<T>(
    target: &Target,
    method: &'static str,
    after: Option<String>,
    limit: u16,
) -> Result<T, CliError>
where
    T: CandidType + DeserializeOwned,
{
    let arguments =
        encode_args((after, limit)).map_err(|source| CliError::Encode { method, source })?;
    call_and_decode(target, method, &arguments, true)
}

fn call_and_decode<T>(
    target: &Target,
    method: &'static str,
    arguments: &[u8],
    query: bool,
) -> Result<T, CliError>
where
    T: CandidType + DeserializeOwned,
{
    let reply = call_raw(target, method, arguments, query)?;
    decode_result(method, &reply)
}

fn decode_result<T>(method: &'static str, reply: &[u8]) -> Result<T, CliError>
where
    T: CandidType + DeserializeOwned,
{
    let result = decode_one::<Result<T, FeedError>>(reply)
        .map_err(|source| CliError::Decode { method, source })?;
    result.map_err(|error| CliError::Canister {
        method,
        error: format!("{error:?}"),
    })
}

fn call_raw(
    target: &Target,
    method: &'static str,
    arguments: &[u8],
    query: bool,
) -> Result<Vec<u8>, CliError> {
    let mut command = Command::new(&target.icp);
    if let Some(project_root) = &target.project_root {
        command.arg("--project-root-override").arg(project_root);
    }
    if let Some(password_file) = &target.identity_password_file {
        command.arg("--identity-password-file").arg(password_file);
    }
    command
        .args(["canister", "call", "--environment"])
        .arg(&target.environment)
        .args(["--args-format", "hex", "--output", "hex"]);
    if let Some(identity) = &target.identity {
        command.args(["--identity", identity]);
    }
    if query {
        command.arg("--query");
    }
    command
        .args([&target.canister, method])
        .arg(encode_hex(arguments));

    let output = command.output().map_err(CliError::StartIcp)?;
    if !output.status.success() {
        let status = output
            .status
            .code()
            .map_or_else(String::new, |code| format!(" (exit {code})"));
        let message = bounded_diagnostic(&output.stderr);
        return Err(CliError::Icp { status, message });
    }
    if output.stdout.len() > MAX_RAW_REPLY_BYTES * 2 + 2 {
        return Err(CliError::RawReply("hexadecimal response exceeded 16 MiB"));
    }
    decode_hex(&output.stdout)
}

fn bounded_diagnostic(bytes: &[u8]) -> String {
    const MAX_DIAGNOSTIC_BYTES: usize = 8 * 1024;
    let visible = &bytes[..bytes.len().min(MAX_DIAGNOSTIC_BYTES)];
    let mut message = String::from_utf8_lossy(visible).trim().to_owned();
    if bytes.len() > MAX_DIAGNOSTIC_BYTES {
        message.push_str("…[truncated]");
    }
    if message.is_empty() {
        "no diagnostic was written to stderr".to_owned()
    } else {
        message
    }
}

fn encode_hex(bytes: &[u8]) -> String {
    const DIGITS: &[u8; 16] = b"0123456789abcdef";
    let mut output = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        output.push(char::from(DIGITS[usize::from(byte >> 4)]));
        output.push(char::from(DIGITS[usize::from(byte & 0x0f)]));
    }
    output
}

fn decode_hex(input: &[u8]) -> Result<Vec<u8>, CliError> {
    let input = std::str::from_utf8(input)
        .map_err(|_| CliError::RawReply("response was not UTF-8 hexadecimal text"))?
        .trim();
    let input = input.strip_prefix("0x").unwrap_or(input);
    if input.len() % 2 != 0 {
        return Err(CliError::RawReply(
            "hexadecimal response had an odd number of digits",
        ));
    }

    input
        .as_bytes()
        .chunks_exact(2)
        .map(|pair| {
            let high = hex_digit(pair[0])?;
            let low = hex_digit(pair[1])?;
            Ok((high << 4) | low)
        })
        .collect()
}

const fn hex_digit(byte: u8) -> Result<u8, CliError> {
    match byte {
        b'0'..=b'9' => Ok(byte - b'0'),
        b'a'..=b'f' => Ok(byte - b'a' + 10),
        b'A'..=b'F' => Ok(byte - b'A' + 10),
        _ => Err(CliError::RawReply(
            "response contained a non-hexadecimal character",
        )),
    }
}

fn list_all_runs(target: &Target, options: &HistoryOptions) -> Result<Value, CliError> {
    let (runs, pages) = collect_history_pages(options, |before, limit| {
        let page = fetch_run_page(target, before, limit)?;
        Ok((page.runs, page.next_before))
    })?;
    Ok(json!({
        "runs": runs,
        "count": runs.len(),
        "pages": pages,
        "next_before": null,
    }))
}

fn list_all_logs(target: &Target, options: &HistoryOptions) -> Result<Value, CliError> {
    let (logs, pages) = collect_history_pages(options, |before, limit| {
        let page = fetch_log_page(target, before, limit)?;
        Ok((page.logs, page.next_before))
    })?;
    Ok(json!({
        "logs": logs,
        "count": logs.len(),
        "pages": pages,
        "next_before": null,
    }))
}

fn list_all_prices(
    target: &Target,
    card_id: &str,
    options: &HistoryOptions,
) -> Result<Value, CliError> {
    let (observations, pages) = collect_history_pages(options, |before, limit| {
        let page = fetch_price_page(target, card_id.to_owned(), before, limit)?;
        Ok((page.observations, page.next_before))
    })?;
    Ok(json!({
        "observations": observations,
        "count": observations.len(),
        "pages": pages,
        "next_before": null,
    }))
}

fn collect_history_pages<T>(
    options: &HistoryOptions,
    mut fetch: impl FnMut(Option<TimeCursor>, u16) -> Result<(Vec<T>, Option<TimeCursor>), CliError>,
) -> Result<(Vec<T>, usize), CliError> {
    let mut before = options.before.clone();
    let mut seen = before
        .as_ref()
        .map(time_cursor_token)
        .into_iter()
        .collect::<HashSet<_>>();
    let mut items = Vec::new();
    let mut pages = 0usize;
    loop {
        enforce_page_budget(pages, options.max_pages)?;
        let (page_items, next_before) = fetch(before, options.limit)?;
        pages += 1;
        items.extend(page_items);
        let Some(next_before) = next_before else {
            return Ok((items, pages));
        };
        validate_time_cursor(&next_before, &mut seen)?;
        before = Some(next_before);
    }
}

const fn enforce_page_budget(pages: usize, maximum: usize) -> Result<(), CliError> {
    if pages == maximum {
        Err(CliError::PaginationLimit(maximum))
    } else {
        Ok(())
    }
}

fn validate_time_cursor(cursor: &TimeCursor, seen: &mut HashSet<String>) -> Result<(), CliError> {
    if !is_valid_local_id(&cursor.id) {
        return Err(CliError::RawReply(
            "next_before contained an invalid local ULID",
        ));
    }
    let token = time_cursor_token(cursor);
    if !seen.insert(token.clone()) {
        return Err(CliError::PaginationStalled(token));
    }
    Ok(())
}

fn time_cursor_token(cursor: &TimeCursor) -> String {
    format!("{}:{}", cursor.timestamp, cursor.id)
}

fn list_all_sets(target: &Target, options: &ListOptions) -> Result<Value, CliError> {
    let (sets, pages) = collect_keyset_pages(options, |after, limit| {
        let page = fetch_set_page(target, after, limit)?;
        Ok((page.sets, page.next_after))
    })?;
    Ok(json!({
        "sets": sets,
        "count": sets.len(),
        "pages": pages,
        "next_after": null,
    }))
}

fn list_all_collections(target: &Target, options: &ListOptions) -> Result<Value, CliError> {
    let (collections, pages) = collect_keyset_pages(options, |after, limit| {
        let page = fetch_collection_page(target, after, limit)?;
        Ok((page.collections, page.next_after))
    })?;
    Ok(json!({
        "collections": collections,
        "count": collections.len(),
        "pages": pages,
        "next_after": null,
    }))
}

fn list_all_cards(target: &Target, options: &ListOptions) -> Result<Value, CliError> {
    let (cards, pages) = collect_keyset_pages(options, |after, limit| {
        let page = fetch_card_page(target, after, limit)?;
        Ok((page.cards, page.next_after))
    })?;
    Ok(json!({
        "cards": cards,
        "count": cards.len(),
        "pages": pages,
        "next_after": null,
    }))
}

fn collect_keyset_pages<T>(
    options: &ListOptions,
    mut fetch: impl FnMut(Option<String>, u16) -> Result<(Vec<T>, Option<String>), CliError>,
) -> Result<(Vec<T>, usize), CliError> {
    let mut after = options.after.clone();
    let mut seen = after.iter().cloned().collect::<HashSet<_>>();
    let mut items = Vec::new();
    let mut pages = 0usize;
    loop {
        enforce_page_budget(pages, options.max_pages)?;
        let (page_items, next_after) = fetch(after, options.limit)?;
        pages += 1;
        items.extend(page_items);
        let Some(next_after) = next_after else {
            return Ok((items, pages));
        };
        validate_reply_cursor(&next_after, &mut seen)?;
        after = Some(next_after);
    }
}

fn validate_reply_cursor(cursor: &str, seen: &mut HashSet<String>) -> Result<(), CliError> {
    if !is_valid_local_id(cursor) {
        return Err(CliError::RawReply("next_after was not a valid local ULID"));
    }
    if !seen.insert(cursor.to_owned()) {
        return Err(CliError::PaginationStalled(cursor.to_owned()));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use candid::{decode_args, encode_one};

    use super::*;

    fn strings(values: &[&str]) -> Vec<String> {
        values.iter().map(ToString::to_string).collect()
    }

    fn invocation(values: &[&str]) -> Invocation {
        match parse_arguments(&strings(values)).expect("arguments should parse") {
            Parsed::Run(invocation) => *invocation,
            Parsed::Help | Parsed::Version => panic!("expected an invocation"),
        }
    }

    #[test]
    fn parses_global_and_automatic_pagination_options() {
        let parsed = invocation(&[
            "--environment",
            "ic",
            "--canister",
            "aaaaa-aa",
            "--identity",
            "operator",
            "--project-root",
            "/srv/toko-feed",
            "--compact",
            "sets",
            "list",
            "--limit",
            "100",
            "--after",
            "01KZ9GFKW3SY1G000000000001",
            "--all",
            "--max-pages",
            "12",
        ]);

        assert_eq!(parsed.target.environment, "ic");
        assert_eq!(parsed.target.canister, "aaaaa-aa");
        assert_eq!(parsed.target.identity.as_deref(), Some("operator"));
        assert_eq!(
            parsed.target.project_root.as_deref(),
            Some(std::path::Path::new("/srv/toko-feed"))
        );
        assert!(parsed.target.compact);
        assert_eq!(
            parsed.action,
            Action::List(
                Resource::Sets,
                ListOptions {
                    limit: 100,
                    after: Some("01KZ9GFKW3SY1G000000000001".to_owned()),
                    all: true,
                    max_pages: 12,
                }
            )
        );
    }

    #[test]
    fn parses_card_ingest_and_get_commands() {
        assert_eq!(
            invocation(&["cards", "ingest"]).action,
            Action::Ingest(Resource::Cards)
        );
        assert_eq!(
            invocation(&["cards", "get", "01KZ9GFKW3SY1G000000000001"]).action,
            Action::Get(Resource::Cards, "01KZ9GFKW3SY1G000000000001".to_owned())
        );
    }

    #[test]
    fn parses_collection_ingest_and_list_commands() {
        assert_eq!(
            invocation(&["collections", "ingest"]).action,
            Action::Ingest(Resource::Collections)
        );
        assert_eq!(
            invocation(&["collections", "list", "--all"]).action,
            Action::List(
                Resource::Collections,
                ListOptions {
                    all: true,
                    ..ListOptions::default()
                }
            )
        );
    }

    #[test]
    fn parses_revisioned_set_curation_and_lock_commands() {
        let id = "01KZ9GFKW3SY1G000000000001";
        assert_eq!(
            invocation(&[
                "sets",
                "curate",
                id,
                "--revision",
                "7",
                "--name",
                "Base Set",
                "--release-date",
                "1999-01-09",
                "--retired",
            ])
            .action,
            Action::CurateSet(CuratePokemonSetArgs {
                id: id.to_owned(),
                expected_revision: 7,
                name: "Base Set".to_owned(),
                release_date: Some("1999-01-09".to_owned()),
                lifecycle_status: SetLifecycleStatus::Retired,
            })
        );
        assert_eq!(
            invocation(&["sets", "lock", id, "--revision", "8"]).action,
            Action::LockSet(LockPokemonSetArgs {
                id: id.to_owned(),
                expected_revision: 8,
            })
        );
        assert!(
            parse_arguments(&strings(&[
                "sets",
                "curate",
                id,
                "--revision",
                "7",
                "--name",
                "Base Set",
            ]))
            .is_err()
        );
    }

    #[test]
    fn parses_scheduler_history_and_price_commands() {
        assert_eq!(invocation(&["scheduler"]).action, Action::Scheduler);
        assert_eq!(
            invocation(&[
                "runs",
                "list",
                "--before-time",
                "123",
                "--before-id",
                "01KZ9GFKW3SY1G000000000001",
                "--all",
            ])
            .action,
            Action::Runs(HistoryOptions {
                before: Some(TimeCursor {
                    timestamp: 123,
                    id: "01KZ9GFKW3SY1G000000000001".to_owned(),
                }),
                all: true,
                ..HistoryOptions::default()
            })
        );
        assert!(matches!(
            invocation(&[
                "cards",
                "prices",
                "01KZ9GFKW3SY1G000000000001",
                "--limit",
                "100",
            ])
            .action,
            Action::Prices(_, HistoryOptions { limit: 100, .. })
        ));
        assert!(parse_arguments(&strings(&["logs", "list", "--before-time", "123",])).is_err());
    }

    #[test]
    fn rejects_invalid_bounds_and_identifiers() {
        assert!(parse_arguments(&strings(&["sets", "list", "--limit", "0"])).is_err());
        assert!(parse_arguments(&strings(&["sets", "list", "--max-pages", "2"])).is_err());
        assert!(parse_arguments(&strings(&["sets", "get", "not-an-id"])).is_err());
    }

    #[test]
    fn accepts_help_at_any_command_depth_and_version_at_the_top_level() {
        assert!(matches!(
            parse_arguments(&strings(&["sets", "list", "--help"])),
            Ok(Parsed::Help)
        ));
        assert!(matches!(
            parse_arguments(&strings(&["--version"])),
            Ok(Parsed::Version)
        ));
    }

    #[test]
    fn encodes_typed_page_arguments_without_candid_text() {
        let bytes = encode_args((Some("01KZ9GFKW3SY1G000000000001".to_owned()), 100_u16))
            .expect("encode page arguments");
        let decoded = decode_args::<(Option<String>, u16)>(&bytes).expect("decode page arguments");

        assert_eq!(
            decoded,
            (Some("01KZ9GFKW3SY1G000000000001".to_owned()), 100)
        );
    }

    #[test]
    fn hexadecimal_transport_round_trips_raw_candid() {
        let bytes = encode_args((None::<String>, 20_u16)).expect("encode arguments");
        assert_eq!(
            decode_hex(format!("0x{}\n", encode_hex(&bytes)).as_bytes()).expect("decode hex"),
            bytes
        );
    }

    #[test]
    fn decodes_a_typed_canister_result() {
        let status = FeedStatus {
            configured: true,
            next_offset: 50,
            ingesting: false,
            last_error_code: None,
            updated_at_ns: 123,
        };
        let reply = encode_one(Ok::<_, FeedError>(status.clone())).expect("encode reply");

        assert_eq!(
            decode_result::<FeedStatus>("toko_feed_status", &reply).expect("decode result"),
            status
        );
    }

    #[test]
    fn rejects_non_hexadecimal_transport_output() {
        assert!(decode_hex(b"not-hex").is_err());
        assert!(decode_hex(b"abc").is_err());
    }

    #[test]
    fn keyset_page_collection_preserves_order_and_rejects_stalled_cursors() {
        const CURSOR: &str = "01KZ9GFKW3SY1G000000000001";
        let options = ListOptions {
            after: None,
            limit: 2,
            all: true,
            max_pages: 3,
        };
        let (items, pages) = collect_keyset_pages(&options, |after, limit| {
            assert_eq!(limit, 2);
            match after.as_deref() {
                None => Ok((vec![1, 2], Some(CURSOR.to_owned()))),
                Some(CURSOR) => Ok((vec![3], None)),
                Some(_) => Err(CliError::RawReply("unexpected test cursor")),
            }
        })
        .expect("bounded pages should collect");
        assert_eq!(items, vec![1, 2, 3]);
        assert_eq!(pages, 2);

        let stalled = ListOptions {
            after: Some(CURSOR.to_owned()),
            ..options
        };
        assert!(matches!(
            collect_keyset_pages::<u8>(&stalled, |_, _| Ok((vec![1], Some(CURSOR.to_owned())))),
            Err(CliError::PaginationStalled(cursor)) if cursor == CURSOR
        ));
    }
}