rialo_subscriber_interface/

instruction.rs

// Copyright (c) Subzero Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

//! Instructions for the Subscriber program.

use std::ops::{Range, RangeInclusive};

use rialo_limits::MAX_STATIC_ACCOUNTS_PER_PACKET;
use rialo_s_instruction::{AccountMeta, Instruction};
use rialo_s_program::system_program;
use rialo_s_pubkey::Pubkey;
use rialo_types::Nonce;

use crate::error::SubscriptionError;

/// Seed used to generate the PDA containing the subscription data. The seed for the PDA should be
/// `RIALO_SUBSCRIBE_SEED` + `subscriber_key` + `subscriber_nonce`.
pub const RIALO_SUBSCRIBE_SEED: &str = "rialo_subscribe";

/// Event log prefix for unsubscribe operations.
///
/// Logs with this prefix are filtered by the `BankMatcher` to remove active subscriptions.
pub const RIALO_UNSUBSCRIBE_SEED: &str = "rialo_unsubscribe";

/// Timestamp range, in milliseconds.
pub type TimestampRange = (u64, u64);

/// Type representing a commit index on Rialo. In Rialo there are currently 3 representations of
/// a block height:
/// - Round: A round is a representation of a block height in the consensus layer. Each round
///   contains a fixed number of blocks. A round is identified by a `u32` index.
/// - Commit index: A commit index is a representation of a block height in the execution layer.
///   Each commit index corresponds to a DAG that is an output from consensus. A commit index is identified
///   by a `u32` index.
/// - Slot: A slot is a representation of a block height in the Solana layer. Each slot corresponds to a block in
///   the Solana layer. A slot is identified by a `u32` index.
///
/// On Rialo we use commit indexes to represent block heights in the execution layer. This makes the value
/// for a Commit Index and a Slot the same, but they represent different concepts. The `Commit` type
/// used here is leverage at execution time around the Virtual Machine, thus it follows the `u64` type
/// used in Solana for slots.
// TODO: Update on naming sanitized for block height at execution time, along with types
pub type Commmit = u64;

/// Subscription kind
#[derive(Clone, Debug, PartialEq, Eq, serde_derive::Deserialize, serde_derive::Serialize)]
pub enum SubscriptionKind {
    /// Persistent subscription which fires every time a matching event is received
    Persistent,
    /// One-shot subscription which fires a single time when a matching event is received
    OneShot,
}

/// Subscriber program instructions
#[derive(Clone, Debug, PartialEq, Eq, serde_derive::Deserialize, serde_derive::Serialize)]
pub enum SubscriberInstruction {
    /// Subscribe to events
    Subscribe {
        /// Nonce to differentiate handlers for the same topic
        nonce: Nonce,
        /// [`Subscription`] data
        handler: Subscription,
    },
    /// Delete subscription
    Unsubscribe {
        /// Nonce to differentiate handlers for the same topic/signer pair
        nonce: Nonce,
    },
    /// Update subscription
    Update {
        /// Nonce used during subscription to differentiate handlers for the same topic/signer pair
        nonce: Nonce,
        /// New [`Subscription`] data
        handler: Subscription,
    },
    /// Destroy subscription account.
    ///
    /// The instruction transfers funds from the subscription account to the subscriber,
    /// marking the subscription account as removable.
    ///
    /// This instruction does not emit any logs.
    Destroy {
        /// Nonce to differentiate handlers for the same topic/signer pair
        nonce: Nonce,
    },
}

/// Represents the predicates set by a subscriber to match the [`Subscription`] to an event.
#[derive(Clone, Debug, PartialEq, Eq, serde_derive::Deserialize, serde_derive::Serialize)]
pub struct Predicate {
    /// The topic the subscription listens to
    topic: String,
    /// The account that stores the event data. Used when
    /// the subscription precisely targets a specific event account.
    event_account: Option<Pubkey>,
    /// The timestamp range to trigger the transaction, if provided.
    /// Left boundary is inclusive, right boundary is exclusive.
    timestamp_range: Option<TimestampRange>,
}

impl Predicate {
    /// Instantiate a new `Predicate` with a topic.
    ///
    /// # Arguments
    ///
    /// * `topic` - The topic the subscription listens to.
    ///
    /// # Returns
    ///
    /// * `Predicate` - A new instance of `Predicate`
    pub fn new(topic: impl Into<String>) -> Self {
        Self {
            topic: topic.into(),
            event_account: None,
            timestamp_range: None,
        }
    }

