From b9596b38f4840dd898f902ec8ecad31fbdff3ff7 Mon Sep 17 00:00:00 2001 From: adefemiesther1-debug Date: Wed, 24 Jun 2026 13:13:46 +0000 Subject: [PATCH 1/2] feat(mobile): design clean social transaction feed item component in capsule format --- mobileapp/app/(personal)/home.tsx | 108 +++++++++++++++++++++++------- 1 file changed, 84 insertions(+), 24 deletions(-) diff --git a/mobileapp/app/(personal)/home.tsx b/mobileapp/app/(personal)/home.tsx index e38f704..f21bbc1 100644 --- a/mobileapp/app/(personal)/home.tsx +++ b/mobileapp/app/(personal)/home.tsx @@ -351,20 +351,31 @@ export default function HomeScreen() { {filteredFeed.map((item) => ( - - {item.sender[0]} + + + {item.sender[0]} + + + {item.receiver[0]} + + - - {item.sender} paid{" "} - {item.receiver} - + + + {item.sender} paid{" "} + {item.receiver} + + + {item.amount} + + {item.timestamp} - {item.amount} + Memo {item.description} @@ -603,24 +614,44 @@ const styles = StyleSheet.create({ }, feedCard: { backgroundColor: COLORS.white, - borderRadius: 16, + borderRadius: 24, padding: 16, - marginBottom: 12, + marginBottom: 14, borderWidth: 1, borderColor: "#ECECEC", + shadowColor: "#0F172A", + shadowOffset: { width: 0, height: 10 }, + shadowOpacity: 0.08, + shadowRadius: 16, + elevation: 3, }, feedHeader: { + flexDirection: "row", + alignItems: "flex-start", + marginBottom: 12, + }, + avatarStack: { flexDirection: "row", alignItems: "center", + marginRight: 12, }, avatar: { width: 40, height: 40, borderRadius: 20, - backgroundColor: "#E2F0D9", justifyContent: "center", alignItems: "center", - marginRight: 12, + borderWidth: 2, + borderColor: COLORS.white, + }, + avatarPrimary: { + backgroundColor: "#E2F0D9", + marginRight: -8, + zIndex: 2, + }, + avatarSecondary: { + backgroundColor: "#FCEEDC", + zIndex: 1, }, avatarText: { color: COLORS.primary, @@ -630,37 +661,66 @@ const styles = StyleSheet.create({ paymentInfo: { flex: 1, }, + paymentRow: { + flexDirection: "row", + justifyContent: "space-between", + alignItems: "flex-start", + flexWrap: "wrap", + }, paymentText: { + flex: 1, fontSize: 15, + lineHeight: 20, fontFamily: "Outfit_400Regular", - color: "#333", + color: "#334155", + flexShrink: 1, + marginRight: 8, }, boldText: { fontFamily: "Outfit_700Bold", - color: "#111", + color: "#111827", }, - timestamp: { - fontSize: 12, - color: "#999", + amountPill: { + backgroundColor: "#F2F9F0", + borderRadius: 999, + paddingHorizontal: 10, + paddingVertical: 6, + borderWidth: 1, + borderColor: "#DDF2DD", + alignSelf: "flex-start", marginTop: 2, }, - paymentAmount: { - fontSize: 16, + amountText: { + fontSize: 13, fontFamily: "Outfit_700Bold", color: "#2E7D32", }, + timestamp: { + fontSize: 12, + color: "#94A3B8", + marginTop: 4, + }, memoContainer: { - marginTop: 12, - backgroundColor: "#F5F8F6", + backgroundColor: "#F4F6F8", + borderColor: "#E7EBEF", + borderWidth: 1, paddingHorizontal: 12, - paddingVertical: 8, - borderRadius: 8, - alignSelf: "flex-start", + paddingVertical: 10, + borderRadius: 16, + }, + memoLabel: { + fontSize: 11, + color: "#64748B", + fontFamily: "Outfit_600SemiBold", + textTransform: "uppercase", + letterSpacing: 0.8, + marginBottom: 2, }, memoText: { fontSize: 14, - color: "#444", + color: "#334155", fontFamily: "Outfit_400Regular", + lineHeight: 19, }, actionsDivider: { height: 1, From 26edd4cdcec736c884869c52360eafa230825cba Mon Sep 17 00:00:00 2001 From: adefemiesther1-debug Date: Wed, 24 Jun 2026 13:22:33 +0000 Subject: [PATCH 2/2] test(backend): design and build stellar event indexer worker for soroban rpc --- backend/migrations/0003_indexer_state.sql | 5 + backend/src/indexer/worker.rs | 269 ++++++++++++++++++++-- backend/src/main.rs | 6 +- 3 files changed, 264 insertions(+), 16 deletions(-) create mode 100644 backend/migrations/0003_indexer_state.sql diff --git a/backend/migrations/0003_indexer_state.sql b/backend/migrations/0003_indexer_state.sql new file mode 100644 index 0000000..a9f0891 --- /dev/null +++ b/backend/migrations/0003_indexer_state.sql @@ -0,0 +1,5 @@ +CREATE TABLE IF NOT EXISTS indexer_state ( + key TEXT PRIMARY KEY, + last_ledger_sequence BIGINT NOT NULL DEFAULT 0, + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); diff --git a/backend/src/indexer/worker.rs b/backend/src/indexer/worker.rs index fa27659..039cb2e 100644 --- a/backend/src/indexer/worker.rs +++ b/backend/src/indexer/worker.rs @@ -1,8 +1,15 @@ use serde::Deserialize; +use serde_json::{json, Value}; use sqlx::{PgPool, Row}; -use std::time::Duration; +use std::{env, error::Error, time::Duration}; use uuid::Uuid; +const INDEXER_CURSOR_KEY: &str = "stellar_event_cursor"; +const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(3); +const INITIAL_BACKOFF: Duration = Duration::from_secs(1); +const MAX_BACKOFF: Duration = Duration::from_secs(30); +const RPC_TIMEOUT: Duration = Duration::from_secs(10); + #[derive(Deserialize, Debug)] pub struct SocialPaymentEvent { pub sender: String, @@ -13,28 +20,65 @@ pub struct SocialPaymentEvent { pub tx_hash: String, } -pub async fn run() -> Result<(), Box> { +pub async fn run( + pool: PgPool, + rpc_url: String, +) -> Result<(), Box> { tracing::info!("Starting Stellar event indexer background worker..."); - // We'll get the pool from config when implemented properly - // For now, this is a placeholder with the parser logic + + let mut cursor = load_or_initialize_cursor(&pool).await?; + let mut backoff_attempt = 0usize; + loop { - // TODO: Implement BE-013 (Poll/Subscribe to Soroban RPC payment events) - // TODO: Implement BE-015 (Stellar cursor tracker to avoid double indexing) - tokio::time::sleep(Duration::from_secs(10)).await; - tracing::debug!("Stellar Indexer heartbeats... polling Soroban RPC for new events."); + match poll_soroban_events(&rpc_url, cursor).await { + Ok((events, latest_ledger)) => { + backoff_attempt = 0; + let mut next_cursor = cursor; + + for event in events { + if let Some(payment_event) = extract_social_payment_event(&event) { + if let Err(err) = process_social_payment_event(payment_event, &pool).await { + tracing::warn!("Failed to process Stellar payment event: {err}"); + } + } + + if let Some(ledger) = event.get("ledger").and_then(Value::as_u64) { + next_cursor = next_cursor.max(ledger as i64); + } + } + + if latest_ledger > 0 { + next_cursor = next_cursor.max(latest_ledger as i64); + } + + if next_cursor > cursor { + cursor = next_cursor; + persist_cursor(&pool, cursor).await?; + tracing::debug!("Indexer cursor advanced to ledger {cursor}"); + } else { + cursor += 1; + persist_cursor(&pool, cursor).await?; + } + + tokio::time::sleep(DEFAULT_POLL_INTERVAL).await; + } + Err(err) => { + let delay = compute_backoff_delay(backoff_attempt); + backoff_attempt += 1; + tracing::warn!("Soroban RPC polling failed, retrying in {:?}: {err}", delay); + tokio::time::sleep(delay).await; + } + } } } -// Parse and process a SocialPaymentEvent into the database pub async fn process_social_payment_event( event: SocialPaymentEvent, pool: &PgPool, ) -> Result<(), Box> { - // Look up sender and receiver UUIDs from their addresses let sender_id = get_or_create_user_id(&event.sender, pool).await?; let receiver_id = get_or_create_user_id(&event.receiver, pool).await?; - // Insert into payments table sqlx::query( r#" INSERT INTO payments (tx_hash, sender_id, receiver_id, amount, currency, memo, visibility) @@ -46,7 +90,7 @@ pub async fn process_social_payment_event( .bind(sender_id) .bind(receiver_id) .bind(event.amount) - .bind("NGN") // Default to NGN for now + .bind("NGN") .bind(&event.memo) .bind(event.visibility.to_uppercase()) .execute(pool) @@ -55,12 +99,157 @@ pub async fn process_social_payment_event( Ok(()) } -// Helper function to get user ID by address, creating if not exists +async fn poll_soroban_events( + rpc_url: &str, + start_ledger: i64, +) -> Result<(Vec, u64), Box> { + let contract_id = env::var("SOCIAL_PAYMENT_CONTRACT_ID").ok(); + let payload = build_get_events_payload(start_ledger, contract_id.as_deref()); + + let response = reqwest::Client::new() + .post(rpc_url) + .timeout(RPC_TIMEOUT) + .json(&payload) + .send() + .await?; + + let status = response.status(); + if !status.is_success() { + return Err(format!("Soroban RPC returned HTTP {status}").into()); + } + + let body: Value = response.json().await?; + let result = body + .get("result") + .ok_or("Soroban RPC response did not include result")?; + let events = result + .get("events") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + let latest_ledger = result + .get("latestLedger") + .and_then(Value::as_u64) + .unwrap_or(0); + + Ok((events, latest_ledger)) +} + +fn build_get_events_payload(start_ledger: i64, contract_id: Option<&str>) -> Value { + let mut filters = vec![json!({ + "topics": [[{ "type": "symbol", "value": "SocialPaymentEvent" }]] + })]; + + if let Some(contract_id) = contract_id.filter(|value| !value.is_empty()) { + let contract_filter = json!({ + "contractIds": [contract_id], + "topics": [[{ "type": "symbol", "value": "SocialPaymentEvent" }]] + }); + filters[0] = contract_filter; + } + + json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "getEvents", + "params": [{ + "startLedger": start_ledger, + "filters": filters + }] + }) +} + +fn compute_backoff_delay(attempt: usize) -> Duration { + let multiplier = 2usize.saturating_pow(attempt.min(5) as u32); + let candidate = INITIAL_BACKOFF.saturating_mul(multiplier as u32); + candidate.min(MAX_BACKOFF) +} + +async fn load_or_initialize_cursor(pool: &PgPool) -> Result> { + let existing = sqlx::query_scalar::<_, i64>( + "SELECT last_ledger_sequence FROM indexer_state WHERE key = $1", + ) + .bind(INDEXER_CURSOR_KEY) + .fetch_optional(pool) + .await?; + + if let Some(cursor) = existing { + return Ok(cursor); + } + + sqlx::query( + "INSERT INTO indexer_state (key, last_ledger_sequence) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING", + ) + .bind(INDEXER_CURSOR_KEY) + .bind(0_i64) + .execute(pool) + .await?; + + Ok(0) +} + +async fn persist_cursor(pool: &PgPool, ledger: i64) -> Result<(), Box> { + sqlx::query( + "INSERT INTO indexer_state (key, last_ledger_sequence, updated_at) VALUES ($1, $2, NOW()) ON CONFLICT (key) DO UPDATE SET last_ledger_sequence = EXCLUDED.last_ledger_sequence, updated_at = NOW()", + ) + .bind(INDEXER_CURSOR_KEY) + .bind(ledger) + .execute(pool) + .await?; + + Ok(()) +} + +fn extract_social_payment_event(value: &Value) -> Option { + let sender = find_nested_string(value, "sender")?; + let receiver = find_nested_string(value, "receiver")?; + let amount = find_nested_i64(value, "amount")?; + let memo = find_nested_string(value, "memo").unwrap_or_else(|| "".to_string()); + let visibility = find_nested_string(value, "visibility").unwrap_or_else(|| "PUBLIC".to_string()); + let tx_hash = find_nested_string(value, "tx_hash") + .or_else(|| find_nested_string(value, "txHash")) + .or_else(|| find_nested_string(value, "transactionHash")) + .unwrap_or_else(|| "unknown".to_string()); + + Some(SocialPaymentEvent { + sender, + receiver, + amount, + memo, + visibility, + tx_hash, + }) +} + +fn find_nested_string(value: &Value, key: &str) -> Option { + match value { + Value::Object(map) => map.get(key).and_then(|item| match item { + Value::String(text) => Some(text.clone()), + Value::Number(number) => Some(number.to_string()), + _ => None, + }).or_else(|| map.values().find_map(|nested| find_nested_string(nested, key))), + Value::Array(items) => items.iter().find_map(|item| find_nested_string(item, key)), + _ => None, + } +} + +fn find_nested_i64(value: &Value, key: &str) -> Option { + match value { + Value::Object(map) => map.get(key).and_then(|item| match item { + Value::Number(number) => number.as_i64(), + Value::String(text) => text.parse::().ok(), + _ => None, + }).or_else(|| map.values().find_map(|nested| find_nested_i64(nested, key))), + Value::Array(items) => items.iter().find_map(|item| find_nested_i64(item, key)), + _ => None, + } +} + async fn get_or_create_user_id( address: &str, pool: &PgPool, ) -> Result> { - let username = format!("u_{}", &address[1..15].to_lowercase()); + let username = slugify_address(address); let row = sqlx::query( r#" INSERT INTO users (address, username, display_name) @@ -78,3 +267,55 @@ async fn get_or_create_user_id( Ok(row.get("id")) } + +fn slugify_address(address: &str) -> String { + let trimmed = address.trim(); + let snippet = trimmed.get(1..15).unwrap_or(trimmed); + format!("u_{}", snippet.to_lowercase()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn builds_payload_with_contract_and_topic_filters() { + let payload = build_get_events_payload(12, Some("CAKE")); + let params = payload["params"].as_array().unwrap(); + let filter = ¶ms[0]["filters"][0]; + + assert_eq!(filter["contractIds"][0].as_str(), Some("CAKE")); + assert_eq!(filter["topics"][0][0]["value"].as_str(), Some("SocialPaymentEvent")); + } + + #[test] + fn backoff_delay_grows_and_caps() { + assert_eq!(compute_backoff_delay(0), INITIAL_BACKOFF); + assert_eq!(compute_backoff_delay(1), Duration::from_secs(2)); + assert_eq!(compute_backoff_delay(6), MAX_BACKOFF); + } + + #[test] + fn extracts_payment_event_from_nested_payload() { + let payload = json!({ + "body": { + "v0": { + "data": { + "sender": "GABC", + "receiver": "GXYZ", + "amount": 2500, + "memo": "Lunch", + "visibility": "PUBLIC", + "tx_hash": "abc123" + } + } + } + }); + + let event = extract_social_payment_event(&payload).unwrap(); + assert_eq!(event.sender, "GABC"); + assert_eq!(event.receiver, "GXYZ"); + assert_eq!(event.amount, 2500); + assert_eq!(event.memo, "Lunch"); + } +} diff --git a/backend/src/main.rs b/backend/src/main.rs index 9db9685..fc1ca25 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -138,8 +138,10 @@ async fn main() { .merge(other_routes); // Spawn indexer in the background - tokio::spawn(async { - if let Err(e) = indexer::worker::run().await { + let indexer_pool = pool.clone(); + let indexer_rpc_url = config.stellar_rpc_url.clone(); + tokio::spawn(async move { + if let Err(e) = indexer::worker::run(indexer_pool, indexer_rpc_url).await { tracing::error!("Stellar Indexer background worker failed: {:?}", e); } });