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
"#;
#[derive(Debug, Error)]
pub enum CliError {
#[error("{0}\n\nRun with --help for usage.")]
Usage(String),
#[error("could not start icp: {0}")]
StartIcp(#[source] io::Error),
#[error("icp call failed{status}: {message}")]
Icp {
status: String,
message: String,
},
#[error("could not encode arguments for `{method}`: {source}")]
Encode {
method: &'static str,
#[source]
source: candid::Error,
},
#[error("could not decode the typed reply from `{method}`: {source}")]
Decode {
method: &'static str,
#[source]
source: candid::Error,
},
#[error("icp returned an invalid raw reply: {0}")]
RawReply(&'static str),
#[error("canister method `{method}` returned {error}")]
Canister {
method: &'static str,
error: String,
},
#[error("could not render JSON output: {0}")]
Json(#[from] serde_json::Error),
#[error("could not write output: {0}")]
Io(#[source] io::Error),
#[error("pagination cursor did not advance: {0}")]
PaginationStalled(String),
#[error("listing still had another page after the --max-pages limit of {0}")]
PaginationLimit(usize),
}
impl CliError {
#[must_use]
pub const fn exit_code(&self) -> i32 {
if matches!(self, Self::Usage(_)) { 2 } else { 1 }
}
#[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>),
}
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
));
}
}