    /// Create a new predicate with a topic and event account.
    ///
    /// # Arguments
    ///
    /// * `topic` - The topic the subscription listens to.
    /// * `event_account` - The account that stores the event data.
    ///
    /// # Returns
    ///
    /// * `Predicate` - A new instance of `Predicate`
    pub fn new_with_event_account(topic: impl Into<String>, event_account: Pubkey) -> Self {
        Self {
            topic: topic.into(),
            event_account: Some(event_account),
            timestamp_range: None,
        }
    }

    /// Create a new predicate with a topic and optional event account.
    ///
    /// # Arguments
    ///
    /// * `topic` - The topic the subscription listens to.
    /// * `event_account` - An optional account that stores the event data.
    ///
    /// # Returns
    ///
    /// * `Predicate` - A new instance of `Predicate`
    pub fn new_with_optional_event_account(
        topic: impl Into<String>,
        event_account: Option<Pubkey>,
    ) -> Self {
        Self {
            topic: topic.into(),
            event_account,
            timestamp_range: None,
        }
    }

    /// Create a new predicate for a timestamp range.
    ///
    /// The topic is set to the clock topic and the event account is set to the clock
    /// sysvar account.
    ///
    /// # Arguments
    ///
    /// * `range` - A range representing the timestamp range in milliseconds. The start is inclusive,
    ///   and the end is exclusive.
    ///
    /// # Returns
    ///
    /// * `Predicate` - A new instance of `Predicate`
    pub fn new_with_timestamp_range(range: Range<u64>) -> Self {
        Self {
            topic: rialo_events_core::types::CLOCK_TOPIC.to_string(),
            event_account: Some(rialo_s_program::sysvar::clock::id()),
            timestamp_range: Some((range.start, range.end)),
        }
    }

    /// Get the topic of the predicate.
    pub fn topic(&self) -> &str {
        &self.topic
    }

    /// Get the event account of the predicate, if any.
    pub fn event_account(&self) -> Option<Pubkey> {
        self.event_account
    }

    /// Get the timestamp range of the predicate, if any.
    pub fn timestamp_range(&self) -> Option<TimestampRange> {
        self.timestamp_range
    }
}

/// [`Subscription`] data stored by a user for a given topic. This is used to generate scheduled
/// transactions to be run if the topic is triggered.
///
/// The instructions are a set of SVM instructions that are meant to be run when the subscription is
/// triggered. The instruction array is converted to a `SanitizedTransaction` and executed in the VM.
#[derive(Clone, Debug, PartialEq, Eq, serde_derive::Deserialize, serde_derive::Serialize)]
pub struct Subscription {
    /// Subscriber key
    pub subscriber: Pubkey,
    /// The predicate set by the subscriber to match the event
    pub predicate: Predicate,
    /// Instructions to be executed
    pub instructions: Vec<Instruction>,
    /// Subscription kind
    pub kind: SubscriptionKind,
    /// Active commits for the subscription
    pub active_commits: RangeInclusive<Commmit>,
}

impl Subscription {
    /// Instantiate a new `Subscription`
    ///
    /// # Arguments
    /// * `subscriber` - The public key of the subscriber
    /// * `predicate` -  The predicate to match the event
    /// * `instructions` - A vector of instructions to be executed
    /// * `kind` - The type of subscription (persistent or one-shot)
    /// * `active_commits` - The range of active commits for the subscription
    ///
    /// # Returns
    /// * `Subscription` - A new instance of `Subscription`
    pub fn new(
        subscriber: Pubkey,
        predicate: Predicate,
        instructions: Vec<Instruction>,
        kind: SubscriptionKind,
        active_commits: RangeInclusive<u64>,
    ) -> Self {
        Self {
            subscriber,
            predicate,
            instructions,
            kind,
            active_commits,
        }
    }

