flow_lib/command/
mod.rs

1//! [`CommandTrait`] and command [`builder`].
2//!
3//! To make a new [`native`][crate::config::CommandType::Native] command:
4//! 1. Implement [`CommandTrait`], 2 ways;
5//!     - Manually implement it to your types.
6//!     - Use [`builder`] helper.
7//! 2. Use [`inventory::submit`] with a [`CommandDescription`] to register the command at compile-time.
8
9use crate::{
10    CommandType, ValueType,
11    config::{
12        CmdInputDescription, CmdOutputDescription, Name, ValueSet,
13        client::{self, NodeData},
14        node::Permissions,
15    },
16    context::CommandContext,
17};
18use futures::future::{Either, LocalBoxFuture, OptionFuture};
19use regex::Regex;
20use serde::{Deserialize, Serialize};
21use std::{borrow::Cow, collections::BTreeMap, future::ready};
22use uuid::Uuid;
23use value::Value;
24
25pub mod builder;
26
27/// Import common types for writing commands.
28pub mod prelude {
29    pub use crate::{
30        CmdInputDescription, CmdInputDescription as Input, CmdOutputDescription,
31        CmdOutputDescription as Output, FlowId, Name, ValueSet, ValueType,
32        command::{
33            CommandDescription, CommandError, CommandTrait, InstructionInfo,
34            builder::{BuildResult, BuilderCache, BuilderError, CmdBuilder},
35        },
36        config::{client::NodeData, node::Permissions},
37        context::CommandContext,
38        solana::Instructions,
39    };
40    pub use async_trait::async_trait;
41    pub use bytes::Bytes;
42    pub use futures::future::Either;
43    pub use serde::{Deserialize, Serialize};
44    pub use serde_json::Value as JsonValue;
45    pub use serde_with::serde_as;
46    pub use solana_keypair::Keypair;
47    pub use solana_pubkey::Pubkey;
48    pub use solana_signature::Signature;
49    pub use thiserror::Error as ThisError;
50    pub use value::{
51        self, Decimal, Value,
52        with::{AsDecimal, AsKeypair, AsPubkey, AsSignature},
53    };
54}
55
56/// Error type of commmands.
57pub type CommandError = anyhow::Error;
58
59/// Generic trait for implementing commands.
60#[async_trait::async_trait(?Send)]
61pub trait CommandTrait: 'static {
62    fn r#type(&self) -> CommandType {
63        CommandType::Native
64    }
65
66    /// Unique name to identify the command.
67    fn name(&self) -> Name;
68
69    /// List of inputs that the command can receive.
70    fn inputs(&self) -> Vec<CmdInputDescription>;
71
72    /// List of outputs that the command will return.
73    fn outputs(&self) -> Vec<CmdOutputDescription>;
74
75    /// Run the command.
76    async fn run(&self, ctx: CommandContext, params: ValueSet) -> Result<ValueSet, CommandError>;
77
78    /// Specify if and how would this command output Solana instructions.
79    fn instruction_info(&self) -> Option<InstructionInfo> {
80        None
81    }
82
83    /// Specify requested permissions of this command.
84    fn permissions(&self) -> Permissions {
85        Permissions::default()
86    }
87
88    /// Async `Drop` method.
89    async fn destroy(&mut self) {}
90
91    /// Specify how [`form_data`][crate::config::NodeConfig::form_data] are read.
92    fn read_form_data(&self, data: serde_json::Value) -> ValueSet {
93        let mut result = ValueSet::new();
94        for i in self.inputs() {
95            if let Some(json) = data.get(&i.name) {
96                let value = Value::from(json.clone());
97                result.insert(i.name.clone(), value);
98            }
99        }
100        result
101    }
102
103    fn node_data(&self) -> NodeData {
104        default_node_data(self)
105    }
106}
107
108/// Specify how to convert inputs into passthrough outputs.
109pub fn passthrough_outputs<T: CommandTrait + ?Sized>(cmd: &T, inputs: &ValueSet) -> ValueSet {
110    let mut res = ValueSet::new();
111    for i in cmd.inputs() {
112        if i.passthrough
113            && let Some(value) = inputs.get(&i.name)
114        {
115            if !i.required && matches!(value, Value::Null) {
116                continue;
117            }
118
119            let value = match i.type_bounds.first() {
120                Some(ValueType::Pubkey) => {
121                    // keypair could be automatically converted into pubkey
122                    // we don't want to passthrough the keypair here, only pubkey
123                    value::pubkey::deserialize(value.clone()).map(Into::into)
124                }
125                _ => Ok(value.clone()),
126            }
127            .unwrap_or_else(|error| {
128                tracing::warn!("error reading passthrough: {}", error);
129                value.clone()
130            });
131            res.insert(i.name, value);
132        }
133    }
134    res
135}
136
137/// Specify how [`form_data`][crate::config::NodeConfig::form_data] are read.
138pub fn default_read_form_data<T: CommandTrait + ?Sized>(
139    cmd: &T,
140    data: serde_json::Value,
141) -> ValueSet {
142    let mut result = ValueSet::new();
143    for i in cmd.inputs() {
144        if let Some(json) = data.get(&i.name) {
145            let value = Value::from(json.clone());
146            result.insert(i.name.clone(), value);
147        }
148    }
149    result
150}
151
152pub fn default_node_data<T: CommandTrait + ?Sized>(cmd: &T) -> NodeData {
153    NodeData {
154        r#type: cmd.r#type(),
155        node_id: cmd.name(),
156        sources: cmd
157            .outputs()
158            .into_iter()
159            .map(|output| client::Source {
160                id: Uuid::nil(),
161                name: output.name,
162                r#type: output.r#type,
163                optional: output.optional,
164            })
165            .collect(),
166        targets: cmd
167            .inputs()
168            .into_iter()
169            .map(|input| client::Target {
170                id: Uuid::nil(),
171                name: input.name,
172                type_bounds: input.type_bounds,
173                required: input.required,
174                passthrough: input.passthrough,
175            })
176            .collect(),
177        targets_form: client::TargetsForm::default(),
178        instruction_info: cmd.instruction_info(),
179    }
180}
181
182pub fn input_is_required<T: CommandTrait + ?Sized>(cmd: &T, name: &str) -> Option<bool> {
183    cmd.inputs()
184        .into_iter()
185        .find_map(|i| (i.name == name).then_some(i.required))
186}
187
188pub fn output_is_optional<T: CommandTrait + ?Sized>(cmd: &T, name: &str) -> Option<bool> {
189    cmd.outputs()
190        .into_iter()
191        .find_map(|o| (o.name == name).then_some(o.optional))
192        .or_else(|| {
193            cmd.inputs()
194                .into_iter()
195                .find_map(|i| (i.name == name && i.passthrough).then_some(!i.required))
196        })
197}
198
199pub fn keypair_outputs<T: CommandTrait + ?Sized>(cmd: &T) -> Vec<String> {
200    cmd.outputs()
201        .iter()
202        .filter(|&o| o.r#type == ValueType::Keypair)
203        .map(|o| o.name.clone())
204        .chain(cmd.inputs().iter().find_map(|i| {
205            i.type_bounds
206                .contains(&ValueType::Keypair)
207                .then(|| i.name.clone())
208        }))
209        .collect()
210}
211
212/// Specify the order with which a command will return its output:
213/// - [`before`][InstructionInfo::before]: list of output names returned before instructions are sent.
214/// - [`signature`][InstructionInfo::signature]: name of the signature output port.
215/// - [`after`][InstructionInfo::after]: list of output names returned after instructions are sent.
216#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
217pub struct InstructionInfo {
218    pub before: Vec<Name>,
219    pub signature: Name,
220    pub after: Vec<Name>,
221}
222
223impl InstructionInfo {
224    /// Simple `InstructionInfo` that can describe most commands:
225    /// - [`before`][InstructionInfo::before]: All passthroughs and outputs, except for `signature`.
226    /// - [`after`][InstructionInfo::after]: empty.
227    pub fn simple<C: CommandTrait>(cmd: &C, signature: &str) -> Self {
228        let before = cmd
229            .inputs()
230            .into_iter()
231            .filter(|i| i.passthrough)
232            .map(|i| i.name)
233            .chain(
234                cmd.outputs()
235                    .into_iter()
236                    .filter(|o| o.name != signature)
237                    .map(|o| o.name),
238            )
239            .collect();
240        Self {
241            before,
242            after: Vec::new(),
243            signature: signature.into(),
244        }
245    }
246}
247
248#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, bincode::Encode)]
249pub enum MatchName {
250    Exact(Cow<'static, str>),
251    Regex(Cow<'static, str>),
252}
253
254// generated by bincode macro
255impl<__Context> ::bincode::Decode<__Context> for MatchName {
256    fn decode<__D: ::bincode::de::Decoder<Context = __Context>>(
257        decoder: &mut __D,
258    ) -> core::result::Result<Self, ::bincode::error::DecodeError> {
259        let variant_index = <u32 as ::bincode::Decode<__D::Context>>::decode(decoder)?;
260        match variant_index {
261            0u32 => core::result::Result::Ok(Self::Exact(
262                ::bincode::Decode::<__D::Context>::decode(decoder)?,
263            )),
264            1u32 => core::result::Result::Ok(Self::Regex(
265                ::bincode::Decode::<__D::Context>::decode(decoder)?,
266            )),
267            variant => {
268                core::result::Result::Err(::bincode::error::DecodeError::UnexpectedVariant {
269                    found: variant,
270                    type_name: "MatchName",
271                    allowed: &::bincode::error::AllowedEnumVariants::Range { min: 0, max: 1 },
272                })
273            }
274        }
275    }
276}
277
278impl<'de, C> bincode::BorrowDecode<'de, C> for MatchName {
279    fn borrow_decode<D: bincode::de::BorrowDecoder<'de, Context = C>>(
280        decoder: &mut D,
281    ) -> Result<Self, bincode::error::DecodeError> {
282        bincode::Decode::decode(decoder)
283    }
284}
285
286impl std::fmt::Debug for MatchName {
287    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
288        match self {
289            Self::Exact(arg0) => arg0.fmt(f),
290            Self::Regex(arg0) => {
291                f.write_str("/")?;
292                f.write_str(arg0)?;
293                f.write_str("/")
294            }
295        }
296    }
297}
298
299impl std::fmt::Display for MatchName {
300    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
301        match self {
302            MatchName::Exact(cow) => cow.fmt(f),
303            MatchName::Regex(cow) => cow.fmt(f),
304        }
305    }
306}
307
308#[derive(Clone, bincode::Encode, bincode::Decode, PartialEq, Eq, PartialOrd, Ord)]
309pub struct MatchCommand {
310    pub r#type: CommandType,
311    pub name: MatchName,
312}
313
314impl std::fmt::Debug for MatchCommand {
315    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
316        self.r#type.fmt(f)?;
317        f.write_str(":")?;
318        self.name.fmt(f)
319    }
320}
321
322impl MatchCommand {
323    pub fn is_match(&self, ty: CommandType, name: &str) -> bool {
324        self.r#type == ty
325            && match &self.name {
326                MatchName::Exact(cow) => cow == name,
327                MatchName::Regex(cow) => Regex::new(cow) // TODO: slow
328                    .map(|re| re.is_match(name))
329                    .ok()
330                    .unwrap_or(false),
331            }
332    }
333}
334
335pub type FnNew =
336    Either<fn(&NodeData) -> FnNewResult, fn(&NodeData) -> LocalBoxFuture<'static, FnNewResult>>;
337
338pub type FnNewResult = Result<Box<dyn CommandTrait>, CommandError>;
339
340/// Use [`inventory::submit`] to register commands at compile-time.
341#[derive(Clone)]
342pub struct CommandDescription {
343    pub matcher: MatchCommand,
344    /// Function to initialize the command from a [`NodeData`].
345    pub fn_new: FnNew,
346}
347
348impl CommandDescription {
349    pub const fn new(name: &'static str, fn_new: fn(&NodeData) -> FnNewResult) -> Self {
350        Self {
351            matcher: MatchCommand {
352                r#type: CommandType::Native,
353                name: MatchName::Exact(Cow::Borrowed(name)),
354            },
355            fn_new: Either::Left(fn_new),
356        }
357    }
358}
359
360inventory::collect!(CommandDescription);
361
362pub fn collect_commands() -> BTreeMap<&'static MatchCommand, &'static CommandDescription> {
363    inventory::iter::<CommandDescription>()
364        .map(|c| (&c.matcher, c))
365        .collect()
366}
367
368#[derive(Debug, Clone)]
369pub struct CommandIndex<T> {
370    pub exact_match: BTreeMap<(CommandType, Cow<'static, str>), T>,
371    pub regex: Vec<(CommandType, regex::Regex, T)>,
372}
373
374impl<T> Default for CommandIndex<T> {
375    fn default() -> Self {
376        Self {
377            exact_match: <_>::default(),
378            regex: <_>::default(),
379        }
380    }
381}
382
383impl<T> FromIterator<(MatchCommand, T)> for CommandIndex<T> {
384    fn from_iter<I: IntoIterator<Item = (MatchCommand, T)>>(iter: I) -> Self {
385        let mut this = Self::default();
386        for (matcher, t) in iter {
387            match &matcher.name {
388                MatchName::Exact(cow) => {
389                    this.exact_match.insert((matcher.r#type, cow.clone()), t);
390                }
391                MatchName::Regex(cow) => {
392                    this.regex
393                        .push((matcher.r#type, Regex::new(cow).expect("invalid regex"), t));
394                }
395            }
396        }
397        this
398    }
399}
400
401impl<T> CommandIndex<T> {
402    pub fn get(&self, ty: CommandType, name: &str) -> Option<&T> {
403        if let Some(d) = self.exact_match.get(&(ty, name.to_owned().into())) {
404            Some(d)
405        } else {
406            let mut matched = None;
407            for r in &self.regex {
408                if r.0 == ty && r.1.is_match(name) {
409                    matched = Some(&r.2);
410                }
411            }
412            matched
413        }
414    }
415
416    pub fn availables(&self) -> impl Iterator<Item = MatchCommand> {
417        self.exact_match
418            .keys()
419            .cloned()
420            .map(|(r#type, name)| MatchCommand {
421                r#type,
422                name: MatchName::Exact(name),
423            })
424            .chain(self.regex.iter().map(|(ty, regex, _)| MatchCommand {
425                r#type: *ty,
426                name: MatchName::Regex(regex.to_string().into()),
427            }))
428    }
429}
430
431#[derive(Clone)]
432pub struct CommandFactory {
433    index: CommandIndex<&'static CommandDescription>,
434}
435
436impl CommandFactory {
437    pub fn collect() -> Self {
438        Self {
439            index: inventory::iter::<CommandDescription>()
440                .map(|c| (c.matcher.clone(), c))
441                .collect(),
442        }
443    }
444
445    pub fn init(
446        &self,
447        nd: &NodeData,
448    ) -> impl Future<Output = Result<Option<Box<dyn CommandTrait>>, CommandError>> + 'static {
449        let cmd = self.index.get(nd.r#type, &nd.node_id);
450
451        let either = cmd.map(|cmd| match cmd.fn_new {
452            Either::Left(fn_new) => Either::Left(ready(fn_new(nd))),
453            Either::Right(async_fn_new) => Either::Right(async_fn_new(nd)),
454        });
455        async move { OptionFuture::from(either).await.transpose() }
456    }
457
458    pub fn availables(&self) -> impl Iterator<Item = MatchCommand> {
459        self.index.availables()
460    }
461}
flow_lib/command/mod.rs

flow_lib/command/
mod.rs
