Webhooks are important infrastructure pieces when working with any data related activities such as alerts, bots, ingestion flows, etc. Even though Mercury is fully equipped to handle most of the logic you can think of that might be listening backend, it can always come in handy to know how to trigger webhooks when working on Soroban.

In today's blog post, we'll explore how to build and test a comprehensive Soroban monitoring system. This service accomplishes two key tasks simultaneously: it provides an API for setting up custom webhook triggers, while also processing real-time events from Stellar Core. The system continuously evaluates incoming events against registered webhooks to determine when notifications should be triggered.

Mercury as a Low-Latency Low-Level Data Layer

Did you know that you can use Mercury's lower level components to build data services on top of them?

Part of the goal for this blog post is to shine light on the fact that Mercury can (and should!) be used also by other higher-level data platforms working on soroban:

Mercury's runtimes (ZVM and rSVM/retorshades SVM) are both open source and are built to simplify development workflows without compromising on latency and maintaining low-level access capabilities.
You have plenty of functionality to build with and customize interactions in terms of DBs (use native Mercury DB or write to your own through web requests), data workflow (which data you are analyzing and how you want to process it is up to you).

This is to say that even if you're looking to build data-related services, using Mercury is the quickest path towards launch.

Creating the Mercury project.

Step 1 is creating the project and setting up all the libraries that are needed:

mercury-cli --jwt d new-project --name webhooks

Note that we'll be using a non-released version of the zephyr SDK (yes, the one with ZVM stack traces!).

[dependencies]
zephyr-sdk = { git = "https://github.com/xycloo/rs-zephyr-toolkit", branch = "stellar-main-2", features = []}
serde = {version="1", features=["derive"]}
serde_json = "1.0"
stellar-strkey = "0.0.8"
hex = "0.4.3"

Setting up the Database

Being a service that receives requests to create webhooks from its users, we need to save those webhook reuqests somewhere in order to fetch them during real-time processing.

To do so, we define how we want the db structure to look like:

#[derive(Deserialize, Serialize)]
pub struct TriggerCondition {
    contract: String,
    topic1: Option<String>,
    topic2: Option<String>,
    // .. add more conditions as your service requires
}

#[derive(DatabaseDerive, Deserialize)]
#[with_name("webhooks")]
pub struct WebhookJob {
    condition: TriggerCondition,
    hook: String
    // add more fields to e.g identify by user.
}

This tells the Zephyr VM that WebhookJob objects are representative of the webhooks table we intend to create and use.

Note: the trigger conditions can be customized as you wish and depending on the number of conditions you're planning to support.

Next up, don't forget to populate the zephyr.toml with the table declaration:

name = "webhooks"

[[tables]]
name = "webhooks"

[[tables.columns]]
name = "condition"
col_type = "BIGINT"

[[tables.columns]]
name = "hook"
col_type = "BIGINT"

Adding Webhook Jobs

We now need to create an API to allow users to create new webhook jobs. Since we won't be handling user subscriptions (even though we could) this is fairly straightforward:

#[no_mangle]
pub extern "C" fn add_new() {
    let env = EnvClient::empty();
    let job: WebhookJob = env.read_request_body();

    env.put(&job);
}

This will instruct the ZVM to create a path users can access through the Mercury API when requesting the add_new function.

Whenever this API is called with the correct parameters (i.e a json body for WebhookJob), the job will be inserted in the database.

Real-time monitoring and Webhook Calls

In the ZephyrVM you need to use the on_close function to access real-time data monitoring as soon as new ledgers close.

In our case we want to: 0. Load all webhook jobs from the database.

For each event, check whether there is a webhook that must be triggered for the event.
If so, send out the webhook.

#[no_mangle]
pub extern "C" fn on_close() {
    let env = EnvClient::new();
    let monitoring_jobs: Vec<WebhookJob> = env.read();

    for (event, txhash) in env.reader().pretty().soroban_events_and_txhash() {
        for job in &monitoring_jobs {
            env.log().debug("Checking job compat", None);
            if job.must_trigger(&event) {
                env.log().debug("Found condition, triggering webhook", None);
                job.trigger(&env, &event, &hex::encode(txhash));
            }
        }
    }
}

The above function does exactly that, and the must_trigger and trigger functions are defined as such:

impl WebhookJob {
    fn must_trigger(&self, event: &PrettyContractEvent) -> bool {
        if self.condition.contract == stellar_strkey::Contract(event.contract).to_string() {
            if let Some(topic) = event.topics.get(0) {
                if self.condition.topic1 == Some(topic.to_xdr_base64(Limits::none()).unwrap()) {
                    return true
                }
            }
            if let Some(topic) = event.topics.get(1) {
                if self.condition.topic2 == Some(topic.to_xdr_base64(Limits::none()).unwrap()) {
                    return true
                }
            }
        }

        false
    }