    /// Sanitize a Subscription
    ///
    /// # Returns
    ///
    /// Result<(), SubscriptionError> - Ok if the subscription is valid, Err otherwise
    pub fn sanitize_instructions(&self) -> Result<(), SubscriptionError> {
        if self.instructions.is_empty() {
            return Err(SubscriptionError::EmptyInstructions);
        }

        // Check that no account that is not the payer is a signer
        if let Some(invalid_signer) = self
            .instructions
            .iter()
            .flat_map(|itx| itx.accounts.iter())
            .find(|account| account.is_signer && account.pubkey != self.subscriber)
        {
            return Err(SubscriptionError::InvalidSigner(invalid_signer.pubkey));
        }

        // Check that we are limiting the  number of accounts across all instructions to MAX_STATIC_ACCOUNTS_PER_PACKET
        let all_pubkeys: std::collections::HashSet<_> = self
            .instructions
            .iter()
            .flat_map(|itx| {
                std::iter::once(itx.program_id).chain(itx.accounts.iter().map(|acc| acc.pubkey))
            })
            .collect();
        let total_accounts = all_pubkeys.len();

        if total_accounts > MAX_STATIC_ACCOUNTS_PER_PACKET as usize {
            return Err(SubscriptionError::TooManyAccounts {
                actual: total_accounts,
                max_accounts: MAX_STATIC_ACCOUNTS_PER_PACKET as usize,
            });
        }

        Ok(())
    }
}

impl SubscriberInstruction {
    /// Create a `SubscriberInstruction::SubscribeToEvent` `Instruction`
    ///
    /// # Account references
    ///   0. `[SIGNER]` Signer (subscriber) account
    ///   1. `[WRITE]` Subscription data account
    ///   2. `[READONLY]` System program account
    ///
    /// # Arguments
    /// * `signer` - The public key of the subscriber
    /// * `nonce` - The nonce to differentiate handlers for the same topic/signer pair
    /// * `handler` - The [`Subscription`] data
    ///
    /// # Returns
    /// * `Instruction` - An SVM instruction to be included in a transaction
    pub fn subscribe_to(signer: Pubkey, nonce: Nonce, mut handler: Subscription) -> Instruction {
        // todo better error handling
        if handler.subscriber != signer {
            panic!("Handler subscriber must be the same as the signer");
        }

        if handler.instructions.is_empty() {
            panic!("Handler instructions must not be empty");
        }

        let subscription_data_account = derive_subscription_address(signer, nonce);

        // append a destroy instruction to one-shot subscription's instructions
        //
        // when the one-shot subscription is matched (`BankMatcher::generate_subscription_matches()`),
        // it is removed from the matcher but the subscription is not automatically removed from the state
        //
        // add destroy instruction as the subscription's last instructions, meaning when all the user-specified
        // instructions have been executed, remove the subscription from the state and return rent to the subscriber
        if let SubscriptionKind::OneShot = handler.kind {
            handler.instructions.push(Instruction::new_with_bincode(
                crate::ID,
                &SubscriberInstruction::Destroy { nonce },
                vec![
                    AccountMeta::new(signer, true),
                    AccountMeta::new(subscription_data_account, false),
                ],
            ));
        }

        Instruction::new_with_bincode(
            crate::ID,
            &SubscriberInstruction::Subscribe { nonce, handler },
            vec![
                AccountMeta::new(signer, true),
                AccountMeta::new(subscription_data_account, false),
                AccountMeta::new_readonly(system_program::id(), false),
            ],
        )
    }

    /// Create a `SubscriberInstruction::Unsubscribe` `Instruction`
    ///
    /// # Account references
    ///   0. `[SIGNER]` Signer (subscriber) account
    ///   1. `[WRITE]`  Subscription data account
    ///
    /// # Arguments
    /// * `signer` - The public key of the subscriber
    /// * `nonce` - The nonce to differentiate handlers for the same topic/signer pair
    ///
    /// # Returns
    /// * `Instruction` - An SVM instruction to be included in a transaction
    pub fn unsubscribe_from(signer: Pubkey, nonce: Nonce) -> Instruction {
        let subscription_data_key = derive_subscription_address(signer, nonce);

        Instruction::new_with_bincode(
            crate::ID,
            &SubscriberInstruction::Unsubscribe { nonce },
            vec![
                AccountMeta::new(signer, true),
                AccountMeta::new(subscription_data_key, false),
            ],
        )
    }

