feat: works e2e

packages/relayer/.env.example

@@ -1,6 +1,7 @@
 CORE_CONTRACT_ADDRESS=           # Address of the deployed wallet contract.
 PRIVATE_KEY=                      # Private key for Relayer's account.
 CHAIN_RPC_PROVIDER=http://127.0.0.1:8545
+CHAIN_RPC_EXPLORER=
 CHAIN_ID=11155111                    # Chain ID of the testnet.
 
 SMTP_SERVER=

packages/relayer/src/database.rs

@@ -182,10 +182,8 @@ impl Database {
         Ok(vec)
     }
 
-    #[named]
     pub async fn get_claims_unexpired(&self, now: i64) -> Result<Vec<Claim>> {
         let mut vec = Vec::new();
-        info!(LOG, "now {}", now; "func" => function_name!());
         let rows =
             sqlx::query("SELECT * FROM claims WHERE expiry_time > $1 AND is_deleted = FALSE")
                 .bind(now)
@@ -217,10 +215,8 @@ impl Database {
         Ok(vec)
     }
 
-    #[named]
     pub async fn get_claims_expired(&self, now: i64) -> Result<Vec<Claim>> {
         let mut vec = Vec::new();
-        info!(LOG, "now {}", now; "func" => function_name!());
         let rows =
             sqlx::query("SELECT * FROM claims WHERE expiry_time < $1 AND is_deleted = FALSE")
                 .bind(now)

packages/relayer/src/lib.rs

@@ -28,13 +28,12 @@ use anyhow::{anyhow, bail, Result};
 use dotenv::dotenv;
 use ethers::prelude::*;
 use lazy_static::lazy_static;
-use relayer_utils::{converters::*, cryptos::*, parse_email::ParsedEmail, Fr, LOG};
+use relayer_utils::{converters::*, cryptos::*, Fr, LOG};
 use slog::{error, info, trace};
 use std::env;
 use std::path::PathBuf;
 use std::sync::atomic::AtomicU32;
 use std::sync::{Arc, OnceLock};
-use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 use tokio::time::{sleep, Duration};
 
 pub static CIRCUITS_DIR_PATH: OnceLock<PathBuf> = OnceLock::new();
@@ -290,13 +289,11 @@ async fn event_listener_fn(
 async fn catch_claims_in_db_fn() -> Result<()> {
     let now = now();
     let claims = DB.get_claims_unexpired(now).await?;
-    info!(LOG, "unexpired claims: {:?}", claims; "func" => function_name!());
     for claim in claims {
         info!(LOG, "Claiming claim for : {}", claim.email_address; "func" => function_name!());
         claim_unclaims(claim.clone()).await?;
     }
     let claims = DB.get_claims_expired(now).await?;
-    info!(LOG, "expired claims: {:?}", claims; "func" => function_name!());
     for claim in claims {
         let email_addr = claim.email_address.clone();
         info!(LOG, "Voiding claim for : {}", email_addr.clone(); "func" => function_name!());

packages/relayer/src/modules/mail.rs

@@ -3,7 +3,7 @@ use handlebars::Handlebars;
 use relayer_utils::AccountKey;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use std::{pin::Pin, sync::atomic::Ordering};
+use std::sync::atomic::Ordering;
 use tokio::fs::read_to_string;
 
 #[derive(Debug, Clone)]

packages/relayer/src/modules/psi.rs

@@ -6,7 +6,6 @@ use axum::Json;
 use ff::Field;
 use serde::{Deserialize, Serialize};
 use tokio::fs::{read_to_string, remove_file};
-use tokio::sync::mpsc::UnboundedSender;
 
 const DELAY: u64 = 300;
 

packages/relayer/src/modules/web_server/rest_api.rs

@@ -1,12 +1,14 @@
 use anyhow::{anyhow, Result};
 use slog::error;
 
-use crate::{handle_email, handle_email_event, render_html, EmailMessage, SafeRequest};
+use crate::{
+    handle_email, handle_email_event, render_html, EmailMessage, EmailWalletEvent, SafeRequest,
+};
 use ethers::types::{Address, U256};
 use relayer_utils::{
     converters::{field2hex, hex2field},
     cryptos::{AccountKey, PaddedEmailAddr, WalletSalt},
-    LOG,
+    ParsedEmail, LOG,
 };
 
 use crate::{CHAIN_RPC_EXPLORER, CLIENT, DB};
@@ -319,7 +321,7 @@ pub async fn delete_safe_owner_api_fn(payload: String) -> Result<()> {
 
 pub async fn receive_email_api_fn(email: String) -> Result<()> {
     tokio::spawn(async move {
-        match handle_email(email).await {
+        match handle_email(email.clone()).await {
             Ok(event) => match handle_email_event(event).await {
                 Ok(_) => {}
                 Err(e) => {
@@ -328,6 +330,19 @@ pub async fn receive_email_api_fn(email: String) -> Result<()> {
             },
             Err(e) => {
                 error!(LOG, "Error handling email: {:?}", e);
+                let parsed_email = ParsedEmail::new_from_raw_email(&email).await.unwrap();
+                let from_addr = parsed_email.get_from_addr().unwrap();
+                match handle_email_event(EmailWalletEvent::Error {
+                    email_addr: from_addr,
+                    error: e.to_string(),
+                })
+                .await
+                {
+                    Ok(_) => {}
+                    Err(e) => {
+                        error!(LOG, "Error handling email event: {:?}", e);
+                    }
+                }
             }
         }
     });

packages/relayer/src/modules/web_server/server.rs

@@ -4,7 +4,6 @@ use std::sync::atomic::Ordering;
 
 use axum::Router;
 use serde::{Deserialize, Serialize};
-use tokio::sync::mpsc::UnboundedSender;
 use tower_http::cors::{AllowHeaders, AllowMethods, Any, CorsLayer};
 
 #[derive(Serialize, Deserialize)]