    fn trigger(&self, env: &EnvClient, event: &PrettyContractEvent, hash: &str) {
        env.send_web_request(AgnosticRequest {
            body: Some(json!({"contract": event.contract, "tx": hash}).to_string()),
            url: self.hook.clone(),
            method: zephyr_sdk::Method::Post,
            headers: vec![("test".into(), "header".into())]
        });
    }
}

That's it! You've created a webhook service which you can immediately deploy to the cloud in less than 30 seconds.

Local Testing

Before deploying though, we want to locally test the service to ensure it does work correctly.

First, we need to add a few dev dependencies we'll use within our tests:

[dev-dependencies]
tokio = {version="1", features = ["full"]}
zephyr-vm = { git = "https://github.com/xycloo/zephyr", rev = "3c897ed", features = [] }

[dev-dependencies.stellar-xdr]
version = "=22.0.0-rc.1.1"

Then, we write the test:

We setup the DB used within the test and create the table used by our program ("webhooks").
Called the function to add a new webhook job and assert that one was indeed created.
Create an artificial transition that acts as if the contract + topics we created the job in step 2 for was in the processed ledger close.
Finally call the on-close function.

use ledger_meta_factory::{Transition, TransitionPretty};
use serde_json::json;
use stellar_xdr::next::{Hash, Int128Parts, Limits, ScAddress, ScSymbol, ScVal, WriteXdr};
use zephyr_sdk::testutils::TestHost;

fn build_transition() -> Transition {
    let mut transition = TransitionPretty::new();
    transition.inner.set_sequence(2000);
    transition
        .contract_event(
            "CAS3J7GYLGXMF6TDJBBYYSE3HQ6BBSMLNUQ34T6TZMYMW2EVH34XOWMA",
            vec![
                ScVal::Symbol(ScSymbol("transfer".try_into().unwrap())),
                ScVal::Address(stellar_xdr::next::ScAddress::Contract(Hash([8; 32]))),
                ScVal::Address(stellar_xdr::next::ScAddress::Contract(Hash([1; 32])))
            ],
            ScVal::I128(Int128Parts {
                hi: 0,
                lo: 100000000,
            }),
        )
        .unwrap();

    transition.inner
}

#[tokio::test]
async fn test_storage() {
    let env = TestHost::default();
    let mut db = env.database("postgres://postgres:postgres@localhost:5432");
    let mut program = env.new_program("./target/wasm32-unknown-unknown/release/webhooks.wasm");
    let transition = build_transition();

    let created = db
        .load_table(0, "webhooks", vec!["condition", "hook"], None)
        .await;
    assert!(created.is_ok());
    assert_eq!(db.get_rows_number(0, "webhooks").await.unwrap(), 0);

    program.set_body(json!({"hook": "https://tdep.requestcatcher.com/test", "condition": {
        "contract": "CAS3J7GYLGXMF6TDJBBYYSE3HQ6BBSMLNUQ34T6TZMYMW2EVH34XOWMA",
        "topic1": ScVal::Symbol(ScSymbol("transfer".try_into().unwrap())).to_xdr_base64(Limits::none()).unwrap()
    }}).to_string());
    
    let invocation = program.invoke_vm("add_new").await;
    assert!(invocation.is_ok());
    let inner_invocation = invocation.unwrap();
    assert!(inner_invocation.is_ok());

    assert_eq!(db.get_rows_number(0, "webhooks").await.unwrap(), 1);

    program.set_transition(transition);

    let invocation = program.invoke_vm("on_close").await;
    assert!(invocation.is_ok());
    let inner_invocation = invocation.unwrap();
    assert!(inner_invocation.is_ok());

    // Drop the connection and all the noise created in the local database.
    db.close().await;
}

Run the test, or deploy and wait for the right events, and you should see your webhook getting triggered!

Considerations

This is a really basic version, but serves as a proof-of-concept that services can use to build on top of Mercury's tech and, optionally, on top of its cloud infrastructure.

There are a few things we could easily add:

balance changes webhooks.
on-chain payments for the service (i.e managing payments within the very same zephyr program).
contract updates webhooks.
entry changes webhooks.
.. and much more!

If you're looking into building on Soroban and want to kickstart your project, let us know by DMing our co-founder and lead dev at https://x.com/heytdep.