    /// Create a `SubscriberInstruction::Update` `Instruction`.
    ///
    /// # Account references
    ///   0. `[SIGNER]` Signer (subscriber) account
    ///   1. `[WRITE]`  Subscription data account
    ///   2. `[READONLY]` System program account
    ///
    /// # Arguments
    /// * `signer` - The public key of the subscriber
    /// * `nonce` - The nonce set during subscription to differentiate handlers for the same topic/signer pair
    /// * `handler` - The [`Subscription`] data
    ///
    /// # Returns
    /// * `Instruction` - An SVM instruction to be included in a transaction
    pub fn update(signer: Pubkey, nonce: Nonce, mut handler: Subscription) -> Instruction {
        let subscription_data_key = derive_subscription_address(signer, nonce);

        // append a destroy instruction to one-shot subscription's instructions
        //
        // when the one-shot subscription is matched (`BankMatcher::generate_subscription_matches()`),
        // it is removed from the matcher but the subscription is not automatically removed from the state
        //
        // add destroy instruction as the subscription's last instructions, meaning when all the user-specified
        // instructions have been executed, remove the subscription from the state and return rent to the subscriber
        if let SubscriptionKind::OneShot = handler.kind {
            handler.instructions.push(Instruction::new_with_bincode(
                crate::ID,
                &SubscriberInstruction::Destroy { nonce },
                vec![
                    AccountMeta::new_readonly(signer, true),
                    AccountMeta::new(subscription_data_key, false),
                ],
            ));
        }

        Instruction::new_with_bincode(
            crate::ID,
            &SubscriberInstruction::Update { nonce, handler },
            vec![
                AccountMeta::new(signer, true),
                AccountMeta::new(subscription_data_key, false),
                AccountMeta::new_readonly(system_program::id(), false),
            ],
        )
    }
}

/// Generates the program-derived address (PDA) for the subscription data account.
///
/// # Arguments
///
/// * `subscriber` - The public key of the subscriber.
/// * `nonce` - A slice of bytes used as a seed to generate the PDA.
///
/// # Returns
///
/// The public key of the subscription data account.
pub fn derive_subscription_address<NONCE: Into<Nonce>>(subscriber: Pubkey, nonce: NONCE) -> Pubkey {
    let nonce = nonce.into();
    Pubkey::find_program_address(
        &[
            RIALO_SUBSCRIBE_SEED.as_bytes(),
            subscriber.as_array(),
            nonce.as_bytes(),
        ],
        &crate::ID,
    )
    .0
}

#[cfg(test)]
mod test {
    use super::*;

    #[test]
    fn test_subscribe_to_instruction() {
        let signer = Pubkey::new_unique();
        let handler = Subscription::new(
            signer,
            Predicate {
                topic: String::from("MySubscription"),
                event_account: Some(Pubkey::new_unique()),
                timestamp_range: Some((1337, 1338)),
            },
            vec![Instruction::new_with_bincode(
                Pubkey::new_unique(),
                &["MyData"],
                vec![],
            )],
            SubscriptionKind::Persistent,
            0..=u64::MAX,
        );

        let instruction =
            SubscriberInstruction::subscribe_to(signer, Nonce::default(), handler.clone());

        assert_eq!(
            instruction.data,
            bincode::serialize(&SubscriberInstruction::Subscribe {
                nonce: Nonce::default(),
                handler,
            })
            .unwrap()
        );
        assert_eq!(instruction.program_id, crate::ID);

        let subscription_data_account = derive_subscription_address(signer, Nonce::default());
        assert_eq!(
            instruction.accounts,
            vec![
                AccountMeta::new(signer, true),
                AccountMeta::new(subscription_data_account, false),
                AccountMeta::new_readonly(system_program::id(), false),
            ]
        )
    }

    #[test]
    fn test_delete_subscription_instruction() {
        let signer = Pubkey::new_unique();

        let instruction = SubscriberInstruction::unsubscribe_from(signer, Nonce::default());

        assert_eq!(
            instruction.data,
            bincode::serialize(&SubscriberInstruction::Unsubscribe {
                nonce: Nonce::default()
            })
            .unwrap()
        );
        assert_eq!(instruction.program_id, crate::ID);

        let subscription_data_account = derive_subscription_address(signer, Nonce::default());

        assert_eq!(
            instruction.accounts,
            vec![
                AccountMeta::new(signer, true),
                AccountMeta::new(subscription_data_account, false),
            ]
        )
    }

    #[test]
    fn test_update_subscription_instruction() {
        let signer = Pubkey::new_unique();
        let handler = Subscription {
            subscriber: signer,
            predicate: Predicate {
                topic: String::from("MySubscription"),
                event_account: Some(Pubkey::new_unique()),
                timestamp_range: Some((1337, 1338)),
            },
            instructions: vec![Instruction::new_with_bincode(
                Pubkey::new_unique(),
                &["MyData"],
                vec![],
            )],
            kind: SubscriptionKind::Persistent,
            active_commits: 0..=u64::MAX,
        };

        let instruction = SubscriberInstruction::update(signer, Nonce::default(), handler.clone());

        assert_eq!(
            instruction.data,
            bincode::serialize(&SubscriberInstruction::Update {
                nonce: Nonce::default(),
                handler
            })
            .unwrap()
        );
        assert_eq!(instruction.program_id, crate::ID);

        let subscription_data_account = derive_subscription_address(signer, Nonce::default());

        assert_eq!(
            instruction.accounts,
            vec![
                AccountMeta::new(signer, true),
                AccountMeta::new(subscription_data_account, false),
                AccountMeta::new_readonly(system_program::id(), false),
            ]
        )
    }

    #[test]
    fn test_one_shot() {
        let signer = Pubkey::new_unique();
        let mut handler = Subscription::new(
            signer,
            Predicate {
                topic: String::from("MySubscription"),
                event_account: Some(Pubkey::new_unique()),
                timestamp_range: None,
            },
            vec![Instruction::new_with_bincode(
                Pubkey::new_unique(),
                &["MyData"],
                vec![],
            )],
            SubscriptionKind::OneShot,
            0..=u64::MAX,
        );

        let instruction =
            SubscriberInstruction::subscribe_to(signer, Nonce::default(), handler.clone());

        handler.instructions.push(Instruction::new_with_bincode(
            crate::ID,
            &SubscriberInstruction::Destroy {
                nonce: Nonce::default(),
            },
            vec![
                AccountMeta::new(signer, true),
                AccountMeta::new(derive_subscription_address(signer, Nonce::default()), false),
            ],
        ));

        assert_eq!(
            instruction.data,
            bincode::serialize(&SubscriberInstruction::Subscribe {
                nonce: Nonce::default(),
                handler,
            })
            .unwrap()
        );
        assert_eq!(instruction.program_id, crate::ID);

        let subscription_data_account = derive_subscription_address(signer, Nonce::default());
        assert_eq!(
            instruction.accounts,
            vec![
                AccountMeta::new(signer, true),
                AccountMeta::new(subscription_data_account, false),
                AccountMeta::new_readonly(system_program::id(), false),
            ]
        )
    }

    #[test]
    fn test_one_shot_with_timestamp_range() {
        let signer = Pubkey::new_unique();
        let mut handler = Subscription::new(
            signer,
            Predicate::new_with_timestamp_range(1337..1338),
            vec![Instruction::new_with_bincode(
                Pubkey::new_unique(),
                &["MyData"],
                vec![],
            )],
            SubscriptionKind::OneShot,
            0..=u64::MAX,
        );

        let instruction =
            SubscriberInstruction::subscribe_to(signer, Nonce::default(), handler.clone());

        handler.instructions.push(Instruction::new_with_bincode(
            crate::ID,
            &SubscriberInstruction::Destroy {
                nonce: Nonce::default(),
            },
            vec![
                AccountMeta::new(signer, true),
                AccountMeta::new(derive_subscription_address(signer, Nonce::default()), false),
            ],
        ));

        assert_eq!(
            instruction.data,
            bincode::serialize(&SubscriberInstruction::Subscribe {
                nonce: Nonce::default(),
                handler,
            })
            .unwrap()
        );
        assert_eq!(instruction.program_id, crate::ID);

        let subscription_data_account = derive_subscription_address(signer, Nonce::default());
        assert_eq!(
            instruction.accounts,
            vec![
                AccountMeta::new(signer, true),
                AccountMeta::new(subscription_data_account, false),
                AccountMeta::new_readonly(system_program::id(), false),
            ]
        )
    }
}